using System.Text.Json; using Microsoft.EntityFrameworkCore; using Microsoft.Extensions.Options; using Prometheus; using StackExchange.Redis; public class TrendDetector { private readonly IConnectionMultiplexer _redis; private readonly IServiceProvider _services; private readonly TrendDetectionOptions _options; private readonly ClinicalMetrics _metrics; private readonly ILogger _logger; public TrendDetector( ILogger logger, ClinicalMetrics metrics, IServiceProvider services, IConnectionMultiplexer redis, IOptions options) { _logger = logger; _metrics = metrics; _services = services; _redis = redis; _options = options.Value; } public async Task ProcessObservationAsync( Guid encounterId, Guid patientId, string observationCode, decimal value, DateTimeOffset recordedAt, CancellationToken ct = default) { if (!TrendCalculator.IsTrendCode(observationCode)) return new TrendResult(TrendOutcome.NotTrendCode); using var timer = _metrics.TrendAnalysisDuration.NewTimer(); var cache = _redis.GetDatabase(); var key = TrendCalculator.HistoryKey(encounterId, observationCode); // Append new entry to history list var entry = new TrendHistoryEntry(value, recordedAt); var historyJson = await cache.StringGetAsync(key); var history = historyJson.HasValue ? JsonSerializer.Deserialize>(historyJson!) ?? new() : new List(); history.Add(entry); // Trim to max entries and evict entries outside window var cutoff = recordedAt.AddMinutes(-_options.WindowMinutes); history = history .Where(e => e.RecordedAt >= cutoff) .TakeLast(_options.MaxHistoryEntries) .ToList(); await cache.StringSetAsync( key, JsonSerializer.Serialize(history), TimeSpan.FromSeconds(_options.HistoryTtlSeconds)); if (history.Count < 2) return new TrendResult(TrendOutcome.InsufficientHistory, observationCode); var rate = TrendCalculator.ComputeRatePerMinute(history, _options.WindowMinutes); if (rate is null) return new TrendResult(TrendOutcome.InsufficientHistory, observationCode); if (!_options.RateThresholdsPerMinute.TryGetValue(observationCode, out var threshold)) return new TrendResult(TrendOutcome.Stable, observationCode, rate); if (!TrendCalculator.ExceedsThreshold(observationCode, rate.Value, threshold)) return new TrendResult(TrendOutcome.Stable, observationCode, rate); var details = TrendCalculator.DescribeTrend(observationCode, rate.Value, value); var created = await TryCreateAlertAsync( encounterId, patientId, observationCode, details, rate.Value, ct); return new TrendResult( created ? TrendOutcome.RapidDeterioration : TrendOutcome.AlertAlreadyOpen, observationCode, rate, created); } private async Task TryCreateAlertAsync( Guid encounterId, Guid patientId, string observationCode, string details, decimal ratePerMinute, CancellationToken ct) { using var scope = _services.CreateScope(); var db = scope.ServiceProvider.GetRequiredService(); await using var tx = await db.Database.BeginTransactionAsync(ct); var alertId = Guid.NewGuid(); var triggeredAt = DateTimeOffset.UtcNow; var fullDetails = $"{details} — velocity {ratePerMinute:F2}/min over {_options.WindowMinutes}min window."; var affected = await db.Database.ExecuteSqlInterpolatedAsync($""" INSERT INTO clinical_alerts (id, encounter_id, patient_id, alert_type, severity, details, observation_code, status, triggered_at) SELECT {alertId}, {encounterId}, {patientId}, 'RAPID_DETERIORATION', 'WARNING', {fullDetails}, {observationCode}, 'OPEN', {triggeredAt} WHERE NOT EXISTS ( SELECT 1 FROM clinical_alerts WHERE encounter_id = {encounterId} AND alert_type = 'RAPID_DETERIORATION' AND observation_code = {observationCode} AND status IN ('OPEN', 'ESCALATED') ) """, ct); if (affected == 0) { await tx.RollbackAsync(ct); return false; } db.OutboxEvents.Add(new OutboxEvent { Id = Guid.NewGuid(), Topic = "alert.generated", Payload = JsonSerializer.Serialize(new { alertId, encounterId, patientId, alertType = "RAPID_DETERIORATION", severity = "Warning", details = fullDetails, triggeredAt, observationCode, ratePerMinute, partitionKey = encounterId.ToString() }), PartitionKey = encounterId.ToString(), CreatedAt = DateTimeOffset.UtcNow }); await db.SaveChangesAsync(ct); await tx.CommitAsync(ct); _metrics.TrendAlertsTotal.WithLabels(observationCode).Inc(); _metrics.ClinicalAlertsTotal .WithLabels("RAPID_DETERIORATION", "Warning").Inc(); return true; } }