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 _logger; private readonly ClinicalMetrics _metrics; public SirsDetector( IConnectionMultiplexer redis, IServiceProvider services, ILogger logger, ClinicalMetrics metrics) { _redis = redis; _services = services; _logger = logger; _metrics = metrics; } public async Task 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 TryCreateAlertAsync( Guid encounterId, Guid patientId, int activeCount, 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 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(); await handler.OnSepsisAlertCreatedAsync(encounterId, alertId, AlertType.SepsisWarning, ct); return true; } }