using System.Net.Http.Json; using System.Text.Json; using Microsoft.EntityFrameworkCore; using Microsoft.Extensions.DependencyInjection; public static class ScenarioReplayHelper { private static readonly JsonSerializerOptions JsonOptions = new() { PropertyNameCaseInsensitive = true, }; public static string ScenariosDirectory => Path.GetFullPath(Path.Combine( AppContext.BaseDirectory, "..", "..", "..", "..", "VigilCare.Simulator", "Scenarios", "List")); public static ScenarioFile Load(string fileName) { var path = Path.Combine(ScenariosDirectory, fileName); var json = File.ReadAllText(path); return JsonSerializer.Deserialize(json, JsonOptions) ?? throw new InvalidOperationException($"Failed to deserialize {path}"); } public static async Task<(Guid PatientId, Guid EncounterId)> ReplayObservationsAsync( HttpClient client, ScenarioFile scenario, IServiceProvider services, CancellationToken ct = default) { var result = await ReplayObservationsCoreAsync(client, scenario, ct); await WaitForOutboxDrainAsync(services, TimeSpan.FromSeconds(30), ct); await RunClinicalEnginesForEncounterAsync(services, result.EncounterId, ct); return result; } /// /// Replays persisted observations through the clinical detectors in encounter order. /// Scenario E2E tests use this after HTTP ingest so assertions do not depend on /// shared Kafka consumer lag across the integration suite. /// public static async Task RunClinicalEnginesForEncounterAsync( IServiceProvider services, Guid encounterId, CancellationToken ct = default) { using var scope = services.CreateScope(); var db = scope.ServiceProvider.GetRequiredService(); var encounter = await db.Encounters .AsNoTracking() .FirstOrDefaultAsync(e => e.Id == encounterId, ct) ?? throw new InvalidOperationException($"Encounter {encounterId} not found"); var observations = await db.Observations .AsNoTracking() .Where(o => o.EncounterId == encounterId) .OrderBy(o => o.RecordedAt) .ThenBy(o => o.CreatedAt) .ToListAsync(ct); var qsofa = scope.ServiceProvider.GetRequiredService(); var gcs = scope.ServiceProvider.GetRequiredService(); var sofa = scope.ServiceProvider.GetRequiredService(); foreach (var obs in observations) { await qsofa.ProcessObservationAsync( encounterId, encounter.PatientId, obs.ObservationCode, obs.Value, ct); var gcsResult = await gcs.ProcessObservationAsync( encounterId, encounter.PatientId, obs.ObservationCode, obs.Value, ct); if (gcsResult.Outcome == GcsOutcome.ScoreComputed) { await sofa.ProcessGcsScoredAsync( encounterId, encounter.PatientId, ct); } await sofa.ProcessObservationAsync( encounterId, encounter.PatientId, obs.ObservationCode, obs.Value, obs.RecordedAt, ct); } await qsofa.SyncAlteredMentationAsync(encounterId, encounter.PatientId, ct); } public static async Task WaitForOutboxDrainAsync( IServiceProvider services, TimeSpan timeout, CancellationToken ct = default) { var deadline = DateTime.UtcNow + timeout; while (DateTime.UtcNow < deadline) { using var scope = services.CreateScope(); var db = scope.ServiceProvider.GetRequiredService(); var pending = await db.OutboxEvents.CountAsync( e => e.ProcessedAt == null && e.FailedAt == null, ct); if (pending == 0) return; await Task.Delay(250, ct); } throw new TimeoutException("Timed out waiting for outbox relay to drain pending events"); } private static async Task<(Guid PatientId, Guid EncounterId)> ReplayObservationsCoreAsync( HttpClient client, ScenarioFile scenario, CancellationToken ct) { var patientResp = await client.PostAsJsonAsync("/api/v1/patients", new { firstName = scenario.Patient.FirstName, lastName = scenario.Patient.LastName, dateOfBirth = scenario.Patient.DateOfBirth, gender = scenario.Patient.Gender, }, ct); patientResp.EnsureSuccessStatusCode(); var patient = (await patientResp.Content.ReadFromJsonAsync>(ct))!.Data; var encounterResp = await client.PostAsJsonAsync( $"/api/v1/patients/{patient.Id}/encounters", new { encounterType = scenario.Encounter.EncounterType, department = scenario.Encounter.Department, attendingPhysician = scenario.Encounter.AttendingPhysician, roomBed = scenario.Encounter.RoomBed, admissionReason = scenario.Encounter.AdmissionReason, }, ct); encounterResp.EnsureSuccessStatusCode(); var encounter = (await encounterResp.Content.ReadFromJsonAsync>(ct))!.Data; var start = DateTimeOffset.UtcNow; foreach (var cluster in scenario.Events .Where(e => e.Type == "observation") .GroupBy(e => e.OffsetMinutes) .OrderBy(g => g.Key)) { var recordedAt = start.AddMinutes(cluster.Key); var batch = cluster.Select(evt => { var code = evt.Data.GetProperty("code").GetString()!; var value = evt.Data.GetProperty("value").GetDecimal(); var unit = evt.Data.GetProperty("unit").GetString()!; var source = evt.Data.TryGetProperty("source", out var s) ? s.GetString()! : "Manual"; return new { observationCode = code, value, unit, source = MapSource(source), recordedAt, }; }).ToList(); foreach (var chunk in batch.Chunk(10)) { var resp = await client.PostAsJsonAsync( $"/api/v1/encounters/{encounter.Id}/observations", new { observations = chunk }, ct); resp.EnsureSuccessStatusCode(); } } return (patient.Id, encounter.Id); } public static async Task WaitForSofaScoreAsync( IServiceProvider services, Guid encounterId, TimeSpan timeout, Func? predicate = null) { var deadline = DateTime.UtcNow + timeout; while (DateTime.UtcNow < deadline) { using var scope = services.CreateScope(); var db = scope.ServiceProvider.GetRequiredService(); var sofa = await db.SofaScores .Where(s => s.EncounterId == encounterId) .OrderByDescending(s => s.CalculatedAt) .FirstOrDefaultAsync(); if (sofa is not null && (predicate is null || predicate(sofa))) return sofa; await Task.Delay(500); } throw new TimeoutException($"Timed out waiting for SOFA score on encounter {encounterId}"); } public static async Task WaitForAlertTypeAsync( IServiceProvider services, Guid encounterId, AlertType alertType, TimeSpan timeout) { var deadline = DateTime.UtcNow + timeout; while (DateTime.UtcNow < deadline) { using var scope = services.CreateScope(); var db = scope.ServiceProvider.GetRequiredService(); if (await db.ClinicalAlerts.AnyAsync(a => a.EncounterId == encounterId && a.AlertType == alertType)) return; await Task.Delay(500); } throw new TimeoutException($"Timed out waiting for {alertType} on encounter {encounterId}"); } public static async Task AssertNoAlertTypeAsync( IServiceProvider services, Guid encounterId, AlertType alertType, TimeSpan settleDelay) { await Task.Delay(settleDelay); using var scope = services.CreateScope(); var db = scope.ServiceProvider.GetRequiredService(); var found = await db.ClinicalAlerts.AnyAsync(a => a.EncounterId == encounterId && a.AlertType == alertType); if (found) throw new InvalidOperationException($"Unexpected {alertType} alert on encounter {encounterId}"); } private static string MapSource(string source) => source.ToLowerInvariant() switch { "device" or "monitor" => "Device", "lab" => "Lab", _ => "Manual", }; private record ApiEnvelope(T Data); private record PatientDto(Guid Id); private record EncounterDto(Guid Id); }