From 899fa7f44c04f4096044789423dda8e27265bf30 Mon Sep 17 00:00:00 2001 From: 666mxvbee Date: Sun, 5 Jul 2026 04:02:51 +0300 Subject: [PATCH 1/6] refactor(bot): extract link update handler --- .../Controllers/UpdatesController.cs | 24 ++++--------- src/LinkTracker.Bot/Program.cs | 4 ++- .../Notifications/ILinkUpdateHandler.cs | 8 +++++ .../TelegramLinkUpdateHandler.cs | 34 +++++++++++++++++++ 4 files changed, 52 insertions(+), 18 deletions(-) create mode 100644 src/LinkTracker.Bot/Services/Notifications/ILinkUpdateHandler.cs create mode 100644 src/LinkTracker.Bot/Services/Notifications/TelegramLinkUpdateHandler.cs diff --git a/src/LinkTracker.Bot/Controllers/UpdatesController.cs b/src/LinkTracker.Bot/Controllers/UpdatesController.cs index d5a197e..75f5bdc 100644 --- a/src/LinkTracker.Bot/Controllers/UpdatesController.cs +++ b/src/LinkTracker.Bot/Controllers/UpdatesController.cs @@ -1,5 +1,5 @@ using Microsoft.AspNetCore.Mvc; -using Telegram.Bot; +using LinkTracker.Bot.Services.Notifications; using LinkTracker.Shared.Models; namespace LinkTracker.Bot.Controllers; @@ -7,29 +7,19 @@ namespace LinkTracker.Bot.Controllers; [ApiController] [Route("updates")] public class UpdatesController( - ITelegramBotClient botClient, + ILinkUpdateHandler linkUpdateHandler, ILogger logger) : ControllerBase { [HttpPost] - public async Task PostUpdate([FromBody] LinkUpdate update) + public async Task 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"); } } -} \ No newline at end of file +} diff --git a/src/LinkTracker.Bot/Program.cs b/src/LinkTracker.Bot/Program.cs index 5f079b4..1042f0b 100644 --- a/src/LinkTracker.Bot/Program.cs +++ b/src/LinkTracker.Bot/Program.cs @@ -2,6 +2,7 @@ using LinkTracker.Bot.Configuration; using LinkTracker.Bot.Repositories; using LinkTracker.Bot.Services; +using LinkTracker.Bot.Services.Notifications; using LinkTracker.Bot.Clients; using Telegram.Bot; using Refit; @@ -29,6 +30,7 @@ builder.Services.AddSingleton(); builder.Services.AddSingleton(); builder.Services.AddSingleton(); +builder.Services.AddSingleton(); builder.Services.AddTransient(); builder.Services.AddTransient(); @@ -42,4 +44,4 @@ app.MapControllers(); -app.Run(); \ No newline at end of file +app.Run(); diff --git a/src/LinkTracker.Bot/Services/Notifications/ILinkUpdateHandler.cs b/src/LinkTracker.Bot/Services/Notifications/ILinkUpdateHandler.cs new file mode 100644 index 0000000..8855dd4 --- /dev/null +++ b/src/LinkTracker.Bot/Services/Notifications/ILinkUpdateHandler.cs @@ -0,0 +1,8 @@ +using LinkTracker.Shared.Models; + +namespace LinkTracker.Bot.Services.Notifications; + +public interface ILinkUpdateHandler +{ + Task HandleAsync(LinkUpdate update, CancellationToken cancellationToken); +} diff --git a/src/LinkTracker.Bot/Services/Notifications/TelegramLinkUpdateHandler.cs b/src/LinkTracker.Bot/Services/Notifications/TelegramLinkUpdateHandler.cs new file mode 100644 index 0000000..64c53ed --- /dev/null +++ b/src/LinkTracker.Bot/Services/Notifications/TelegramLinkUpdateHandler.cs @@ -0,0 +1,34 @@ +using LinkTracker.Shared.Models; +using Telegram.Bot; +using Telegram.Bot.Types.Enums; + +namespace LinkTracker.Bot.Services.Notifications; + +public class TelegramLinkUpdateHandler( + ITelegramBotClient botClient, + ILogger logger) : ILinkUpdateHandler +{ + public async Task HandleAsync(LinkUpdate update, CancellationToken cancellationToken) + { + logger.LogInformation( + "Sending link update notification for URL {Url} to {ChatCount} chats", + update.Url, + update.TgChatIds.Length); + + foreach (var chatId in update.TgChatIds) + { + await botClient.SendMessage( + chatId: chatId, + text: BuildMessage(update), + parseMode: ParseMode.Markdown, + cancellationToken: cancellationToken); + } + } + + private static string BuildMessage(LinkUpdate update) + { + return $"🔔 *Update found!*\n\n" + + $"Source: {update.Url}\n" + + $"Description: {update.Description}"; + } +} From 486289cc35368d842327124f4dcf9d269971b7fe Mon Sep 17 00:00:00 2001 From: 666mxvbee Date: Tue, 7 Jul 2026 20:24:14 +0300 Subject: [PATCH 2/6] feat: add kafka notification transport --- .env.example | 3 +- docker-compose.yml | 150 ++++++++++- .../Configuration/NotificationOptions.cs | 32 +++ src/LinkTracker.Bot/LinkTracker.Bot.csproj | 1 + src/LinkTracker.Bot/Program.cs | 27 ++ .../INotificationDeduplicationStore.cs | 8 + .../InMemoryNotificationDeduplicationStore.cs | 18 ++ .../Services/Notifications/KafkaDlqMessage.cs | 10 + .../Notifications/KafkaLinkUpdateConsumer.cs | 252 ++++++++++++++++++ src/LinkTracker.Bot/appsettings.json | 11 + .../Configuration/NotificationOptions.cs | 28 ++ .../LinkTracker.Scrapper.csproj | 1 + src/LinkTracker.Scrapper/Program.cs | 35 ++- .../Notifications/KafkaMessageSender.cs | 43 +++ src/LinkTracker.Scrapper/appsettings.json | 11 +- 15 files changed, 625 insertions(+), 5 deletions(-) create mode 100644 src/LinkTracker.Bot/Configuration/NotificationOptions.cs create mode 100644 src/LinkTracker.Bot/Services/Notifications/INotificationDeduplicationStore.cs create mode 100644 src/LinkTracker.Bot/Services/Notifications/InMemoryNotificationDeduplicationStore.cs create mode 100644 src/LinkTracker.Bot/Services/Notifications/KafkaDlqMessage.cs create mode 100644 src/LinkTracker.Bot/Services/Notifications/KafkaLinkUpdateConsumer.cs create mode 100644 src/LinkTracker.Scrapper/Configuration/NotificationOptions.cs create mode 100644 src/LinkTracker.Scrapper/Services/Notifications/KafkaMessageSender.cs diff --git a/.env.example b/.env.example index 7fe65e8..6f2d89f 100644 --- a/.env.example +++ b/.env.example @@ -3,4 +3,5 @@ POSTGRES_DB=linktracker POSTGRES_USER=linktracker POSTGRES_PASSWORD=linktracker -GITHUB_TOKEN=your-github-token \ No newline at end of file +GITHUB_TOKEN=your-github-token +NOTIFICATION_TRANSPORT=HTTP diff --git a/docker-compose.yml b/docker-compose.yml index 6393ef3..a994a01 100644 --- a/docker-compose.yml +++ b/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,11 @@ 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 depends_on: postgres: condition: service_healthy @@ -47,5 +59,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: \ No newline at end of file + postgres-data: diff --git a/src/LinkTracker.Bot/Configuration/NotificationOptions.cs b/src/LinkTracker.Bot/Configuration/NotificationOptions.cs new file mode 100644 index 0000000..0372f19 --- /dev/null +++ b/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; +} diff --git a/src/LinkTracker.Bot/LinkTracker.Bot.csproj b/src/LinkTracker.Bot/LinkTracker.Bot.csproj index 1f45112..199b85b 100644 --- a/src/LinkTracker.Bot/LinkTracker.Bot.csproj +++ b/src/LinkTracker.Bot/LinkTracker.Bot.csproj @@ -9,6 +9,7 @@ + diff --git a/src/LinkTracker.Bot/Program.cs b/src/LinkTracker.Bot/Program.cs index 1042f0b..da3075d 100644 --- a/src/LinkTracker.Bot/Program.cs +++ b/src/LinkTracker.Bot/Program.cs @@ -7,12 +7,19 @@ using Telegram.Bot; using Refit; using Microsoft.Extensions.Options; +using Confluent.Kafka; var builder = WebApplication.CreateBuilder(args); builder.Services.AddControllers(); builder.Services.Configure(builder.Configuration.GetSection(BotOptions.SectionName)); +builder.Services.Configure( + builder.Configuration.GetSection(NotificationOptions.SectionName)); + +var notificationOptions = builder.Configuration + .GetSection(NotificationOptions.SectionName) + .Get() ?? new NotificationOptions(); builder.Services.AddSingleton(sp => { @@ -32,6 +39,26 @@ builder.Services.AddSingleton(); builder.Services.AddSingleton(); +if (notificationOptions.Transport.Equals(NotificationTransports.Kafka, StringComparison.OrdinalIgnoreCase)) +{ + builder.Services.AddSingleton>(_ => + new ProducerBuilder(new ProducerConfig + { + BootstrapServers = notificationOptions.Kafka.BootstrapServers, + Acks = Acks.All, + EnableIdempotence = true, + ClientId = "linktracker-bot-dlq", + }).Build()); + + builder.Services.AddSingleton(); + builder.Services.AddHostedService(); +} +else if (!notificationOptions.Transport.Equals(NotificationTransports.Http, StringComparison.OrdinalIgnoreCase)) +{ + throw new InvalidOperationException( + $"Unknown notification transport: {notificationOptions.Transport}. Use HTTP or Kafka."); +} + builder.Services.AddTransient(); builder.Services.AddTransient(); builder.Services.AddTransient(); diff --git a/src/LinkTracker.Bot/Services/Notifications/INotificationDeduplicationStore.cs b/src/LinkTracker.Bot/Services/Notifications/INotificationDeduplicationStore.cs new file mode 100644 index 0000000..6c08bfb --- /dev/null +++ b/src/LinkTracker.Bot/Services/Notifications/INotificationDeduplicationStore.cs @@ -0,0 +1,8 @@ +namespace LinkTracker.Bot.Services.Notifications; + +public interface INotificationDeduplicationStore +{ + bool IsProcessed(string fingerprint); + + void MarkProcessed(string fingerprint); +} diff --git a/src/LinkTracker.Bot/Services/Notifications/InMemoryNotificationDeduplicationStore.cs b/src/LinkTracker.Bot/Services/Notifications/InMemoryNotificationDeduplicationStore.cs new file mode 100644 index 0000000..08f4c0a --- /dev/null +++ b/src/LinkTracker.Bot/Services/Notifications/InMemoryNotificationDeduplicationStore.cs @@ -0,0 +1,18 @@ +using System.Collections.Concurrent; + +namespace LinkTracker.Bot.Services.Notifications; + +public sealed class InMemoryNotificationDeduplicationStore : INotificationDeduplicationStore +{ + private readonly ConcurrentDictionary _processedMessages = new(); + + public bool IsProcessed(string fingerprint) + { + return _processedMessages.ContainsKey(fingerprint); + } + + public void MarkProcessed(string fingerprint) + { + _processedMessages.TryAdd(fingerprint, 0); + } +} diff --git a/src/LinkTracker.Bot/Services/Notifications/KafkaDlqMessage.cs b/src/LinkTracker.Bot/Services/Notifications/KafkaDlqMessage.cs new file mode 100644 index 0000000..7250def --- /dev/null +++ b/src/LinkTracker.Bot/Services/Notifications/KafkaDlqMessage.cs @@ -0,0 +1,10 @@ +namespace LinkTracker.Bot.Services.Notifications; + +public sealed record KafkaDlqMessage( + string Reason, + string Error, + string? Payload, + string SourceTopic, + int SourcePartition, + long SourceOffset, + DateTimeOffset FailedAt); diff --git a/src/LinkTracker.Bot/Services/Notifications/KafkaLinkUpdateConsumer.cs b/src/LinkTracker.Bot/Services/Notifications/KafkaLinkUpdateConsumer.cs new file mode 100644 index 0000000..10ebf7a --- /dev/null +++ b/src/LinkTracker.Bot/Services/Notifications/KafkaLinkUpdateConsumer.cs @@ -0,0 +1,252 @@ +using System.Security.Cryptography; +using System.Text; +using System.Text.Json; +using Confluent.Kafka; +using LinkTracker.Bot.Configuration; +using LinkTracker.Shared.Models; +using Microsoft.Extensions.Options; + +namespace LinkTracker.Bot.Services.Notifications; + +public sealed class KafkaLinkUpdateConsumer( + IOptions options, + ILinkUpdateHandler linkUpdateHandler, + INotificationDeduplicationStore deduplicationStore, + IProducer dlqProducer, + ILogger logger) : BackgroundService +{ + private static readonly JsonSerializerOptions JsonOptions = new(JsonSerializerDefaults.Web); + + protected override Task ExecuteAsync(CancellationToken stoppingToken) + { + return Task.Run(() => ConsumeLoop(stoppingToken), stoppingToken); + } + + private void ConsumeLoop(CancellationToken cancellationToken) + { + var kafkaOptions = options.Value.Kafka; + + using var consumer = new ConsumerBuilder(new ConsumerConfig + { + BootstrapServers = kafkaOptions.BootstrapServers, + GroupId = kafkaOptions.GroupId, + EnableAutoCommit = false, + AutoOffsetReset = AutoOffsetReset.Earliest, + AllowAutoCreateTopics = false, + }).Build(); + + consumer.Subscribe(kafkaOptions.Topic); + logger.LogInformation("Kafka link update consumer subscribed to topic {Topic}", kafkaOptions.Topic); + + try + { + while (!cancellationToken.IsCancellationRequested) + { + try + { + var consumeResult = consumer.Consume(cancellationToken); + + ProcessMessageAsync(consumeResult, cancellationToken) + .GetAwaiter() + .GetResult(); + + consumer.Commit(consumeResult); + } + catch (ConsumeException ex) + { + logger.LogError(ex, "Kafka consume error: {Reason}", ex.Error.Reason); + } + } + } + catch (OperationCanceledException) + { + logger.LogInformation("Kafka link update consumer is stopping"); + } + finally + { + consumer.Close(); + } + } + + private async Task ProcessMessageAsync( + ConsumeResult consumeResult, + CancellationToken cancellationToken) + { + if (!TryDeserialize(consumeResult, out var update, out var deserializationError)) + { + await SendToDlqAsync( + consumeResult, + "DeserializationError", + deserializationError, + cancellationToken); + return; + } + + if (!TryValidate(update, out var validationError)) + { + await SendToDlqAsync( + consumeResult, + "ValidationError", + validationError, + cancellationToken); + return; + } + + var fingerprint = BuildFingerprint(update); + + if (deduplicationStore.IsProcessed(fingerprint)) + { + logger.LogInformation("Skipping already processed link update {Fingerprint}", fingerprint); + return; + } + + await HandleWithRetriesAsync(update, consumeResult, cancellationToken); + deduplicationStore.MarkProcessed(fingerprint); + } + + private static bool TryDeserialize( + ConsumeResult consumeResult, + out LinkUpdate update, + out string error) + { + try + { + var result = JsonSerializer.Deserialize(consumeResult.Message.Value, JsonOptions); + + if (result is null) + { + update = default!; + error = "Message payload is empty."; + return false; + } + + update = result; + error = string.Empty; + return true; + } + catch (JsonException ex) + { + update = default!; + error = ex.Message; + return false; + } + } + + private static bool TryValidate(LinkUpdate update, out string error) + { + if (string.IsNullOrWhiteSpace(update.Url)) + { + error = "Url is required."; + return false; + } + + if (!Uri.TryCreate(update.Url, UriKind.Absolute, out _)) + { + error = "Url must be absolute."; + return false; + } + + if (string.IsNullOrWhiteSpace(update.Description)) + { + error = "Description is required."; + return false; + } + + if (update.TgChatIds is null || update.TgChatIds.Length == 0) + { + error = "At least one Telegram chat id is required."; + return false; + } + + error = string.Empty; + return true; + } + + private async Task HandleWithRetriesAsync( + LinkUpdate update, + ConsumeResult consumeResult, + CancellationToken cancellationToken) + { + var kafkaOptions = options.Value.Kafka; + var maxAttempts = Math.Max(1, kafkaOptions.MaxRetryAttempts); + var retryDelay = TimeSpan.FromMilliseconds(Math.Max(0, kafkaOptions.RetryDelayMilliseconds)); + + for (var attempt = 1; attempt <= maxAttempts; attempt++) + { + try + { + await linkUpdateHandler.HandleAsync(update, cancellationToken); + return; + } + catch (Exception ex) when (attempt < maxAttempts) + { + logger.LogWarning( + ex, + "Failed to process Kafka link update {Url}. Attempt {Attempt}/{MaxAttempts}", + update.Url, + attempt, + maxAttempts); + + if (retryDelay > TimeSpan.Zero) + { + await Task.Delay(retryDelay, cancellationToken); + } + } + catch (Exception ex) + { + logger.LogError( + ex, + "Failed to process Kafka link update {Url}. Sending message to DLQ", + update.Url); + + await SendToDlqAsync( + consumeResult, + "ProcessingError", + ex.Message, + cancellationToken); + return; + } + } + } + + private async Task SendToDlqAsync( + ConsumeResult consumeResult, + string reason, + string error, + CancellationToken cancellationToken) + { + var kafkaOptions = options.Value.Kafka; + + var message = new KafkaDlqMessage( + Reason: reason, + Error: error, + Payload: consumeResult.Message.Value, + SourceTopic: consumeResult.Topic, + SourcePartition: consumeResult.Partition.Value, + SourceOffset: consumeResult.Offset.Value, + FailedAt: DateTimeOffset.UtcNow); + + await dlqProducer.ProduceAsync( + kafkaOptions.DlqTopic, + new Message + { + Key = consumeResult.Message.Key, + Value = JsonSerializer.Serialize(message, JsonOptions), + }, + cancellationToken); + + logger.LogWarning( + "Sent Kafka message from {TopicPartitionOffset} to DLQ topic {DlqTopic}. Reason: {Reason}", + consumeResult.TopicPartitionOffset, + kafkaOptions.DlqTopic, + reason); + } + + private static string BuildFingerprint(LinkUpdate update) + { + var payload = $"{update.Id}|{update.Url}|{update.Description}|{string.Join(',', update.TgChatIds)}"; + var hash = SHA256.HashData(Encoding.UTF8.GetBytes(payload)); + + return Convert.ToHexString(hash); + } +} diff --git a/src/LinkTracker.Bot/appsettings.json b/src/LinkTracker.Bot/appsettings.json index 5723112..f6b56ea 100644 --- a/src/LinkTracker.Bot/appsettings.json +++ b/src/LinkTracker.Bot/appsettings.json @@ -8,5 +8,16 @@ "BotConfiguration": { "BotToken": "", "ScrapperUrl": "http://localhost:5000" + }, + "Notifications": { + "Transport": "HTTP", + "Kafka": { + "BootstrapServers": "localhost:9092,localhost:9093,localhost:9094", + "Topic": "link-updates", + "DlqTopic": "link-updates-dlq", + "GroupId": "linktracker-bot", + "MaxRetryAttempts": 3, + "RetryDelayMilliseconds": 500 + } } } diff --git a/src/LinkTracker.Scrapper/Configuration/NotificationOptions.cs b/src/LinkTracker.Scrapper/Configuration/NotificationOptions.cs new file mode 100644 index 0000000..21235cb --- /dev/null +++ b/src/LinkTracker.Scrapper/Configuration/NotificationOptions.cs @@ -0,0 +1,28 @@ +namespace LinkTracker.Scrapper.Configuration; + +public sealed class NotificationOptions +{ + public const string SectionName = "Notifications"; + + public string Transport { get; init; } = NotificationTransports.Http; + + public KafkaProducerOptions Kafka { get; init; } = new(); +} + +public static class NotificationTransports +{ + public const string Http = "HTTP"; + + public const string Kafka = "Kafka"; +} + +public sealed class KafkaProducerOptions +{ + public string BootstrapServers { get; init; } = "localhost:9092"; + + public string Topic { get; init; } = "link-updates"; + + public string DlqTopic { get; init; } = "link-updates-dlq"; + + public int LingerMs { get; init; } = 10; +} diff --git a/src/LinkTracker.Scrapper/LinkTracker.Scrapper.csproj b/src/LinkTracker.Scrapper/LinkTracker.Scrapper.csproj index c206ac9..fa7d9bd 100644 --- a/src/LinkTracker.Scrapper/LinkTracker.Scrapper.csproj +++ b/src/LinkTracker.Scrapper/LinkTracker.Scrapper.csproj @@ -8,6 +8,7 @@ + diff --git a/src/LinkTracker.Scrapper/Program.cs b/src/LinkTracker.Scrapper/Program.cs index f3c8815..322d1cc 100644 --- a/src/LinkTracker.Scrapper/Program.cs +++ b/src/LinkTracker.Scrapper/Program.cs @@ -9,6 +9,7 @@ using LinkTracker.Scrapper.Services.Updates; using Microsoft.EntityFrameworkCore; using System.Net.Http.Headers; +using Confluent.Kafka; using Npgsql; using Quartz; @@ -24,6 +25,9 @@ builder.Services.Configure( builder.Configuration.GetSection(ScrapperOptions.SectionName)); +builder.Services.Configure( + builder.Configuration.GetSection(NotificationOptions.SectionName)); + var databaseOptions = builder.Configuration .GetSection(DatabaseOptions.SectionName) .Get() ?? new DatabaseOptions(); @@ -32,6 +36,10 @@ .GetSection(ScrapperOptions.SectionName) .Get() ?? new ScrapperOptions(); +var notificationOptions = builder.Configuration + .GetSection(NotificationOptions.SectionName) + .Get() ?? new NotificationOptions(); + builder.Services.AddSingleton(_ => NpgsqlDataSource.Create(databaseOptions.ConnectionString)); builder.Services.AddDbContext(options => @@ -79,7 +87,30 @@ client.BaseAddress = new Uri(botUrl); }); -builder.Services.AddScoped(); +if (notificationOptions.Transport.Equals(NotificationTransports.Http, StringComparison.OrdinalIgnoreCase)) +{ + builder.Services.AddScoped(); +} +else if (notificationOptions.Transport.Equals(NotificationTransports.Kafka, StringComparison.OrdinalIgnoreCase)) +{ + builder.Services.AddSingleton>(_ => + new ProducerBuilder(new ProducerConfig + { + BootstrapServers = notificationOptions.Kafka.BootstrapServers, + Acks = Acks.All, + EnableIdempotence = true, + LingerMs = notificationOptions.Kafka.LingerMs, + ClientId = "linktracker-scrapper", + }).Build()); + + builder.Services.AddSingleton(); +} +else +{ + throw new InvalidOperationException( + $"Unknown notification transport: {notificationOptions.Transport}. Use HTTP or Kafka."); +} + builder.Services.AddScoped(); builder.Services.AddQuartz(q => @@ -110,4 +141,4 @@ app.MapControllers(); -app.Run(); \ No newline at end of file +app.Run(); diff --git a/src/LinkTracker.Scrapper/Services/Notifications/KafkaMessageSender.cs b/src/LinkTracker.Scrapper/Services/Notifications/KafkaMessageSender.cs new file mode 100644 index 0000000..9f8918d --- /dev/null +++ b/src/LinkTracker.Scrapper/Services/Notifications/KafkaMessageSender.cs @@ -0,0 +1,43 @@ +using System.Globalization; +using System.Text.Json; +using Confluent.Kafka; +using LinkTracker.Scrapper.Configuration; +using LinkTracker.Shared.Models; +using Microsoft.Extensions.Options; + +namespace LinkTracker.Scrapper.Services.Notifications; + +public sealed class KafkaMessageSender( + IProducer producer, + IOptions options, + ILogger logger) : IMessageSender +{ + private static readonly JsonSerializerOptions JsonOptions = new(JsonSerializerDefaults.Web); + + public async Task SendAsync(LinkUpdate update, CancellationToken cancellationToken) + { + var topic = options.Value.Kafka.Topic; + var payload = JsonSerializer.Serialize(update, JsonOptions); + + var result = await producer.ProduceAsync( + topic, + new Message + { + Key = BuildMessageKey(update), + Value = payload, + }, + cancellationToken); + + logger.LogInformation( + "Published link update to Kafka topic {Topic} at {TopicPartitionOffset}", + topic, + result.TopicPartitionOffset); + } + + private static string BuildMessageKey(LinkUpdate update) + { + return update.Id > 0 + ? update.Id.ToString(CultureInfo.InvariantCulture) + : update.Url; + } +} diff --git a/src/LinkTracker.Scrapper/appsettings.json b/src/LinkTracker.Scrapper/appsettings.json index ee8df37..f4a431f 100644 --- a/src/LinkTracker.Scrapper/appsettings.json +++ b/src/LinkTracker.Scrapper/appsettings.json @@ -17,5 +17,14 @@ "GitHubBaseUrl": "https://api.github.com/", "StackOverflowBaseUrl": "https://api.stackexchange.com/2.3/" }, + "Notifications": { + "Transport": "HTTP", + "Kafka": { + "BootstrapServers": "localhost:9092,localhost:9093,localhost:9094", + "Topic": "link-updates", + "DlqTopic": "link-updates-dlq", + "LingerMs": 10 + } + }, "AllowedHosts": "*" -} \ No newline at end of file +} From f75c14da4519c688689cb4e3e970d79a864e5a18 Mon Sep 17 00:00:00 2001 From: 666mxvbee Date: Tue, 7 Jul 2026 20:29:41 +0300 Subject: [PATCH 3/6] feat: add notification outbox --- docker-compose.yml | 2 + migrations/002_notification_outbox.sql | 18 +++ .../Configuration/NotificationOptions.cs | 4 + src/LinkTracker.Scrapper/Program.cs | 4 +- .../Notifications/IKafkaMessagePublisher.cs | 10 ++ .../Notifications/KafkaMessagePublisher.cs | 29 ++++ .../Notifications/KafkaMessageSender.cs | 20 +-- .../Notifications/KafkaOutboxDispatcher.cs | 148 ++++++++++++++++++ .../Notifications/OutboxMessageSender.cs | 40 +++++ src/LinkTracker.Scrapper/appsettings.json | 4 +- 10 files changed, 262 insertions(+), 17 deletions(-) create mode 100644 migrations/002_notification_outbox.sql create mode 100644 src/LinkTracker.Scrapper/Services/Notifications/IKafkaMessagePublisher.cs create mode 100644 src/LinkTracker.Scrapper/Services/Notifications/KafkaMessagePublisher.cs create mode 100644 src/LinkTracker.Scrapper/Services/Notifications/KafkaOutboxDispatcher.cs create mode 100644 src/LinkTracker.Scrapper/Services/Notifications/OutboxMessageSender.cs diff --git a/docker-compose.yml b/docker-compose.yml index a994a01..587dc64 100644 --- a/docker-compose.yml +++ b/docker-compose.yml @@ -38,6 +38,8 @@ services: - 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 diff --git a/migrations/002_notification_outbox.sql b/migrations/002_notification_outbox.sql new file mode 100644 index 0000000..4bb2dfb --- /dev/null +++ b/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'; diff --git a/src/LinkTracker.Scrapper/Configuration/NotificationOptions.cs b/src/LinkTracker.Scrapper/Configuration/NotificationOptions.cs index 21235cb..e1d0f17 100644 --- a/src/LinkTracker.Scrapper/Configuration/NotificationOptions.cs +++ b/src/LinkTracker.Scrapper/Configuration/NotificationOptions.cs @@ -25,4 +25,8 @@ public sealed class KafkaProducerOptions public string DlqTopic { get; init; } = "link-updates-dlq"; public int LingerMs { get; init; } = 10; + + public int OutboxBatchSize { get; init; } = 100; + + public int OutboxDispatchIntervalSeconds { get; init; } = 5; } diff --git a/src/LinkTracker.Scrapper/Program.cs b/src/LinkTracker.Scrapper/Program.cs index 322d1cc..12f9481 100644 --- a/src/LinkTracker.Scrapper/Program.cs +++ b/src/LinkTracker.Scrapper/Program.cs @@ -103,7 +103,9 @@ ClientId = "linktracker-scrapper", }).Build()); - builder.Services.AddSingleton(); + builder.Services.AddSingleton(); + builder.Services.AddScoped(); + builder.Services.AddHostedService(); } else { diff --git a/src/LinkTracker.Scrapper/Services/Notifications/IKafkaMessagePublisher.cs b/src/LinkTracker.Scrapper/Services/Notifications/IKafkaMessagePublisher.cs new file mode 100644 index 0000000..294ef37 --- /dev/null +++ b/src/LinkTracker.Scrapper/Services/Notifications/IKafkaMessagePublisher.cs @@ -0,0 +1,10 @@ +namespace LinkTracker.Scrapper.Services.Notifications; + +public interface IKafkaMessagePublisher +{ + Task PublishAsync( + string topic, + string? key, + string payload, + CancellationToken cancellationToken); +} diff --git a/src/LinkTracker.Scrapper/Services/Notifications/KafkaMessagePublisher.cs b/src/LinkTracker.Scrapper/Services/Notifications/KafkaMessagePublisher.cs new file mode 100644 index 0000000..9e186f5 --- /dev/null +++ b/src/LinkTracker.Scrapper/Services/Notifications/KafkaMessagePublisher.cs @@ -0,0 +1,29 @@ +using Confluent.Kafka; + +namespace LinkTracker.Scrapper.Services.Notifications; + +public sealed class KafkaMessagePublisher( + IProducer producer, + ILogger logger) : IKafkaMessagePublisher +{ + public async Task PublishAsync( + string topic, + string? key, + string payload, + CancellationToken cancellationToken) + { + var result = await producer.ProduceAsync( + topic, + new Message + { + Key = key ?? string.Empty, + Value = payload, + }, + cancellationToken); + + logger.LogInformation( + "Published Kafka message to topic {Topic} at {TopicPartitionOffset}", + topic, + result.TopicPartitionOffset); + } +} diff --git a/src/LinkTracker.Scrapper/Services/Notifications/KafkaMessageSender.cs b/src/LinkTracker.Scrapper/Services/Notifications/KafkaMessageSender.cs index 9f8918d..e683684 100644 --- a/src/LinkTracker.Scrapper/Services/Notifications/KafkaMessageSender.cs +++ b/src/LinkTracker.Scrapper/Services/Notifications/KafkaMessageSender.cs @@ -1,6 +1,5 @@ using System.Globalization; using System.Text.Json; -using Confluent.Kafka; using LinkTracker.Scrapper.Configuration; using LinkTracker.Shared.Models; using Microsoft.Extensions.Options; @@ -8,9 +7,8 @@ namespace LinkTracker.Scrapper.Services.Notifications; public sealed class KafkaMessageSender( - IProducer producer, - IOptions options, - ILogger logger) : IMessageSender + IKafkaMessagePublisher publisher, + IOptions options) : IMessageSender { private static readonly JsonSerializerOptions JsonOptions = new(JsonSerializerDefaults.Web); @@ -19,19 +17,11 @@ public async Task SendAsync(LinkUpdate update, CancellationToken cancellationTok var topic = options.Value.Kafka.Topic; var payload = JsonSerializer.Serialize(update, JsonOptions); - var result = await producer.ProduceAsync( + await publisher.PublishAsync( topic, - new Message - { - Key = BuildMessageKey(update), - Value = payload, - }, + BuildMessageKey(update), + payload, cancellationToken); - - logger.LogInformation( - "Published link update to Kafka topic {Topic} at {TopicPartitionOffset}", - topic, - result.TopicPartitionOffset); } private static string BuildMessageKey(LinkUpdate update) diff --git a/src/LinkTracker.Scrapper/Services/Notifications/KafkaOutboxDispatcher.cs b/src/LinkTracker.Scrapper/Services/Notifications/KafkaOutboxDispatcher.cs new file mode 100644 index 0000000..4102696 --- /dev/null +++ b/src/LinkTracker.Scrapper/Services/Notifications/KafkaOutboxDispatcher.cs @@ -0,0 +1,148 @@ +using LinkTracker.Scrapper.Configuration; +using Microsoft.Extensions.Options; +using Npgsql; + +namespace LinkTracker.Scrapper.Services.Notifications; + +public sealed class KafkaOutboxDispatcher( + NpgsqlDataSource dataSource, + IKafkaMessagePublisher publisher, + IOptions options, + ILogger logger) : BackgroundService +{ + protected override async Task ExecuteAsync(CancellationToken stoppingToken) + { + var interval = TimeSpan.FromSeconds(Math.Max(1, options.Value.Kafka.OutboxDispatchIntervalSeconds)); + + while (!stoppingToken.IsCancellationRequested) + { + try + { + await DispatchPendingBatchAsync(stoppingToken); + } + catch (OperationCanceledException) when (stoppingToken.IsCancellationRequested) + { + break; + } + catch (Exception ex) + { + logger.LogError(ex, "Failed to dispatch Kafka outbox batch"); + } + + await Task.Delay(interval, stoppingToken); + } + } + + internal async Task DispatchPendingBatchAsync(CancellationToken cancellationToken) + { + var batchSize = Math.Clamp(options.Value.Kafka.OutboxBatchSize, 1, 500); + + await using var connection = await dataSource.OpenConnectionAsync(cancellationToken); + await using var transaction = await connection.BeginTransactionAsync(cancellationToken); + + var messages = await LoadPendingMessagesAsync(connection, transaction, batchSize, cancellationToken); + + foreach (var message in messages) + { + await DispatchMessageAsync(connection, transaction, message, cancellationToken); + } + + await transaction.CommitAsync(cancellationToken); + } + + private async Task DispatchMessageAsync( + NpgsqlConnection connection, + NpgsqlTransaction transaction, + OutboxMessage message, + CancellationToken cancellationToken) + { + try + { + await publisher.PublishAsync(message.Topic, message.Key, message.Payload, cancellationToken); + await MarkProcessedAsync(connection, transaction, message.Id, cancellationToken); + } + catch (Exception ex) + { + logger.LogWarning( + ex, + "Failed to publish outbox message {OutboxMessageId}. It will remain pending", + message.Id); + + await MarkFailedAttemptAsync(connection, transaction, message.Id, ex.Message, cancellationToken); + } + } + + private static async Task LoadPendingMessagesAsync( + NpgsqlConnection connection, + NpgsqlTransaction transaction, + int batchSize, + CancellationToken cancellationToken) + { + await using var command = new NpgsqlCommand(""" + SELECT id, topic, message_key, payload::text + FROM notification_outbox + WHERE status = 'Pending' + ORDER BY created_at, id + LIMIT @batchSize + FOR UPDATE SKIP LOCKED; + """, connection, transaction); + + command.Parameters.AddWithValue("batchSize", batchSize); + + await using var reader = await command.ExecuteReaderAsync(cancellationToken); + var messages = new List(); + + while (await reader.ReadAsync(cancellationToken)) + { + messages.Add(new OutboxMessage( + reader.GetInt64(0), + reader.GetString(1), + reader.IsDBNull(2) ? null : reader.GetString(2), + reader.GetString(3))); + } + + return messages.ToArray(); + } + + private static async Task MarkProcessedAsync( + NpgsqlConnection connection, + NpgsqlTransaction transaction, + long id, + CancellationToken cancellationToken) + { + await using var command = new NpgsqlCommand(""" + UPDATE notification_outbox + SET status = 'Processed', + attempt_count = attempt_count + 1, + last_error = NULL, + processed_at = NOW() + WHERE id = @id; + """, connection, transaction); + + command.Parameters.AddWithValue("id", id); + + await command.ExecuteNonQueryAsync(cancellationToken); + } + + private static async Task MarkFailedAttemptAsync( + NpgsqlConnection connection, + NpgsqlTransaction transaction, + long id, + string error, + CancellationToken cancellationToken) + { + await using var command = new NpgsqlCommand(""" + UPDATE notification_outbox + SET attempt_count = attempt_count + 1, + last_error = @lastError + WHERE id = @id; + """, connection, transaction); + + command.Parameters.AddWithValue("id", id); + command.Parameters.AddWithValue("lastError", error); + + await command.ExecuteNonQueryAsync(cancellationToken); + } + + private sealed record OutboxMessage(long Id, string Topic, string? Key, string Payload); +} diff --git a/src/LinkTracker.Scrapper/Services/Notifications/OutboxMessageSender.cs b/src/LinkTracker.Scrapper/Services/Notifications/OutboxMessageSender.cs new file mode 100644 index 0000000..3603a71 --- /dev/null +++ b/src/LinkTracker.Scrapper/Services/Notifications/OutboxMessageSender.cs @@ -0,0 +1,40 @@ +using System.Globalization; +using System.Text.Json; +using LinkTracker.Scrapper.Configuration; +using LinkTracker.Shared.Models; +using Microsoft.Extensions.Options; +using Npgsql; +using NpgsqlTypes; + +namespace LinkTracker.Scrapper.Services.Notifications; + +public sealed class OutboxMessageSender( + NpgsqlDataSource dataSource, + IOptions options) : IMessageSender +{ + private static readonly JsonSerializerOptions JsonOptions = new(JsonSerializerDefaults.Web); + + public async Task SendAsync(LinkUpdate update, CancellationToken cancellationToken) + { + var payload = JsonSerializer.Serialize(update, JsonOptions); + + await using var command = dataSource.CreateCommand(""" + INSERT INTO notification_outbox (message_id, topic, message_key, payload) + VALUES (@messageId, @topic, @messageKey, @payload); + """); + + command.Parameters.AddWithValue("messageId", Guid.NewGuid()); + command.Parameters.AddWithValue("topic", options.Value.Kafka.Topic); + command.Parameters.AddWithValue("messageKey", BuildMessageKey(update)); + command.Parameters.Add("payload", NpgsqlDbType.Jsonb).Value = payload; + + await command.ExecuteNonQueryAsync(cancellationToken); + } + + private static string BuildMessageKey(LinkUpdate update) + { + return update.Id > 0 + ? update.Id.ToString(CultureInfo.InvariantCulture) + : update.Url; + } +} diff --git a/src/LinkTracker.Scrapper/appsettings.json b/src/LinkTracker.Scrapper/appsettings.json index f4a431f..12a43e3 100644 --- a/src/LinkTracker.Scrapper/appsettings.json +++ b/src/LinkTracker.Scrapper/appsettings.json @@ -23,7 +23,9 @@ "BootstrapServers": "localhost:9092,localhost:9093,localhost:9094", "Topic": "link-updates", "DlqTopic": "link-updates-dlq", - "LingerMs": 10 + "LingerMs": 10, + "OutboxBatchSize": 100, + "OutboxDispatchIntervalSeconds": 5 } }, "AllowedHosts": "*" From ee4e3c1877a0ed40860b417d48817315a70e288c Mon Sep 17 00:00:00 2001 From: 666mxvbee Date: Tue, 7 Jul 2026 20:35:40 +0300 Subject: [PATCH 4/6] test: cover kafka notification flow --- .../LinkTracker.Scrapper.csproj | 6 + .../Kafka/KafkaCollection.cs | 7 + .../Kafka/KafkaFixture.cs | 21 ++ .../Kafka/KafkaLinkUpdateConsumerTests.cs | 233 ++++++++++++++++++ .../Kafka/KafkaPublisherTests.cs | 42 ++++ .../Kafka/KafkaTestSupport.cs | 85 +++++++ .../LinkTracker.Scrapper.Tests.csproj | 3 + .../Postgres/MigrationTests.cs | 1 + .../Postgres/OutboxTests.cs | 161 ++++++++++++ .../Postgres/PostgresFixture.cs | 4 +- 10 files changed, 561 insertions(+), 2 deletions(-) create mode 100644 tests/LinkTracker.Scrapper.Tests/Kafka/KafkaCollection.cs create mode 100644 tests/LinkTracker.Scrapper.Tests/Kafka/KafkaFixture.cs create mode 100644 tests/LinkTracker.Scrapper.Tests/Kafka/KafkaLinkUpdateConsumerTests.cs create mode 100644 tests/LinkTracker.Scrapper.Tests/Kafka/KafkaPublisherTests.cs create mode 100644 tests/LinkTracker.Scrapper.Tests/Kafka/KafkaTestSupport.cs create mode 100644 tests/LinkTracker.Scrapper.Tests/Postgres/OutboxTests.cs diff --git a/src/LinkTracker.Scrapper/LinkTracker.Scrapper.csproj b/src/LinkTracker.Scrapper/LinkTracker.Scrapper.csproj index fa7d9bd..f3aa499 100644 --- a/src/LinkTracker.Scrapper/LinkTracker.Scrapper.csproj +++ b/src/LinkTracker.Scrapper/LinkTracker.Scrapper.csproj @@ -7,6 +7,12 @@ Linux + + + <_Parameter1>LinkTracker.Scrapper.Tests + + + diff --git a/tests/LinkTracker.Scrapper.Tests/Kafka/KafkaCollection.cs b/tests/LinkTracker.Scrapper.Tests/Kafka/KafkaCollection.cs new file mode 100644 index 0000000..555a50d --- /dev/null +++ b/tests/LinkTracker.Scrapper.Tests/Kafka/KafkaCollection.cs @@ -0,0 +1,7 @@ +namespace LinkTracker.Scrapper.Tests.Kafka; + +[CollectionDefinition(Name)] +public sealed class KafkaCollection : ICollectionFixture +{ + public const string Name = "kafka"; +} diff --git a/tests/LinkTracker.Scrapper.Tests/Kafka/KafkaFixture.cs b/tests/LinkTracker.Scrapper.Tests/Kafka/KafkaFixture.cs new file mode 100644 index 0000000..302e045 --- /dev/null +++ b/tests/LinkTracker.Scrapper.Tests/Kafka/KafkaFixture.cs @@ -0,0 +1,21 @@ +using Testcontainers.Kafka; + +namespace LinkTracker.Scrapper.Tests.Kafka; + +public sealed class KafkaFixture : IAsyncLifetime +{ + private readonly KafkaContainer _container = new KafkaBuilder("confluentinc/cp-kafka:7.6.1") + .Build(); + + public string BootstrapServers => _container.GetBootstrapAddress(); + + public Task InitializeAsync() + { + return _container.StartAsync(); + } + + public Task DisposeAsync() + { + return _container.DisposeAsync().AsTask(); + } +} diff --git a/tests/LinkTracker.Scrapper.Tests/Kafka/KafkaLinkUpdateConsumerTests.cs b/tests/LinkTracker.Scrapper.Tests/Kafka/KafkaLinkUpdateConsumerTests.cs new file mode 100644 index 0000000..f3ba1f7 --- /dev/null +++ b/tests/LinkTracker.Scrapper.Tests/Kafka/KafkaLinkUpdateConsumerTests.cs @@ -0,0 +1,233 @@ +using System.Collections.Concurrent; +using System.Text.Json; +using Confluent.Kafka; +using LinkTracker.Bot.Configuration; +using LinkTracker.Bot.Services.Notifications; +using LinkTracker.Shared.Models; +using Microsoft.Extensions.Logging.Abstractions; +using Microsoft.Extensions.Options; + +namespace LinkTracker.Scrapper.Tests.Kafka; + +[Collection(KafkaCollection.Name)] +public class KafkaLinkUpdateConsumerTests(KafkaFixture fixture) +{ + private static readonly JsonSerializerOptions JsonOptions = new(JsonSerializerDefaults.Web); + + [Fact] + public async Task Consumer_HandlesValidLinkUpdate() + { + var topics = await CreateTopicsAsync(); + var handler = new CapturingLinkUpdateHandler(); + using var consumer = CreateConsumer(topics, handler); + + await consumer.StartAsync(CancellationToken.None); + + try + { + var update = new LinkUpdate( + 1, + "https://github.com/octo/repo", + "new issue", + [1001]); + + await KafkaTestSupport.ProduceAsync( + fixture.BootstrapServers, + topics.Topic, + "link-1", + JsonSerializer.Serialize(update, JsonOptions)); + + await WaitUntilAsync(() => handler.Handled.Count == 1); + + Assert.Equal(update.Url, handler.Handled.Single().Url); + Assert.Equal(update.Description, handler.Handled.Single().Description); + } + finally + { + await consumer.StopAsync(CancellationToken.None); + } + } + + [Fact] + public async Task Consumer_SendsInvalidJsonToDlq() + { + var topics = await CreateTopicsAsync(); + var handler = new CapturingLinkUpdateHandler(); + using var consumer = CreateConsumer(topics, handler); + + await consumer.StartAsync(CancellationToken.None); + + try + { + await KafkaTestSupport.ProduceAsync( + fixture.BootstrapServers, + topics.Topic, + "bad-json", + "{not valid json"); + + var dlqMessage = KafkaTestSupport.ConsumeSingle( + fixture.BootstrapServers, + topics.DlqTopic, + TimeSpan.FromSeconds(15)); + + Assert.Contains("DeserializationError", dlqMessage.Message.Value); + Assert.Empty(handler.Handled); + } + finally + { + await consumer.StopAsync(CancellationToken.None); + } + } + + [Fact] + public async Task Consumer_SendsInvalidDataToDlq() + { + var topics = await CreateTopicsAsync(); + var handler = new CapturingLinkUpdateHandler(); + using var consumer = CreateConsumer(topics, handler); + + await consumer.StartAsync(CancellationToken.None); + + try + { + var invalidUpdate = new LinkUpdate(1, "", "new issue", [1001]); + + await KafkaTestSupport.ProduceAsync( + fixture.BootstrapServers, + topics.Topic, + "invalid-data", + JsonSerializer.Serialize(invalidUpdate, JsonOptions)); + + var dlqMessage = KafkaTestSupport.ConsumeSingle( + fixture.BootstrapServers, + topics.DlqTopic, + TimeSpan.FromSeconds(15)); + + Assert.Contains("ValidationError", dlqMessage.Message.Value); + Assert.Empty(handler.Handled); + } + finally + { + await consumer.StopAsync(CancellationToken.None); + } + } + + [Fact] + public async Task Consumer_RetriesProcessingErrorAndSendsMessageToDlq() + { + var topics = await CreateTopicsAsync(); + var handler = new CapturingLinkUpdateHandler(shouldFail: true); + using var consumer = CreateConsumer(topics, handler, maxRetryAttempts: 2); + + await consumer.StartAsync(CancellationToken.None); + + try + { + var update = new LinkUpdate( + 1, + "https://github.com/octo/repo", + "new issue", + [1001]); + + await KafkaTestSupport.ProduceAsync( + fixture.BootstrapServers, + topics.Topic, + "processing-error", + JsonSerializer.Serialize(update, JsonOptions)); + + var dlqMessage = KafkaTestSupport.ConsumeSingle( + fixture.BootstrapServers, + topics.DlqTopic, + TimeSpan.FromSeconds(15)); + + Assert.Equal(2, handler.Attempts); + Assert.Contains("ProcessingError", dlqMessage.Message.Value); + } + finally + { + await consumer.StopAsync(CancellationToken.None); + } + } + + private async Task CreateTopicsAsync() + { + var topics = new KafkaTopics( + $"link-updates-{Guid.NewGuid():N}", + $"link-updates-dlq-{Guid.NewGuid():N}"); + + await KafkaTestSupport.CreateTopicAsync(fixture.BootstrapServers, topics.Topic); + await KafkaTestSupport.CreateTopicAsync(fixture.BootstrapServers, topics.DlqTopic); + + return topics; + } + + private KafkaLinkUpdateConsumer CreateConsumer( + KafkaTopics topics, + ILinkUpdateHandler handler, + int maxRetryAttempts = 3) + { + var producer = new ProducerBuilder(new ProducerConfig + { + BootstrapServers = fixture.BootstrapServers, + Acks = Acks.All + }).Build(); + + return new KafkaLinkUpdateConsumer( + Options.Create(new NotificationOptions + { + Transport = NotificationTransports.Kafka, + Kafka = new KafkaConsumerOptions + { + BootstrapServers = fixture.BootstrapServers, + Topic = topics.Topic, + DlqTopic = topics.DlqTopic, + GroupId = $"linktracker-bot-test-{Guid.NewGuid():N}", + MaxRetryAttempts = maxRetryAttempts, + RetryDelayMilliseconds = 10 + } + }), + handler, + new InMemoryNotificationDeduplicationStore(), + producer, + NullLogger.Instance); + } + + private static async Task WaitUntilAsync(Func condition) + { + var deadline = DateTimeOffset.UtcNow + TimeSpan.FromSeconds(15); + + while (DateTimeOffset.UtcNow < deadline) + { + if (condition()) + { + return; + } + + await Task.Delay(100); + } + + throw new TimeoutException("Expected Kafka consumer condition was not reached."); + } + + private sealed class CapturingLinkUpdateHandler(bool shouldFail = false) : ILinkUpdateHandler + { + public ConcurrentBag Handled { get; } = []; + + public int Attempts { get; private set; } + + public Task HandleAsync(LinkUpdate update, CancellationToken cancellationToken) + { + Attempts++; + + if (shouldFail) + { + throw new InvalidOperationException("Telegram unavailable"); + } + + Handled.Add(update); + return Task.CompletedTask; + } + } + + private sealed record KafkaTopics(string Topic, string DlqTopic); +} diff --git a/tests/LinkTracker.Scrapper.Tests/Kafka/KafkaPublisherTests.cs b/tests/LinkTracker.Scrapper.Tests/Kafka/KafkaPublisherTests.cs new file mode 100644 index 0000000..0a22334 --- /dev/null +++ b/tests/LinkTracker.Scrapper.Tests/Kafka/KafkaPublisherTests.cs @@ -0,0 +1,42 @@ +using Confluent.Kafka; +using LinkTracker.Scrapper.Services.Notifications; +using Microsoft.Extensions.Logging.Abstractions; + +namespace LinkTracker.Scrapper.Tests.Kafka; + +[Collection(KafkaCollection.Name)] +public class KafkaPublisherTests(KafkaFixture fixture) +{ + [Fact] + public async Task KafkaMessagePublisher_PublishesMessageToTopic() + { + var topic = $"link-updates-{Guid.NewGuid():N}"; + await KafkaTestSupport.CreateTopicAsync(fixture.BootstrapServers, topic); + + using var producer = new ProducerBuilder(new ProducerConfig + { + BootstrapServers = fixture.BootstrapServers, + Acks = Acks.All + }).Build(); + + var publisher = new KafkaMessagePublisher( + producer, + NullLogger.Instance); + + await publisher.PublishAsync( + topic, + "link-42", + """{"url":"https://github.com/octo/repo"}""", + CancellationToken.None); + + producer.Flush(TimeSpan.FromSeconds(5)); + + var consumed = KafkaTestSupport.ConsumeSingle( + fixture.BootstrapServers, + topic, + TimeSpan.FromSeconds(15)); + + Assert.Equal("link-42", consumed.Message.Key); + Assert.Contains("github.com/octo/repo", consumed.Message.Value); + } +} diff --git a/tests/LinkTracker.Scrapper.Tests/Kafka/KafkaTestSupport.cs b/tests/LinkTracker.Scrapper.Tests/Kafka/KafkaTestSupport.cs new file mode 100644 index 0000000..35de43c --- /dev/null +++ b/tests/LinkTracker.Scrapper.Tests/Kafka/KafkaTestSupport.cs @@ -0,0 +1,85 @@ +using Confluent.Kafka; +using Confluent.Kafka.Admin; + +namespace LinkTracker.Scrapper.Tests.Kafka; + +internal static class KafkaTestSupport +{ + public static async Task CreateTopicAsync(string bootstrapServers, string topic) + { + using var admin = new AdminClientBuilder(new AdminClientConfig + { + BootstrapServers = bootstrapServers + }).Build(); + + try + { + await admin.CreateTopicsAsync( + [ + new TopicSpecification + { + Name = topic, + NumPartitions = 1, + ReplicationFactor = 1 + } + ]); + } + catch (CreateTopicsException ex) when ( + ex.Results.All(result => result.Error.Code == ErrorCode.TopicAlreadyExists)) + { + } + } + + public static async Task ProduceAsync( + string bootstrapServers, + string topic, + string key, + string value) + { + using var producer = new ProducerBuilder(new ProducerConfig + { + BootstrapServers = bootstrapServers, + Acks = Acks.All + }).Build(); + + await producer.ProduceAsync( + topic, + new Message + { + Key = key, + Value = value + }); + + producer.Flush(TimeSpan.FromSeconds(5)); + } + + public static ConsumeResult ConsumeSingle( + string bootstrapServers, + string topic, + TimeSpan timeout) + { + using var consumer = new ConsumerBuilder(new ConsumerConfig + { + BootstrapServers = bootstrapServers, + GroupId = $"test-{Guid.NewGuid():N}", + AutoOffsetReset = AutoOffsetReset.Earliest, + EnableAutoCommit = false + }).Build(); + + consumer.Subscribe(topic); + + var deadline = DateTimeOffset.UtcNow + timeout; + + while (DateTimeOffset.UtcNow < deadline) + { + var result = consumer.Consume(TimeSpan.FromMilliseconds(500)); + + if (result is not null) + { + return result; + } + } + + throw new TimeoutException($"Kafka message was not consumed from topic {topic}."); + } +} diff --git a/tests/LinkTracker.Scrapper.Tests/LinkTracker.Scrapper.Tests.csproj b/tests/LinkTracker.Scrapper.Tests/LinkTracker.Scrapper.Tests.csproj index ed868bc..ff9516e 100644 --- a/tests/LinkTracker.Scrapper.Tests/LinkTracker.Scrapper.Tests.csproj +++ b/tests/LinkTracker.Scrapper.Tests/LinkTracker.Scrapper.Tests.csproj @@ -8,8 +8,10 @@ + + @@ -17,6 +19,7 @@ + diff --git a/tests/LinkTracker.Scrapper.Tests/Postgres/MigrationTests.cs b/tests/LinkTracker.Scrapper.Tests/Postgres/MigrationTests.cs index 4c7eb5a..3ca92a6 100644 --- a/tests/LinkTracker.Scrapper.Tests/Postgres/MigrationTests.cs +++ b/tests/LinkTracker.Scrapper.Tests/Postgres/MigrationTests.cs @@ -29,6 +29,7 @@ FROM information_schema.tables Assert.Contains("chat_links", tables); Assert.Contains("tags", tables); Assert.Contains("chat_link_tags", tables); + Assert.Contains("notification_outbox", tables); Assert.Contains("schemaversions", tables); } } diff --git a/tests/LinkTracker.Scrapper.Tests/Postgres/OutboxTests.cs b/tests/LinkTracker.Scrapper.Tests/Postgres/OutboxTests.cs new file mode 100644 index 0000000..f028e8d --- /dev/null +++ b/tests/LinkTracker.Scrapper.Tests/Postgres/OutboxTests.cs @@ -0,0 +1,161 @@ +using LinkTracker.Scrapper.Configuration; +using LinkTracker.Scrapper.Services.Notifications; +using LinkTracker.Shared.Models; +using Microsoft.Extensions.Logging.Abstractions; +using Microsoft.Extensions.Options; +using Npgsql; + +namespace LinkTracker.Scrapper.Tests.Postgres; + +[Collection(PostgresCollection.Name)] +public class OutboxTests(PostgresFixture fixture) +{ + [Fact] + public async Task OutboxMessageSender_WritesPendingMessage() + { + await fixture.ResetDatabaseAsync(); + + var sender = CreateOutboxSender(); + + await sender.SendAsync( + new LinkUpdate(42, "https://github.com/octo/repo", "new issue", [1001]), + CancellationToken.None); + + var row = await ReadSingleOutboxRowAsync(); + + Assert.Equal("link-updates-test", row.Topic); + Assert.Equal("42", row.MessageKey); + Assert.Equal("Pending", row.Status); + Assert.Contains("github.com/octo/repo", row.Payload); + } + + [Fact] + public async Task KafkaOutboxDispatcher_MarksMessageProcessedAfterSuccessfulPublish() + { + await fixture.ResetDatabaseAsync(); + + var sender = CreateOutboxSender(); + var publisher = new CapturingKafkaPublisher(); + var dispatcher = CreateDispatcher(publisher); + + await sender.SendAsync( + new LinkUpdate(0, "https://github.com/octo/repo", "new issue", [1001]), + CancellationToken.None); + + await dispatcher.DispatchPendingBatchAsync(CancellationToken.None); + + var row = await ReadSingleOutboxRowAsync(); + + Assert.Equal("Processed", row.Status); + Assert.Equal(1, row.AttemptCount); + Assert.Null(row.LastError); + Assert.Single(publisher.Messages); + Assert.Equal("https://github.com/octo/repo", publisher.Messages.Single().Key); + } + + [Fact] + public async Task KafkaOutboxDispatcher_KeepsMessagePendingWhenPublishFails() + { + await fixture.ResetDatabaseAsync(); + + var sender = CreateOutboxSender(); + var publisher = new CapturingKafkaPublisher(shouldFail: true); + var dispatcher = CreateDispatcher(publisher); + + await sender.SendAsync( + new LinkUpdate(0, "https://github.com/octo/repo", "new issue", [1001]), + CancellationToken.None); + + await dispatcher.DispatchPendingBatchAsync(CancellationToken.None); + + var row = await ReadSingleOutboxRowAsync(); + + Assert.Equal("Pending", row.Status); + Assert.Equal(1, row.AttemptCount); + Assert.Contains("Kafka unavailable", row.LastError); + Assert.Empty(publisher.Messages); + } + + private OutboxMessageSender CreateOutboxSender() + { + return new OutboxMessageSender( + fixture.DataSource, + Options.Create(CreateOptions())); + } + + private KafkaOutboxDispatcher CreateDispatcher(IKafkaMessagePublisher publisher) + { + return new KafkaOutboxDispatcher( + fixture.DataSource, + publisher, + Options.Create(CreateOptions()), + NullLogger.Instance); + } + + private static NotificationOptions CreateOptions() + { + return new NotificationOptions + { + Transport = NotificationTransports.Kafka, + Kafka = new KafkaProducerOptions + { + Topic = "link-updates-test", + OutboxBatchSize = 10, + OutboxDispatchIntervalSeconds = 1 + } + }; + } + + private async Task ReadSingleOutboxRowAsync() + { + await using var connection = await fixture.DataSource.OpenConnectionAsync(); + await using var command = new NpgsqlCommand(""" + SELECT topic, message_key, payload::text, status, attempt_count, last_error + FROM notification_outbox; + """, connection); + + await using var reader = await command.ExecuteReaderAsync(); + + Assert.True(await reader.ReadAsync()); + + var row = new OutboxRow( + reader.GetString(0), + reader.GetString(1), + reader.GetString(2), + reader.GetString(3), + reader.GetInt32(4), + reader.IsDBNull(5) ? null : reader.GetString(5)); + + Assert.False(await reader.ReadAsync()); + + return row; + } + + private sealed class CapturingKafkaPublisher(bool shouldFail = false) : IKafkaMessagePublisher + { + public List<(string Topic, string? Key, string Payload)> Messages { get; } = []; + + public Task PublishAsync( + string topic, + string? key, + string payload, + CancellationToken cancellationToken) + { + if (shouldFail) + { + throw new InvalidOperationException("Kafka unavailable"); + } + + Messages.Add((topic, key, payload)); + return Task.CompletedTask; + } + } + + private sealed record OutboxRow( + string Topic, + string MessageKey, + string Payload, + string Status, + int AttemptCount, + string? LastError); +} diff --git a/tests/LinkTracker.Scrapper.Tests/Postgres/PostgresFixture.cs b/tests/LinkTracker.Scrapper.Tests/Postgres/PostgresFixture.cs index d557289..97ed7ce 100644 --- a/tests/LinkTracker.Scrapper.Tests/Postgres/PostgresFixture.cs +++ b/tests/LinkTracker.Scrapper.Tests/Postgres/PostgresFixture.cs @@ -8,8 +8,7 @@ namespace LinkTracker.Scrapper.Tests.Postgres; public sealed class PostgresFixture : IAsyncLifetime { - private readonly PostgreSqlContainer _container = new PostgreSqlBuilder() - .WithImage("postgres:16") + private readonly PostgreSqlContainer _container = new PostgreSqlBuilder("postgres:16") .WithDatabase("linktracker_tests") .WithUsername("linktracker") .WithPassword("linktracker") @@ -46,6 +45,7 @@ public async Task ResetDatabaseAsync() await using var connection = await DataSource.OpenConnectionAsync(); await using var command = new NpgsqlCommand(""" TRUNCATE TABLE + notification_outbox, chat_link_tags, chat_links, tags, From 8e25921f4df816eb1c6cd779d3e1228c26250c3e Mon Sep 17 00:00:00 2001 From: 666mxvbee Date: Tue, 7 Jul 2026 20:37:52 +0300 Subject: [PATCH 5/6] docs: update async notification setup --- README.md | 45 ++++++++++++++++++++++++++++++++++++++++++++- 1 file changed, 44 insertions(+), 1 deletion(-) diff --git a/README.md b/README.md index 4fa3454..dfaac1b 100644 --- a/README.md +++ b/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; another implementation such as Kafka can be added later without changing the scheduler business logic. +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`. From 3d7004124584c148bcc44aa5bbf1d2998847308b Mon Sep 17 00:00:00 2001 From: 666mxvbee Date: Tue, 7 Jul 2026 20:46:03 +0300 Subject: [PATCH 6/6] test: pull kafka image when missing --- tests/LinkTracker.Scrapper.Tests/Kafka/KafkaFixture.cs | 2 ++ 1 file changed, 2 insertions(+) diff --git a/tests/LinkTracker.Scrapper.Tests/Kafka/KafkaFixture.cs b/tests/LinkTracker.Scrapper.Tests/Kafka/KafkaFixture.cs index 302e045..99bdcc3 100644 --- a/tests/LinkTracker.Scrapper.Tests/Kafka/KafkaFixture.cs +++ b/tests/LinkTracker.Scrapper.Tests/Kafka/KafkaFixture.cs @@ -1,3 +1,4 @@ +using DotNet.Testcontainers.Images; using Testcontainers.Kafka; namespace LinkTracker.Scrapper.Tests.Kafka; @@ -5,6 +6,7 @@ namespace LinkTracker.Scrapper.Tests.Kafka; public sealed class KafkaFixture : IAsyncLifetime { private readonly KafkaContainer _container = new KafkaBuilder("confluentinc/cp-kafka:7.6.1") + .WithImagePullPolicy(PullPolicy.Missing) .Build(); public string BootstrapServers => _container.GetBootstrapAddress();