diff --git a/.dockerignore b/.dockerignore index b347bcd..408f9b5 100644 --- a/.dockerignore +++ b/.dockerignore @@ -4,4 +4,7 @@ .vs/ .idea/ *.user -*.suo \ No newline at end of file +*.suo +**/.env +**/.env.* +!**/.env.template diff --git a/Directory.Build.props b/Directory.Build.props new file mode 100644 index 0000000..fbbe1b8 --- /dev/null +++ b/Directory.Build.props @@ -0,0 +1,14 @@ + + + + net9.0 + latest + enable + enable + true + true + latest + false + + + diff --git a/OBSERVABILITY.md b/OBSERVABILITY.md index 88d9687..7170840 100644 --- a/OBSERVABILITY.md +++ b/OBSERVABILITY.md @@ -32,7 +32,8 @@ Rate limiter применяется точечно, политикой `public-a | `api_requests_total` | Counter | `source` | Счётчик запросов к API (лейбл — шаблон маршрута, не сырой путь) | | `request_duration_ms_total` | Histogram | `scope`, `scope_type` | Длительность операции в мс (RED: Duration) | | `errors_total` | Counter | `scope`, `scope_type`, `reason` | **Ошибки (RED: Errors)** | -| `sent_updates_total` | Counter | — | Количество обновлений, отправленных в Bot | +| `sent_updates_total` | Counter | — | Количество обновлений, фактически отправленных в Bot (в outbox-режиме инкрементируется при успешной отправке из outbox) | +| `outbox_enqueued_updates_total` | Counter | — | Количество обновлений, записанных в transactional outbox | | `db_queries_total` | Counter | `operation` | Количество запросов к БД | | `db_errors_total` | Counter | `operation` | Количество ошибок БД | | `db_query_duration_ms_total` | Histogram | `operation` | Длительность запроса к БД в мс | @@ -58,6 +59,20 @@ Rate limiter применяется точечно, политикой `public-a | `process_memory_working_set_bytes` | Gauge | — | **Потребление RAM (working set), метрика для алерта** | | `process_memory_managed_bytes` | Gauge | — | Управляемая память (GC) | +### AI Agent + +| Метрика | Тип | Лейблы | Описание | +|---|---|---|---| +| `kafka_consumed_total` | Counter | `topic` | Сообщений обработано из Kafka | +| `kafka_consume_errors_total` | Counter | `topic` | Ошибок обработки из Kafka | +| `kafka_consume_duration_ms_total` | Histogram | `topic` | Длительность обработки из Kafka в мс | +| `kafka_dead_letter_total` | Counter | `topic` | Сообщений отправлено в DLQ | +| `kafka_dead_letter_errors_total` | Counter | `topic` | Неудачных отправок в DLQ | +| `summarizations_total` | Counter | — | Успешных суммаризаций через Yandex AI | +| `summarization_fallbacks_total` | Counter | `reason` | **Суммаризаций, деградировавших до обрезки текста** | +| `process_memory_working_set_bytes` | Gauge | — | **Потребление RAM (working set), метрика для алерта** | +| `process_memory_managed_bytes` | Gauge | — | Управляемая память (GC) | + > Histogram-метрики дают в Prometheus три серии: `_bucket`, `_sum`, `_count`. ## Конфигурация сбора diff --git a/README.md b/README.md index acd5f08..e82d2c7 100644 --- a/README.md +++ b/README.md @@ -87,9 +87,9 @@ dotnet run --project src/LinkTracker.AiAgent.Api ### Для запуска в контейнерах 9. В [bot.Docker.appesettings](src/LinkTracker.Bot.Api/appsettings.Docker.json), [scrapper.Docker.appsettings](src/LinkTracker.Scrapper.Api/appsettings.Docker.json) и [aiagent.Docker.appesettings](src/LinkTracker.AiAgent.Api/appsettings.Docker.json) в ветках Scrapper, Bot и AiAgent соответственно выбрать валидные параметры. -10. Выполнить поочердено запуск сначала [docker-compose](docker-compose.yml), [Scrapper](src/LinkTracker.Scrapper.Api/Program.cs), а затем [Bot](src/LinkTracker.Bot.Api/Program.cs) с помощью команд. +10. Поднять инфраструктуру и все три сервиса одной командой — порядок запуска обеспечивают `depends_on` в compose-файлах. ``` -docker compose -f docker-compose.yml -f docker-compose.apps.yml up +docker compose -f docker-compose.yml -f docker-compose.apps.yml up ``` diff --git a/docker-compose.apps.yml b/docker-compose.apps.yml index dba379d..80606a6 100644 --- a/docker-compose.apps.yml +++ b/docker-compose.apps.yml @@ -41,6 +41,8 @@ services: condition: service_completed_successfully schema-registry: condition: service_started + valkey-cluster-init: + condition: service_completed_successfully expose: - "8091" - "8092" diff --git a/migrations/006_links_last_checked_at.sql b/migrations/006_links_last_checked_at.sql new file mode 100644 index 0000000..38828ec --- /dev/null +++ b/migrations/006_links_last_checked_at.sql @@ -0,0 +1,4 @@ +ALTER TABLE links + ADD COLUMN IF NOT EXISTS last_checked_at TIMESTAMPTZ NOT NULL DEFAULT '0001-01-01 00:00:00+00'; + +CREATE INDEX IF NOT EXISTS ix_links_last_checked_at ON links (last_checked_at, id); diff --git a/src/LinkTracker.AiAgent.Api/LinkTracker.AiAgent.Api.csproj b/src/LinkTracker.AiAgent.Api/LinkTracker.AiAgent.Api.csproj index 49a6112..e03a233 100644 --- a/src/LinkTracker.AiAgent.Api/LinkTracker.AiAgent.Api.csproj +++ b/src/LinkTracker.AiAgent.Api/LinkTracker.AiAgent.Api.csproj @@ -1,10 +1,4 @@ - - net9.0 - enable - enable - - diff --git a/src/LinkTracker.AiAgent.Api/appsettings.Docker.json b/src/LinkTracker.AiAgent.Api/appsettings.Docker.json index e1c2e96..c0979c5 100644 --- a/src/LinkTracker.AiAgent.Api/appsettings.Docker.json +++ b/src/LinkTracker.AiAgent.Api/appsettings.Docker.json @@ -52,10 +52,5 @@ "Grouping": { "WindowMs": 30000 } - }, - "YandexAi": { - "BaseUrl": "https://ai.api.cloud.yandex.net", - "ModelId": "aliceai-llm", - "TimeoutSeconds": 120 } } \ No newline at end of file diff --git a/src/LinkTracker.AiAgent.Api/appsettings.json b/src/LinkTracker.AiAgent.Api/appsettings.json index 6bd95d6..bc1e0f4 100644 --- a/src/LinkTracker.AiAgent.Api/appsettings.json +++ b/src/LinkTracker.AiAgent.Api/appsettings.json @@ -64,10 +64,5 @@ "Grouping": { "WindowMs": 30000 } - }, - "YandexAi": { - "BaseUrl": "https://ai.api.cloud.yandex.net", - "ModelId": "aliceai-llm", - "TimeoutSeconds": 120 } } \ No newline at end of file diff --git a/src/LinkTracker.AiAgent.Application/LinkTracker.AiAgent.Application.csproj b/src/LinkTracker.AiAgent.Application/LinkTracker.AiAgent.Application.csproj index 6c1c2b1..4af4800 100644 --- a/src/LinkTracker.AiAgent.Application/LinkTracker.AiAgent.Application.csproj +++ b/src/LinkTracker.AiAgent.Application/LinkTracker.AiAgent.Application.csproj @@ -1,10 +1,4 @@ - - net9.0 - enable - enable - - diff --git a/src/LinkTracker.AiAgent.Application/Telemetry/Abstractions/IAiAgentMetrics.cs b/src/LinkTracker.AiAgent.Application/Telemetry/Abstractions/IAiAgentMetrics.cs index a458c45..319411d 100644 --- a/src/LinkTracker.AiAgent.Application/Telemetry/Abstractions/IAiAgentMetrics.cs +++ b/src/LinkTracker.AiAgent.Application/Telemetry/Abstractions/IAiAgentMetrics.cs @@ -11,4 +11,8 @@ public interface IAiAgentMetrics void IncrementKafkaDeadLetter(string topic); void IncrementKafkaDeadLetterError(string topic); + + void IncrementSummarization(); + + void IncrementSummarizationFallback(string reason); } diff --git a/src/LinkTracker.AiAgent.Infrastructure/Clients/Kafka/RawUpdatesKafkaConsumer.cs b/src/LinkTracker.AiAgent.Infrastructure/Clients/Kafka/RawUpdatesKafkaConsumer.cs index 511f77b..fae6ebe 100644 --- a/src/LinkTracker.AiAgent.Infrastructure/Clients/Kafka/RawUpdatesKafkaConsumer.cs +++ b/src/LinkTracker.AiAgent.Infrastructure/Clients/Kafka/RawUpdatesKafkaConsumer.cs @@ -17,10 +17,16 @@ internal sealed class RawUpdatesKafkaConsumer( ILogger logger) : BackgroundService { private static readonly TimeSpan PollTimeout = TimeSpan.FromMilliseconds(500); + private static readonly TimeSpan ConsumeErrorBackoff = TimeSpan.FromSeconds(1); protected override Task ExecuteAsync(CancellationToken stoppingToken) { - return Task.Run(() => ConsumeLoopAsync(stoppingToken), stoppingToken); + return Task.Factory.StartNew( + () => ConsumeLoopAsync(stoppingToken), + stoppingToken, + TaskCreationOptions.LongRunning | TaskCreationOptions.DenyChildAttach, + TaskScheduler.Default) + .Unwrap(); } private async Task ConsumeLoopAsync(CancellationToken stoppingToken) @@ -45,7 +51,8 @@ private async Task ConsumeLoopAsync(CancellationToken stoppingToken) } catch (ConsumeException ex) { - logger.LogWarning(ex, "Kafka consume завершился ошибкой."); + logger.LogWarning(ex, "Kafka consume завершился ошибкой. Пауза перед повтором."); + await Task.Delay(ConsumeErrorBackoff, stoppingToken); continue; } diff --git a/src/LinkTracker.AiAgent.Infrastructure/Clients/Registration/ClientsModule.cs b/src/LinkTracker.AiAgent.Infrastructure/Clients/Registration/ClientsModule.cs index d652995..53fbf84 100644 --- a/src/LinkTracker.AiAgent.Infrastructure/Clients/Registration/ClientsModule.cs +++ b/src/LinkTracker.AiAgent.Infrastructure/Clients/Registration/ClientsModule.cs @@ -24,11 +24,10 @@ public static IServiceCollection AddAiAgentInfrastructure( services .AddOptions() .Bind(configuration.GetSection("AiAgent")) - .Validate(o => o.Filtering != null, "AiAgent:Filtering must be set") - .Validate(o => o.Summarization != null, "AiAgent:Summarization must be set") - .Validate(o => o.Prioritization != null, "AiAgent:Prioritization must be set") - .Validate(o => o.Grouping.WindowMs > 0, "AiAgent:Grouping:WindowMs must be greater that 0") - .Validate(o => o.Grouping.FlushIntervalMs > 0, "AiAgent:Grouping.FlushIntervalMs must be greater that 0") + .Validate(o => o.Filtering.MinLength >= 0, "AiAgent:Filtering:MinLength must not be negative") + .Validate(o => o.Summarization.Threshold > 0, "AiAgent:Summarization:Threshold must be greater than 0") + .Validate(o => o.Grouping.WindowMs > 0, "AiAgent:Grouping:WindowMs must be greater than 0") + .Validate(o => o.Grouping.FlushIntervalMs > 0, "AiAgent:Grouping:FlushIntervalMs must be greater than 0") .ValidateOnStart(); services @@ -52,6 +51,13 @@ public static IServiceCollection AddAiAgentInfrastructure( services .AddOptions() .Bind(configuration.GetSection("YandexAi")) + .Validate(o => !string.IsNullOrWhiteSpace(o.ApiKey), "YandexAi:ApiKey must be set") + .Validate(o => !string.IsNullOrWhiteSpace(o.FolderId), "YandexAi:FolderId must be set") + .Validate(o => !string.IsNullOrWhiteSpace(o.ModelId), "YandexAi:ModelId must be set") + .Validate( + o => Uri.TryCreate(o.BaseUrl, UriKind.Absolute, out _), + "YandexAi:BaseUrl must be an absolute URI") + .Validate(o => o.TimeoutSeconds > 0, "YandexAi:TimeoutSeconds must be greater than 0") .ValidateOnStart(); services.AddSingleton(); diff --git a/src/LinkTracker.AiAgent.Infrastructure/Clients/YandexAi/YandexAiHttpClient.cs b/src/LinkTracker.AiAgent.Infrastructure/Clients/YandexAi/YandexAiHttpClient.cs index f352aa9..6eb4248 100644 --- a/src/LinkTracker.AiAgent.Infrastructure/Clients/YandexAi/YandexAiHttpClient.cs +++ b/src/LinkTracker.AiAgent.Infrastructure/Clients/YandexAi/YandexAiHttpClient.cs @@ -1,6 +1,7 @@ using System.Net.Http.Headers; using System.Net.Http.Json; using LinkTracker.AiAgent.Application.Abstractions; +using LinkTracker.AiAgent.Application.Telemetry.Abstractions; using LinkTracker.AiAgent.Infrastructure.Clients.YandexAi.Contracts; using LinkTracker.AiAgent.Infrastructure.Configuration.AiAgent; using LinkTracker.AiAgent.Infrastructure.Configuration.YandexAi; @@ -13,8 +14,11 @@ internal sealed class YandexAiHttpClient( IHttpClientFactory httpClientFactory, IOptions yandexOptions, IOptions agentOptions, + IAiAgentMetrics metrics, ILogger logger) : ILinkUpdateSummarizer { + private const string Instructions = "You are a concise summarizer. Summarize the given update in 2-3 sentences."; + public async Task SummarizeAsync(string text, CancellationToken ct) { var threshold = agentOptions.Value.Summarization.Threshold; @@ -26,7 +30,18 @@ public async Task SummarizeAsync(string text, CancellationToken ct) try { - return await CallApiAsync(text, ct); + var summary = await CallApiAsync(text, ct); + + if (string.IsNullOrWhiteSpace(summary)) + { + logger.LogWarning("Yandex AI вернул пустой ответ. Используется обрезка текста."); + metrics.IncrementSummarizationFallback("empty_response"); + + return FallbackTruncate(text, threshold); + } + + metrics.IncrementSummarization(); + return summary; } catch (OperationCanceledException) when (ct.IsCancellationRequested) { @@ -34,16 +49,18 @@ public async Task SummarizeAsync(string text, CancellationToken ct) } catch (Exception ex) { - logger.LogWarning(ex, "Yandex AI суммаризация завершилась ошибкой ({Type}). Используется заглушка.", ex.GetType().Name); + logger.LogWarning(ex, "Yandex AI суммаризация завершилась ошибкой ({Type}). Используется обрезка текста.", ex.GetType().Name); + metrics.IncrementSummarizationFallback(ex.GetType().Name); + return FallbackTruncate(text, threshold); } } - private async Task CallApiAsync(string text, CancellationToken ct) + private async Task CallApiAsync(string text, CancellationToken ct) { var opts = yandexOptions.Value; - var requestBody = new YandexResponsesRequest { Model = $"gpt://{opts.FolderId}/{opts.ModelId}/latest", Instructions = "You are a concise summarizer. Summarize the given update in 2-3 sentences.", Input = text }; + var requestBody = new YandexResponsesRequest { Model = $"gpt://{opts.FolderId}/{opts.ModelId}/latest", Instructions = Instructions, Input = text }; var httpClient = httpClientFactory.CreateClient(nameof(YandexAiHttpClient)); @@ -57,15 +74,7 @@ private async Task CallApiAsync(string text, CancellationToken ct) var result = await response.Content.ReadFromJsonAsync(ct); - var responseText = result?.Output?.FirstOrDefault()?.Content?.FirstOrDefault()?.Text; - - if (!string.IsNullOrWhiteSpace(responseText)) - { - return responseText; - } - - logger.LogWarning("Yandex AI вернул пустой ответ."); - return text; + return result?.Output?.FirstOrDefault()?.Content?.FirstOrDefault()?.Text; } private static string FallbackTruncate(string text, int threshold) @@ -84,4 +93,4 @@ private static string FallbackTruncate(string text, int threshold) return string.Concat(text.AsSpan(0, cutAt).TrimEnd(), "\n..."); } -} \ No newline at end of file +} diff --git a/src/LinkTracker.AiAgent.Infrastructure/Configuration/AiAgent/AiAgentOptions.cs b/src/LinkTracker.AiAgent.Infrastructure/Configuration/AiAgent/AiAgentOptions.cs index 0cbc1b8..60b1365 100644 --- a/src/LinkTracker.AiAgent.Infrastructure/Configuration/AiAgent/AiAgentOptions.cs +++ b/src/LinkTracker.AiAgent.Infrastructure/Configuration/AiAgent/AiAgentOptions.cs @@ -2,9 +2,9 @@ namespace LinkTracker.AiAgent.Infrastructure.Configuration.AiAgent; public sealed class AiAgentOptions { - public FilteringOptions? Filtering { get; set; } - public SummarizationOptions? Summarization { get; set; } - public PrioritizationOptions? Prioritization { get; set; } + public FilteringOptions Filtering { get; set; } = new(); + public SummarizationOptions Summarization { get; set; } = new(); + public PrioritizationOptions Prioritization { get; set; } = new(); public GroupingOptions Grouping { get; set; } = new(); } diff --git a/src/LinkTracker.AiAgent.Infrastructure/LinkTracker.AiAgent.Infrastructure.csproj b/src/LinkTracker.AiAgent.Infrastructure/LinkTracker.AiAgent.Infrastructure.csproj index 9b08bf9..4269c14 100644 --- a/src/LinkTracker.AiAgent.Infrastructure/LinkTracker.AiAgent.Infrastructure.csproj +++ b/src/LinkTracker.AiAgent.Infrastructure/LinkTracker.AiAgent.Infrastructure.csproj @@ -1,10 +1,4 @@ - - net9.0 - enable - enable - - diff --git a/src/LinkTracker.AiAgent.Infrastructure/Telemetry/AiAgentMetrics.cs b/src/LinkTracker.AiAgent.Infrastructure/Telemetry/AiAgentMetrics.cs index 5db8dcb..9b42b4f 100644 --- a/src/LinkTracker.AiAgent.Infrastructure/Telemetry/AiAgentMetrics.cs +++ b/src/LinkTracker.AiAgent.Infrastructure/Telemetry/AiAgentMetrics.cs @@ -16,6 +16,8 @@ public sealed class AiAgentMetrics : IAiAgentMetrics, IDisposable private readonly Counter _kafkaConsumeErrors; private readonly Counter _kafkaDeadLetterErrors; private readonly Counter _kafkaDeadLetters; + private readonly Counter _summarizationFallbacks; + private readonly Counter _summarizations; private readonly Meter _meter; @@ -45,6 +47,14 @@ public AiAgentMetrics() "Длительность обработки сообщения из Kafka в миллисекундах", advice: new InstrumentAdvice { HistogramBucketBoundaries = DurationBuckets }); + _summarizations = _meter.CreateCounter( + "summarizations_total", + description: "Количество успешных суммаризаций через Yandex AI"); + + _summarizationFallbacks = _meter.CreateCounter( + "summarization_fallbacks_total", + description: "Количество суммаризаций, деградировавших до обрезки текста, с разбивкой по причине"); + _meter.CreateObservableGauge( "process_memory_working_set_bytes", static () => Process.GetCurrentProcess().WorkingSet64, @@ -93,6 +103,18 @@ public void IncrementKafkaDeadLetterError(string topic) new KeyValuePair("topic", topic)); } + public void IncrementSummarization() + { + _summarizations.Add(1); + } + + public void IncrementSummarizationFallback(string reason) + { + _summarizationFallbacks.Add( + 1, + new KeyValuePair("reason", reason)); + } + public void Dispose() { _meter.Dispose(); diff --git a/src/LinkTracker.Bot.Api/LinkTracker.Bot.Api.csproj b/src/LinkTracker.Bot.Api/LinkTracker.Bot.Api.csproj index c25eea4..ea6f66d 100644 --- a/src/LinkTracker.Bot.Api/LinkTracker.Bot.Api.csproj +++ b/src/LinkTracker.Bot.Api/LinkTracker.Bot.Api.csproj @@ -8,10 +8,4 @@ - - net9.0 - enable - enable - - diff --git a/src/LinkTracker.Bot.Api/Program.cs b/src/LinkTracker.Bot.Api/Program.cs index 243d7cd..78c9d7a 100644 --- a/src/LinkTracker.Bot.Api/Program.cs +++ b/src/LinkTracker.Bot.Api/Program.cs @@ -29,7 +29,7 @@ builder.Services.AddCommands(); builder.Services.AddUpdateRouting(); builder.Services.AddDialogs(); -builder.Services.AddDialogStorage(); +builder.Services.AddDialogStorage(builder.Configuration); builder.Services.AddClients(builder.Configuration); builder.Services.AddTelegramPresentation(builder.Configuration); builder.Services.AddTelegramNotifications(); diff --git a/src/LinkTracker.Bot.Api/appsettings.Docker.json b/src/LinkTracker.Bot.Api/appsettings.Docker.json index 05f63a5..e0d1fc0 100644 --- a/src/LinkTracker.Bot.Api/appsettings.Docker.json +++ b/src/LinkTracker.Bot.Api/appsettings.Docker.json @@ -21,6 +21,12 @@ "Serialization": "Json", "SchemaRegistryUrl": "http://schema-registry:8071" }, + "Valkey": { + "Enabled": true, + "ConnectionString": "valkey-node-1:6379,valkey-node-2:6379,valkey-node-3:6379,password=valkey", + "InstanceName": "linktracker", + "DialogTtlSeconds": 3600 + }, "Resilience": { "Http": { "TimeoutMilliseconds": 1000, diff --git a/src/LinkTracker.Bot.Api/appsettings.json b/src/LinkTracker.Bot.Api/appsettings.json index b7ff7a2..66482ad 100644 --- a/src/LinkTracker.Bot.Api/appsettings.json +++ b/src/LinkTracker.Bot.Api/appsettings.json @@ -36,6 +36,12 @@ "Serialization": "Json", "SchemaRegistryUrl": "http://localhost:8071" }, + "Valkey": { + "Enabled": true, + "ConnectionString": "localhost:7379,localhost:7380,localhost:7381,password=valkey", + "InstanceName": "linktracker", + "DialogTtlSeconds": 3600 + }, "Resilience": { "Http": { "TimeoutMilliseconds": 1000, diff --git a/src/LinkTracker.Bot.Application/Dialogs/Abstractions/DialogContext.cs b/src/LinkTracker.Bot.Application/Dialogs/Abstractions/DialogContext.cs index 8b6ab28..47399c6 100644 --- a/src/LinkTracker.Bot.Application/Dialogs/Abstractions/DialogContext.cs +++ b/src/LinkTracker.Bot.Application/Dialogs/Abstractions/DialogContext.cs @@ -1,3 +1,5 @@ +using System.Text.Json.Serialization; + namespace LinkTracker.Bot.Application.Dialogs.Abstractions; public sealed class DialogContext @@ -7,10 +9,9 @@ public sealed class DialogContext public string? ActiveDialogId { get; set; } public string? ActiveNodeId { get; set; } - public Dictionary Data { get; } = new(); - - public DateTimeOffset UpdatedAt { get; set; } = DateTimeOffset.UtcNow; + public Dictionary Data { get; init; } = new(); + [JsonIgnore] public bool HasActiveDialog => !string.IsNullOrWhiteSpace(ActiveDialogId) && !string.IsNullOrWhiteSpace(ActiveNodeId); diff --git a/src/LinkTracker.Bot.Application/LinkTracker.Bot.Application.csproj b/src/LinkTracker.Bot.Application/LinkTracker.Bot.Application.csproj index e5fb9c0..3bfa9b3 100644 --- a/src/LinkTracker.Bot.Application/LinkTracker.Bot.Application.csproj +++ b/src/LinkTracker.Bot.Application/LinkTracker.Bot.Application.csproj @@ -1,4 +1,4 @@ - + @@ -9,10 +9,4 @@ - - net9.0 - enable - enable - - \ No newline at end of file diff --git a/src/LinkTracker.Bot.Infrastructure/Clients/Kafka/LinkUpdatesKafkaConsumer.cs b/src/LinkTracker.Bot.Infrastructure/Clients/Kafka/LinkUpdatesKafkaConsumer.cs index 2565ee5..407207b 100644 --- a/src/LinkTracker.Bot.Infrastructure/Clients/Kafka/LinkUpdatesKafkaConsumer.cs +++ b/src/LinkTracker.Bot.Infrastructure/Clients/Kafka/LinkUpdatesKafkaConsumer.cs @@ -16,15 +16,17 @@ internal sealed class LinkUpdatesKafkaConsumer( IBotMetrics metrics, ILogger logger) : BackgroundService { - protected override Task ExecuteAsync(CancellationToken stoppingToken) - { - return ConsumeLoopAsync(stoppingToken); - } + private static readonly TimeSpan PollTimeout = TimeSpan.FromMilliseconds(500); + private static readonly TimeSpan ConsumeErrorBackoff = TimeSpan.FromSeconds(1); - public override Task StopAsync(CancellationToken cancellationToken) + protected override Task ExecuteAsync(CancellationToken stoppingToken) { - consumer.Close(); - return base.StopAsync(cancellationToken); + return Task.Factory.StartNew( + () => ConsumeLoopAsync(stoppingToken), + stoppingToken, + TaskCreationOptions.LongRunning | TaskCreationOptions.DenyChildAttach, + TaskScheduler.Default) + .Unwrap(); } private async Task ConsumeLoopAsync(CancellationToken stoppingToken) @@ -45,11 +47,12 @@ private async Task ConsumeLoopAsync(CancellationToken stoppingToken) try { - result = consumer.Consume(stoppingToken); + result = consumer.Consume(PollTimeout); } catch (ConsumeException ex) { - logger.LogWarning(ex, "Kafka consume завершился ошибкой."); + logger.LogWarning(ex, "Kafka consume завершился ошибкой. Пауза перед повтором."); + await Task.Delay(ConsumeErrorBackoff, stoppingToken); continue; } @@ -65,6 +68,10 @@ private async Task ConsumeLoopAsync(CancellationToken stoppingToken) { logger.LogInformation("Kafka consumer остановлен."); } + finally + { + consumer.Close(); + } } private async Task ProcessMessageAsync(ConsumeResult result, CancellationToken ct) diff --git a/src/LinkTracker.Bot.Infrastructure/Configuration/Valkey/DialogStateStoreOptions.cs b/src/LinkTracker.Bot.Infrastructure/Configuration/Valkey/DialogStateStoreOptions.cs new file mode 100644 index 0000000..b88c02d --- /dev/null +++ b/src/LinkTracker.Bot.Infrastructure/Configuration/Valkey/DialogStateStoreOptions.cs @@ -0,0 +1,14 @@ +namespace LinkTracker.Bot.Infrastructure.Configuration.Valkey; + +public sealed class DialogStateStoreOptions +{ + public const string SectionName = "Valkey"; + + public bool Enabled { get; init; } = true; + + public string ConnectionString { get; init; } = string.Empty; + + public string InstanceName { get; init; } = "linktracker"; + + public int DialogTtlSeconds { get; init; } = 3600; +} diff --git a/src/LinkTracker.Bot.Infrastructure/LinkTracker.Bot.Infrastructure.csproj b/src/LinkTracker.Bot.Infrastructure/LinkTracker.Bot.Infrastructure.csproj index 56078a2..0dbea97 100644 --- a/src/LinkTracker.Bot.Infrastructure/LinkTracker.Bot.Infrastructure.csproj +++ b/src/LinkTracker.Bot.Infrastructure/LinkTracker.Bot.Infrastructure.csproj @@ -1,4 +1,4 @@ - + @@ -16,12 +16,7 @@ + - - net9.0 - enable - enable - - diff --git a/src/LinkTracker.Bot.Infrastructure/Storage/InMemory/InMemoryDialogStateStore.cs b/src/LinkTracker.Bot.Infrastructure/Storage/InMemory/InMemoryDialogStateStore.cs index 9980d3f..037c255 100644 --- a/src/LinkTracker.Bot.Infrastructure/Storage/InMemory/InMemoryDialogStateStore.cs +++ b/src/LinkTracker.Bot.Infrastructure/Storage/InMemory/InMemoryDialogStateStore.cs @@ -1,19 +1,38 @@ using System.Collections.Concurrent; using LinkTracker.Bot.Application.Dialogs.Abstractions; +using LinkTracker.Bot.Infrastructure.Configuration.Valkey; +using Microsoft.Extensions.Options; namespace LinkTracker.Bot.Infrastructure.Storage.InMemory; public sealed class InMemoryDialogStateStore : IDialogStateStore { - private readonly ConcurrentDictionary _states = new(); + private readonly ConcurrentDictionary _states = new(); + private readonly TimeProvider _timeProvider; + private readonly TimeSpan _ttl; + + public InMemoryDialogStateStore(IOptions options, TimeProvider timeProvider) + { + ArgumentNullException.ThrowIfNull(options); + + _ttl = TimeSpan.FromSeconds(options.Value.DialogTtlSeconds); + _timeProvider = timeProvider; + } public Task GetOrCreateAsync(long chatId, CancellationToken ct) { ct.ThrowIfCancellationRequested(); - var context = _states.GetOrAdd(chatId, static id => new DialogContext { ChatId = id }); + EvictExpired(); - return Task.FromResult(context); + var now = _timeProvider.GetUtcNow(); + + if (_states.TryGetValue(chatId, out var entry) && entry.ExpiresAt > now) + { + return Task.FromResult(entry.Context); + } + + return Task.FromResult(new DialogContext { ChatId = chatId }); } public Task SaveAsync(DialogContext ctx, CancellationToken ct) @@ -21,7 +40,8 @@ public Task SaveAsync(DialogContext ctx, CancellationToken ct) ct.ThrowIfCancellationRequested(); ArgumentNullException.ThrowIfNull(ctx); - _states[ctx.ChatId] = ctx; + _states[ctx.ChatId] = new Entry(ctx, _timeProvider.GetUtcNow() + _ttl); + return Task.CompletedTask; } @@ -30,6 +50,22 @@ public Task ResetAsync(long chatId, CancellationToken ct) ct.ThrowIfCancellationRequested(); _states.TryRemove(chatId, out _); + return Task.CompletedTask; } -} \ No newline at end of file + + private void EvictExpired() + { + var now = _timeProvider.GetUtcNow(); + + foreach (var (chatId, entry) in _states) + { + if (entry.ExpiresAt <= now) + { + _states.TryRemove(new KeyValuePair(chatId, entry)); + } + } + } + + private sealed record Entry(DialogContext Context, DateTimeOffset ExpiresAt); +} diff --git a/src/LinkTracker.Bot.Infrastructure/Storage/Registration/DialogStorageModule.cs b/src/LinkTracker.Bot.Infrastructure/Storage/Registration/DialogStorageModule.cs index acecef4..7e12b34 100644 --- a/src/LinkTracker.Bot.Infrastructure/Storage/Registration/DialogStorageModule.cs +++ b/src/LinkTracker.Bot.Infrastructure/Storage/Registration/DialogStorageModule.cs @@ -1,15 +1,53 @@ using LinkTracker.Bot.Application.Dialogs.Abstractions; +using LinkTracker.Bot.Infrastructure.Configuration.Valkey; using LinkTracker.Bot.Infrastructure.Storage.InMemory; +using LinkTracker.Bot.Infrastructure.Storage.Valkey; +using LinkTracker.Shared.Infrastructure.Valkey; +using Microsoft.Extensions.Configuration; using Microsoft.Extensions.DependencyInjection; +using Microsoft.Extensions.DependencyInjection.Extensions; +using Microsoft.Extensions.Options; +using StackExchange.Redis; namespace LinkTracker.Bot.Infrastructure.Storage.Registration; public static class DialogStorageModule { - public static IServiceCollection AddDialogStorage(this IServiceCollection services) + public static IServiceCollection AddDialogStorage(this IServiceCollection services, IConfiguration configuration) { - services.AddSingleton(); + services + .AddOptions() + .Bind(configuration.GetSection(DialogStateStoreOptions.SectionName)) + .Validate( + o => o.DialogTtlSeconds > 0, + "Valkey:DialogTtlSeconds must be positive.") + .Validate( + o => !o.Enabled || !string.IsNullOrWhiteSpace(o.ConnectionString), + "Valkey connection string is required when the Valkey dialog state store is enabled.") + .ValidateOnStart(); + + services.TryAddSingleton(TimeProvider.System); + + var options = configuration + .GetSection(DialogStateStoreOptions.SectionName) + .Get() ?? new DialogStateStoreOptions(); + + if (!options.Enabled) + { + services.AddSingleton(); + + return services; + } + + services.AddSingleton(sp => + { + var storeOptions = sp.GetRequiredService>().Value; + + return ConnectionMultiplexer.Connect(ValkeyConfiguration.Parse(storeOptions.ConnectionString)); + }); + + services.AddSingleton(); return services; } -} \ No newline at end of file +} diff --git a/src/LinkTracker.Bot.Infrastructure/Storage/Valkey/ValkeyDialogStateStore.cs b/src/LinkTracker.Bot.Infrastructure/Storage/Valkey/ValkeyDialogStateStore.cs new file mode 100644 index 0000000..795f24e --- /dev/null +++ b/src/LinkTracker.Bot.Infrastructure/Storage/Valkey/ValkeyDialogStateStore.cs @@ -0,0 +1,98 @@ +using System.Text.Json; +using LinkTracker.Bot.Application.Dialogs.Abstractions; +using LinkTracker.Bot.Infrastructure.Configuration.Valkey; +using Microsoft.Extensions.Logging; +using Microsoft.Extensions.Options; +using StackExchange.Redis; + +namespace LinkTracker.Bot.Infrastructure.Storage.Valkey; + +internal sealed class ValkeyDialogStateStore( + IConnectionMultiplexer connection, + IOptions options, + ILogger logger) : IDialogStateStore +{ + private static readonly JsonSerializerOptions SerializerOptions = new(JsonSerializerDefaults.Web); + + private readonly DialogStateStoreOptions _options = options.Value; + + public async Task GetOrCreateAsync(long chatId, CancellationToken ct) + { + ct.ThrowIfCancellationRequested(); + + var key = BuildKey(chatId); + + try + { + var value = await connection.GetDatabase().StringGetAsync(key); + + if (value.IsNullOrEmpty) + { + return new DialogContext { ChatId = chatId }; + } + + var context = JsonSerializer.Deserialize((string)value!, SerializerOptions); + + return context ?? new DialogContext { ChatId = chatId }; + } + catch (Exception ex) when (ex is RedisException or JsonException) + { + logger.LogError( + ex, + "Не удалось прочитать состояние диалога. Диалог начнётся заново. ChatId={ChatId}, Key={Key}", + chatId, + key); + + return new DialogContext { ChatId = chatId }; + } + } + + public async Task SaveAsync(DialogContext ctx, CancellationToken ct) + { + ct.ThrowIfCancellationRequested(); + ArgumentNullException.ThrowIfNull(ctx); + + var key = BuildKey(ctx.ChatId); + + try + { + await connection.GetDatabase().StringSetAsync( + key, + JsonSerializer.Serialize(ctx, SerializerOptions), + TimeSpan.FromSeconds(_options.DialogTtlSeconds)); + } + catch (RedisException ex) + { + logger.LogError( + ex, + "Не удалось сохранить состояние диалога. ChatId={ChatId}, Key={Key}", + ctx.ChatId, + key); + } + } + + public async Task ResetAsync(long chatId, CancellationToken ct) + { + ct.ThrowIfCancellationRequested(); + + var key = BuildKey(chatId); + + try + { + await connection.GetDatabase().KeyDeleteAsync(key); + } + catch (RedisException ex) + { + logger.LogError( + ex, + "Не удалось сбросить состояние диалога. ChatId={ChatId}, Key={Key}", + chatId, + key); + } + } + + private string BuildKey(long chatId) + { + return $"{_options.InstanceName}:dialog:{{chat:{chatId}}}"; + } +} diff --git a/src/LinkTracker.Bot.Presentation/LinkTracker.Bot.Presentation.csproj b/src/LinkTracker.Bot.Presentation/LinkTracker.Bot.Presentation.csproj index 9582273..8ebbc2f 100644 --- a/src/LinkTracker.Bot.Presentation/LinkTracker.Bot.Presentation.csproj +++ b/src/LinkTracker.Bot.Presentation/LinkTracker.Bot.Presentation.csproj @@ -1,4 +1,4 @@ - + @@ -11,10 +11,4 @@ - - net9.0 - enable - enable - - diff --git a/src/LinkTracker.Bot.Presentation/Telegram/Registration/TelegramPresentationModule.cs b/src/LinkTracker.Bot.Presentation/Telegram/Registration/TelegramPresentationModule.cs index 5891ae0..f9b3fba 100644 --- a/src/LinkTracker.Bot.Presentation/Telegram/Registration/TelegramPresentationModule.cs +++ b/src/LinkTracker.Bot.Presentation/Telegram/Registration/TelegramPresentationModule.cs @@ -1,7 +1,5 @@ -using LinkTracker.Bot.Application.Updates.Abstractions; using LinkTracker.Bot.Presentation.Telegram.Configuration; using LinkTracker.Bot.Presentation.Telegram.Hosting; -using LinkTracker.Bot.Presentation.Telegram.Notifications; using LinkTracker.Bot.Presentation.Telegram.Updates; using Microsoft.Extensions.Configuration; using Microsoft.Extensions.DependencyInjection; @@ -31,8 +29,6 @@ public static IServiceCollection AddTelegramPresentation( services.AddSingleton(); services.AddSingleton(); - services.AddSingleton(); - services.AddHostedService(); services.AddHostedService(); diff --git a/src/LinkTracker.EnvReader/LinkTracker.EnvReader.csproj b/src/LinkTracker.EnvReader/LinkTracker.EnvReader.csproj index 02e15a6..1dd950a 100644 --- a/src/LinkTracker.EnvReader/LinkTracker.EnvReader.csproj +++ b/src/LinkTracker.EnvReader/LinkTracker.EnvReader.csproj @@ -1,10 +1,4 @@ - - - net9.0 - enable - enable - - + diff --git a/src/LinkTracker.Scrapper.Api/LinkTracker.Scrapper.Api.csproj b/src/LinkTracker.Scrapper.Api/LinkTracker.Scrapper.Api.csproj index dc6d6c3..e133576 100644 --- a/src/LinkTracker.Scrapper.Api/LinkTracker.Scrapper.Api.csproj +++ b/src/LinkTracker.Scrapper.Api/LinkTracker.Scrapper.Api.csproj @@ -9,10 +9,4 @@ - - net9.0 - enable - enable - - diff --git a/src/LinkTracker.Scrapper.Api/appsettings.json b/src/LinkTracker.Scrapper.Api/appsettings.json index 108076c..1aa3283 100644 --- a/src/LinkTracker.Scrapper.Api/appsettings.json +++ b/src/LinkTracker.Scrapper.Api/appsettings.json @@ -23,15 +23,31 @@ "Transport": "Kafka" }, "Scheduling": { - "IntervalSeconds": 30, + "IntervalSeconds": 300, "BatchSize": 100, "MaxDegreeOfParallelism": 4 }, "GitHub": { - "BaseUrl": "https://api.github.com" + "BaseUrl": "https://api.github.com", + "RateLimit": { + "Enabled": true, + "TokenLimit": 200, + "TokensPerPeriod": 80, + "ReplenishmentPeriodSeconds": 60, + "QueueLimit": 1000, + "AcquireTimeoutSeconds": 60 + } }, "StackOverflow": { - "BaseUrl": "https://api.stackexchange.com" + "BaseUrl": "https://api.stackexchange.com", + "RateLimit": { + "Enabled": true, + "TokenLimit": 30, + "TokensPerPeriod": 12, + "ReplenishmentPeriodSeconds": 3600, + "QueueLimit": 1000, + "AcquireTimeoutSeconds": 60 + } }, "Database": { "Host": "localhost", diff --git a/src/LinkTracker.Scrapper.Application/Clients/GitHub/IGitHubClient.cs b/src/LinkTracker.Scrapper.Application/Clients/GitHub/IGitHubClient.cs index aedc9da..6410f1b 100644 --- a/src/LinkTracker.Scrapper.Application/Clients/GitHub/IGitHubClient.cs +++ b/src/LinkTracker.Scrapper.Application/Clients/GitHub/IGitHubClient.cs @@ -6,7 +6,14 @@ public interface IGitHubClient { Task GetRepositoryAsync(string owner, string repository, CancellationToken ct = default); - Task> GetIssuesAsync(string owner, string repository, CancellationToken ct = default); + Task> GetIssuesAsync( + string owner, + string repository, + DateTimeOffset? since = null, + CancellationToken ct = default); - Task> GetPullRequestsAsync(string owner, string repository, CancellationToken ct = default); -} \ No newline at end of file + Task> GetPullRequestsAsync( + string owner, + string repository, + CancellationToken ct = default); +} diff --git a/src/LinkTracker.Scrapper.Application/LinkTracker.Scrapper.Application.csproj b/src/LinkTracker.Scrapper.Application/LinkTracker.Scrapper.Application.csproj index 32b4518..123358b 100644 --- a/src/LinkTracker.Scrapper.Application/LinkTracker.Scrapper.Application.csproj +++ b/src/LinkTracker.Scrapper.Application/LinkTracker.Scrapper.Application.csproj @@ -1,4 +1,4 @@ - + @@ -11,10 +11,4 @@ - - net9.0 - enable - enable - - \ No newline at end of file diff --git a/src/LinkTracker.Scrapper.Application/Services/Updates/Clients/GitHubLinkUpdateHandler.cs b/src/LinkTracker.Scrapper.Application/Services/Updates/Clients/GitHubLinkUpdateHandler.cs index 6be5580..4bc4a72 100644 --- a/src/LinkTracker.Scrapper.Application/Services/Updates/Clients/GitHubLinkUpdateHandler.cs +++ b/src/LinkTracker.Scrapper.Application/Services/Updates/Clients/GitHubLinkUpdateHandler.cs @@ -34,7 +34,7 @@ protected override async Task> GetNewEventsAsync( { TryParseRepository(subscription.Url, out var owner, out var repository); - var issuesTask = gitHubClient.GetIssuesAsync(owner, repository, ct); + var issuesTask = gitHubClient.GetIssuesAsync(owner, repository, lastSeenAt, ct); var pullRequestsTask = gitHubClient.GetPullRequestsAsync(owner, repository, ct); await Task.WhenAll(issuesTask, pullRequestsTask); diff --git a/src/LinkTracker.Scrapper.Application/Services/Updates/Clients/StackOverflowLinkUpdateHandler.cs b/src/LinkTracker.Scrapper.Application/Services/Updates/Clients/StackOverflowLinkUpdateHandler.cs index bf83dbd..49fd0f3 100644 --- a/src/LinkTracker.Scrapper.Application/Services/Updates/Clients/StackOverflowLinkUpdateHandler.cs +++ b/src/LinkTracker.Scrapper.Application/Services/Updates/Clients/StackOverflowLinkUpdateHandler.cs @@ -37,6 +37,13 @@ protected override async Task> GetNewEventsAsync( { TryParseQuestionId(subscription.Url, out var questionId); + var questionResponse = await stackOverflowClient.GetQuestionAsync(questionId, ct); + + if (questionResponse is null || questionResponse.LastActivityDate <= lastSeenAt) + { + return []; + } + var answersTask = stackOverflowClient.GetAnswersAsync(questionId, ct); var commentsTask = stackOverflowClient.GetCommentsAsync(questionId, ct); @@ -45,19 +52,7 @@ protected override async Task> GetNewEventsAsync( var answers = await answersTask; var comments = await commentsTask; - var hasNewAnswers = answers.Any(x => - IsAfterCursor(x.CreationDate, $"answer:{x.AnswerId}", lastSeenAt, lastEventKey)); - - var hasNewComments = comments.Any(x => - IsAfterCursor(x.CreationDate, $"comment:{x.CommentId}", lastSeenAt, lastEventKey)); - - if (!hasNewAnswers && !hasNewComments) - { - return []; - } - - var questionResponse = await stackOverflowClient.GetQuestionAsync(questionId, ct); - var title = questionResponse?.Title ?? subscription.Url.AbsoluteUri; + var title = questionResponse.Title; var events = new List(); diff --git a/src/LinkTracker.Scrapper.Contracts/LinkTracker.Scrapper.Contracts.csproj b/src/LinkTracker.Scrapper.Contracts/LinkTracker.Scrapper.Contracts.csproj index bcaf623..4634abc 100644 --- a/src/LinkTracker.Scrapper.Contracts/LinkTracker.Scrapper.Contracts.csproj +++ b/src/LinkTracker.Scrapper.Contracts/LinkTracker.Scrapper.Contracts.csproj @@ -1,10 +1,4 @@ - - - - net9.0 - enable - enable - + diff --git a/src/LinkTracker.Scrapper.Infrastructure/Cache/Helpers/LinksResponseCacheKeyBuilder.cs b/src/LinkTracker.Scrapper.Infrastructure/Cache/Helpers/LinksResponseCacheKeyBuilder.cs index 33a4502..f6d205b 100644 --- a/src/LinkTracker.Scrapper.Infrastructure/Cache/Helpers/LinksResponseCacheKeyBuilder.cs +++ b/src/LinkTracker.Scrapper.Infrastructure/Cache/Helpers/LinksResponseCacheKeyBuilder.cs @@ -5,17 +5,10 @@ namespace LinkTracker.Scrapper.Infrastructure.Cache.Helpers; internal sealed class LinksResponseCacheKeyBuilder(IOptions options) { - private const string LinksCacheHashTag = "linktracker-links"; - private readonly ValkeyOptions _options = options.Value; public string Build(long chatId) { - return $"{BuildPrefix()}{chatId}"; - } - - public string BuildPrefix() - { - return $"{_options.InstanceName}:{{{LinksCacheHashTag}}}:links:chat:"; + return $"{_options.InstanceName}:links:{{chat:{chatId}}}"; } -} \ No newline at end of file +} diff --git a/src/LinkTracker.Scrapper.Infrastructure/Cache/Implementation/ValkeyConnectionProvider.cs b/src/LinkTracker.Scrapper.Infrastructure/Cache/Implementation/ValkeyConnectionProvider.cs index 8eafcaa..187046c 100644 --- a/src/LinkTracker.Scrapper.Infrastructure/Cache/Implementation/ValkeyConnectionProvider.cs +++ b/src/LinkTracker.Scrapper.Infrastructure/Cache/Implementation/ValkeyConnectionProvider.cs @@ -1,5 +1,6 @@ using LinkTracker.Scrapper.Infrastructure.Cache.Abstractions; using LinkTracker.Scrapper.Infrastructure.Configuration.Valkey; +using LinkTracker.Shared.Infrastructure.Valkey; using Microsoft.Extensions.Logging; using Microsoft.Extensions.Options; using StackExchange.Redis; @@ -41,17 +42,7 @@ public async Task GetConnectionAsync(CancellationToken c current?.Dispose(); - var configuration = ConfigurationOptions.Parse(options.Value.ConnectionString); - - configuration.AbortOnConnectFail = false; - configuration.AllowAdmin = true; - configuration.ConnectRetry = 10; - configuration.ConnectTimeout = 15000; - configuration.SyncTimeout = 15000; - configuration.AsyncTimeout = 15000; - configuration.ResolveDns = true; - configuration.KeepAlive = 30; - configuration.ReconnectRetryPolicy = new ExponentialRetry(1000); + var configuration = ValkeyConfiguration.Parse(options.Value.ConnectionString); logger.LogInformation( "Подключение к Valkey. Endpoints={Endpoints}", diff --git a/src/LinkTracker.Scrapper.Infrastructure/Clients/GitHub/GitHubHttpClient.cs b/src/LinkTracker.Scrapper.Infrastructure/Clients/GitHub/GitHubHttpClient.cs index 46627cb..4287c53 100644 --- a/src/LinkTracker.Scrapper.Infrastructure/Clients/GitHub/GitHubHttpClient.cs +++ b/src/LinkTracker.Scrapper.Infrastructure/Clients/GitHub/GitHubHttpClient.cs @@ -1,4 +1,5 @@ using System.Diagnostics; +using System.Globalization; using System.Net.Http.Json; using LinkTracker.Scrapper.Application.Clients.GitHub; using LinkTracker.Scrapper.Application.Clients.GitHub.Contracts; @@ -10,83 +11,51 @@ public sealed class GitHubHttpClient(HttpClient httpClient, ScrapperMetrics metr { private const string Scope = "external_source"; private const string ScopeType = "github.com"; + private const int PageSize = 100; - public async Task GetRepositoryAsync(string owner, string repository, CancellationToken ct = default) + public Task GetRepositoryAsync(string owner, string repository, CancellationToken ct = default) { - var sw = Stopwatch.StartNew(); - - try - { - using var response = await httpClient.GetAsync($"/repos/{owner}/{repository}", ct); - response.EnsureSuccessStatusCode(); - - var body = await response.Content.ReadFromJsonAsync(ct); - return body ?? throw new InvalidOperationException("GitHub returned an empty response body."); - } - catch - { - metrics.Errors.Add( - 1, - new KeyValuePair("scope", Scope), - new KeyValuePair("scope_type", ScopeType), - new KeyValuePair("reason", "exception")); - throw; - } - finally - { - metrics.RequestDuration.Record( - sw.Elapsed.TotalMilliseconds, - new KeyValuePair("scope", Scope), - new KeyValuePair("scope_type", ScopeType)); - } + return SendAsync($"/repos/{owner}/{repository}", ct); } public async Task> GetIssuesAsync( string owner, string repository, + DateTimeOffset? since = null, CancellationToken ct = default) { - var sw = Stopwatch.StartNew(); + var requestUri = + $"/repos/{owner}/{repository}/issues?state=all&sort=updated&direction=desc&per_page={PageSize}"; - try - { - using var response = await httpClient.GetAsync($"/repos/{owner}/{repository}/issues", ct); - response.EnsureSuccessStatusCode(); - - var body = await response.Content.ReadFromJsonAsync>(ct); - return body ?? throw new InvalidOperationException("GitHub returned an empty response body."); - } - catch - { - metrics.Errors.Add( - 1, - new KeyValuePair("scope", Scope), - new KeyValuePair("scope_type", ScopeType), - new KeyValuePair("reason", "exception")); - throw; - } - finally + if (since is not null) { - metrics.RequestDuration.Record( - sw.Elapsed.TotalMilliseconds, - new KeyValuePair("scope", Scope), - new KeyValuePair("scope_type", ScopeType)); + var sinceValue = since.Value.UtcDateTime.ToString("yyyy-MM-ddTHH:mm:ssZ", CultureInfo.InvariantCulture); + requestUri = $"{requestUri}&since={sinceValue}"; } + + return await SendAsync>(requestUri, ct); } public async Task> GetPullRequestsAsync( string owner, string repository, CancellationToken ct = default) + { + return await SendAsync>( + $"/repos/{owner}/{repository}/pulls?state=all&sort=updated&direction=desc&per_page={PageSize}", + ct); + } + + private async Task SendAsync(string requestUri, CancellationToken ct) { var sw = Stopwatch.StartNew(); try { - using var response = await httpClient.GetAsync($"/repos/{owner}/{repository}/pulls", ct); + using var response = await httpClient.GetAsync(requestUri, ct); response.EnsureSuccessStatusCode(); - var body = await response.Content.ReadFromJsonAsync>(ct); + var body = await response.Content.ReadFromJsonAsync(ct); return body ?? throw new InvalidOperationException("GitHub returned an empty response body."); } catch @@ -106,4 +75,4 @@ public async Task> GetPullRequestsAsync new KeyValuePair("scope_type", ScopeType)); } } -} \ No newline at end of file +} diff --git a/src/LinkTracker.Scrapper.Infrastructure/Clients/RateLimiting/ExternalApiRateLimitedException.cs b/src/LinkTracker.Scrapper.Infrastructure/Clients/RateLimiting/ExternalApiRateLimitedException.cs new file mode 100644 index 0000000..0d42928 --- /dev/null +++ b/src/LinkTracker.Scrapper.Infrastructure/Clients/RateLimiting/ExternalApiRateLimitedException.cs @@ -0,0 +1,7 @@ +namespace LinkTracker.Scrapper.Infrastructure.Clients.RateLimiting; + +public sealed class ExternalApiRateLimitedException(string apiName) + : Exception($"Rate limit for external API '{apiName}' is exhausted; the request was not sent.") +{ + public string ApiName { get; } = apiName; +} diff --git a/src/LinkTracker.Scrapper.Infrastructure/Clients/RateLimiting/ExternalApiRateLimiter.cs b/src/LinkTracker.Scrapper.Infrastructure/Clients/RateLimiting/ExternalApiRateLimiter.cs new file mode 100644 index 0000000..2b7f37c --- /dev/null +++ b/src/LinkTracker.Scrapper.Infrastructure/Clients/RateLimiting/ExternalApiRateLimiter.cs @@ -0,0 +1,74 @@ +using System.Threading.RateLimiting; +using LinkTracker.Scrapper.Infrastructure.Configuration.Clients; + +namespace LinkTracker.Scrapper.Infrastructure.Clients.RateLimiting; + +public sealed class ExternalApiRateLimiter : IDisposable +{ + private readonly TokenBucketRateLimiter _limiter; + private readonly TimeProvider _timeProvider; + + private long _cooldownUntilTicks; + + public ExternalApiRateLimiter(string apiName, ExternalApiRateLimitOptions options, TimeProvider timeProvider) + { + ArgumentNullException.ThrowIfNull(options); + + ApiName = apiName; + AcquireTimeout = TimeSpan.FromSeconds(options.AcquireTimeoutSeconds); + + _timeProvider = timeProvider; + _limiter = new TokenBucketRateLimiter(new TokenBucketRateLimiterOptions + { + TokenLimit = options.TokenLimit, + TokensPerPeriod = options.TokensPerPeriod, + ReplenishmentPeriod = TimeSpan.FromSeconds(options.ReplenishmentPeriodSeconds), + QueueLimit = options.QueueLimit, + QueueProcessingOrder = QueueProcessingOrder.OldestFirst, + AutoReplenishment = true + }); + } + + public string ApiName { get; } + + public TimeSpan AcquireTimeout { get; } + + public TimeSpan RemainingCooldown + { + get + { + var until = new DateTimeOffset(Interlocked.Read(ref _cooldownUntilTicks), TimeSpan.Zero); + var remaining = until - _timeProvider.GetUtcNow(); + + return remaining > TimeSpan.Zero ? remaining : TimeSpan.Zero; + } + } + + public ValueTask AcquireAsync(CancellationToken ct) + { + return _limiter.AcquireAsync(1, ct); + } + + public void Cooldown(DateTimeOffset until) + { + var ticks = until.UtcTicks; + + long current; + + do + { + current = Interlocked.Read(ref _cooldownUntilTicks); + + if (ticks <= current) + { + return; + } + } + while (Interlocked.CompareExchange(ref _cooldownUntilTicks, ticks, current) != current); + } + + public void Dispose() + { + _limiter.Dispose(); + } +} diff --git a/src/LinkTracker.Scrapper.Infrastructure/Clients/RateLimiting/ExternalApiRateLimitingHandler.cs b/src/LinkTracker.Scrapper.Infrastructure/Clients/RateLimiting/ExternalApiRateLimitingHandler.cs new file mode 100644 index 0000000..7deac76 --- /dev/null +++ b/src/LinkTracker.Scrapper.Infrastructure/Clients/RateLimiting/ExternalApiRateLimitingHandler.cs @@ -0,0 +1,147 @@ +using System.Globalization; +using System.Net; +using System.Threading.RateLimiting; +using LinkTracker.Scrapper.Infrastructure.Telemetry; +using Microsoft.Extensions.Logging; + +namespace LinkTracker.Scrapper.Infrastructure.Clients.RateLimiting; + +public sealed class ExternalApiRateLimitingHandler( + ExternalApiRateLimiter rateLimiter, + TimeProvider timeProvider, + ScrapperMetrics metrics, + ILogger logger) : DelegatingHandler +{ + private const string RemainingHeader = "X-RateLimit-Remaining"; + private const string ResetHeader = "X-RateLimit-Reset"; + + protected override async Task SendAsync( + HttpRequestMessage request, + CancellationToken cancellationToken) + { + await WaitForCooldownAsync(cancellationToken); + + using var timeout = CancellationTokenSource.CreateLinkedTokenSource(cancellationToken); + timeout.CancelAfter(rateLimiter.AcquireTimeout); + + using var lease = await AcquireAsync(timeout.Token, cancellationToken); + + if (!lease.IsAcquired) + { + throw new ExternalApiRateLimitedException(rateLimiter.ApiName); + } + + var response = await base.SendAsync(request, cancellationToken); + + ApplyThrottlingHints(response); + + return response; + } + + private async Task AcquireAsync( + CancellationToken acquireToken, + CancellationToken requestToken) + { + try + { + return await rateLimiter.AcquireAsync(acquireToken); + } + catch (OperationCanceledException) when (!requestToken.IsCancellationRequested) + { + throw new ExternalApiRateLimitedException(rateLimiter.ApiName); + } + } + + private async Task WaitForCooldownAsync(CancellationToken ct) + { + var remaining = rateLimiter.RemainingCooldown; + + if (remaining <= TimeSpan.Zero) + { + return; + } + + if (remaining > rateLimiter.AcquireTimeout) + { + throw new ExternalApiRateLimitedException(rateLimiter.ApiName); + } + + logger.LogWarning( + "Ожидание сброса лимита внешнего API. Api={Api}, DelaySeconds={DelaySeconds}", + rateLimiter.ApiName, + remaining.TotalSeconds); + + await Task.Delay(remaining, timeProvider, ct); + } + + private void ApplyThrottlingHints(HttpResponseMessage response) + { + var cooldownUntil = TryGetResetAt(response) ?? TryGetRetryAfter(response); + + if (cooldownUntil is null) + { + return; + } + + rateLimiter.Cooldown(cooldownUntil.Value); + + metrics.Errors.Add( + 1, + new KeyValuePair("scope", "external_source"), + new KeyValuePair("scope_type", rateLimiter.ApiName), + new KeyValuePair("reason", "rate_limited")); + + logger.LogWarning( + "Внешний API сообщил об исчерпании лимита. Api={Api}, Status={Status}, CooldownUntil={CooldownUntil}", + rateLimiter.ApiName, + (int)response.StatusCode, + cooldownUntil.Value); + } + + private DateTimeOffset? TryGetResetAt(HttpResponseMessage response) + { + if (!TryGetHeaderValue(response, RemainingHeader, out var remainingValue) || + !long.TryParse(remainingValue, NumberStyles.Integer, CultureInfo.InvariantCulture, out var remaining) || + remaining > 0) + { + return null; + } + + if (!TryGetHeaderValue(response, ResetHeader, out var resetValue) || + !long.TryParse(resetValue, NumberStyles.Integer, CultureInfo.InvariantCulture, out var resetAtUnixSeconds)) + { + return null; + } + + return DateTimeOffset.FromUnixTimeSeconds(resetAtUnixSeconds); + } + + private DateTimeOffset? TryGetRetryAfter(HttpResponseMessage response) + { + if (response.StatusCode is not (HttpStatusCode.TooManyRequests or HttpStatusCode.Forbidden)) + { + return null; + } + + var retryAfter = response.Headers.RetryAfter; + + if (retryAfter?.Delta is { } delta) + { + return timeProvider.GetUtcNow() + delta; + } + + return retryAfter?.Date; + } + + private static bool TryGetHeaderValue(HttpResponseMessage response, string name, out string? value) + { + if (response.Headers.TryGetValues(name, out var values)) + { + value = values.FirstOrDefault(); + return value is not null; + } + + value = null; + return false; + } +} diff --git a/src/LinkTracker.Scrapper.Infrastructure/Clients/Registration/ClientsModule.cs b/src/LinkTracker.Scrapper.Infrastructure/Clients/Registration/ClientsModule.cs index b156823..f253613 100644 --- a/src/LinkTracker.Scrapper.Infrastructure/Clients/Registration/ClientsModule.cs +++ b/src/LinkTracker.Scrapper.Infrastructure/Clients/Registration/ClientsModule.cs @@ -10,23 +10,30 @@ using LinkTracker.Scrapper.Application.Clients.StackOverflow; using LinkTracker.Scrapper.Infrastructure.Clients.Bot; using LinkTracker.Scrapper.Infrastructure.Clients.GitHub; +using LinkTracker.Scrapper.Infrastructure.Clients.RateLimiting; using LinkTracker.Scrapper.Infrastructure.Clients.StackOverflow; using LinkTracker.Scrapper.Infrastructure.Configuration.Bot; using LinkTracker.Scrapper.Infrastructure.Configuration.Clients; using LinkTracker.Scrapper.Infrastructure.Configuration.Kafka; using LinkTracker.Scrapper.Infrastructure.Kafka.Abstractions; using LinkTracker.Scrapper.Infrastructure.Kafka.Serialization; +using LinkTracker.Scrapper.Infrastructure.Telemetry; using LinkTracker.Shared.Infrastructure; using LinkTracker.Shared.Infrastructure.Authentication; using LinkTracker.Shared.Infrastructure.Resilience; using Microsoft.Extensions.Configuration; using Microsoft.Extensions.DependencyInjection; +using Microsoft.Extensions.DependencyInjection.Extensions; +using Microsoft.Extensions.Logging; using Microsoft.Extensions.Options; namespace LinkTracker.Scrapper.Infrastructure.Clients.Registration; public static class ClientsModule { + private const string GitHubApiName = "github.com"; + private const string StackOverflowApiName = "stackoverflow.com"; + public static IServiceCollection AddClients(this IServiceCollection services, IConfiguration configuration) { services @@ -44,14 +51,29 @@ public static IServiceCollection AddClients(this IServiceCollection services, IC services .AddOptions() - .Bind(configuration.GetSection("GitHub")); + .Bind(configuration.GetSection("GitHub")) + .Validate(o => !string.IsNullOrWhiteSpace(o.BaseUrl), "GitHub:BaseUrl must be set") + .ValidateOnStart(); services .AddOptions() - .Bind(configuration.GetSection("StackOverflow")); + .Bind(configuration.GetSection("StackOverflow")) + .Validate(o => !string.IsNullOrWhiteSpace(o.BaseUrl), "StackOverflow:BaseUrl must be set") + .ValidateOnStart(); var httpResilienceOptions = configuration.GetHttpResilienceOptions(); + services.TryAddSingleton(TimeProvider.System); + + var gitHubRateLimit = configuration.GetSection("GitHub").Get()?.RateLimit + ?? new GitHubOptions().RateLimit; + + var stackOverflowRateLimit = configuration.GetSection("StackOverflow").Get()?.RateLimit + ?? new StackOverflowOptions().RateLimit; + + services.AddExternalApiRateLimiter(GitHubApiName, gitHubRateLimit); + services.AddExternalApiRateLimiter(StackOverflowApiName, stackOverflowRateLimit); + services.AddHttpClient((sp, client) => { var options = sp.GetRequiredService>().Value; @@ -128,7 +150,8 @@ public static IServiceCollection AddClients(this IServiceCollection services, IC client.DefaultRequestHeaders.Authorization = new AuthenticationHeaderValue("Bearer", options.Token); } }) - .AddConfiguredHttpResilience("github", httpResilienceOptions); + .AddConfiguredHttpResilience("github", httpResilienceOptions) + .AddExternalApiRateLimiting(GitHubApiName, gitHubRateLimit); services.AddHttpClient((sp, client) => { @@ -136,8 +159,43 @@ public static IServiceCollection AddClients(this IServiceCollection services, IC client.BaseAddress = new Uri(options.BaseUrl); client.DefaultRequestHeaders.UserAgent.Add(new ProductInfoHeaderValue("LinkTracker", "1.0")); }) - .AddConfiguredHttpResilience("stackoverflow", httpResilienceOptions); + .AddConfiguredHttpResilience("stackoverflow", httpResilienceOptions) + .AddExternalApiRateLimiting(StackOverflowApiName, stackOverflowRateLimit); + + return services; + } + + private static IServiceCollection AddExternalApiRateLimiter( + this IServiceCollection services, + string apiName, + ExternalApiRateLimitOptions options) + { + if (!options.Enabled) + { + return services; + } + + services.AddKeyedSingleton( + apiName, + (sp, _) => new ExternalApiRateLimiter(apiName, options, sp.GetRequiredService())); return services; } + + private static IHttpClientBuilder AddExternalApiRateLimiting( + this IHttpClientBuilder builder, + string apiName, + ExternalApiRateLimitOptions options) + { + if (!options.Enabled) + { + return builder; + } + + return builder.AddHttpMessageHandler(sp => new ExternalApiRateLimitingHandler( + sp.GetRequiredKeyedService(apiName), + sp.GetRequiredService(), + sp.GetRequiredService(), + sp.GetRequiredService>())); + } } \ No newline at end of file diff --git a/src/LinkTracker.Scrapper.Infrastructure/Clients/StackOverflow/StackOverflowHttpClient.cs b/src/LinkTracker.Scrapper.Infrastructure/Clients/StackOverflow/StackOverflowHttpClient.cs index 15803fc..23a1c54 100644 --- a/src/LinkTracker.Scrapper.Infrastructure/Clients/StackOverflow/StackOverflowHttpClient.cs +++ b/src/LinkTracker.Scrapper.Infrastructure/Clients/StackOverflow/StackOverflowHttpClient.cs @@ -10,6 +10,7 @@ public sealed class StackOverflowHttpClient(HttpClient httpClient, ScrapperMetri { private const string Scope = "external_source"; private const string ScopeType = "stackoverflow.com"; + private const int PageSize = 100; public async Task GetQuestionAsync(long questionId, CancellationToken ct = default) { @@ -48,7 +49,7 @@ public async Task> GetAnswersAsync(lo try { using var response = await httpClient.GetAsync( - $"/2.3/questions/{questionId}/answers?site=stackoverflow&sort=creation&order=desc&filter=withbody", + $"/2.3/questions/{questionId}/answers?site=stackoverflow&sort=creation&order=desc&pagesize={PageSize}&filter=withbody", ct); response.EnsureSuccessStatusCode(); @@ -80,7 +81,7 @@ public async Task> GetCommentsAsync( try { using var response = await httpClient.GetAsync( - $"/2.3/questions/{questionId}/comments?site=stackoverflow&sort=creation&order=desc&filter=withbody", + $"/2.3/questions/{questionId}/comments?site=stackoverflow&sort=creation&order=desc&pagesize={PageSize}&filter=withbody", ct); response.EnsureSuccessStatusCode(); diff --git a/src/LinkTracker.Scrapper.Infrastructure/Configuration/Clients/ExternalApiRateLimitOptions.cs b/src/LinkTracker.Scrapper.Infrastructure/Configuration/Clients/ExternalApiRateLimitOptions.cs new file mode 100644 index 0000000..0b0c613 --- /dev/null +++ b/src/LinkTracker.Scrapper.Infrastructure/Configuration/Clients/ExternalApiRateLimitOptions.cs @@ -0,0 +1,16 @@ +namespace LinkTracker.Scrapper.Infrastructure.Configuration.Clients; + +public sealed class ExternalApiRateLimitOptions +{ + public bool Enabled { get; init; } = true; + + public int TokenLimit { get; init; } = 100; + + public int TokensPerPeriod { get; init; } = 80; + + public int ReplenishmentPeriodSeconds { get; init; } = 60; + + public int QueueLimit { get; init; } = 1_000; + + public int AcquireTimeoutSeconds { get; init; } = 60; +} diff --git a/src/LinkTracker.Scrapper.Infrastructure/Configuration/Clients/GitHubOptions.cs b/src/LinkTracker.Scrapper.Infrastructure/Configuration/Clients/GitHubOptions.cs index 1ac87d0..029f926 100644 --- a/src/LinkTracker.Scrapper.Infrastructure/Configuration/Clients/GitHubOptions.cs +++ b/src/LinkTracker.Scrapper.Infrastructure/Configuration/Clients/GitHubOptions.cs @@ -5,4 +5,11 @@ public sealed class GitHubOptions public string BaseUrl { get; init; } = "https://api.github.com"; public string? Token { get; init; } -} \ No newline at end of file + + public ExternalApiRateLimitOptions RateLimit { get; init; } = new() + { + TokenLimit = 200, + TokensPerPeriod = 80, + ReplenishmentPeriodSeconds = 60 + }; +} diff --git a/src/LinkTracker.Scrapper.Infrastructure/Configuration/Clients/StackOverflowOptions.cs b/src/LinkTracker.Scrapper.Infrastructure/Configuration/Clients/StackOverflowOptions.cs index 52c99f6..9b3f012 100644 --- a/src/LinkTracker.Scrapper.Infrastructure/Configuration/Clients/StackOverflowOptions.cs +++ b/src/LinkTracker.Scrapper.Infrastructure/Configuration/Clients/StackOverflowOptions.cs @@ -3,4 +3,11 @@ namespace LinkTracker.Scrapper.Infrastructure.Configuration.Clients; public sealed class StackOverflowOptions { public string BaseUrl { get; init; } = "https://api.stackexchange.com"; -} \ No newline at end of file + + public ExternalApiRateLimitOptions RateLimit { get; init; } = new() + { + TokenLimit = 30, + TokensPerPeriod = 12, + ReplenishmentPeriodSeconds = 3600 + }; +} diff --git a/src/LinkTracker.Scrapper.Infrastructure/Configuration/Database/DatabaseOptions.cs b/src/LinkTracker.Scrapper.Infrastructure/Configuration/Database/DatabaseOptions.cs index 7ccd613..01ff690 100644 --- a/src/LinkTracker.Scrapper.Infrastructure/Configuration/Database/DatabaseOptions.cs +++ b/src/LinkTracker.Scrapper.Infrastructure/Configuration/Database/DatabaseOptions.cs @@ -13,7 +13,6 @@ public sealed class DatabaseOptions public string MigrationsPath { get; init; } = "migrations"; public bool RunMigrations { get; init; } = true; public SslMode SslMode { get; init; } = SslMode.Disable; - public bool TrustServerCertificate { get; init; } public string BuildConnectionString() { @@ -24,8 +23,7 @@ public string BuildConnectionString() Database = Name, Username = User, Password = Password, - SslMode = SslMode, - TrustServerCertificate = TrustServerCertificate + SslMode = SslMode }; return builder.ConnectionString; diff --git a/src/LinkTracker.Scrapper.Infrastructure/LinkTracker.Scrapper.Infrastructure.csproj b/src/LinkTracker.Scrapper.Infrastructure/LinkTracker.Scrapper.Infrastructure.csproj index d70107f..3709ccb 100644 --- a/src/LinkTracker.Scrapper.Infrastructure/LinkTracker.Scrapper.Infrastructure.csproj +++ b/src/LinkTracker.Scrapper.Infrastructure/LinkTracker.Scrapper.Infrastructure.csproj @@ -1,4 +1,4 @@ - + @@ -30,10 +30,4 @@ - - net9.0 - enable - enable - - diff --git a/src/LinkTracker.Scrapper.Infrastructure/Outbox/Jobs/OutboxDispatchJob.cs b/src/LinkTracker.Scrapper.Infrastructure/Outbox/Jobs/OutboxDispatchJob.cs index 654ea4a..6089b28 100644 --- a/src/LinkTracker.Scrapper.Infrastructure/Outbox/Jobs/OutboxDispatchJob.cs +++ b/src/LinkTracker.Scrapper.Infrastructure/Outbox/Jobs/OutboxDispatchJob.cs @@ -1,6 +1,7 @@ using LinkTracker.Scrapper.Infrastructure.Clients.Bot; using LinkTracker.Scrapper.Infrastructure.Outbox.Abstractions; using LinkTracker.Scrapper.Infrastructure.Outbox.Configuration; +using LinkTracker.Scrapper.Infrastructure.Telemetry; using Microsoft.Extensions.Logging; using Microsoft.Extensions.Options; using Quartz; @@ -12,6 +13,7 @@ internal sealed class OutboxDispatchJob( IOutboxStore outboxStore, IBotDirectClient botClient, IOptions outboxOptions, + ScrapperMetrics metrics, ILogger logger) : IJob { public async Task Execute(IJobExecutionContext context) @@ -39,6 +41,8 @@ public async Task Execute(IJobExecutionContext context) await botClient.SendUpdateAsync(message.Payload, ct); await outboxStore.MarkProcessedAsync(message.Id, ct); + metrics.SentUpdates.Add(1); + logger.LogDebug( "Outbox сообщение отправлено. OutboxMessageId={OutboxMessageId}", message.Id); diff --git a/src/LinkTracker.Scrapper.Infrastructure/Quartz/Configuration/LinkUpdatesSchedulingOptions.cs b/src/LinkTracker.Scrapper.Infrastructure/Quartz/Configuration/LinkUpdatesSchedulingOptions.cs index 063bba9..a42505f 100644 --- a/src/LinkTracker.Scrapper.Infrastructure/Quartz/Configuration/LinkUpdatesSchedulingOptions.cs +++ b/src/LinkTracker.Scrapper.Infrastructure/Quartz/Configuration/LinkUpdatesSchedulingOptions.cs @@ -2,7 +2,7 @@ namespace LinkTracker.Scrapper.Infrastructure.Quartz.Configuration; public sealed class LinkUpdatesSchedulingOptions { - public int IntervalSeconds { get; init; } = 30; + public int IntervalSeconds { get; init; } = 300; public int BatchSize { get; init; } = 100; - public int MaxDegreeOfParallelism { get; init; } = 1; -} \ No newline at end of file + public int MaxDegreeOfParallelism { get; init; } = 4; +} diff --git a/src/LinkTracker.Scrapper.Infrastructure/Quartz/Jobs/LinkUpdatesJob.cs b/src/LinkTracker.Scrapper.Infrastructure/Quartz/Jobs/LinkUpdatesJob.cs index 5d2ff06..ac2a532 100644 --- a/src/LinkTracker.Scrapper.Infrastructure/Quartz/Jobs/LinkUpdatesJob.cs +++ b/src/LinkTracker.Scrapper.Infrastructure/Quartz/Jobs/LinkUpdatesJob.cs @@ -23,6 +23,7 @@ internal sealed class LinkUpdatesJob( IOutboxStore outboxStore, IOptions schedulingOptions, IOptions outboxOptions, + TimeProvider timeProvider, ILogger logger, ScrapperMetrics metrics) : IJob { @@ -33,12 +34,12 @@ internal sealed class LinkUpdatesJob( public async Task Execute(IJobExecutionContext context) { var ct = context.CancellationToken; - var afterLinkId = (long?)null; + var runStartedAt = timeProvider.GetUtcNow(); var linkCountBySource = new Dictionary(StringComparer.OrdinalIgnoreCase); while (!ct.IsCancellationRequested) { - var batch = await trackingStore.GetSubscriptionsBatchAsync(afterLinkId, _batchSize, ct); + var batch = await trackingStore.GetSubscriptionsDueForCheckAsync(runStartedAt, _batchSize, ct); if (batch.Count == 0) { @@ -52,8 +53,8 @@ public async Task Execute(IJobExecutionContext context) } logger.LogDebug( - "Начата обработка батча ссылок. AfterLinkId={AfterLinkId}, BatchSize={BatchSize}, ActualCount={ActualCount}, MaxDegreeOfParallelism={MaxDegreeOfParallelism}", - afterLinkId, + "Начата обработка батча ссылок. CheckedBefore={CheckedBefore}, BatchSize={BatchSize}, ActualCount={ActualCount}, MaxDegreeOfParallelism={MaxDegreeOfParallelism}", + runStartedAt, _batchSize, batch.Count, _maxDegreeOfParallelism); @@ -81,6 +82,11 @@ await Parallel.ForEachAsync( } }); + await trackingStore.MarkCheckedAsync( + batch.Select(x => x.Id).ToArray(), + timeProvider.GetUtcNow(), + ct); + if (!failedSubscriptions.IsEmpty) { var failed = failedSubscriptions @@ -97,8 +103,6 @@ await Parallel.ForEachAsync( await SendFailedReportsAsync(failed, ct); } - - afterLinkId = batch[^1].Id; } foreach (var (source, count) in linkCountBySource) @@ -259,7 +263,7 @@ await outboxStore.AddRangeAndSetCursorAsync( updates, ct); - metrics.SentUpdates.Add(updates.Count); + metrics.OutboxEnqueuedUpdates.Add(updates.Count); logger.LogDebug( "Обновления сохранены в transactional outbox. LinkId={LinkId}, Url={Url}, UpdateCount={UpdateCount}, ChatCount={ChatCount}", diff --git a/src/LinkTracker.Scrapper.Infrastructure/Quartz/Registration/QuartzModule.cs b/src/LinkTracker.Scrapper.Infrastructure/Quartz/Registration/QuartzModule.cs index c1d5e15..8d672b8 100644 --- a/src/LinkTracker.Scrapper.Infrastructure/Quartz/Registration/QuartzModule.cs +++ b/src/LinkTracker.Scrapper.Infrastructure/Quartz/Registration/QuartzModule.cs @@ -4,6 +4,7 @@ using LinkTracker.Scrapper.Infrastructure.Quartz.Jobs; using Microsoft.Extensions.Configuration; using Microsoft.Extensions.DependencyInjection; +using Microsoft.Extensions.DependencyInjection.Extensions; using Quartz; namespace LinkTracker.Scrapper.Infrastructure.Quartz.Registration; @@ -27,6 +28,8 @@ public static IServiceCollection AddQuartzScheduling( "Scheduling:MaxDegreeOfParallelism must be greater than zero.") .ValidateOnStart(); + services.TryAddSingleton(TimeProvider.System); + var schedulingOptions = schedulingSection.Get() ?? new LinkUpdatesSchedulingOptions(); var outboxOptions = outboxSection.Get() ?? new OutboxOptions(); diff --git a/src/LinkTracker.Scrapper.Infrastructure/Telemetry/ScrapperMetrics.cs b/src/LinkTracker.Scrapper.Infrastructure/Telemetry/ScrapperMetrics.cs index 970669f..26abb47 100644 --- a/src/LinkTracker.Scrapper.Infrastructure/Telemetry/ScrapperMetrics.cs +++ b/src/LinkTracker.Scrapper.Infrastructure/Telemetry/ScrapperMetrics.cs @@ -33,7 +33,11 @@ public ScrapperMetrics() SentUpdates = _meter.CreateCounter( "sent_updates_total", - description: "Количество обновлений, отправленных в Bot Service"); + description: "Количество обновлений, фактически отправленных в Bot Service"); + + OutboxEnqueuedUpdates = _meter.CreateCounter( + "outbox_enqueued_updates_total", + description: "Количество обновлений, записанных в transactional outbox"); RequestDuration = _meter.CreateHistogram( "request_duration_ms_total", @@ -90,6 +94,8 @@ public ScrapperMetrics() public Counter SentUpdates { get; } + public Counter OutboxEnqueuedUpdates { get; } + public Histogram RequestDuration { get; } public Counter Errors { get; } diff --git a/src/LinkTracker.Scrapper.Presentation/LinkTracker.Scrapper.Presentation.csproj b/src/LinkTracker.Scrapper.Presentation/LinkTracker.Scrapper.Presentation.csproj index ee6c20f..0658774 100644 --- a/src/LinkTracker.Scrapper.Presentation/LinkTracker.Scrapper.Presentation.csproj +++ b/src/LinkTracker.Scrapper.Presentation/LinkTracker.Scrapper.Presentation.csproj @@ -1,4 +1,4 @@ - + @@ -10,10 +10,4 @@ - - net9.0 - enable - enable - - diff --git a/src/LinkTracker.Scrapper.Storage.Abstractions/LinkTracker.Scrapper.Storage.Abstractions.csproj b/src/LinkTracker.Scrapper.Storage.Abstractions/LinkTracker.Scrapper.Storage.Abstractions.csproj index 5061485..9dd6b7b 100644 --- a/src/LinkTracker.Scrapper.Storage.Abstractions/LinkTracker.Scrapper.Storage.Abstractions.csproj +++ b/src/LinkTracker.Scrapper.Storage.Abstractions/LinkTracker.Scrapper.Storage.Abstractions.csproj @@ -1,10 +1,4 @@ - - - - net9.0 - enable - enable - + diff --git a/src/LinkTracker.Scrapper.Storage.Abstractions/Models/ILinkTrackingStore.cs b/src/LinkTracker.Scrapper.Storage.Abstractions/Models/ILinkTrackingStore.cs index ed9a28b..4c07607 100644 --- a/src/LinkTracker.Scrapper.Storage.Abstractions/Models/ILinkTrackingStore.cs +++ b/src/LinkTracker.Scrapper.Storage.Abstractions/Models/ILinkTrackingStore.cs @@ -28,13 +28,16 @@ public interface ILinkTrackingStore Task TryDeleteTagAsync(long chatId, string tag, CancellationToken ct = default); - Task> GetAllSubscriptionsAsync(CancellationToken ct = default); - - Task> GetSubscriptionsBatchAsync( - long? afterLinkId, + Task> GetSubscriptionsDueForCheckAsync( + DateTimeOffset checkedBefore, int batchSize, CancellationToken ct = default); + Task MarkCheckedAsync( + IReadOnlyCollection linkIds, + DateTimeOffset checkedAt, + CancellationToken ct = default); + Task SetCursorAsync( long linkId, DateTimeOffset lastUpdatedAt, diff --git a/src/LinkTracker.Scrapper.Storage.Orm/Configurations/LinkEntityConfiguration.cs b/src/LinkTracker.Scrapper.Storage.Orm/Configurations/LinkEntityConfiguration.cs index c5cb9aa..14bac96 100644 --- a/src/LinkTracker.Scrapper.Storage.Orm/Configurations/LinkEntityConfiguration.cs +++ b/src/LinkTracker.Scrapper.Storage.Orm/Configurations/LinkEntityConfiguration.cs @@ -29,7 +29,13 @@ public void Configure(EntityTypeBuilder builder) builder.Property(x => x.LastEventKey) .HasColumnName("last_event_key"); + builder.Property(x => x.LastCheckedAt) + .HasColumnName("last_checked_at"); + builder.HasIndex(x => x.NormalizedUrl) .IsUnique(); + + builder.HasIndex(x => new { x.LastCheckedAt, x.Id }) + .HasDatabaseName("ix_links_last_checked_at"); } } \ No newline at end of file diff --git a/src/LinkTracker.Scrapper.Storage.Orm/Entities/LinkEntity.cs b/src/LinkTracker.Scrapper.Storage.Orm/Entities/LinkEntity.cs index cdff6ad..6419b3e 100644 --- a/src/LinkTracker.Scrapper.Storage.Orm/Entities/LinkEntity.cs +++ b/src/LinkTracker.Scrapper.Storage.Orm/Entities/LinkEntity.cs @@ -10,6 +10,7 @@ public sealed class LinkEntity public DateTimeOffset? LastUpdatedAt { get; set; } public string? LastEventKey { get; set; } + public DateTimeOffset LastCheckedAt { get; set; } = DateTimeOffset.MinValue; public ICollection Subscriptions { get; set; } = []; } \ No newline at end of file diff --git a/src/LinkTracker.Scrapper.Storage.Orm/LinkTracker.Scrapper.Storage.Orm.csproj b/src/LinkTracker.Scrapper.Storage.Orm/LinkTracker.Scrapper.Storage.Orm.csproj index bdccdf5..54838ac 100644 --- a/src/LinkTracker.Scrapper.Storage.Orm/LinkTracker.Scrapper.Storage.Orm.csproj +++ b/src/LinkTracker.Scrapper.Storage.Orm/LinkTracker.Scrapper.Storage.Orm.csproj @@ -1,10 +1,4 @@ - - - - net9.0 - enable - enable - + diff --git a/src/LinkTracker.Scrapper.Storage.Orm/OrmLinkTrackingStore.cs b/src/LinkTracker.Scrapper.Storage.Orm/OrmLinkTrackingStore.cs index da39241..294c34f 100644 --- a/src/LinkTracker.Scrapper.Storage.Orm/OrmLinkTrackingStore.cs +++ b/src/LinkTracker.Scrapper.Storage.Orm/OrmLinkTrackingStore.cs @@ -398,26 +398,8 @@ public async Task TryDeleteTagAsync(long chatId, string tag, CancellationT } } - public async Task> GetAllSubscriptionsAsync(CancellationToken ct = default) - { - await using var dbContext = await dbContextFactory.CreateDbContextAsync(ct); - - return await dbContext.Subscriptions - .AsNoTracking() - .GroupBy(x => new { x.LinkId, x.Link.Url, x.Link.LastUpdatedAt }) - .Select(group => new TrackedLinkSubscription - { - Id = group.Key.LinkId, - Url = new Uri(group.Key.Url), - LastUpdatedAt = group.Key.LastUpdatedAt, - LastEventKey = group.First().Link.LastEventKey, - TgChatIds = group.Select(x => x.ChatId).Distinct().ToArray() - }) - .ToArrayAsync(ct); - } - - public async Task> GetSubscriptionsBatchAsync( - long? afterLinkId, + public async Task> GetSubscriptionsDueForCheckAsync( + DateTimeOffset checkedBefore, int batchSize, CancellationToken ct = default) { @@ -426,8 +408,9 @@ public async Task> GetSubscriptionsBatchA return await dbContext.Links .AsNoTracking() .Where(x => x.Subscriptions.Any()) - .Where(x => afterLinkId == null || x.Id > afterLinkId.Value) - .OrderBy(x => x.Id) + .Where(x => x.LastCheckedAt < checkedBefore) + .OrderBy(x => x.LastCheckedAt) + .ThenBy(x => x.Id) .Take(batchSize) .Select(x => new TrackedLinkSubscription { @@ -443,6 +426,25 @@ public async Task> GetSubscriptionsBatchA .ToArrayAsync(ct); } + public async Task MarkCheckedAsync( + IReadOnlyCollection linkIds, + DateTimeOffset checkedAt, + CancellationToken ct = default) + { + ArgumentNullException.ThrowIfNull(linkIds); + + if (linkIds.Count == 0) + { + return; + } + + await using var dbContext = await dbContextFactory.CreateDbContextAsync(ct); + + await dbContext.Links + .Where(x => linkIds.Contains(x.Id)) + .ExecuteUpdateAsync(setters => setters.SetProperty(x => x.LastCheckedAt, checkedAt), ct); + } + public async Task SetCursorAsync(long linkId, DateTimeOffset lastUpdatedAt, string? lastEventKey, CancellationToken ct = default) { await using var dbContext = await dbContextFactory.CreateDbContextAsync(ct); diff --git a/src/LinkTracker.Scrapper.Storage.Sql/LinkTracker.Scrapper.Storage.Sql.csproj b/src/LinkTracker.Scrapper.Storage.Sql/LinkTracker.Scrapper.Storage.Sql.csproj index b421dea..adb7818 100644 --- a/src/LinkTracker.Scrapper.Storage.Sql/LinkTracker.Scrapper.Storage.Sql.csproj +++ b/src/LinkTracker.Scrapper.Storage.Sql/LinkTracker.Scrapper.Storage.Sql.csproj @@ -1,10 +1,4 @@ - - - - net9.0 - enable - enable - + diff --git a/src/LinkTracker.Scrapper.Storage.Sql/SqlLinkTrackingStore.cs b/src/LinkTracker.Scrapper.Storage.Sql/SqlLinkTrackingStore.cs index 9aa8caf..2bdf4f0 100644 --- a/src/LinkTracker.Scrapper.Storage.Sql/SqlLinkTrackingStore.cs +++ b/src/LinkTracker.Scrapper.Storage.Sql/SqlLinkTrackingStore.cs @@ -381,32 +381,41 @@ await SqlTagsHelper.DeleteOrphanTagByNameAsync( } } - public async Task> GetAllSubscriptionsAsync(CancellationToken ct = default) + public async Task> GetSubscriptionsDueForCheckAsync( + DateTimeOffset checkedBefore, + int batchSize, + CancellationToken ct = default) { await using var connection = await dataSource.OpenConnectionAsync(ct); var rows = await connection.QueryAsync( new CommandDefinition( - SqlLinkTrackingStoreCommands.GetSubscriptionRows, + SqlLinkTrackingStoreCommands.GetSubscriptionsDueForCheckRows, + new { checkedBefore, batchSize }, cancellationToken: ct)); return MapSubscriptions(rows); } - public async Task> GetSubscriptionsBatchAsync( - long? afterLinkId, - int batchSize, + public async Task MarkCheckedAsync( + IReadOnlyCollection linkIds, + DateTimeOffset checkedAt, CancellationToken ct = default) { + ArgumentNullException.ThrowIfNull(linkIds); + + if (linkIds.Count == 0) + { + return; + } + await using var connection = await dataSource.OpenConnectionAsync(ct); - var rows = await connection.QueryAsync( + await connection.ExecuteAsync( new CommandDefinition( - SqlLinkTrackingStoreCommands.GetSubscriptionBatchRows, - new { afterLinkId, batchSize }, + SqlLinkTrackingStoreCommands.MarkChecked, + new { linkIds = linkIds.ToArray(), checkedAt }, cancellationToken: ct)); - - return MapSubscriptions(rows); } public async Task SetCursorAsync(long linkId, DateTimeOffset lastUpdatedAt, string? lastEventKey, CancellationToken ct = default) diff --git a/src/LinkTracker.Scrapper.Storage.Sql/SqlLinkTrackingStoreCommands.cs b/src/LinkTracker.Scrapper.Storage.Sql/SqlLinkTrackingStoreCommands.cs index decaaf9..1c0d454 100644 --- a/src/LinkTracker.Scrapper.Storage.Sql/SqlLinkTrackingStoreCommands.cs +++ b/src/LinkTracker.Scrapper.Storage.Sql/SqlLinkTrackingStoreCommands.cs @@ -48,19 +48,6 @@ GROUP BY ORDER BY l.id; """; - public const string GetSubscriptionRows = - """ - SELECT - l.id AS Id, - l.url AS Url, - l.last_updated_at AS LastUpdatedAt, - l.last_event_key AS LastEventKey, - s.chat_id AS ChatId - FROM subscriptions s - JOIN links l ON l.id = s.link_id - ORDER BY l.id, s.chat_id; - """; - public const string SetCursor = """ UPDATE links @@ -252,22 +239,23 @@ FROM subscription_tags st ); """; - public const string GetSubscriptionBatchRows = + public const string GetSubscriptionsDueForCheckRows = """ WITH target_links AS ( SELECT l.id AS Id, l.url AS Url, l.last_updated_at AS LastUpdatedAt, - l.last_event_key AS LastEventKey + l.last_event_key AS LastEventKey, + l.last_checked_at AS LastCheckedAt FROM links l - WHERE (@afterLinkId IS NULL OR l.id > @afterLinkId) + WHERE l.last_checked_at < @checkedBefore AND EXISTS ( SELECT 1 FROM subscriptions s WHERE s.link_id = l.id ) - ORDER BY l.id + ORDER BY l.last_checked_at, l.id LIMIT @batchSize ) SELECT @@ -278,6 +266,13 @@ LIMIT @batchSize s.chat_id AS ChatId FROM target_links tl JOIN subscriptions s ON s.link_id = tl.Id - ORDER BY tl.Id, s.chat_id; + ORDER BY tl.LastCheckedAt, tl.Id, s.chat_id; + """; + + public const string MarkChecked = + """ + UPDATE links + SET last_checked_at = @checkedAt + WHERE id = ANY(@linkIds); """; } \ No newline at end of file diff --git a/src/LinkTracker.Shared/Infrastructure/Valkey/ValkeyConfiguration.cs b/src/LinkTracker.Shared/Infrastructure/Valkey/ValkeyConfiguration.cs new file mode 100644 index 0000000..eb417a6 --- /dev/null +++ b/src/LinkTracker.Shared/Infrastructure/Valkey/ValkeyConfiguration.cs @@ -0,0 +1,17 @@ +using StackExchange.Redis; + +namespace LinkTracker.Shared.Infrastructure.Valkey; + +public static class ValkeyConfiguration +{ + private static readonly ValkeyDefaultOptionsProvider Defaults = new(); + + public static ConfigurationOptions Parse(string connectionString) + { + var configuration = ConfigurationOptions.Parse(connectionString); + + configuration.Defaults = Defaults; + + return configuration; + } +} diff --git a/src/LinkTracker.Shared/Infrastructure/Valkey/ValkeyDefaultOptionsProvider.cs b/src/LinkTracker.Shared/Infrastructure/Valkey/ValkeyDefaultOptionsProvider.cs new file mode 100644 index 0000000..1ce3602 --- /dev/null +++ b/src/LinkTracker.Shared/Infrastructure/Valkey/ValkeyDefaultOptionsProvider.cs @@ -0,0 +1,21 @@ +using StackExchange.Redis; +using StackExchange.Redis.Configuration; + +namespace LinkTracker.Shared.Infrastructure.Valkey; + +public sealed class ValkeyDefaultOptionsProvider : DefaultOptionsProvider +{ + public override bool AbortOnConnectFail => false; + + public override int ConnectRetry => 10; + + public override TimeSpan? ConnectTimeout => TimeSpan.FromSeconds(15); + + public override TimeSpan SyncTimeout => TimeSpan.FromSeconds(15); + + public override TimeSpan KeepAliveInterval => TimeSpan.FromSeconds(30); + + public override bool ResolveDns => true; + + public override IReconnectRetryPolicy ReconnectRetryPolicy => new ExponentialRetry(1000); +} diff --git a/src/LinkTracker.Shared/LinkTracker.Shared.csproj b/src/LinkTracker.Shared/LinkTracker.Shared.csproj index 7405786..99f7016 100644 --- a/src/LinkTracker.Shared/LinkTracker.Shared.csproj +++ b/src/LinkTracker.Shared/LinkTracker.Shared.csproj @@ -1,10 +1,4 @@ - - - - net9.0 - enable - enable - + @@ -23,6 +17,7 @@ + diff --git a/src/LinkTracker.Tests/LinkTracker.Tests.csproj b/src/LinkTracker.Tests/LinkTracker.Tests.csproj index 5135d0d..196a3bb 100644 --- a/src/LinkTracker.Tests/LinkTracker.Tests.csproj +++ b/src/LinkTracker.Tests/LinkTracker.Tests.csproj @@ -1,10 +1,6 @@ - net9.0 - enable - enable - false true diff --git a/src/LinkTracker.Tests/Scrapper/Integration/Cache/ScrapperLinksCacheApiTests.cs b/src/LinkTracker.Tests/Scrapper/Integration/Cache/ScrapperLinksCacheApiTests.cs index b5a7540..28dda36 100644 --- a/src/LinkTracker.Tests/Scrapper/Integration/Cache/ScrapperLinksCacheApiTests.cs +++ b/src/LinkTracker.Tests/Scrapper/Integration/Cache/ScrapperLinksCacheApiTests.cs @@ -318,7 +318,7 @@ private static TrackedLinkRecord CreateRecord( private static string BuildLinksCacheKey(long chatId) { - return $"{InstanceName}:{{linktracker-links}}:links:chat:{chatId}"; + return $"{InstanceName}:links:{{chat:{chatId}}}"; } private static async Task WaitUntilAsync(Func> condition) diff --git a/src/LinkTracker.Tests/Scrapper/Integration/Http/GitHubHttpClientTests.cs b/src/LinkTracker.Tests/Scrapper/Integration/Http/GitHubHttpClientTests.cs index 9418103..c1de470 100644 --- a/src/LinkTracker.Tests/Scrapper/Integration/Http/GitHubHttpClientTests.cs +++ b/src/LinkTracker.Tests/Scrapper/Integration/Http/GitHubHttpClientTests.cs @@ -134,7 +134,7 @@ public async Task GetIssues_WhenResponseIsValid_ReturnsDeserializedBody() await wireMock.StubAsync(new { - request = new { method = "GET", url = "/repos/user/repo/issues" }, + request = new { method = "GET", urlPath = "/repos/user/repo/issues" }, response = new { status = 200, @@ -174,7 +174,7 @@ public async Task GetIssues_WhenStatusIsNotSuccess_ThrowsHttpRequestException() await wireMock.StubAsync(new { - request = new { method = "GET", url = "/repos/user/repo/issues" }, + request = new { method = "GET", urlPath = "/repos/user/repo/issues" }, response = new { status = (int)HttpStatusCode.BadGateway, @@ -204,7 +204,7 @@ public async Task GetPullRequests_WhenResponseIsValid_ReturnsDeserializedBody() await wireMock.StubAsync(new { - request = new { method = "GET", url = "/repos/user/repo/pulls" }, + request = new { method = "GET", urlPath = "/repos/user/repo/pulls" }, response = new { status = 200, @@ -244,7 +244,7 @@ public async Task GetPullRequests_WhenStatusIsNotSuccess_ThrowsHttpRequestExcept await wireMock.StubAsync(new { - request = new { method = "GET", url = "/repos/user/repo/pulls" }, + request = new { method = "GET", urlPath = "/repos/user/repo/pulls" }, response = new { status = (int)HttpStatusCode.BadGateway, diff --git a/src/LinkTracker.Tests/Scrapper/Integration/Http/StackOverflowHttpClientTests.cs b/src/LinkTracker.Tests/Scrapper/Integration/Http/StackOverflowHttpClientTests.cs index 24ebc4d..8ed9a72 100644 --- a/src/LinkTracker.Tests/Scrapper/Integration/Http/StackOverflowHttpClientTests.cs +++ b/src/LinkTracker.Tests/Scrapper/Integration/Http/StackOverflowHttpClientTests.cs @@ -85,7 +85,7 @@ public async Task GetAnswers_WhenResponseIsValid_ReturnsAnswers() await wireMock.StubAsync(new { - request = new { method = "GET", url = "/2.3/questions/123/answers?site=stackoverflow&sort=creation&order=desc&filter=withbody" }, + request = new { method = "GET", urlPath = "/2.3/questions/123/answers" }, response = new { status = 200, @@ -129,7 +129,7 @@ public async Task GetComments_WhenResponseIsValid_ReturnsComments() await wireMock.StubAsync(new { - request = new { method = "GET", url = "/2.3/questions/123/comments?site=stackoverflow&sort=creation&order=desc&filter=withbody" }, + request = new { method = "GET", urlPath = "/2.3/questions/123/comments" }, response = new { status = 200, diff --git a/src/LinkTracker.Tests/Scrapper/Integration/Storage/LinkTrackingStoreContractTests.cs b/src/LinkTracker.Tests/Scrapper/Integration/Storage/LinkTrackingStoreContractTests.cs index 8b0a9a8..213e8e6 100644 --- a/src/LinkTracker.Tests/Scrapper/Integration/Storage/LinkTrackingStoreContractTests.cs +++ b/src/LinkTracker.Tests/Scrapper/Integration/Storage/LinkTrackingStoreContractTests.cs @@ -4,6 +4,8 @@ namespace LinkTracker.Tests.Scrapper.Integration.Storage; public abstract class LinkTrackingStoreContractTests { + private static readonly DateTimeOffset FarFuture = new(2100, 1, 1, 0, 0, 0, TimeSpan.Zero); + protected abstract Task ExecuteWithSut(Func test); [Fact] @@ -93,7 +95,7 @@ await ExecuteWithSut(async sut => var removed = await sut.TryRemoveAsync(firstChatId, url); var firstChatLinks = await sut.GetAllTrackedLinkRecordsAsync(firstChatId); var secondChatLinks = await sut.GetAllTrackedLinkRecordsAsync(secondChatId); - var subscriptions = await sut.GetAllSubscriptionsAsync(); + var subscriptions = await sut.GetSubscriptionsDueForCheckAsync(FarFuture, 100); Assert.NotNull(removed); Assert.Equal(url, removed!.Url); @@ -148,7 +150,7 @@ await ExecuteWithSut(async sut => Assert.Equal(updatedAt, only.LastUpdatedAt); Assert.Equal(eventKey, only.LastEventKey); - var subscriptions = await sut.GetAllSubscriptionsAsync(); + var subscriptions = await sut.GetSubscriptionsDueForCheckAsync(FarFuture, 100); var subscription = Assert.Single(subscriptions); Assert.Equal(updatedAt, subscription.LastUpdatedAt); @@ -191,7 +193,7 @@ await ExecuteWithSut(async sut => Assert.Equal(updatedAt, secondOnly.LastUpdatedAt); Assert.Equal(eventKey, secondOnly.LastEventKey); - var subscriptions = await sut.GetAllSubscriptionsAsync(); + var subscriptions = await sut.GetSubscriptionsDueForCheckAsync(FarFuture, 100); var subscription = Assert.Single(subscriptions); Assert.Equal(firstAdded.Id, subscription.Id); @@ -222,7 +224,7 @@ await ExecuteWithSut(async sut => Assert.Equal(updatedAt, only.LastUpdatedAt); Assert.Null(only.LastEventKey); - var subscriptions = await sut.GetAllSubscriptionsAsync(); + var subscriptions = await sut.GetSubscriptionsDueForCheckAsync(FarFuture, 100); var subscription = Assert.Single(subscriptions); Assert.Equal(updatedAt, subscription.LastUpdatedAt); @@ -337,7 +339,7 @@ await ExecuteWithSut(async sut => } [Fact] - public async Task GetSubscriptionsBatchAsync_ReturnsSubscriptionsOrderedById() + public async Task GetSubscriptionsDueForCheckAsync_ReturnsSubscriptionsOrderedById() { await ExecuteWithSut(async sut => { @@ -368,7 +370,7 @@ await ExecuteWithSut(async sut => Assert.NotNull(second); Assert.NotNull(third); - var batch = await sut.GetSubscriptionsBatchAsync(null, 10); + var batch = await sut.GetSubscriptionsDueForCheckAsync(FarFuture, 10); Assert.Equal(3, batch.Count); @@ -380,7 +382,7 @@ await ExecuteWithSut(async sut => } [Fact] - public async Task GetSubscriptionsBatchAsync_ReturnsOnlyItemsAfterLinkId() + public async Task GetSubscriptionsDueForCheckAsync_SkipsLinksCheckedAfterTheCutoff() { await ExecuteWithSut(async sut => { @@ -388,6 +390,9 @@ await ExecuteWithSut(async sut => const long secondChatId = 8102; const long thirdChatId = 8103; + var checkedAt = new DateTimeOffset(2026, 3, 22, 12, 0, 0, TimeSpan.Zero); + var cutoff = checkedAt.AddMinutes(-1); + await sut.TryRegisterChatAsync(firstChatId); await sut.TryRegisterChatAsync(secondChatId); await sut.TryRegisterChatAsync(thirdChatId); @@ -411,7 +416,9 @@ await ExecuteWithSut(async sut => Assert.NotNull(second); Assert.NotNull(third); - var batch = await sut.GetSubscriptionsBatchAsync(first!.Id, 10); + await sut.MarkCheckedAsync([first!.Id], checkedAt); + + var batch = await sut.GetSubscriptionsDueForCheckAsync(cutoff, 10); Assert.Equal( [second!.Id, third!.Id], @@ -419,11 +426,41 @@ await ExecuteWithSut(async sut => }); } + [Fact] + public async Task GetSubscriptionsDueForCheckAsync_ReturnsLeastRecentlyCheckedFirst() + { + await ExecuteWithSut(async sut => + { + const long firstChatId = 8501; + const long secondChatId = 8502; + + await sut.TryRegisterChatAsync(firstChatId); + await sut.TryRegisterChatAsync(secondChatId); + + var first = await sut.TryAddAsync(firstChatId, new Uri("https://github.com/user/repo-1"), []); + var second = await sut.TryAddAsync(secondChatId, new Uri("https://github.com/user/repo-2"), []); + + Assert.NotNull(first); + Assert.NotNull(second); + + var recent = new DateTimeOffset(2026, 3, 22, 12, 0, 0, TimeSpan.Zero); + + await sut.MarkCheckedAsync([first!.Id], recent); + await sut.MarkCheckedAsync([second!.Id], recent.AddHours(-1)); + + var batch = await sut.GetSubscriptionsDueForCheckAsync(recent.AddHours(1), 10); + + Assert.Equal( + [second.Id, first.Id], + batch.Select(x => x.Id).ToArray()); + }); + } + [Theory] [InlineData(1)] [InlineData(2)] [InlineData(3)] - public async Task GetSubscriptionsBatchAsync_RespectsBatchSize(int batchSize) + public async Task GetSubscriptionsDueForCheckAsync_RespectsBatchSize(int batchSize) { await ExecuteWithSut(async sut => { @@ -442,14 +479,14 @@ await ExecuteWithSut(async sut => await sut.TryAddAsync(thirdChatId, new Uri("https://github.com/user/repo-3"), []); await sut.TryAddAsync(fourthChatId, new Uri("https://github.com/user/repo-4"), []); - var batch = await sut.GetSubscriptionsBatchAsync(null, batchSize); + var batch = await sut.GetSubscriptionsDueForCheckAsync(FarFuture, batchSize); Assert.Equal(batchSize, batch.Count); }); } [Fact] - public async Task GetSubscriptionsBatchAsync_ReturnsSharedLinkOnceWithAllSubscribers() + public async Task GetSubscriptionsDueForCheckAsync_ReturnsSharedLinkOnceWithAllSubscribers() { await ExecuteWithSut(async sut => { @@ -467,7 +504,7 @@ await ExecuteWithSut(async sut => Assert.NotNull(second); Assert.Equal(first!.Id, second!.Id); - var batch = await sut.GetSubscriptionsBatchAsync(null, 10); + var batch = await sut.GetSubscriptionsDueForCheckAsync(FarFuture, 10); var subscription = Assert.Single(batch); @@ -480,7 +517,7 @@ await ExecuteWithSut(async sut => } [Fact] - public async Task GetSubscriptionsBatchAsync_ReturnsCursorFields() + public async Task GetSubscriptionsDueForCheckAsync_ReturnsCursorFields() { await ExecuteWithSut(async sut => { @@ -496,7 +533,7 @@ await ExecuteWithSut(async sut => await sut.SetCursorAsync(added!.Id, updatedAt, eventKey); - var batch = await sut.GetSubscriptionsBatchAsync(null, 10); + var batch = await sut.GetSubscriptionsDueForCheckAsync(FarFuture, 10); var subscription = Assert.Single(batch); diff --git a/src/LinkTracker.Tests/Scrapper/Unit/Application/Services/Updates/Clients/GitHubLinkUpdateHandlerTests.cs b/src/LinkTracker.Tests/Scrapper/Unit/Application/Services/Updates/Clients/GitHubLinkUpdateHandlerTests.cs index ae2a581..26aec49 100644 --- a/src/LinkTracker.Tests/Scrapper/Unit/Application/Services/Updates/Clients/GitHubLinkUpdateHandlerTests.cs +++ b/src/LinkTracker.Tests/Scrapper/Unit/Application/Services/Updates/Clients/GitHubLinkUpdateHandlerTests.cs @@ -37,7 +37,7 @@ await gitHubClient.Received(1) .GetRepositoryAsync("user", "repo", Arg.Any()); await gitHubClient.DidNotReceive() - .GetIssuesAsync(Arg.Any(), Arg.Any(), Arg.Any()); + .GetIssuesAsync(Arg.Any(), Arg.Any(), Arg.Any(), Arg.Any()); await gitHubClient.DidNotReceive() .GetPullRequestsAsync(Arg.Any(), Arg.Any(), Arg.Any()); @@ -51,7 +51,7 @@ public async Task CheckAsync_WhenNoNewIssuesOrPullRequests_ReturnsNoChangesAndKe var lastSeenAt = new DateTimeOffset(2025, 3, 9, 12, 0, 0, TimeSpan.Zero); - gitHubClient.GetIssuesAsync("user", "repo", Arg.Any()) + gitHubClient.GetIssuesAsync("user", "repo", Arg.Any(), Arg.Any()) .Returns( [ new GitHubIssueResponse @@ -106,7 +106,7 @@ public async Task CheckAsync_WhenSingleNewEventExists_ReturnsMappedEvent(string if (expectedKind == LinkEventKind.Issue) { - gitHubClient.GetIssuesAsync("user", "repo", Arg.Any()) + gitHubClient.GetIssuesAsync("user", "repo", Arg.Any(), Arg.Any()) .Returns( [ new GitHubIssueResponse @@ -124,7 +124,7 @@ public async Task CheckAsync_WhenSingleNewEventExists_ReturnsMappedEvent(string } else { - gitHubClient.GetIssuesAsync("user", "repo", Arg.Any()) + gitHubClient.GetIssuesAsync("user", "repo", Arg.Any(), Arg.Any()) .Returns(Array.Empty()); gitHubClient.GetPullRequestsAsync("user", "repo", Arg.Any()) @@ -184,7 +184,7 @@ public async Task CheckAsync_WhenIssueAndPullRequestExist_ReturnsOrderedEventsWi var issueCreatedAt = new DateTimeOffset(2025, 3, 10, 10, 0, 0, TimeSpan.Zero); var prCreatedAt = new DateTimeOffset(2025, 3, 10, 11, 0, 0, TimeSpan.Zero); - gitHubClient.GetIssuesAsync("user", "repo", Arg.Any()) + gitHubClient.GetIssuesAsync("user", "repo", Arg.Any(), Arg.Any()) .Returns( [ new GitHubIssueResponse diff --git a/src/LinkTracker.Tests/Scrapper/Unit/Application/Services/Updates/Clients/StackOverflowLinkUpdateHandlerTests.cs b/src/LinkTracker.Tests/Scrapper/Unit/Application/Services/Updates/Clients/StackOverflowLinkUpdateHandlerTests.cs index 569f850..447a1ec 100644 --- a/src/LinkTracker.Tests/Scrapper/Unit/Application/Services/Updates/Clients/StackOverflowLinkUpdateHandlerTests.cs +++ b/src/LinkTracker.Tests/Scrapper/Unit/Application/Services/Updates/Clients/StackOverflowLinkUpdateHandlerTests.cs @@ -47,13 +47,47 @@ await stackOverflowClient.DidNotReceive() } [Fact] - public async Task CheckAsync_WhenNoNewAnswersOrComments_ReturnsNoChangesAndKeepsLastUpdatedAt() + public async Task CheckAsync_WhenQuestionHasNoActivitySinceCursor_SkipsAnswerAndCommentRequests() { var stackOverflowClient = Substitute.For(); var logger = Substitute.For>(); var lastSeenAt = new DateTimeOffset(2025, 3, 9, 12, 0, 0, TimeSpan.Zero); + stackOverflowClient.GetQuestionAsync(123, Arg.Any()) + .Returns(new StackOverflowQuestionResponse { QuestionId = 123, Title = "How to test this?", Link = new Uri("https://stackoverflow.com/questions/123/how-to-test-this"), LastActivityDateUnix = lastSeenAt.AddMinutes(-1).ToUnixTimeSeconds() }); + + var subscription = new TrackedLinkSubscription { Id = 1, Url = new Uri("https://stackoverflow.com/questions/123/how-to-test-this"), TgChatIds = [1001L], LastUpdatedAt = lastSeenAt }; + + var sut = new StackOverflowLinkUpdateHandler(stackOverflowClient, logger); + + var result = await sut.CheckAsync(subscription); + + Assert.False(result.HasChanges); + Assert.Empty(result.Events); + Assert.Equal(lastSeenAt, result.NewLastUpdatedAt); + + await stackOverflowClient.Received(1) + .GetQuestionAsync(123, Arg.Any()); + + await stackOverflowClient.DidNotReceive() + .GetAnswersAsync(Arg.Any(), Arg.Any()); + + await stackOverflowClient.DidNotReceive() + .GetCommentsAsync(Arg.Any(), Arg.Any()); + } + + [Fact] + public async Task CheckAsync_WhenActivityIsNewerButItemsAreOld_ReturnsNoChanges() + { + var stackOverflowClient = Substitute.For(); + var logger = Substitute.For>(); + + var lastSeenAt = new DateTimeOffset(2025, 3, 9, 12, 0, 0, TimeSpan.Zero); + + stackOverflowClient.GetQuestionAsync(123, Arg.Any()) + .Returns(new StackOverflowQuestionResponse { QuestionId = 123, Title = "How to test this?", Link = new Uri("https://stackoverflow.com/questions/123/how-to-test-this"), LastActivityDateUnix = lastSeenAt.AddMinutes(5).ToUnixTimeSeconds() }); + stackOverflowClient.GetAnswersAsync(123, Arg.Any()) .Returns( [ @@ -89,9 +123,6 @@ public async Task CheckAsync_WhenNoNewAnswersOrComments_ReturnsNoChangesAndKeeps Assert.False(result.HasChanges); Assert.Empty(result.Events); Assert.Equal(lastSeenAt, result.NewLastUpdatedAt); - - await stackOverflowClient.DidNotReceive() - .GetQuestionAsync(Arg.Any(), Arg.Any()); } [Theory] diff --git a/src/LinkTracker.Tests/Scrapper/Unit/Infrastructure/Outbox/Jobs/OutboxDispatchJobTests.cs b/src/LinkTracker.Tests/Scrapper/Unit/Infrastructure/Outbox/Jobs/OutboxDispatchJobTests.cs index 1db746e..473b114 100644 --- a/src/LinkTracker.Tests/Scrapper/Unit/Infrastructure/Outbox/Jobs/OutboxDispatchJobTests.cs +++ b/src/LinkTracker.Tests/Scrapper/Unit/Infrastructure/Outbox/Jobs/OutboxDispatchJobTests.cs @@ -4,6 +4,7 @@ using LinkTracker.Scrapper.Infrastructure.Outbox.Abstractions; using LinkTracker.Scrapper.Infrastructure.Outbox.Configuration; using LinkTracker.Scrapper.Infrastructure.Outbox.Jobs; +using LinkTracker.Scrapper.Infrastructure.Telemetry; using LinkTracker.Scrapper.Infrastructure.Outbox.Models; using LinkTracker.Shared.Contracts.Bot; using LinkTracker.Shared.Infrastructure; @@ -180,6 +181,7 @@ private static OutboxDispatchJob CreateSut( outboxStore, botClient, outboxOptions, + new ScrapperMetrics(), NullLogger.Instance); } diff --git a/src/LinkTracker.Tests/Scrapper/Unit/Infrastructure/Quartz/Jobs/LinkUpdatesJobTests.cs b/src/LinkTracker.Tests/Scrapper/Unit/Infrastructure/Quartz/Jobs/LinkUpdatesJobTests.cs index 690bd90..31d82b7 100644 --- a/src/LinkTracker.Tests/Scrapper/Unit/Infrastructure/Quartz/Jobs/LinkUpdatesJobTests.cs +++ b/src/LinkTracker.Tests/Scrapper/Unit/Infrastructure/Quartz/Jobs/LinkUpdatesJobTests.cs @@ -46,11 +46,8 @@ public async Task Execute_WhenTrackedLinkHasEvents_SendsUpdateOnlyToSubscribers_ LastEventKey = "issue:122" }; - trackingStore.GetSubscriptionsBatchAsync(null, DefaultBatchSize, Arg.Any()) - .Returns([subscription]); - - trackingStore.GetSubscriptionsBatchAsync(subscription.Id, DefaultBatchSize, Arg.Any()) - .Returns([]); + trackingStore.GetSubscriptionsDueForCheckAsync(Arg.Any(), DefaultBatchSize, Arg.Any()) + .Returns(Batch(subscription), Batch()); githubHandler.CanHandle(subscription.Url).Returns(true); stackOverflowHandler.CanHandle(subscription.Url).Returns(false); @@ -124,11 +121,8 @@ public async Task Execute_WhenHandlerThrows_SendsFailedReport_AndDoesNotUpdateCu var subscription = new TrackedLinkSubscription { Id = 10, Url = new Uri("https://github.com/user/repo"), TgChatIds = [1001L] }; - trackingStore.GetSubscriptionsBatchAsync(null, DefaultBatchSize, Arg.Any()) - .Returns([subscription]); - - trackingStore.GetSubscriptionsBatchAsync(subscription.Id, DefaultBatchSize, Arg.Any()) - .Returns([]); + trackingStore.GetSubscriptionsDueForCheckAsync(Arg.Any(), DefaultBatchSize, Arg.Any()) + .Returns(Batch(subscription), Batch()); githubHandler.CanHandle(subscription.Url).Returns(true); stackOverflowHandler.CanHandle(subscription.Url).Returns(false); @@ -188,11 +182,8 @@ public async Task Execute_WhenMultipleEvents_SendsEachEventAsSeparateUpdate_AndU LastEventKey = "issue:120" }; - trackingStore.GetSubscriptionsBatchAsync(null, DefaultBatchSize, Arg.Any()) - .Returns([subscription]); - - trackingStore.GetSubscriptionsBatchAsync(subscription.Id, DefaultBatchSize, Arg.Any()) - .Returns([]); + trackingStore.GetSubscriptionsDueForCheckAsync(Arg.Any(), DefaultBatchSize, Arg.Any()) + .Returns(Batch(subscription), Batch()); githubHandler.CanHandle(subscription.Url).Returns(true); stackOverflowHandler.CanHandle(subscription.Url).Returns(false); @@ -278,11 +269,8 @@ public async Task Execute_WhenNoEvents_DoesNotSendUpdate_AndUpdatesCursorOnlyWhe LastEventKey = "issue:122" }; - trackingStore.GetSubscriptionsBatchAsync(null, DefaultBatchSize, Arg.Any()) - .Returns([subscription]); - - trackingStore.GetSubscriptionsBatchAsync(subscription.Id, DefaultBatchSize, Arg.Any()) - .Returns([]); + trackingStore.GetSubscriptionsDueForCheckAsync(Arg.Any(), DefaultBatchSize, Arg.Any()) + .Returns(Batch(subscription), Batch()); githubHandler.CanHandle(subscription.Url).Returns(true); stackOverflowHandler.CanHandle(subscription.Url).Returns(false); @@ -337,11 +325,8 @@ public async Task Execute_WhenNoHandlerCanHandleUrl_SkipsSubscription() var subscription = new TrackedLinkSubscription { Id = 10, Url = new Uri("https://example.com/page"), TgChatIds = [1001L] }; - trackingStore.GetSubscriptionsBatchAsync(null, DefaultBatchSize, Arg.Any()) - .Returns([subscription]); - - trackingStore.GetSubscriptionsBatchAsync(subscription.Id, DefaultBatchSize, Arg.Any()) - .Returns([]); + trackingStore.GetSubscriptionsDueForCheckAsync(Arg.Any(), DefaultBatchSize, Arg.Any()) + .Returns(Batch(subscription), Batch()); githubHandler.CanHandle(subscription.Url).Returns(false); stackOverflowHandler.CanHandle(subscription.Url).Returns(false); @@ -388,14 +373,8 @@ public async Task Execute_WhenSubscriptionsExceedBatchSize_ProcessesMultipleBatc var second = new TrackedLinkSubscription { Id = 20, Url = new Uri("https://github.com/user/repo2"), TgChatIds = [1002L] }; - trackingStore.GetSubscriptionsBatchAsync(null, 1, Arg.Any()) - .Returns([first]); - - trackingStore.GetSubscriptionsBatchAsync(first.Id, 1, Arg.Any()) - .Returns([second]); - - trackingStore.GetSubscriptionsBatchAsync(second.Id, 1, Arg.Any()) - .Returns([]); + trackingStore.GetSubscriptionsDueForCheckAsync(Arg.Any(), 1, Arg.Any()) + .Returns(Batch(first), Batch(second), Batch()); handler.CanHandle(first.Url).Returns(true); handler.CanHandle(second.Url).Returns(true); @@ -419,14 +398,8 @@ public async Task Execute_WhenSubscriptionsExceedBatchSize_ProcessesMultipleBatc await sut.Execute(quartzContext); - await trackingStore.Received(1) - .GetSubscriptionsBatchAsync(null, 1, Arg.Any()); - - await trackingStore.Received(1) - .GetSubscriptionsBatchAsync(first.Id, 1, Arg.Any()); - - await trackingStore.Received(1) - .GetSubscriptionsBatchAsync(second.Id, 1, Arg.Any()); + await trackingStore.Received(3) + .GetSubscriptionsDueForCheckAsync(Arg.Any(), 1, Arg.Any()); await handler.Received(1) .CheckAsync(first, Arg.Any()); @@ -450,11 +423,8 @@ public async Task Execute_WhenOneSubscriptionFails_ContinuesWithRemainingSubscri var eventCreatedAt = new DateTimeOffset(2025, 3, 10, 12, 0, 0, TimeSpan.Zero); - trackingStore.GetSubscriptionsBatchAsync(null, DefaultBatchSize, Arg.Any()) - .Returns([failedSubscription, successfulSubscription]); - - trackingStore.GetSubscriptionsBatchAsync(successfulSubscription.Id, DefaultBatchSize, Arg.Any()) - .Returns([]); + trackingStore.GetSubscriptionsDueForCheckAsync(Arg.Any(), DefaultBatchSize, Arg.Any()) + .Returns(Batch(failedSubscription, successfulSubscription), Batch()); handler.CanHandle(failedSubscription.Url).Returns(true); handler.CanHandle(successfulSubscription.Url).Returns(true); @@ -545,8 +515,8 @@ public async Task Execute_WhenStoreReturnsEmptyBatch_StopsProcessing() var outboxStore = Substitute.For(); var logger = Substitute.For>(); - trackingStore.GetSubscriptionsBatchAsync(null, DefaultBatchSize, Arg.Any()) - .Returns([]); + trackingStore.GetSubscriptionsDueForCheckAsync(Arg.Any(), DefaultBatchSize, Arg.Any()) + .Returns(Batch()); var quartzContext = Substitute.For(); quartzContext.CancellationToken.Returns(CancellationToken.None); @@ -579,8 +549,8 @@ public async Task Execute_UsesConfiguredBatchSize(int configuredBatchSize) var outboxStore = Substitute.For(); var logger = Substitute.For>(); - trackingStore.GetSubscriptionsBatchAsync(null, configuredBatchSize, Arg.Any()) - .Returns([]); + trackingStore.GetSubscriptionsDueForCheckAsync(Arg.Any(), configuredBatchSize, Arg.Any()) + .Returns(Batch()); var quartzContext = Substitute.For(); quartzContext.CancellationToken.Returns(CancellationToken.None); @@ -596,7 +566,7 @@ public async Task Execute_UsesConfiguredBatchSize(int configuredBatchSize) await sut.Execute(quartzContext); await trackingStore.Received(1) - .GetSubscriptionsBatchAsync(null, configuredBatchSize, Arg.Any()); + .GetSubscriptionsDueForCheckAsync(Arg.Any(), configuredBatchSize, Arg.Any()); } [Theory] @@ -615,11 +585,8 @@ public async Task Execute_AcceptsConfiguredParallelismAndProcessesBatch(int conf var secondSubscription = new TrackedLinkSubscription { Id = 20, Url = new Uri("https://github.com/user/repo2"), TgChatIds = [1002L] }; - trackingStore.GetSubscriptionsBatchAsync(null, DefaultBatchSize, Arg.Any()) - .Returns([firstSubscription, secondSubscription]); - - trackingStore.GetSubscriptionsBatchAsync(secondSubscription.Id, DefaultBatchSize, Arg.Any()) - .Returns([]); + trackingStore.GetSubscriptionsDueForCheckAsync(Arg.Any(), DefaultBatchSize, Arg.Any()) + .Returns(Batch(firstSubscription, secondSubscription), Batch()); handler.CanHandle(firstSubscription.Url).Returns(true); handler.CanHandle(secondSubscription.Url).Returns(true); @@ -643,8 +610,8 @@ public async Task Execute_AcceptsConfiguredParallelismAndProcessesBatch(int conf await sut.Execute(quartzContext); - await trackingStore.Received(1) - .GetSubscriptionsBatchAsync(null, DefaultBatchSize, Arg.Any()); + await trackingStore.Received(2) + .GetSubscriptionsDueForCheckAsync(Arg.Any(), DefaultBatchSize, Arg.Any()); await handler.Received(1) .CheckAsync(firstSubscription, Arg.Any()); @@ -666,11 +633,8 @@ public async Task Execute_WhenParallelismIsGreaterThanOne_ProcessesMultipleSubsc var secondSubscription = new TrackedLinkSubscription { Id = 20, Url = new Uri("https://github.com/user/repo2"), TgChatIds = [1002L] }; - trackingStore.GetSubscriptionsBatchAsync(null, DefaultBatchSize, Arg.Any()) - .Returns([firstSubscription, secondSubscription]); - - trackingStore.GetSubscriptionsBatchAsync(secondSubscription.Id, DefaultBatchSize, Arg.Any()) - .Returns([]); + trackingStore.GetSubscriptionsDueForCheckAsync(Arg.Any(), DefaultBatchSize, Arg.Any()) + .Returns(Batch(firstSubscription, secondSubscription), Batch()); handler.CanHandle(firstSubscription.Url).Returns(true); handler.CanHandle(secondSubscription.Url).Returns(true); @@ -724,11 +688,8 @@ public async Task Execute_WhenSeveralSubscriptionsFail_GroupsFailedReportsByChat var secondFailedSubscription = new TrackedLinkSubscription { Id = 20, Url = new Uri("https://github.com/user/repo2"), TgChatIds = [1002L] }; - trackingStore.GetSubscriptionsBatchAsync(null, DefaultBatchSize, Arg.Any()) - .Returns([firstFailedSubscription, secondFailedSubscription]); - - trackingStore.GetSubscriptionsBatchAsync(secondFailedSubscription.Id, DefaultBatchSize, Arg.Any()) - .Returns([]); + trackingStore.GetSubscriptionsDueForCheckAsync(Arg.Any(), DefaultBatchSize, Arg.Any()) + .Returns(Batch(firstFailedSubscription, secondFailedSubscription), Batch()); handler.CanHandle(firstFailedSubscription.Url).Returns(true); handler.CanHandle(secondFailedSubscription.Url).Returns(true); @@ -796,11 +757,8 @@ public async Task Execute_WhenEventBodyLongerThan200_TrimsPreviewInSentUpdate() var subscription = new TrackedLinkSubscription { Id = 10, Url = new Uri("https://github.com/user/repo"), TgChatIds = [1001L] }; - trackingStore.GetSubscriptionsBatchAsync(null, DefaultBatchSize, Arg.Any()) - .Returns([subscription]); - - trackingStore.GetSubscriptionsBatchAsync(subscription.Id, DefaultBatchSize, Arg.Any()) - .Returns([]); + trackingStore.GetSubscriptionsDueForCheckAsync(Arg.Any(), DefaultBatchSize, Arg.Any()) + .Returns(Batch(subscription), Batch()); handler.CanHandle(subscription.Url).Returns(true); @@ -868,11 +826,8 @@ public async Task Execute_WhenOutboxEnabledAndTrackedLinkHasEvents_SavesUpdatesT LastEventKey = "issue:122" }; - trackingStore.GetSubscriptionsBatchAsync(null, DefaultBatchSize, Arg.Any()) - .Returns([subscription]); - - trackingStore.GetSubscriptionsBatchAsync(subscription.Id, DefaultBatchSize, Arg.Any()) - .Returns([]); + trackingStore.GetSubscriptionsDueForCheckAsync(Arg.Any(), DefaultBatchSize, Arg.Any()) + .Returns(Batch(subscription), Batch()); handler.CanHandle(subscription.Url).Returns(true); @@ -945,11 +900,8 @@ public async Task Execute_WhenOutboxEnabledAndHandlerThrows_SendsFailedReportVia var subscription = new TrackedLinkSubscription { Id = 10, Url = new Uri("https://github.com/user/repo"), TgChatIds = [1001L] }; - trackingStore.GetSubscriptionsBatchAsync(null, DefaultBatchSize, Arg.Any()) - .Returns([subscription]); - - trackingStore.GetSubscriptionsBatchAsync(subscription.Id, DefaultBatchSize, Arg.Any()) - .Returns([]); + trackingStore.GetSubscriptionsDueForCheckAsync(Arg.Any(), DefaultBatchSize, Arg.Any()) + .Returns(Batch(subscription), Batch()); handler.CanHandle(subscription.Url).Returns(true); @@ -985,6 +937,90 @@ await botClient.Received(1).SendUpdateAsync( Arg.Any()); } + [Fact] + public async Task Execute_MarksEveryProcessedLinkAsChecked_EvenWhenTheCheckFails() + { + var trackingStore = Substitute.For(); + var handler = Substitute.For(); + var botClient = Substitute.For(); + var outboxStore = Substitute.For(); + var logger = Substitute.For>(); + + var failing = new TrackedLinkSubscription { Id = 10, Url = new Uri("https://github.com/user/repo-1"), TgChatIds = [1001L] }; + var succeeding = new TrackedLinkSubscription { Id = 20, Url = new Uri("https://github.com/user/repo-2"), TgChatIds = [1002L] }; + + trackingStore.GetSubscriptionsDueForCheckAsync(Arg.Any(), DefaultBatchSize, Arg.Any()) + .Returns(Batch(failing, succeeding), Batch()); + + handler.CanHandle(Arg.Any()).Returns(true); + + handler.CheckAsync(failing, Arg.Any()) + .Returns(_ => throw new HttpRequestException("boom")); + + handler.CheckAsync(succeeding, Arg.Any()) + .Returns(new LinkCheckResult()); + + var quartzContext = Substitute.For(); + quartzContext.CancellationToken.Returns(CancellationToken.None); + + var sut = CreateSut(trackingStore, [handler], botClient, outboxStore, logger); + + await sut.Execute(quartzContext); + + await trackingStore.Received(1) + .MarkCheckedAsync( + Arg.Is>(ids => ids.SequenceEqual(new[] { failing.Id, succeeding.Id })), + Arg.Any(), + Arg.Any()); + } + + [Fact] + public async Task Execute_QueriesEveryBatchWithTheSameCutoff() + { + var trackingStore = Substitute.For(); + var handler = Substitute.For(); + var botClient = Substitute.For(); + var outboxStore = Substitute.For(); + var logger = Substitute.For>(); + + var first = new TrackedLinkSubscription { Id = 10, Url = new Uri("https://github.com/user/repo-1"), TgChatIds = [1001L] }; + var second = new TrackedLinkSubscription { Id = 20, Url = new Uri("https://github.com/user/repo-2"), TgChatIds = [1002L] }; + + var cutoffs = new List(); + + trackingStore.GetSubscriptionsDueForCheckAsync(Arg.Any(), 1, Arg.Any()) + .Returns(callInfo => + { + cutoffs.Add(callInfo.Arg()); + + return cutoffs.Count switch + { + 1 => Batch(first), + 2 => Batch(second), + _ => Batch() + }; + }); + + handler.CanHandle(Arg.Any()).Returns(true); + handler.CheckAsync(Arg.Any(), Arg.Any()) + .Returns(new LinkCheckResult()); + + var quartzContext = Substitute.For(); + quartzContext.CancellationToken.Returns(CancellationToken.None); + + var sut = CreateSut(trackingStore, [handler], botClient, outboxStore, logger, batchSize: 1); + + await sut.Execute(quartzContext); + + Assert.Equal(3, cutoffs.Count); + Assert.Single(cutoffs.Distinct()); + } + + private static IReadOnlyList Batch(params TrackedLinkSubscription[] subscriptions) + { + return subscriptions; + } + private static LinkUpdatesJob CreateSut( ILinkTrackingStore trackingStore, IEnumerable handlers, @@ -1008,6 +1044,7 @@ private static LinkUpdatesJob CreateSut( outboxStore, schedulingOptions, outboxOptions, + TimeProvider.System, logger, metrics); }