vigilcare-clinical/VigilCareClinicalAPI/Sepsis/SirsDetector.cs

using System.Text.Json;
using Microsoft.EntityFrameworkCore;
using Serilog.Context;
using StackExchange.Redis;

public class SirsDetector
{
    // 30 minutes in seconds. This is a clinical parameter: SIRS criteria evaluated
    // outside a 30-minute window are clinically stale. The TTL enforces the window
    // automatically — no cleanup job required.
    private const int SirsTtlSeconds = 1800;

    private readonly IConnectionMultiplexer _redis;
    private readonly IServiceProvider _services;
    private readonly ILogger<SirsDetector> _logger;
    private readonly ClinicalMetrics _metrics;

    public SirsDetector(
        IConnectionMultiplexer redis,
        IServiceProvider services,
        ILogger<SirsDetector> logger,
        ClinicalMetrics metrics)
    {
        _redis = redis;
        _services = services;
        _logger = logger;
        _metrics = metrics;
    }

    public async Task<SirsResult> ProcessObservationAsync(
        Guid encounterId,
        Guid patientId,
        string observationCode,
        decimal value,
        CancellationToken ct = default)
    {
        // Fast exit for non-SIRS codes. The sepsis engine subscribes to the full
        // observation.recorded stream — the majority of messages (SpO2, potassium, glucose)
        // are not SIRS-relevant and are discarded here without touching Redis or PostgreSQL.
        if (!SirsEvaluator.SirsCodes.Contains(observationCode))
            return SirsResult.NotSirsCode;

        var cache = _redis.GetDatabase();
        var key = SirsEvaluator.CriterionKey(encounterId, observationCode);

        if (SirsEvaluator.MeetsCriterion(observationCode, value))
        {
            // SET with EX refreshes the TTL on every qualifying observation.
            // A patient with tachycardia posting a reading every 60 seconds will keep
            // sirs:{id}:HEART_RATE alive for 30 minutes after the LAST qualifying reading,
            // not the first — the window slides forward with each new abnormal value.
            await cache.StringSetAsync(key, "1", TimeSpan.FromSeconds(SirsTtlSeconds));

            _logger.LogDebug("SIRS criterion set: {Key} (TTL={Ttl}s)", key, SirsTtlSeconds);
        }
        else
        {
            // Criterion no longer met — remove the key immediately rather than waiting
            // for TTL expiry. If a patient's temperature normalises at 37.0 °C, the
            // fever criterion must stop contributing to the count right away.
            // Without this DEL, a recovered criterion could persist for up to 30 minutes
            // and falsely sustain a SEPSIS_WARNING count.
            await cache.KeyDeleteAsync(key);

            _logger.LogDebug("SIRS criterion cleared: {Key}", key);
        }

        // Count active criteria in one MGET round-trip.
        // MGET is O(N) where N = number of keys requested (4 here, always).
        // Never use KEYS pattern for this check: KEYS scans the entire keyspace
        // and blocks all other Redis operations until the scan completes.
        var allKeys = SirsEvaluator.AllCriterionKeys(encounterId);
        var values = await cache.StringGetAsync(allKeys);
        var activeCount = values.Count(v => v.HasValue);

        _logger.LogDebug(
            "SIRS state for encounter {Id}: {Active}/4 criteria active after {Code}={Value}",
            encounterId, activeCount, observationCode, value);

        if (activeCount < 2)
            return SirsResult.InsufficientCriteria(activeCount);

        // Two or more criteria are active — attempt to create the alert.
        var created = await TryCreateAlertAsync(encounterId, patientId, activeCount, ct);
        return created ? SirsResult.AlertCreated : SirsResult.AlertAlreadyOpen;
    }

    // Creates the SEPSIS_WARNING alert and its outbox event in one atomic transaction.
    // The INSERT WHERE NOT EXISTS pattern makes this safe under at-least-once delivery:
    // if the consumer crashes after the INSERT but before committing the Kafka offset,
    // the observation is reprocessed on restart. The second run hits the WHERE NOT EXISTS
    // subquery, finds the existing open alert, inserts 0 rows, and returns false — no
    // duplicate alert, no duplicate outbox event.
    private async Task<bool> TryCreateAlertAsync(
        Guid encounterId,
        Guid patientId,
        int activeCount,
        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 details =
            $"SIRS criteria met: {activeCount} of 4 criteria active within the 30-minute window.";

        // One SQL round-trip: check + insert atomically.
        // status IN ('OPEN', 'ESCALATED') prevents re-creating an alert that has been
        // escalated but not yet resolved — the patient is still in danger.
        var affected = await db.Database.ExecuteSqlInterpolatedAsync($"""
            INSERT INTO clinical_alerts
                (id, encounter_id, patient_id, alert_type, severity, details, status, triggered_at)
            SELECT {alertId}, {encounterId}, {patientId},
                   'SEPSIS_WARNING', 'CRITICAL', {details}, 'OPEN', {triggeredAt}
            WHERE NOT EXISTS (
                SELECT 1 FROM clinical_alerts
                WHERE  encounter_id = {encounterId}
                  AND  alert_type   = 'SEPSIS_WARNING'
                  AND  status       IN ('OPEN', 'ESCALATED')
            )
            """, ct);

        if (affected == 0)
        {
            await tx.RollbackAsync(ct);
            _logger.LogDebug(
                "SEPSIS_WARNING already open for encounter {Id} — no new alert", encounterId);
            return false;
        }

        // Alert was created — write the outbox event in the same transaction.
        // The relay (Phase 3) will publish to alert.generated, which Phase 6's
        // notification worker reads to page the attending physician via RabbitMQ.
        db.OutboxEvents.Add(new OutboxEvent
        {
            Id = Guid.NewGuid(),
            Topic = "alert.generated",
            Payload = JsonSerializer.Serialize(new
            {
                alertId,
                encounterId,
                patientId,
                alertType = AlertType.SepsisWarning.ToDbString(),
                severity  = "Critical",
                details,
                triggeredAt,
                partitionKey = encounterId.ToString()
            }),
            PartitionKey = encounterId.ToString(),
            CreatedAt = DateTimeOffset.UtcNow
        });

        await db.SaveChangesAsync(ct);
        await tx.CommitAsync(ct);

        _metrics.SirsDetectionsTotal.Inc();
        _metrics.ClinicalAlertsTotal
            .WithLabels(AlertType.SepsisWarning.ToDbString(), AlertSeverity.Critical.ToDbString())
            .Inc();

        using (LogContext.PushProperty("EncounterId", encounterId))
        using (LogContext.PushProperty("PatientId", patientId))
        {
            _logger.LogWarning(
                "SEPSIS_WARNING created. ActiveCriteriaCount={Count} AlertId={AlertId}",
                activeCount, alertId);
        }

        var handler = scope.ServiceProvider.GetRequiredService<SepsisAlertHandler>();
        await handler.OnSepsisAlertCreatedAsync(encounterId, alertId, AlertType.SepsisWarning, ct);

        return true;
    }
}