diff --git a/VigilCareClinicalAPI.Tests/SirsDetectorTests.cs b/VigilCareClinicalAPI.Tests/SirsDetectorTests.cs new file mode 100644 index 0000000..f3c46dd --- /dev/null +++ b/VigilCareClinicalAPI.Tests/SirsDetectorTests.cs @@ -0,0 +1,214 @@ +using FluentAssertions; +using Microsoft.EntityFrameworkCore; +using Microsoft.Extensions.DependencyInjection; +using StackExchange.Redis; + +[Collection("Integration")] +public class SirsDetectorTests : IAsyncLifetime +{ + private readonly ApiFixture _fixture; + private Guid _encounterId; + private Guid _patientId; + + public SirsDetectorTests(ApiFixture fixture) => _fixture = fixture; + + public async Task InitializeAsync() + { + // Reset PostgreSQL test data + using var scope = _fixture.Services.CreateScope(); + var db = scope.ServiceProvider.GetRequiredService(); + await DbResetHelper.ResetAsync(db); + + var patient = new Patient + { + Id = Guid.NewGuid(), Mrn = "MRN-SIRS-001", FirstName = "SIRS", LastName = "Test", + DateOfBirth = new DateOnly(1960, 1, 1), Gender = "M", + CreatedAt = DateTimeOffset.UtcNow + }; + var encounter = new Encounter + { + Id = Guid.NewGuid(), PatientId = patient.Id, EncounterType = EncounterType.Inpatient, + Status = EncounterStatus.Active, Department = "ICU", + AttendingPhysician = "Dr. SIRS", AdmittedAt = DateTimeOffset.UtcNow, + CreatedAt = DateTimeOffset.UtcNow + }; + db.Patients.Add(patient); + db.Encounters.Add(encounter); + await db.SaveChangesAsync(); + + _patientId = patient.Id; + _encounterId = encounter.Id; + + // Flush all SIRS keys for this encounter from Redis + var redis = scope.ServiceProvider.GetRequiredService(); + var cache = redis.GetDatabase(); + foreach (var key in SirsEvaluator.AllCriterionKeys(_encounterId)) + await cache.KeyDeleteAsync(key); + } + + public Task DisposeAsync() => Task.CompletedTask; + + private SirsDetector CreateDetector() + { + using var scope = _fixture.Services.CreateScope(); + return scope.ServiceProvider.GetRequiredService(); + } + + // Test 1: one criterion met — insufficient for alert + [Fact] + public async Task OneCriterion_NoAlert() + { + using var scope = _fixture.Services.CreateScope(); + var detector = scope.ServiceProvider.GetRequiredService(); + + var result = await detector.ProcessObservationAsync( + _encounterId, _patientId, "HEART_RATE", 95m); + + result.Outcome.Should().Be(SirsOutcome.InsufficientCriteria); + result.ActiveCount.Should().Be(1); + + // Verify Redis key was set + var redis = scope.ServiceProvider.GetRequiredService(); + var ttl = await redis.GetDatabase() + .KeyTimeToLiveAsync(SirsEvaluator.CriterionKey(_encounterId, "HEART_RATE")); + ttl.Should().NotBeNull().And.BeGreaterThan(TimeSpan.Zero); + } + + // Test 2: two criteria met — SEPSIS_WARNING alert created + [Fact] + public async Task TwoCriteriaMet_AlertCreated() + { + using var scope = _fixture.Services.CreateScope(); + var detector = scope.ServiceProvider.GetRequiredService(); + var db = scope.ServiceProvider.GetRequiredService(); + + // Tachycardia + var r1 = await detector.ProcessObservationAsync( + _encounterId, _patientId, "HEART_RATE", 95m); + r1.Outcome.Should().Be(SirsOutcome.InsufficientCriteria); + + // Fever — now count = 2 → alert + var r2 = await detector.ProcessObservationAsync( + _encounterId, _patientId, "TEMP_C", 38.5m); + r2.Outcome.Should().Be(SirsOutcome.AlertCreated); + + // Verify clinical_alert row in PostgreSQL + var alert = await db.ClinicalAlerts.SingleAsync(); + alert.AlertType.Should().Be(AlertType.SepsisWarning); + alert.Severity.Should().Be(AlertSeverity.Critical); + alert.Status.Should().Be(AlertStatus.Open); + alert.EncounterId.Should().Be(_encounterId); + alert.PatientId.Should().Be(_patientId); + + // Verify outbox event was written in the same transaction + var outbox = await db.OutboxEvents.SingleAsync(); + outbox.Topic.Should().Be("alert.generated"); + outbox.PartitionKey.Should().Be(_encounterId.ToString()); + } + + // Test 3: third criterion met while alert already open — no second alert + [Fact] + public async Task ThreeCriteriaMet_NoSecondAlert() + { + using var scope = _fixture.Services.CreateScope(); + var detector = scope.ServiceProvider.GetRequiredService(); + var db = scope.ServiceProvider.GetRequiredService(); + + await detector.ProcessObservationAsync(_encounterId, _patientId, "HEART_RATE", 95m); + await detector.ProcessObservationAsync(_encounterId, _patientId, "TEMP_C", 38.5m); + // Third criterion (tachypnea) — alert already open + var r3 = await detector.ProcessObservationAsync( + _encounterId, _patientId, "RESP_RATE", 22m); + + r3.Outcome.Should().Be(SirsOutcome.AlertAlreadyOpen); + + // Still exactly one alert — the WHERE NOT EXISTS prevented a duplicate + var alertCount = await db.ClinicalAlerts.CountAsync(); + alertCount.Should().Be(1, "a second SEPSIS_WARNING must not be created while one is already open"); + } + + // Test 4: criterion clears — Redis key deleted, alert remains open + // + // This is the most important test in Phase 5. It verifies: + // (a) The DEL path works when a criterion is no longer met. + // (b) Clearing a criterion does not resolve the existing alert — the clinical + // workflow requires explicit acknowledgment. A patient whose temperature + // normalises may still be septic; the alert is for the clinician to evaluate. + [Fact] + public async Task CriterionClears_KeyDeleted_AlertRemainsOpen() + { + using var scope = _fixture.Services.CreateScope(); + var detector = scope.ServiceProvider.GetRequiredService(); + var db = scope.ServiceProvider.GetRequiredService(); + var redis = scope.ServiceProvider.GetRequiredService(); + var cache = redis.GetDatabase(); + + // Establish two criteria and create the alert + await detector.ProcessObservationAsync(_encounterId, _patientId, "HEART_RATE", 95m); + await detector.ProcessObservationAsync(_encounterId, _patientId, "TEMP_C", 38.5m); + + // Temperature normalises + var result = await detector.ProcessObservationAsync( + _encounterId, _patientId, "TEMP_C", 37.0m); + + // TEMP_C key must be gone from Redis + var tempKeyExists = await cache.KeyExistsAsync( + SirsEvaluator.CriterionKey(_encounterId, "TEMP_C")); + tempKeyExists.Should().BeFalse("a cleared criterion must be deleted from Redis immediately"); + + // HEART_RATE key must still exist (criterion still met) + var hrKeyExists = await cache.KeyExistsAsync( + SirsEvaluator.CriterionKey(_encounterId, "HEART_RATE")); + hrKeyExists.Should().BeTrue("an active criterion must remain until its own TTL or a clearing observation"); + + // Active count is now 1, but alert must stay open + result.Outcome.Should().Be(SirsOutcome.InsufficientCriteria); + result.ActiveCount.Should().Be(1); + + var alert = await db.ClinicalAlerts.SingleAsync(); + alert.Status.Should().Be(AlertStatus.Open, + "the existing alert is not auto-resolved when criteria drop below 2 — " + + "clinical acknowledgment is required"); + } + + // Test 5: at-least-once redelivery — idempotency under simulated crash + [Fact] + public async Task DuplicateObservationEvent_AlertCreatedOnce() + { + using var scope = _fixture.Services.CreateScope(); + var detector = scope.ServiceProvider.GetRequiredService(); + var db = scope.ServiceProvider.GetRequiredService(); + + await detector.ProcessObservationAsync(_encounterId, _patientId, "HEART_RATE", 95m); + + // Simulate: the consumer crashes after this call completes but before committing + // the Kafka offset. On restart, the same observation is redelivered. + await detector.ProcessObservationAsync(_encounterId, _patientId, "TEMP_C", 38.5m); + await detector.ProcessObservationAsync(_encounterId, _patientId, "TEMP_C", 38.5m); // duplicate + + var alertCount = await db.ClinicalAlerts.CountAsync(); + alertCount.Should().Be(1, "WHERE NOT EXISTS prevents duplicate alert on redelivery"); + + var outboxCount = await db.OutboxEvents.CountAsync(); + outboxCount.Should().Be(1, "outbox event must be written exactly once"); + } + + // Test 6: non-SIRS code — engine ignores it entirely + [Fact] + public async Task NonSirsCode_NoRedisInteraction() + { + using var scope = _fixture.Services.CreateScope(); + var detector = scope.ServiceProvider.GetRequiredService(); + var redis = scope.ServiceProvider.GetRequiredService(); + + var result = await detector.ProcessObservationAsync( + _encounterId, _patientId, "POTASSIUM_MEQ_L", 3.2m); + + result.Outcome.Should().Be(SirsOutcome.NotSirsCode); + + // No SIRS keys were created for this encounter + var anyKey = await redis.GetDatabase() + .KeyExistsAsync(SirsEvaluator.CriterionKey(_encounterId, "POTASSIUM_MEQ_L")); + anyKey.Should().BeFalse(); + } +} \ No newline at end of file diff --git a/VigilCareClinicalAPI.Tests/SirsEvaluatorTests.cs b/VigilCareClinicalAPI.Tests/SirsEvaluatorTests.cs new file mode 100644 index 0000000..9be202e --- /dev/null +++ b/VigilCareClinicalAPI.Tests/SirsEvaluatorTests.cs @@ -0,0 +1,30 @@ +using FluentAssertions; + +public class SirsEvaluatorTests +{ + [Theory] + [InlineData("TEMP_C", 38.4, true)] // fever + [InlineData("TEMP_C", 35.9, true)] // hypothermia + [InlineData("TEMP_C", 37.0, false)] // normal + [InlineData("HEART_RATE", 91, true)] + [InlineData("HEART_RATE", 90, false)] // boundary: 90 is NOT tachycardia (> not >=) + [InlineData("RESP_RATE", 21, true)] + [InlineData("RESP_RATE", 20, false)] // boundary + [InlineData("WBC_K_UL", 12.1, true)] // leukocytosis + [InlineData("WBC_K_UL", 3.9, true)] // leukopenia + [InlineData("WBC_K_UL", 8.0, false)] // normal + [InlineData("POTASSIUM_MEQ_L", 4.0, false)] // not a SIRS code + public void MeetsCriterion_ReturnsExpected(string code, double value, bool expected) + { + SirsEvaluator.MeetsCriterion(code, (decimal)value).Should().Be(expected); + } + + [Fact] + public void AllCriterionKeys_ReturnsFourKeys_AllDistinct() + { + var id = Guid.NewGuid(); + var keys = SirsEvaluator.AllCriterionKeys(id); + keys.Should().HaveCount(4); + keys.Select(k => k.ToString()).Should().OnlyHaveUniqueItems(); + } +} \ No newline at end of file diff --git a/VigilCareClinicalAPI/BackgroundServices/SepsisEngineService.cs b/VigilCareClinicalAPI/BackgroundServices/SepsisEngineService.cs new file mode 100644 index 0000000..e9995be --- /dev/null +++ b/VigilCareClinicalAPI/BackgroundServices/SepsisEngineService.cs @@ -0,0 +1,98 @@ +using System.Text.Json; +using Confluent.Kafka; +using Microsoft.Extensions.Options; + +public class SepsisEngineService : BackgroundService +{ + private static readonly JsonSerializerOptions EventJsonOptions = new() + { + PropertyNameCaseInsensitive = true + }; + + private readonly IServiceProvider _services; + private readonly KafkaOptions _kafkaOptions; + private readonly ILogger _logger; + + public SepsisEngineService( + IServiceProvider services, + IOptions kafkaOptions, + ILogger logger) + { + _services = services; + _kafkaOptions = kafkaOptions.Value; + _logger = logger; + } + + protected override async Task ExecuteAsync(CancellationToken stoppingToken) + { + var config = new ConsumerConfig + { + BootstrapServers = _kafkaOptions.BootstrapServers, + GroupId = "sepsis-engine", + AutoOffsetReset = AutoOffsetReset.Earliest, + EnableAutoCommit = false + }; + + using var consumer = new ConsumerBuilder(config).Build(); + // Subscribes to observation.recorded only. + // The es-indexer consumes all three topics; the sepsis engine only needs one. + // Subscribing to a superset of needed topics would waste CPU deserializing + // alert and encounter events that this engine discards immediately. + consumer.Subscribe(_kafkaOptions.Topics.ObservationRecorded); + + _logger.LogInformation("SepsisEngineService started — consumer group: sepsis-engine"); + + try + { + while (!stoppingToken.IsCancellationRequested) + { + ConsumeResult? result = null; + try + { + result = consumer.Consume(stoppingToken); + + var evt = JsonSerializer.Deserialize( + result.Message.Value, EventJsonOptions)!; + + // Create a scope per message — SirsDetector is scoped and + // owns a fresh DbContext for each observation processed. + using var scope = _services.CreateScope(); + var detector = scope.ServiceProvider.GetRequiredService(); + + var outcome = await detector.ProcessObservationAsync( + evt.EncounterId, + evt.PatientId, + evt.ObservationCode, + evt.Value, + stoppingToken); + + if (outcome.Outcome == SirsOutcome.AlertCreated) + _logger.LogWarning( + "SEPSIS_WARNING created via SepsisEngine " + + "— encounter={EncounterId} code={Code} value={Value}", + evt.EncounterId, evt.ObservationCode, evt.Value); + + // Commit only after successful processing. + consumer.Commit(result); + } + catch (OperationCanceledException) + { + break; + } + catch (Exception ex) + { + _logger.LogError(ex, + "SepsisEngine failed on topic={Topic} offset={Offset} — not committing", + result?.Topic, result?.Offset.Value); + // Back off before retrying so a persistent failure (e.g., Redis down) + // does not spin the loop at maximum throughput. + await Task.Delay(2000, stoppingToken); + } + } + } + finally + { + consumer.Close(); + } + } +} \ No newline at end of file diff --git a/VigilCareClinicalAPI/Data/Seed/DataSeeder.cs b/VigilCareClinicalAPI/Data/Seed/DataSeeder.cs index a804d3b..3d0d689 100644 --- a/VigilCareClinicalAPI/Data/Seed/DataSeeder.cs +++ b/VigilCareClinicalAPI/Data/Seed/DataSeeder.cs @@ -62,6 +62,20 @@ public static class DataSeeder Id = Guid.NewGuid(), ObservationCode = "SPO2", DisplayName = "Oxygen Saturation", Unit = "%", CriticalLow = 88, WarningLow = 92, WarningHigh = null, CriticalHigh = null, CreatedAt = DateTimeOffset.UtcNow + }, + new AlertThreshold + { + Id = Guid.NewGuid(), ObservationCode = "RESP_RATE", + DisplayName = "Respiratory Rate", Unit = "breaths/min", + CriticalLow = null, WarningLow = 12m, WarningHigh = 20m, CriticalHigh = 30m, + CreatedAt = DateTimeOffset.UtcNow + }, + new AlertThreshold + { + Id = Guid.NewGuid(), ObservationCode = "WBC_K_UL", + DisplayName = "White Blood Cell Count", Unit = "k/µL", + CriticalLow = 2.0m, WarningLow = 4.0m, WarningHigh = 12.0m, CriticalHigh = 20.0m, + CreatedAt = DateTimeOffset.UtcNow } }; db.AlertThresholds.AddRange(thresholds); diff --git a/VigilCareClinicalAPI/Models/Records/Sepsis/SepsisObservationEvent.cs b/VigilCareClinicalAPI/Models/Records/Sepsis/SepsisObservationEvent.cs new file mode 100644 index 0000000..bdf7a6d --- /dev/null +++ b/VigilCareClinicalAPI/Models/Records/Sepsis/SepsisObservationEvent.cs @@ -0,0 +1,6 @@ +// Subset of the observation.recorded payload — only the fields the sepsis engine needs. +public record SepsisObservationEvent( + Guid EncounterId, + Guid PatientId, + string ObservationCode, + decimal Value); \ No newline at end of file diff --git a/VigilCareClinicalAPI/Models/Records/Sepsis/SirsOutcome.cs b/VigilCareClinicalAPI/Models/Records/Sepsis/SirsOutcome.cs new file mode 100644 index 0000000..d1e1705 --- /dev/null +++ b/VigilCareClinicalAPI/Models/Records/Sepsis/SirsOutcome.cs @@ -0,0 +1,7 @@ +public enum SirsOutcome +{ + NotSirsCode, + InsufficientCriteria, + AlertCreated, + AlertAlreadyOpen +} \ No newline at end of file diff --git a/VigilCareClinicalAPI/Models/Records/Sepsis/SirsResult.cs b/VigilCareClinicalAPI/Models/Records/Sepsis/SirsResult.cs new file mode 100644 index 0000000..3719fbc --- /dev/null +++ b/VigilCareClinicalAPI/Models/Records/Sepsis/SirsResult.cs @@ -0,0 +1,11 @@ +// Discriminated result — allows tests and callers to assert the exact outcome +// without inspecting PostgreSQL or Redis directly. +public record SirsResult(SirsOutcome Outcome, int ActiveCount = 0) +{ + public static readonly SirsResult NotSirsCode = new(SirsOutcome.NotSirsCode); + public static readonly SirsResult AlertCreated = new(SirsOutcome.AlertCreated); + public static readonly SirsResult AlertAlreadyOpen = new(SirsOutcome.AlertAlreadyOpen); + + public static SirsResult InsufficientCriteria(int count) => + new(SirsOutcome.InsufficientCriteria, count); +} \ No newline at end of file diff --git a/VigilCareClinicalAPI/Program.cs b/VigilCareClinicalAPI/Program.cs index 2ca9f70..e79f8ec 100644 --- a/VigilCareClinicalAPI/Program.cs +++ b/VigilCareClinicalAPI/Program.cs @@ -12,10 +12,15 @@ try { var builder = WebApplication.CreateBuilder(args); - builder.Host.UseSerilog((ctx, services, config) => - config.ReadFrom.Configuration(ctx.Configuration) - .ReadFrom.Services(services) - .Enrich.FromLogContext()); + // Serilog's reloadable logger can only be frozen once per process; skip in + // integration tests where WebApplicationFactory may build multiple hosts. + if (!builder.Environment.IsEnvironment("Testing")) + { + builder.Host.UseSerilog((ctx, services, config) => + config.ReadFrom.Configuration(ctx.Configuration) + .ReadFrom.Services(services) + .Enrich.FromLogContext()); + } builder.Services.AddDbContext(opts => opts.UseNpgsql(builder.Configuration.GetConnectionString("DefaultConnection"))); @@ -43,12 +48,14 @@ try builder.Services.AddScoped(); builder.Services.AddScoped(); builder.Services.AddScoped(); + builder.Services.AddScoped(); builder.Services.AddHostedService(); builder.Services.AddHostedService(); builder.Services.AddHostedService(); builder.Services.AddHostedService(); builder.Services.AddHostedService(); + builder.Services.AddHostedService(); builder.Services.AddControllers() .AddJsonOptions(opts => @@ -70,11 +77,14 @@ try await DataSeeder.SeedAsync(db, redis); } - app.UseSerilogRequestLogging(options => + if (!app.Environment.IsEnvironment("Testing")) { - options.MessageTemplate = - "HTTP {RequestMethod} {RequestPath} responded {StatusCode} in {Elapsed:0.000}ms"; - }); + app.UseSerilogRequestLogging(options => + { + options.MessageTemplate = + "HTTP {RequestMethod} {RequestPath} responded {StatusCode} in {Elapsed:0.000}ms"; + }); + } app.UseMiddleware(); app.UseMiddleware(); diff --git a/VigilCareClinicalAPI/Sepsis/SirsDetector.cs b/VigilCareClinicalAPI/Sepsis/SirsDetector.cs new file mode 100644 index 0000000..8b4d5f4 --- /dev/null +++ b/VigilCareClinicalAPI/Sepsis/SirsDetector.cs @@ -0,0 +1,161 @@ +using System.Text.Json; +using Microsoft.EntityFrameworkCore; +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; + + public SirsDetector( + IConnectionMultiplexer redis, + IServiceProvider services, + ILogger logger) + { + _redis = redis; + _services = services; + _logger = logger; + } + + 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", + triggeredAt, + partitionKey = encounterId.ToString() + }), + PartitionKey = encounterId.ToString(), + CreatedAt = DateTimeOffset.UtcNow + }); + + await db.SaveChangesAsync(ct); + await tx.CommitAsync(ct); + + _logger.LogWarning( + "SEPSIS_WARNING alert {AlertId} created for encounter {EncounterId} " + + "— {Active}/4 SIRS criteria active", + alertId, encounterId, activeCount); + + return true; + } +} \ No newline at end of file diff --git a/VigilCareClinicalAPI/Sepsis/SirsEvaluator.cs b/VigilCareClinicalAPI/Sepsis/SirsEvaluator.cs new file mode 100644 index 0000000..3d75fab --- /dev/null +++ b/VigilCareClinicalAPI/Sepsis/SirsEvaluator.cs @@ -0,0 +1,38 @@ +using StackExchange.Redis; + +public static class SirsEvaluator +{ + // The four SIRS codes defined by this project's simplified SIRS criteria. + // Observations for any other code are ignored by the sepsis engine entirely — + // they pass through to the outbox and Elasticsearch but do not affect SIRS state. + public static readonly IReadOnlySet SirsCodes = + new HashSet { "TEMP_C", "HEART_RATE", "RESP_RATE", "WBC_K_UL" }; + + // Returns true if the observation value meets the SIRS criterion for its code. + // These thresholds are clinical parameters, not configuration — changing them + // requires clinical review, not a config file edit. They live here as named constants. + public static bool MeetsCriterion(string observationCode, decimal value) => + observationCode switch + { + // Fever (> 38.3 °C) or hypothermia (< 36.0 °C) + "TEMP_C" => value > 38.3m || value < 36.0m, + // Tachycardia + "HEART_RATE" => value > 90m, + // Tachypnea + "RESP_RATE" => value > 20m, + // Leukocytosis or leukopenia + "WBC_K_UL" => value > 12.0m || value < 4.0m, + _ => false + }; + + // Redis key for one SIRS criterion for one encounter. + public static string CriterionKey(Guid encounterId, string code) => + $"sirs:{encounterId}:{code}"; + + // All four Redis keys for one encounter — used in MGET to count active criteria. + // The order is stable so the MGET result array always maps to the same codes. + public static RedisKey[] AllCriterionKeys(Guid encounterId) => + SirsCodes + .Select(code => (RedisKey)CriterionKey(encounterId, code)) + .ToArray(); +} \ No newline at end of file diff --git a/docs/decisions/sepsis-engine-design.md b/docs/decisions/sepsis-engine-design.md new file mode 100644 index 0000000..7b486f4 --- /dev/null +++ b/docs/decisions/sepsis-engine-design.md @@ -0,0 +1,53 @@ +```markdown +# Sepsis Engine Design Decisions + +## Why Redis for SIRS state, not a PostgreSQL time-range query + +At a medium hospital with 200 concurrent inpatients, each generating five observations +per patient per minute, the sepsis engine processes approximately 17 observation events +per second at steady state. + +A PostgreSQL alternative would look like this on every event: + +```sql +SELECT observation_code +FROM observations +WHERE encounter_id = :encounterId + AND observation_code IN ('TEMP_C', 'HEART_RATE', 'RESP_RATE', 'WBC_K_UL') + AND recorded_at >= NOW() - INTERVAL '30 minutes' + AND ( + (observation_code = 'TEMP_C' AND (value > 38.3 OR value < 36.0)) OR + (observation_code = 'HEART_RATE' AND value > 90) OR + (observation_code = 'RESP_RATE' AND value > 20) OR + (observation_code = 'WBC_K_UL' AND (value > 12.0 OR value < 4.0)) + ); +``` + +This query hits the `observations` table on every event. Under load it competes for +I/O with the ingest path writing new rows — both want the same composite index. With +Redis: four `SET`/`DEL` operations and one `MGET`, all O(1), all in-memory. No disk +I/O, no lock contention with the write path. + +The TTL enforces the 30-minute sliding window automatically. Without Redis (or an +equivalent in-memory store), a background job would be needed to clean up stale +criteria — another failure point, another deployment concern. + +## Why not Apache Flink + +Flink is a distributed stream processor designed for stateful computation at scale +(millions of events per second across a fleet). It brings real costs: + +- A Flink cluster (JobManager + TaskManagers) is infrastructure that must be deployed, + monitored, and upgraded independently of the application. +- Flink state backends (RocksDB, heap) add operational complexity that is not + justified unless the stream volume saturates what a single consumer thread can handle. +- Flink's exactly-once semantics require Kafka transactions, which add latency and + require tuning separate from the rest of the application. + +At a single hospital (200 inpatients, ~17 observations/second), a Kafka consumer + +Redis state store handles the volume with single-digit millisecond latency per event +and no additional infrastructure. The trade-off: if this system needed to scale to a +multi-hospital network with 50,000+ concurrent inpatients (~5,000 observations/second), +Flink would become the right choice. The architecture decision is correct at this scale +and defensible at interview with a clear scale inflection point named. +``` \ No newline at end of file diff --git a/scripts/run-sepsis-sirs-tests.sh b/scripts/run-sepsis-sirs-tests.sh new file mode 100755 index 0000000..945d0b5 --- /dev/null +++ b/scripts/run-sepsis-sirs-tests.sh @@ -0,0 +1,544 @@ +#!/usr/bin/env bash + +set -euo pipefail + +SCRIPT_DIR="$(cd "$(dirname "${BASH_SOURCE[0]}")" && pwd)" +COMPOSE_FILE="${COMPOSE_FILE:-${SCRIPT_DIR}/../docker-compose.yml}" + +BASE_URL="${BASE_URL:-http://localhost:5270}" +REDIS_PORT="${REDIS_PORT:-6382}" +ES_URL="${ES_URL:-http://localhost:9200}" +PGHOST="${PGHOST:-localhost}" +PGPORT="${PGPORT:-5436}" +PGDATABASE="${PGDATABASE:-vigilcare}" +PGUSER="${PGUSER:-postgres}" +PGPASSWORD="${PGPASSWORD:-password}" + +SEPSIS_CONSUMER_GROUP="${SEPSIS_CONSUMER_GROUP:-sepsis-engine}" +ES_CONSUMER_GROUP="${ES_CONSUMER_GROUP:-es-indexer}" +SEPSIS_WAIT_SECS="${SEPSIS_WAIT_SECS:-45}" +INDEX_WAIT_SECS="${INDEX_WAIT_SECS:-60}" +RELAY_WAIT_SECS="${RELAY_WAIT_SECS:-45}" +KAFKA_READY_WAIT_SECS="${KAFKA_READY_WAIT_SECS:-30}" +TTL_DECAY_WAIT_SECS="${TTL_DECAY_WAIT_SECS:-5}" + +SCRIPT_RUN_ID="$(date -u +"%Y%m%d%H%M%S")" +RECORDED_AT="$(date -u +"%Y-%m-%dT%H:%M:%SZ")" + +TMP_FILES=() + +cleanup() { + local f + for f in "${TMP_FILES[@]}"; do + rm -f "${f}" "${f}.status" 2>/dev/null || true + done +} +trap cleanup EXIT + +if ! command -v curl >/dev/null 2>&1; then + echo "Missing dependency: curl" + exit 1 +fi + +if ! command -v jq >/dev/null 2>&1; then + echo "Missing dependency: jq" + exit 1 +fi + +if ! command -v docker >/dev/null 2>&1 || [[ ! -f "${COMPOSE_FILE}" ]]; then + echo "Missing dependency: docker compose (${COMPOSE_FILE})" + exit 1 +fi + +compose() { + docker compose -f "${COMPOSE_FILE}" "$@" +} + +kafka_exec() { + compose exec -T kafka "$@" +} + +redis_cmd() { + if command -v redis-cli >/dev/null 2>&1; then + redis-cli -p "${REDIS_PORT}" "$@" + else + compose exec -T redis redis-cli "$@" + fi +} + +psql_cmd() { + local sql="$1" + if command -v psql >/dev/null 2>&1; then + PGPASSWORD="${PGPASSWORD}" psql -h "${PGHOST}" -p "${PGPORT}" -U "${PGUSER}" -d "${PGDATABASE}" -tAc "${sql}" + else + compose exec -T postgres psql -U "${PGUSER}" -d "${PGDATABASE}" -tAc "${sql}" + fi +} + +request() { + local method="$1" + local url="$2" + local body="${3:-}" + local tmp_body + tmp_body="$(mktemp)" + TMP_FILES+=("${tmp_body}") + local status + + if [[ -n "${body}" ]]; then + status="$(curl -sS -o "${tmp_body}" -w "%{http_code}" -X "${method}" "${url}" \ + -H "Content-Type: application/json" -d "${body}")" + else + status="$(curl -sS -o "${tmp_body}" -w "%{http_code}" -X "${method}" "${url}")" + fi + + echo "${status}" > "${tmp_body}.status" + echo "${tmp_body}" +} + +assert_status() { + local expected="$1" + local body_file="$2" + local status + status="$(cat "${body_file}.status")" + if [[ "${status}" != "${expected}" ]]; then + echo "Expected HTTP ${expected}, got ${status}" + echo "Response body:" + cat "${body_file}" + echo + return 1 + fi +} + +ingest_observation() { + local encounter_id="$1" + local code="$2" + local value="$3" + local unit="$4" + local source="$5" + local recorded_at="$6" + local idempotency_key="$7" + + local payload + payload="$(jq -nc \ + --arg code "${code}" \ + --argjson value "${value}" \ + --arg unit "${unit}" \ + --arg source "${source}" \ + --arg recordedAt "${recorded_at}" \ + --arg key "${idempotency_key}" \ + '{observations:[{observationCode:$code,value:$value,unit:$unit,source:$source,recordedAt:$recordedAt,idempotencyKey:$key}]}')" + + local resp + resp="$(request POST "${BASE_URL}/api/v1/encounters/${encounter_id}/observations" "${payload}")" + assert_status "201" "${resp}" +} + +wait_for_kafka() { + local elapsed=0 + while (( elapsed < KAFKA_READY_WAIT_SECS )); do + if kafka_exec /opt/kafka/bin/kafka-broker-api-versions.sh --bootstrap-server localhost:9092 >/dev/null 2>&1; then + return 0 + fi + sleep 2 + elapsed=$((elapsed + 2)) + done + echo "Kafka did not become ready within ${KAFKA_READY_WAIT_SECS}s" + return 1 +} + +consumer_group_lag() { + local group="$1" + kafka_exec /opt/kafka/bin/kafka-consumer-groups.sh \ + --bootstrap-server localhost:9092 \ + --describe \ + --group "${group}" 2>/dev/null | \ + awk '/observation.recorded/ { sum += $6 } END { print sum + 0 }' +} + +wait_for_consumer_lag_zero() { + local group="$1" + local elapsed=0 + while (( elapsed < SEPSIS_WAIT_SECS )); do + local lag + lag="$(consumer_group_lag "${group}")" + if [[ "${lag}" == "0" ]]; then + return 0 + fi + sleep 1 + elapsed=$((elapsed + 1)) + done + echo "Consumer group ${group} lag did not reach zero within ${SEPSIS_WAIT_SECS}s (lag=${lag:-unknown})" + return 1 +} + +wait_for_sepsis_alert() { + local encounter_id="$1" + local elapsed=0 + while (( elapsed < SEPSIS_WAIT_SECS )); do + local count + count="$(psql_cmd "SELECT COUNT(*) FROM clinical_alerts WHERE encounter_id = '${encounter_id}' AND alert_type = 'SEPSIS_WARNING' AND status = 'OPEN'")" + if [[ "${count}" == "1" ]]; then + return 0 + fi + sleep 1 + elapsed=$((elapsed + 1)) + done + echo "SEPSIS_WARNING alert not found for encounter ${encounter_id} within ${SEPSIS_WAIT_SECS}s" + return 1 +} + +wait_for_outbox_processed() { + local outbox_id="$1" + local elapsed=0 + while (( elapsed < RELAY_WAIT_SECS )); do + local processed + processed="$(psql_cmd "SELECT processed_at IS NOT NULL FROM outbox_events WHERE id = '${outbox_id}'")" + if [[ "${processed}" == "t" ]]; then + return 0 + fi + sleep 1 + elapsed=$((elapsed + 1)) + done + echo "Outbox row ${outbox_id} was not processed within ${RELAY_WAIT_SECS}s" + return 1 +} + +latest_observation_outbox_id() { + local encounter_id="$1" + psql_cmd "SELECT id FROM outbox_events WHERE topic = 'observation.recorded' AND partition_key = '${encounter_id}' ORDER BY created_at DESC LIMIT 1" +} + +# Observation ingest writes outbox first; relay publishes to Kafka; sepsis-engine consumes. +# Consumer lag alone is not enough — lag can be 0 before the relay publishes the new event. +wait_for_observation_pipeline() { + local encounter_id="$1" + local outbox_id + outbox_id="$(latest_observation_outbox_id "${encounter_id}")" + if [[ -z "${outbox_id}" ]]; then + echo "No observation.recorded outbox row for encounter ${encounter_id}" + return 1 + fi + wait_for_outbox_processed "${outbox_id}" + wait_for_consumer_lag_zero "${SEPSIS_CONSUMER_GROUP}" +} + +redis_get() { + local value + value="$(redis_cmd GET "$1" | tr -d '\r')" + if [[ "${value}" == "(nil)" ]]; then + echo "" + else + echo "${value}" + fi +} + +redis_ttl() { + redis_cmd TTL "$1" | tr -d '\r' +} + +sirs_key() { + local encounter_id="$1" + local code="$2" + echo "sirs:${encounter_id}:${code}" +} + +TOTAL_STEPS=10 + +echo "Running sepsis / SIRS verification against ${BASE_URL}" +echo "Script run id: ${SCRIPT_RUN_ID}" + +echo "" +echo "[0/${TOTAL_STEPS}] Preflight — API, Postgres, Redis, Kafka, and Elasticsearch reachable" +preflight_status="$(curl -sS -o /dev/null -w "%{http_code}" "${BASE_URL}/api/v1/alert-thresholds" || true)" +if [[ "${preflight_status}" != "200" ]]; then + echo "API not reachable at ${BASE_URL} (HTTP ${preflight_status})." + echo "Start the API with: dotnet run --project VigilCareClinicalAPI" + exit 1 +fi + +if ! psql_cmd "SELECT 1" >/dev/null 2>&1; then + echo "Postgres not reachable on ${PGHOST}:${PGPORT}." + echo "Start the stack with: docker compose up -d" + exit 1 +fi + +if ! redis_cmd PING >/dev/null 2>&1; then + echo "Redis not reachable on port ${REDIS_PORT}." + exit 1 +fi + +if ! wait_for_kafka; then + exit 1 +fi + +es_health_status="$(curl -sS "${ES_URL}/_cluster/health" | jq -r '.status' || true)" +if [[ "${es_health_status}" != "green" && "${es_health_status}" != "yellow" ]]; then + echo "Elasticsearch cluster health is '${es_health_status}' (expected green or yellow)." + exit 1 +fi +echo "OK: API, Postgres, Redis, Kafka, and Elasticsearch are up (cluster=${es_health_status})" + +echo "" +echo "[1/${TOTAL_STEPS}] Creating patient, encounter, and baseline critical potassium alert" +patient_payload='{"firstName":"SIRS","lastName":"Verifier","dateOfBirth":"1975-04-12","gender":"M"}' +resp="$(request POST "${BASE_URL}/api/v1/patients" "${patient_payload}")" +assert_status "201" "${resp}" +patient_id="$(jq -r '.data.id' "${resp}")" + +enc_payload='{"encounterType":"Inpatient","department":"ICU","attendingPhysician":"Dr. SIRS"}' +resp="$(request POST "${BASE_URL}/api/v1/patients/${patient_id}/encounters" "${enc_payload}")" +assert_status "201" "${resp}" +encounter_id="$(jq -r '.data.id' "${resp}")" +echo "OK: patient ${patient_id}, encounter ${encounter_id}" + +critical_payload="$(jq -nc \ + --arg recordedAt "${RECORDED_AT}" \ + --arg key "sirs-potassium-${SCRIPT_RUN_ID}" \ + '{observations:[{observationCode:"POTASSIUM_MEQ_L",value:2.1,unit:"mEq/L",source:"LAB",recordedAt:$recordedAt,idempotencyKey:$key}]}')" +resp="$(request POST "${BASE_URL}/api/v1/encounters/${encounter_id}/observations" "${critical_payload}")" +assert_status "201" "${resp}" +if [[ "$(jq -r '.data.alertGenerated' "${resp}")" != "true" ]]; then + echo "Expected critical potassium ingest to generate an alert" + exit 1 +fi +echo "OK: baseline critical alert ingested (for openAlertCount=2 after sepsis)" + +echo "" +echo "[2/${TOTAL_STEPS}] End-to-end SIRS — tachycardia sets Redis key, no SEPSIS_WARNING yet" +ingest_observation "${encounter_id}" "HEART_RATE" 95 "bpm" "DEVICE" "${RECORDED_AT}" \ + "sirs-hr1-${SCRIPT_RUN_ID}" + +if ! wait_for_observation_pipeline "${encounter_id}"; then + exit 1 +fi + +hr_key="$(sirs_key "${encounter_id}" "HEART_RATE")" +hr_value="$(redis_get "${hr_key}")" +if [[ "${hr_value}" != "1" ]]; then + echo "Expected Redis ${hr_key} = 1, got '${hr_value}'" + exit 1 +fi + +hr_ttl="$(redis_ttl "${hr_key}")" +if [[ "${hr_ttl}" -lt 1500 || "${hr_ttl}" -gt 1800 ]]; then + echo "Expected HEART_RATE TTL between 1500 and 1800, got ${hr_ttl}" + exit 1 +fi + +sepsis_count="$(psql_cmd "SELECT COUNT(*) FROM clinical_alerts WHERE encounter_id = '${encounter_id}' AND alert_type = 'SEPSIS_WARNING'")" +if [[ "${sepsis_count}" != "0" ]]; then + echo "Expected no SEPSIS_WARNING after one criterion, got count=${sepsis_count}" + exit 1 +fi +echo "OK: HEART_RATE key set (TTL=${hr_ttl}s), no sepsis alert yet" + +echo "" +echo "[3/${TOTAL_STEPS}] End-to-end SIRS — fever triggers SEPSIS_WARNING through Kafka pipeline" +recorded_fever="$(date -u +"%Y-%m-%dT%H:%M:%SZ")" +ingest_observation "${encounter_id}" "TEMP_C" 38.5 "°C" "DEVICE" "${recorded_fever}" \ + "sirs-temp-${SCRIPT_RUN_ID}" + +if ! wait_for_observation_pipeline "${encounter_id}"; then + exit 1 +fi + +if ! wait_for_sepsis_alert "${encounter_id}"; then + exit 1 +fi + +alert_row="$(psql_cmd "SELECT alert_type || '|' || severity || '|' || status FROM clinical_alerts WHERE encounter_id = '${encounter_id}' AND alert_type = 'SEPSIS_WARNING' LIMIT 1")" +if [[ "${alert_row}" != "SEPSIS_WARNING|CRITICAL|OPEN" ]]; then + echo "Unexpected SEPSIS_WARNING row: '${alert_row}'" + exit 1 +fi +echo "OK: SEPSIS_WARNING created (type=SEPSIS_WARNING, severity=CRITICAL, status=OPEN)" + +echo "" +echo "[4/${TOTAL_STEPS}] Verifying alert.generated outbox row for sepsis alert" +outbox_row="$(psql_cmd "SELECT topic || '|' || partition_key FROM outbox_events WHERE topic = 'alert.generated' AND partition_key = '${encounter_id}' ORDER BY created_at DESC LIMIT 1")" +if [[ "${outbox_row}" != "alert.generated|${encounter_id}" ]]; then + echo "Expected alert.generated outbox row for encounter ${encounter_id}, got '${outbox_row}'" + exit 1 +fi + +sepsis_outbox_id="$(psql_cmd "SELECT id FROM outbox_events WHERE topic = 'alert.generated' AND partition_key = '${encounter_id}' ORDER BY created_at DESC LIMIT 1")" +wait_for_outbox_processed "${sepsis_outbox_id}" +echo "OK: alert.generated outbox row exists and was relayed" + +echo "" +echo "[5/${TOTAL_STEPS}] Verifying SirsDetector uses MGET (one round-trip for four keys)" +monitor_out="$(mktemp)" +TMP_FILES+=("${monitor_out}") + +redis_cmd MONITOR > "${monitor_out}" 2>&1 & +monitor_pid=$! +sleep 0.5 + +recorded_resp="$(date -u +"%Y-%m-%dT%H:%M:%SZ")" +ingest_observation "${encounter_id}" "RESP_RATE" 22 "breaths/min" "DEVICE" "${recorded_resp}" \ + "sirs-resp-monitor-${SCRIPT_RUN_ID}" + +if ! wait_for_observation_pipeline "${encounter_id}"; then + kill "${monitor_pid}" 2>/dev/null || true + exit 1 +fi + +sleep 1 +kill "${monitor_pid}" 2>/dev/null || true +wait "${monitor_pid}" 2>/dev/null || true + +if grep -qi '"keys"' "${monitor_out}"; then + echo "MONITOR output contains KEYS — SirsDetector must not scan the keyspace" + exit 1 +fi + +mget_line="$(grep -i 'mget' "${monitor_out}" | grep -F "sirs:${encounter_id}" | head -n 1 || true)" +if [[ -z "${mget_line}" ]]; then + echo "No MGET command found in Redis MONITOR output for encounter ${encounter_id}" + echo "MONITOR tail:" + tail -n 20 "${monitor_out}" + exit 1 +fi + +for code in TEMP_C HEART_RATE RESP_RATE WBC_K_UL; do + if ! grep -F "sirs:${encounter_id}:${code}" <<< "${mget_line}" >/dev/null; then + echo "MGET line missing key sirs:${encounter_id}:${code}" + echo "MGET line: ${mget_line}" + exit 1 + fi +done + +sirs_get_count="$(grep -cE '"get".*sirs:'"${encounter_id}" "${monitor_out}" || true)" +if [[ "${sirs_get_count}" -ge 4 ]]; then + echo "Found ${sirs_get_count} individual GET commands on SIRS keys — expected MGET instead" + exit 1 +fi +echo "OK: single MGET with all four SIRS keys observed" + +echo "" +echo "[6/${TOTAL_STEPS}] Verifying TTL sliding window on HEART_RATE key" +ttl_before_decay="$(redis_ttl "${hr_key}")" +if (( TTL_DECAY_WAIT_SECS > 0 )); then + sleep "${TTL_DECAY_WAIT_SECS}" + ttl_after_decay="$(redis_ttl "${hr_key}")" + expected_max="$((ttl_before_decay - TTL_DECAY_WAIT_SECS + 2))" + if [[ "${ttl_after_decay}" -gt "${expected_max}" ]]; then + echo "Expected TTL to decay after ${TTL_DECAY_WAIT_SECS}s (${ttl_before_decay} -> <=${expected_max}), got ${ttl_after_decay}" + exit 1 + fi + echo "OK: TTL decayed (${ttl_before_decay}s -> ${ttl_after_decay}s over ${TTL_DECAY_WAIT_SECS}s wait)" +fi + +recorded_hr2="$(date -u +"%Y-%m-%dT%H:%M:%SZ")" +ingest_observation "${encounter_id}" "HEART_RATE" 96 "bpm" "DEVICE" "${recorded_hr2}" \ + "sirs-hr2-${SCRIPT_RUN_ID}" +if ! wait_for_observation_pipeline "${encounter_id}"; then + exit 1 +fi + +ttl_after_refresh="$(redis_ttl "${hr_key}")" +if [[ "${ttl_after_refresh}" -lt 1500 || "${ttl_after_refresh}" -gt 1800 ]]; then + echo "Expected HEART_RATE TTL to reset near 1800 after re-qualifying observation, got ${ttl_after_refresh}" + exit 1 +fi +echo "OK: TTL reset after qualifying HEART_RATE observation (TTL=${ttl_after_refresh}s)" + +echo "" +echo "[7/${TOTAL_STEPS}] Verifying consumer group independence (sepsis-engine and es-indexer)" +if ! wait_for_consumer_lag_zero "${SEPSIS_CONSUMER_GROUP}"; then + exit 1 +fi +if ! wait_for_consumer_lag_zero "${ES_CONSUMER_GROUP}"; then + exit 1 +fi +echo "OK: sepsis-engine and es-indexer both have LAG=0 on observation.recorded" + +echo "" +echo "[8/${TOTAL_STEPS}] Verifying SEPSIS_WARNING in Elasticsearch and openAlertCount=2" +elapsed=0 +while (( elapsed < INDEX_WAIT_SECS )); do + es_hits="$(curl -sS "${ES_URL}/clinical_alerts/_search" \ + -H "Content-Type: application/json" \ + -d "{\"query\":{\"bool\":{\"must\":[{\"term\":{\"encounterId\":\"${encounter_id}\"}},{\"term\":{\"alertType\":\"SEPSIS_WARNING\"}}]}}}" \ + | jq -r '.hits.total.value')" + if [[ "${es_hits}" == "1" ]]; then + break + fi + sleep 2 + elapsed=$((elapsed + 2)) +done + +if [[ "${es_hits:-0}" != "1" ]]; then + echo "Expected 1 SEPSIS_WARNING document in clinical_alerts for encounter ${encounter_id}, got ${es_hits:-0}" + exit 1 +fi + +es_alert="$(curl -sS "${ES_URL}/clinical_alerts/_search" \ + -H "Content-Type: application/json" \ + -d "{\"query\":{\"bool\":{\"must\":[{\"term\":{\"encounterId\":\"${encounter_id}\"}},{\"term\":{\"alertType\":\"SEPSIS_WARNING\"}}]}}}" \ + | jq -r '.hits.hits[0]._source | [.severity, .status] | @tsv')" +if [[ "${es_alert}" != $'Critical\tOpen' ]]; then + echo "Expected Elasticsearch alert severity=Critical status=Open, got '${es_alert}'" + exit 1 +fi + +open_alert_count="$(curl -sS "${ES_URL}/patient_encounters/_source/${encounter_id}" | jq -r '.openAlertCount')" +if [[ "${open_alert_count}" != "2" ]]; then + echo "Expected openAlertCount=2 (potassium + sepsis), got ${open_alert_count}" + exit 1 +fi +echo "OK: SEPSIS_WARNING indexed (Critical/Open) and openAlertCount=2" + +echo "" +echo "[9/${TOTAL_STEPS}] Verifying criterion clears immediately but alert remains open" +ingest_observation "${encounter_id}" "TEMP_C" 38.5 "°C" "DEVICE" "$(date -u +"%Y-%m-%dT%H:%M:%SZ")" \ + "sirs-temp-qual-${SCRIPT_RUN_ID}" +if ! wait_for_observation_pipeline "${encounter_id}"; then + exit 1 +fi + +temp_key="$(sirs_key "${encounter_id}" "TEMP_C")" +if [[ "$(redis_get "${temp_key}")" != "1" ]]; then + echo "Expected qualifying temperature to set ${temp_key}" + exit 1 +fi + +ingest_observation "${encounter_id}" "TEMP_C" 37.0 "°C" "DEVICE" "$(date -u +"%Y-%m-%dT%H:%M:%SZ")" \ + "sirs-temp-normal-${SCRIPT_RUN_ID}" +if ! wait_for_observation_pipeline "${encounter_id}"; then + exit 1 +fi + +temp_value="$(redis_get "${temp_key}")" +if [[ -n "${temp_value}" ]]; then + echo "Expected ${temp_key} to be cleared after normal temperature, got '${temp_value}'" + exit 1 +fi + +alert_status="$(psql_cmd "SELECT status FROM clinical_alerts WHERE encounter_id = '${encounter_id}' AND alert_type = 'SEPSIS_WARNING' LIMIT 1")" +if [[ "${alert_status}" != "OPEN" ]]; then + echo "Expected SEPSIS_WARNING to remain OPEN after criterion cleared, got '${alert_status}'" + exit 1 +fi +echo "OK: TEMP_C key deleted on normalisation; SEPSIS_WARNING remains OPEN" + +echo "" +echo "[10/${TOTAL_STEPS}] Verifying non-SIRS observation is ignored by the sepsis engine" +ingest_observation "${encounter_id}" "POTASSIUM_MEQ_L" 3.2 "mEq/L" "LAB" "$(date -u +"%Y-%m-%dT%H:%M:%SZ")" \ + "sirs-non-sirs-${SCRIPT_RUN_ID}" +if ! wait_for_observation_pipeline "${encounter_id}"; then + exit 1 +fi + +pot_key="$(sirs_key "${encounter_id}" "POTASSIUM_MEQ_L")" +if redis_cmd EXISTS "${pot_key}" | grep -q '^1'; then + echo "Non-SIRS code must not create a Redis SIRS key (${pot_key})" + exit 1 +fi +echo "OK: non-SIRS observation did not create SIRS Redis keys" + +echo "" +echo "All ${TOTAL_STEPS} sepsis / SIRS checks passed." +echo "" +echo "Prerequisites: docker compose up -d && dotnet run --project VigilCareClinicalAPI" +echo "Optional: set TTL_DECAY_WAIT_SECS=60 to match the full sliding-window decay check in the plan."