vigilcare-clinical/VigilCareClinicalAPI/Trend/TrendDetector.cs

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<TrendDetector> _logger;

    public TrendDetector(
        ILogger<TrendDetector> logger, 
        ClinicalMetrics metrics,
        IServiceProvider services, 
        IConnectionMultiplexer redis,
        IOptions<TrendDetectionOptions> options)
    {
        _logger = logger;
        _metrics = metrics;
        _services = services;
        _redis = redis;
        _options = options.Value;
    }

    public async Task<TrendResult> 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<List<TrendHistoryEntry>>(historyJson!) ?? new()
            : new List<TrendHistoryEntry>();

        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<bool> TryCreateAlertAsync(
        Guid encounterId, Guid patientId,
        string observationCode, string details, decimal ratePerMinute,
        CancellationToken ct)
    {
        using var scope = _services.CreateScope();
        var db = scope.ServiceProvider.GetRequiredService<AppDbContext>();

        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;
    }
}