Echways · Echways · Aug 16, 2026 · Aug 16, 2026 · Aug 16, 2026
diff --git a/docker-compose.yml b/docker-compose.yml
@@ -103,33 +103,24 @@ services:
       - /bin/sh
       - -c
       - |
-        /opt/kafka/bin/kafka-topics.sh \
-          --bootstrap-server kafka-1:9092 \
-          --create \
-          --if-not-exists \
-          --topic link.processed-updates \
-          --partitions 6 \
-          --replication-factor 3 \
-          --config min.insync.replicas=2
-
-        /opt/kafka/bin/kafka-topics.sh \
-          --bootstrap-server kafka-1:9092 \
-          --create \
-          --if-not-exists \
-          --topic link.raw-updates \
-          --partitions 6 \
-          --replication-factor 3 \
-          --config min.insync.replicas=2
-
-        /opt/kafka/bin/kafka-topics.sh \
-          --bootstrap-server kafka-1:9092 \
-          --create \
-          --if-not-exists \
-          --topic link.raw-updates-dlq \
-          --partitions 6 \
-          --replication-factor 3 \
-          --config min.insync.replicas=2
+        set -e
 
+        for topic in \
+          link.raw-updates \
+          link.raw-updates-dlq \
+          link.processed-updates \
+          link.processed-updates-dlq
+        do
+          /opt/kafka/bin/kafka-topics.sh \
+            --bootstrap-server kafka-1:9092 \
+            --create \
+            --if-not-exists \
+            --topic "$$topic" \
+            --partitions 6 \
+            --replication-factor 3 \
+            --config min.insync.replicas=2
+        done
+
   schema-registry:
     image: confluentinc/cp-schema-registry:7.6.1
     container_name: linktracker-schema-registry

diff --git a/monitoring/grafana/provisioning/alerting/rules.yml b/monitoring/grafana/provisioning/alerting/rules.yml
@@ -26,7 +26,7 @@ groups:
             model:
               refId: A
               editorMode: code
-              expr: max by (job, instance) (process_memory_working_set_bytes{job=~"scrapper|bot"})
+              expr: max by (job, instance) (process_memory_working_set_bytes{job=~"scrapper|bot|aiagent"})
               instant: true
               intervalMs: 1000
               maxDataPoints: 43200
@@ -44,4 +44,92 @@ groups:
                   evaluator:
                     type: gt
                     params:
-                      - 524288000
+                      - 524288000
+
+  - orgId: 1
+    name: kafka
+    folder: LinkTracker
+    interval: 1m
+    rules:
+      - uid: kafka_dlq_publish_failed
+        title: Kafka DLQ is unavailable
+        condition: C
+        for: 2m
+        noDataState: OK
+        execErrState: Error
+        labels:
+          severity: critical
+        annotations:
+          summary: "{{ $labels.job }} не может писать в DLQ (topic {{ $labels.topic }})"
+          description: >-
+            kafka_dead_letter_errors_total растет: отправка в DLQ падает, offset не коммитится,
+            и «ядовитое» сообщение переигрывается бесконечно. Проверь, что DLQ-топик существует
+            (kafka-init) и что DeadLetterTopic в конфиге сервиса совпадает с ним.
+        data:
+          - refId: A
+            relativeTimeRange:
+              from: 600
+              to: 0
+            datasourceUid: prometheus
+            model:
+              refId: A
+              editorMode: code
+              expr: sum by (job, topic) (increase(kafka_dead_letter_errors_total{job=~"bot|aiagent"}[5m]))
+              instant: true
+              intervalMs: 1000
+              maxDataPoints: 43200
+          - refId: C
+            relativeTimeRange:
+              from: 600
+              to: 0
+            datasourceUid: __expr__
+            model:
+              refId: C
+              type: threshold
+              expression: A
+              conditions:
+                - type: query
+                  evaluator:
+                    type: gt
+                    params:
+                      - 0
+
+      - uid: kafka_dlq_growth
+        title: Kafka DLQ is filling up
+        condition: C
+        for: 5m
+        noDataState: OK
+        execErrState: Error
+        labels:
+          severity: warning
+        annotations:
+          summary: "Сообщения уходят в DLQ у {{ $labels.job }} (topic {{ $labels.topic }})"
+          description: "kafka_dead_letter_total растет: сообщения не проходят десериализацию, валидацию или обработку."
+        data:
+          - refId: A
+            relativeTimeRange:
+              from: 600
+              to: 0
+            datasourceUid: prometheus
+            model:
+              refId: A
+              editorMode: code
+              expr: sum by (job, topic) (increase(kafka_dead_letter_total{job=~"bot|aiagent"}[5m]))
+              instant: true
+              intervalMs: 1000
+              maxDataPoints: 43200
+          - refId: C
+            relativeTimeRange:
+              from: 600
+              to: 0
+            datasourceUid: __expr__
+            model:
+              refId: C
+              type: threshold
+              expression: A
+              conditions:
+                - type: query
+                  evaluator:
+                    type: gt
+                    params:
+                      - 0
diff --git a/monitoring/prometheus.yml b/monitoring/prometheus.yml
@@ -9,3 +9,7 @@ scrape_configs:
   - job_name: bot
     static_configs:
       - targets: ["bot:8011"]
