feature: Trend Detection & Alert Suppression Windows

2026-06-18 21:18:13 +08:00
parent e6f7989298
commit 3c54feb38a
31 changed files with 2751 additions and 18 deletions
@@ -0,0 +1,85 @@
+using System.Text.Json;
+using Confluent.Kafka;
+using Microsoft.Extensions.Options;
+
+public class TrendAnalyzerService : BackgroundService
+{
+    private readonly IServiceProvider _services;
+    private readonly KafkaOptions _kafkaOptions;
+    private readonly ILogger<TrendAnalyzerService> _logger;
+
+    public TrendAnalyzerService(
+        IServiceProvider services,
+        IOptions<KafkaOptions> kafkaOptions,
+        ILogger<TrendAnalyzerService> logger)
+    {
+        _services = services;
+        _kafkaOptions = kafkaOptions.Value;
+        _logger = logger;
+    }
+
+    protected override async Task ExecuteAsync(CancellationToken stoppingToken)
+    {
+        var config = new ConsumerConfig
+        {
+            BootstrapServers = _kafkaOptions.BootstrapServers,
+            GroupId = "trend-analyzer",
+            AutoOffsetReset = AutoOffsetReset.Earliest,
+            EnableAutoCommit = false
+        };
+
+        using var consumer = new ConsumerBuilder<string, string>(config).Build();
+        consumer.Subscribe(_kafkaOptions.Topics.ObservationRecorded);
+
+        _logger.LogInformation("TrendAnalyzerService started — consumer group: trend-analyzer");
+
+        try
+        {
+            while (!stoppingToken.IsCancellationRequested)
+            {
+                ConsumeResult<string, string>? result = null;
+                try
+                {
+                    result = consumer.Consume(stoppingToken);
+
+                    var evt = JsonSerializer.Deserialize<TrendObservationEvent>(
+                        result.Message.Value,
+                        new JsonSerializerOptions { PropertyNameCaseInsensitive = true })!;
+
+                    using var scope = _services.CreateScope();
+                    var detector = scope.ServiceProvider.GetRequiredService<TrendDetector>();
+
+                    var outcome = await detector.ProcessObservationAsync(
+                        evt.EncounterId,
+                        evt.PatientId,
+                        evt.ObservationCode,
+                        evt.Value,
+                        evt.RecordedAt,
+                        stoppingToken);
+
+                    if (outcome.Outcome == TrendOutcome.RapidDeterioration)
+                        _logger.LogInformation(
+                            "RAPID_DETERIORATION alert via consumer — encounter={Id} code={Code} rate={Rate}/min",
+                            evt.EncounterId, outcome.ObservationCode, outcome.RatePerMinute);
+
+                    consumer.Commit(result);
+                }
+                catch (OperationCanceledException)
+                {
+                    break;
+                }
+                catch (Exception ex)
+                {
+                    _logger.LogError(ex,
+                        "TrendAnalyzerService failed on topic={Topic} offset={Offset} — not committing",
+                        result?.Topic, result?.Offset.Value);
+                    await Task.Delay(2000, stoppingToken);
+                }
+            }
+        }
+        finally
+        {
+            consumer.Close();
+        }
+    }
+}