Add asynchronous Kafka notifications with transactional outbox

.env.example

@@ -3,4 +3,5 @@
 POSTGRES_DB=linktracker
 POSTGRES_USER=linktracker
 POSTGRES_PASSWORD=linktracker
-GITHUB_TOKEN=your-github-token
+GITHUB_TOKEN=your-github-token
+NOTIFICATION_TRANSPORT=HTTP

README.md

@@ -6,6 +6,7 @@
 [![Release](https://img.shields.io/github/v/release/666mxvbee/LinkTracker?include_prereleases&sort=semver)](https://github.com/666mxvbee/LinkTracker/releases)
 [![.NET 9](https://img.shields.io/badge/.NET-9.0-512BD4?logo=dotnet)](https://dotnet.microsoft.com/)
 [![PostgreSQL](https://img.shields.io/badge/PostgreSQL-16-4169E1?logo=postgresql&logoColor=white)](https://www.postgresql.org/)
+[![Apache Kafka](https://img.shields.io/badge/Apache%20Kafka-3.x-231F20?logo=apachekafka&logoColor=white)](docker-compose.yml)
 [![Docker](https://img.shields.io/badge/Docker-Compose-2496ED?logo=docker&logoColor=white)](docker-compose.yml)
 [![Tests](https://img.shields.io/badge/tests-xUnit%20%2B%20Testcontainers-5A2D82)](tests/LinkTracker.Scrapper.Tests)
 [![Platform](https://img.shields.io/badge/platform-Windows%20%7C%20Linux%20%7C%20Docker-lightgrey)](docker-compose.yml)
@@ -37,6 +38,8 @@ TELEGRAM_BOT_TOKEN=your-telegram-bot-token
 POSTGRES_DB=linktracker
 POSTGRES_USER=linktracker
 POSTGRES_PASSWORD=linktracker
+GITHUB_TOKEN=your-github-token
+NOTIFICATION_TRANSPORT=HTTP
 ```
 
 `.env` is ignored by git. Use `.env.example` as the template.
@@ -72,6 +75,29 @@ Scheduler settings:
 
 `BatchSize` is clamped by the application to `50..500`. `Parallelism` controls how many links are processed concurrently.
 
+Notification transport settings:
+
+```json
+"Notifications": {
+  "Transport": "HTTP",
+  "Kafka": {
+    "BootstrapServers": "localhost:9092,localhost:9093,localhost:9094",
+    "Topic": "link-updates",
+    "DlqTopic": "link-updates-dlq",
+    "LingerMs": 10,
+    "OutboxBatchSize": 100,
+    "OutboxDispatchIntervalSeconds": 5
+  }
+}
+```
+
+`Transport` can be:
+
+```text
+HTTP   Scrapper sends updates directly to Bot over HTTP
+Kafka  Scrapper writes updates to notification_outbox, then publishes them to Kafka in batches
+```
+
 ## Run With Docker Compose
 
 Run all services:
@@ -86,6 +112,7 @@ Endpoints:
 Bot API:       http://localhost:5100
 Scrapper API: http://localhost:5000
 PostgreSQL:   localhost:5433
+Kafka UI:     http://localhost:8085
 ```
 
 Inside Docker, Scrapper connects to PostgreSQL by service name:
@@ -111,6 +138,12 @@ dotnet run --project src\LinkTracker.Bot
 
 Scrapper applies SQL migrations from `migrations/` automatically when `Database:RunMigrations` is `true`.
 
+To run asynchronous notifications locally, set this in `.env` before starting Docker Compose:
+
+```env
+NOTIFICATION_TRANSPORT=Kafka
+```
+
 ## Useful Manual Checks
 
 List database tables:
@@ -127,6 +160,7 @@ links
 chat_links
 tags
 chat_link_tags
+notification_outbox
 ```
 
 DbUp also creates:
@@ -183,4 +217,13 @@ creation time
 text preview limited to 200 characters
 ```
 
-The notification sender is abstracted behind `IMessageSender`. The current implementation is HTTP from Scrapper to Bot.
+The notification sender is abstracted behind `IMessageSender`.
+
+Available transports:
+
+```text
+HTTP   Scrapper calls Bot /updates directly
+Kafka  Scrapper stores updates in notification_outbox and a dispatcher publishes them to link-updates
+```
+
+In Kafka mode, Bot consumes `link-updates`, sends valid notifications to Telegram, and sends deserialization, validation, or exhausted processing failures to `link-updates-dlq`.

docker-compose.yml

@@ -9,6 +9,13 @@ services:
       - ASPNETCORE_ENVIRONMENT=Development
       - BotConfiguration__BotToken=${TELEGRAM_BOT_TOKEN}
       - BotConfiguration__ScrapperUrl=http://scrapper:8080
+      - Notifications__Transport=${NOTIFICATION_TRANSPORT:-HTTP}
+      - Notifications__Kafka__BootstrapServers=kafka-1:29092,kafka-2:29092,kafka-3:29092
+      - Notifications__Kafka__Topic=link-updates
+      - Notifications__Kafka__DlqTopic=link-updates-dlq
+      - Notifications__Kafka__GroupId=linktracker-bot
+      - Notifications__Kafka__MaxRetryAttempts=3
+      - Notifications__Kafka__RetryDelayMilliseconds=500
     depends_on:
       - scrapper
 
@@ -26,6 +33,13 @@ services:
       - Database__ConnectionString=Host=postgres;Port=5432;Database=${POSTGRES_DB};Username=${POSTGRES_USER};Password=${POSTGRES_PASSWORD}
       - Database__RunMigrations=true
       - Scrapper__GitHubToken=${GITHUB_TOKEN}
+      - Notifications__Transport=${NOTIFICATION_TRANSPORT:-HTTP}
+      - Notifications__Kafka__BootstrapServers=kafka-1:29092,kafka-2:29092,kafka-3:29092
+      - Notifications__Kafka__Topic=link-updates
+      - Notifications__Kafka__DlqTopic=link-updates-dlq
+      - Notifications__Kafka__LingerMs=10
+      - Notifications__Kafka__OutboxBatchSize=100
+      - Notifications__Kafka__OutboxDispatchIntervalSeconds=5
     depends_on:
       postgres:
         condition: service_healthy
@@ -47,5 +61,141 @@ services:
       timeout: 5s
       retries: 10
 
+  zookeeper:
+    image: confluentinc/cp-zookeeper:7.6.1
+    container_name: linktracker-zookeeper
+    environment:
+      ZOOKEEPER_CLIENT_PORT: 2181
+      ZOOKEEPER_TICK_TIME: 2000
+    ports:
+      - "2181:2181"
+
+  kafka-1:
+    image: confluentinc/cp-kafka:7.6.1
+    container_name: linktracker-kafka-1
+    depends_on:
+      - zookeeper
+    ports:
+      - "9092:9092"
+    environment:
+      KAFKA_BROKER_ID: 1
+      KAFKA_ZOOKEEPER_CONNECT: zookeeper:2181
+      KAFKA_LISTENERS: INTERNAL://0.0.0.0:29092,EXTERNAL://0.0.0.0:9092
+      KAFKA_ADVERTISED_LISTENERS: INTERNAL://kafka-1:29092,EXTERNAL://localhost:9092
+      KAFKA_LISTENER_SECURITY_PROTOCOL_MAP: INTERNAL:PLAINTEXT,EXTERNAL:PLAINTEXT
+      KAFKA_INTER_BROKER_LISTENER_NAME: INTERNAL
+      KAFKA_OFFSETS_TOPIC_REPLICATION_FACTOR: 3
+      KAFKA_TRANSACTION_STATE_LOG_REPLICATION_FACTOR: 3
+      KAFKA_TRANSACTION_STATE_LOG_MIN_ISR: 2
+      KAFKA_MIN_INSYNC_REPLICAS: 2
+      KAFKA_AUTO_CREATE_TOPICS_ENABLE: false
+      KAFKA_GROUP_INITIAL_REBALANCE_DELAY_MS: 0
+      KAFKA_CONFLUENT_SUPPORT_METRICS_ENABLE: false
+    healthcheck:
+      test: ["CMD", "kafka-broker-api-versions", "--bootstrap-server", "localhost:9092"]
+      interval: 10s
+      timeout: 10s
+      retries: 10
+
+  kafka-2:
+    image: confluentinc/cp-kafka:7.6.1
+    container_name: linktracker-kafka-2
+    depends_on:
+      - zookeeper
+    ports:
+      - "9093:9093"
+    environment:
+      KAFKA_BROKER_ID: 2
+      KAFKA_ZOOKEEPER_CONNECT: zookeeper:2181
+      KAFKA_LISTENERS: INTERNAL://0.0.0.0:29092,EXTERNAL://0.0.0.0:9093
+      KAFKA_ADVERTISED_LISTENERS: INTERNAL://kafka-2:29092,EXTERNAL://localhost:9093
+      KAFKA_LISTENER_SECURITY_PROTOCOL_MAP: INTERNAL:PLAINTEXT,EXTERNAL:PLAINTEXT
+      KAFKA_INTER_BROKER_LISTENER_NAME: INTERNAL
+      KAFKA_OFFSETS_TOPIC_REPLICATION_FACTOR: 3
+      KAFKA_TRANSACTION_STATE_LOG_REPLICATION_FACTOR: 3
+      KAFKA_TRANSACTION_STATE_LOG_MIN_ISR: 2
+      KAFKA_MIN_INSYNC_REPLICAS: 2
+      KAFKA_AUTO_CREATE_TOPICS_ENABLE: false
+      KAFKA_GROUP_INITIAL_REBALANCE_DELAY_MS: 0
+      KAFKA_CONFLUENT_SUPPORT_METRICS_ENABLE: false
+    healthcheck:
+      test: ["CMD", "kafka-broker-api-versions", "--bootstrap-server", "localhost:9093"]
+      interval: 10s
+      timeout: 10s
+      retries: 10
+
+  kafka-3:
+    image: confluentinc/cp-kafka:7.6.1
+    container_name: linktracker-kafka-3
+    depends_on:
+      - zookeeper
+    ports:
+      - "9094:9094"
+    environment:
+      KAFKA_BROKER_ID: 3
+      KAFKA_ZOOKEEPER_CONNECT: zookeeper:2181
+      KAFKA_LISTENERS: INTERNAL://0.0.0.0:29092,EXTERNAL://0.0.0.0:9094
+      KAFKA_ADVERTISED_LISTENERS: INTERNAL://kafka-3:29092,EXTERNAL://localhost:9094
+      KAFKA_LISTENER_SECURITY_PROTOCOL_MAP: INTERNAL:PLAINTEXT,EXTERNAL:PLAINTEXT
+      KAFKA_INTER_BROKER_LISTENER_NAME: INTERNAL
+      KAFKA_OFFSETS_TOPIC_REPLICATION_FACTOR: 3
+      KAFKA_TRANSACTION_STATE_LOG_REPLICATION_FACTOR: 3
+      KAFKA_TRANSACTION_STATE_LOG_MIN_ISR: 2
+      KAFKA_MIN_INSYNC_REPLICAS: 2
+      KAFKA_AUTO_CREATE_TOPICS_ENABLE: false
+      KAFKA_GROUP_INITIAL_REBALANCE_DELAY_MS: 0
+      KAFKA_CONFLUENT_SUPPORT_METRICS_ENABLE: false
+    healthcheck:
+      test: ["CMD", "kafka-broker-api-versions", "--bootstrap-server", "localhost:9094"]
+      interval: 10s
+      timeout: 10s
+      retries: 10
+
+  kafka-init:
+    image: confluentinc/cp-kafka:7.6.1
+    container_name: linktracker-kafka-init
+    depends_on:
+      kafka-1:
+        condition: service_healthy
+      kafka-2:
+        condition: service_healthy
+      kafka-3:
+        condition: service_healthy
+    command:
+      - /bin/sh
+      - -c
+      - |
+        kafka-topics --bootstrap-server kafka-1:29092 \
+          --create --if-not-exists \
+          --topic link-updates \
+          --partitions 3 \
+          --replication-factor 3 \
+          --config min.insync.replicas=2 \
+          --config cleanup.policy=delete \
+          --config retention.ms=604800000
+
+        kafka-topics --bootstrap-server kafka-1:29092 \
+          --create --if-not-exists \
+          --topic link-updates-dlq \
+          --partitions 3 \
+          --replication-factor 3 \
+          --config min.insync.replicas=2 \
+          --config cleanup.policy=delete \
+          --config retention.ms=2592000000
+
+        kafka-topics --bootstrap-server kafka-1:29092 --list
+
+  kafka-ui:
+    image: provectuslabs/kafka-ui:v0.7.2
+    container_name: linktracker-kafka-ui
+    depends_on:
+      kafka-init:
+        condition: service_completed_successfully
+    ports:
+      - "8085:8080"
+    environment:
+      KAFKA_CLUSTERS_0_NAME: linktracker
+      KAFKA_CLUSTERS_0_BOOTSTRAPSERVERS: kafka-1:29092,kafka-2:29092,kafka-3:29092
+
 volumes:
-  postgres-data:
+  postgres-data:

migrations/002_notification_outbox.sql

@@ -0,0 +1,18 @@
+CREATE TABLE IF NOT EXISTS notification_outbox (
+    id BIGSERIAL PRIMARY KEY,
+    message_id UUID NOT NULL UNIQUE,
+    topic TEXT NOT NULL,
+    message_key TEXT NULL,
+    payload JSONB NOT NULL,
+    status TEXT NOT NULL DEFAULT 'Pending',
+    attempt_count INTEGER NOT NULL DEFAULT 0,
+    last_error TEXT NULL,
+    created_at TIMESTAMPTZ NOT NULL DEFAULT NOW(),
+    processed_at TIMESTAMPTZ NULL,
+    CONSTRAINT ck_notification_outbox_status
+        CHECK (status IN ('Pending', 'Processed'))
+);
+
+CREATE INDEX IF NOT EXISTS ix_notification_outbox_pending
+    ON notification_outbox (created_at, id)
+    WHERE status = 'Pending';

src/LinkTracker.Bot/Configuration/NotificationOptions.cs

@@ -0,0 +1,32 @@
+namespace LinkTracker.Bot.Configuration;
+
+public sealed class NotificationOptions
+{
+    public const string SectionName = "Notifications";
+
+    public string Transport { get; init; } = NotificationTransports.Http;
+
+    public KafkaConsumerOptions Kafka { get; init; } = new();
+}
+
+public static class NotificationTransports
+{
+    public const string Http = "HTTP";
+
+    public const string Kafka = "Kafka";
+}
+
+public sealed class KafkaConsumerOptions
+{
+    public string BootstrapServers { get; init; } = "localhost:9092";
+
+    public string Topic { get; init; } = "link-updates";
+
+    public string DlqTopic { get; init; } = "link-updates-dlq";
+
+    public string GroupId { get; init; } = "linktracker-bot";
+
+    public int MaxRetryAttempts { get; init; } = 3;
+
+    public int RetryDelayMilliseconds { get; init; } = 500;
+}

src/LinkTracker.Bot/Controllers/UpdatesController.cs

@@ -1,35 +1,25 @@
 using Microsoft.AspNetCore.Mvc;
-using Telegram.Bot;
+using LinkTracker.Bot.Services.Notifications;
 using LinkTracker.Shared.Models;
 
 namespace LinkTracker.Bot.Controllers;
 
 [ApiController]
 [Route("updates")]
 public class UpdatesController(
-    ITelegramBotClient botClient,
+    ILinkUpdateHandler linkUpdateHandler,
     ILogger<UpdatesController> logger) : ControllerBase
 {
     [HttpPost]
-    public async Task<IActionResult> PostUpdate([FromBody] LinkUpdate update)
+    public async Task<IActionResult> PostUpdate(
+        [FromBody] LinkUpdate update,
+        CancellationToken cancellationToken)
     {
         logger.LogInformation("Received update for URL: {Url}", update.Url);
 
         try
         {
-            foreach (var chatId in update.TgChatIds)
-            {
-                var message = $"🔔 *Update found!*\n\n" +
-                              $"Source: {update.Url}\n" +
-                              $"Description: {update.Description}";
-
-                await botClient.SendMessage(
-                    chatId: chatId,
-                    text: message,
-                    parseMode: Telegram.Bot.Types.Enums.ParseMode.Markdown
-                );
-            }
-
+            await linkUpdateHandler.HandleAsync(update, cancellationToken);
             return Ok();
         }
         catch (Exception ex)
@@ -38,4 +28,4 @@ await botClient.SendMessage(
             return StatusCode(500, "Internal Server Error");
         }
     }
-}
+}

src/LinkTracker.Bot/LinkTracker.Bot.csproj

@@ -9,6 +9,7 @@
   </PropertyGroup>
 
     <ItemGroup>
+        <PackageReference Include="Confluent.Kafka" Version="2.15.0" />
         <PackageReference Include="Microsoft.AspNetCore.OpenApi" Version="9.0.10" />
         <PackageReference Include="Refit" Version="10.0.1" />
         <PackageReference Include="Refit.HttpClientFactory" Version="10.0.1" />