+
+  - job_name: aiagent
+    static_configs:
+      - targets: ["aiagent:8102"]
diff --git a/src/LinkTracker.AiAgent.Api/Program.cs b/src/LinkTracker.AiAgent.Api/Program.cs
@@ -1,16 +1,21 @@
 using LinkTracker.AiAgent.Application.Registration;
 using LinkTracker.AiAgent.Infrastructure.Clients.Registration;
+using LinkTracker.AiAgent.Infrastructure.Telemetry.Registration;
 using LinkTracker.EnvReader;
+using LinkTracker.Shared.Infrastructure.Telemetry;
 
 var builder = WebApplication.CreateBuilder(args);
 
 builder.AddLocalDotEnv();
 
 builder.Services.AddAiAgentApplication();
 builder.Services.AddAiAgentInfrastructure(builder.Configuration);
+builder.Services.AddTelemetry(builder.Configuration);
 
 var app = builder.Build();
 
 app.MapGet("/health", () => Results.Ok());
 
-await app.RunAsync();
+app.MapMetricsEndpoint().RequireHost("*:8102");
+
+await app.RunAsync();
diff --git a/src/LinkTracker.AiAgent.Api/appsettings.json b/src/LinkTracker.AiAgent.Api/appsettings.json
@@ -6,6 +6,18 @@
     }
   },
   "AllowedHosts": "*",
+  "Kestrel": {
+    "Endpoints": {
+      "Rest": {
+        "Url": "http://0.0.0.0:8101",
+        "Protocols": "Http1"
+      },
+      "Metrics": {
+        "Url": "http://0.0.0.0:8102",
+        "Protocols": "Http1"
+      }
+    }
+  },
   "Kafka": {
     "Consumer": {
       "BootstrapServers": "localhost:9094,localhost:9095,localhost:9096",

diff --git a/src/LinkTracker.AiAgent.Application/Abstractions/IGroupingBuffer.cs b/src/LinkTracker.AiAgent.Application/Abstractions/IGroupingBuffer.cs
@@ -2,8 +2,15 @@
 
 namespace LinkTracker.AiAgent.Application.Abstractions;
 
+public sealed record BufferedLinkUpdate(ProcessedLinkUpdate Update, IMessageAck Ack);
+
+public sealed record GroupingBucket(long ChatId, IReadOnlyList<BufferedLinkUpdate> Updates);
+
 public interface IGroupingBuffer
 {
-    void Add(long tgChatId, ProcessedLinkUpdate update);
-    IReadOnlyList<(long ChatId, IReadOnlyList<ProcessedLinkUpdate> Updates)> Flush();
+    void Add(long tgChatId, ProcessedLinkUpdate update, IMessageAck ack);
+
+    IReadOnlyList<GroupingBucket> Flush(bool force = false);
+
+    void Requeue(GroupingBucket bucket);
 }
diff --git a/src/LinkTracker.AiAgent.Application/Abstractions/ILinkUpdateProcessingService.cs b/src/LinkTracker.AiAgent.Application/Abstractions/ILinkUpdateProcessingService.cs
@@ -4,5 +4,5 @@ namespace LinkTracker.AiAgent.Application.Abstractions;
 
 public interface ILinkUpdateProcessingService
 {
-    Task ProcessAsync(LinkUpdate update, CancellationToken ct = default);
+    Task ProcessAsync(LinkUpdate update, IMessageAck ack, CancellationToken ct = default);
 }
diff --git a/src/LinkTracker.AiAgent.Application/Abstractions/IMessageAck.cs b/src/LinkTracker.AiAgent.Application/Abstractions/IMessageAck.cs
@@ -0,0 +1,8 @@
+namespace LinkTracker.AiAgent.Application.Abstractions;
+
+public interface IMessageAck
+{
+    void Retain();
+
+    void Release();
+}
diff --git a/src/LinkTracker.AiAgent.Application/Services/LinkUpdateProcessingService.cs b/src/LinkTracker.AiAgent.Application/Services/LinkUpdateProcessingService.cs
@@ -10,10 +10,17 @@ public sealed class LinkUpdateProcessingService(
     ILinkUpdateSummarizer summarizer,
     ILinkUpdatePrioritizer prioritizer,
     IGroupingBuffer groupingBuffer,
+    IProcessedUpdatePublisher publisher,
     ILogger<LinkUpdateProcessingService> logger) : ILinkUpdateProcessingService
 {
-    public async Task ProcessAsync(LinkUpdate update, CancellationToken ct)
+    public async Task ProcessAsync(LinkUpdate update, IMessageAck ack, CancellationToken ct = default)
     {
+        if (update.Kind == LinkUpdateKind.SystemReport)
+        {
+            await PublishSystemReportAsync(update, ct);
+            return;
+        }
+
         if (filter.ShouldFilter(update))
         {
             logger.LogDebug(
@@ -27,18 +34,41 @@ public async Task ProcessAsync(LinkUpdate update, CancellationToken ct)
 
         foreach (var chatId in update.TgChatIds)
         {
-            groupingBuffer.Add(chatId, new ProcessedLinkUpdate
-            {
-                Id = update.Id,
-                Url = update.Url,
-                Description = description,
-                TgChatIds = [chatId],
-                Priority = priority
-            });
+            groupingBuffer.Add(
+                chatId,
+                new ProcessedLinkUpdate
+                {
+                    Id = update.Id,
+                    Url = update.Url,
+                    Description = description,
+                    TgChatIds = [chatId],
+                    Priority = priority
+                },
+                ack);
         }
 
         logger.LogDebug(
             "Обновление добавлено в буфер. UpdateId={UpdateId}, Priority={Priority}",
             update.Id, priority);
     }
-}
+
+    private async Task PublishSystemReportAsync(LinkUpdate update, CancellationToken ct)
+    {
+        foreach (var chatId in update.TgChatIds)
+        {
+            await publisher.PublishAsync(
+                new ProcessedLinkUpdate
+                {
+                    Id = update.Id,
+                    Url = update.Url,
+                    Description = update.Description,
+                    TgChatIds = [chatId],
+                    Priority = update.Priority,
+                    Kind = LinkUpdateKind.SystemReport
+                },
+                ct);
+        }
+
+        logger.LogDebug("Служебный отчёт опубликован без обработки. UpdateId={UpdateId}", update.Id);
+    }
+}
diff --git a/src/LinkTracker.AiAgent.Application/Telemetry/Abstractions/IAiAgentMetrics.cs b/src/LinkTracker.AiAgent.Application/Telemetry/Abstractions/IAiAgentMetrics.cs
@@ -0,0 +1,14 @@
+namespace LinkTracker.AiAgent.Application.Telemetry.Abstractions;
+
+public interface IAiAgentMetrics
+{
+    void IncrementKafkaConsumed(string topic);
+
+    void IncrementKafkaConsumeError(string topic);
+
+    void ObserveKafkaConsumeDuration(string topic, double milliseconds);
+
+    void IncrementKafkaDeadLetter(string topic);
+
+    void IncrementKafkaDeadLetterError(string topic);
+}
diff --git a/src/LinkTracker.AiAgent.Infrastructure/Clients/Kafka/IRawUpdatesKafkaMessageHandler.cs b/src/LinkTracker.AiAgent.Infrastructure/Clients/Kafka/IRawUpdatesKafkaMessageHandler.cs
@@ -1,8 +1,9 @@
 using Confluent.Kafka;
+using LinkTracker.AiAgent.Application.Abstractions;
 
 namespace LinkTracker.AiAgent.Infrastructure.Clients.Kafka;
 
 internal interface IRawUpdatesKafkaMessageHandler
 {
-    Task<bool> HandleAsync(ConsumeResult<string, byte[]> result, CancellationToken ct);
+    Task<bool> HandleAsync(ConsumeResult<string, byte[]> result, IMessageAck ack, CancellationToken ct);
 }