fix errors of Degraded Operations Visibility

2026-06-24 00:46:27 +08:00
parent 4399996448
commit 2e80700bf0
7 changed files with 177 additions and 40 deletions
@@ -32,13 +32,14 @@ public class ClinicalRefactorEndToEndTests : IAsyncLifetime
    public async Task SofaProgressionScenario_EndToEnd()
    {
        var scenario = ScenarioReplayHelper.Load("sepsis-sofa-progression-01.json");
-        var (_, encounterId) = await ScenarioReplayHelper.ReplayObservationsAsync(_client, scenario);
+        var (_, encounterId) = await ScenarioReplayHelper.ReplayObservationsAsync(
+            _client, scenario, _fixture.Services);

        await ScenarioReplayHelper.WaitForAlertTypeAsync(
-            _fixture.Services, encounterId, AlertType.QsofaScreen, TimeSpan.FromSeconds(30));
+            _fixture.Services, encounterId, AlertType.QsofaScreen, TimeSpan.FromSeconds(60));

        await ScenarioReplayHelper.WaitForAlertTypeAsync(
-            _fixture.Services, encounterId, AlertType.SofaSepsis, TimeSpan.FromSeconds(30));
+            _fixture.Services, encounterId, AlertType.SofaSepsis, TimeSpan.FromSeconds(60));

        SepsisBundle? bundle = null;
        var deadline = DateTime.UtcNow + TimeSpan.FromSeconds(10);
@@ -58,13 +59,14 @@ public class ClinicalRefactorEndToEndTests : IAsyncLifetime
    public async Task GcsDeclineScenario_EndToEnd()
    {
        var scenario = ScenarioReplayHelper.Load("neurological-decline-gcs-01.json");
-        var (_, encounterId) = await ScenarioReplayHelper.ReplayObservationsAsync(_client, scenario);
+        var (_, encounterId) = await ScenarioReplayHelper.ReplayObservationsAsync(
+            _client, scenario, _fixture.Services);

        await ScenarioReplayHelper.WaitForAlertTypeAsync(
-            _fixture.Services, encounterId, AlertType.GcsWarning, TimeSpan.FromSeconds(30));
+            _fixture.Services, encounterId, AlertType.GcsWarning, TimeSpan.FromSeconds(60));

        await ScenarioReplayHelper.WaitForAlertTypeAsync(
-            _fixture.Services, encounterId, AlertType.GcsCritical, TimeSpan.FromSeconds(30));
+            _fixture.Services, encounterId, AlertType.GcsCritical, TimeSpan.FromSeconds(60));

        await ScenarioReplayHelper.AssertNoAlertTypeAsync(
            _fixture.Services, encounterId, AlertType.SofaSepsis, TimeSpan.FromSeconds(5));
@@ -77,11 +79,12 @@ public class ClinicalRefactorEndToEndTests : IAsyncLifetime
    public async Task PartialSofaScenario_StalenessFlags()
    {
        var scenario = ScenarioReplayHelper.Load("sofa-partial-spo2-fallback-01.json");
-        var (_, encounterId) = await ScenarioReplayHelper.ReplayObservationsAsync(_client, scenario);
+        var (_, encounterId) = await ScenarioReplayHelper.ReplayObservationsAsync(
+            _client, scenario, _fixture.Services);

        var jsonOpts = new JsonSerializerOptions { PropertyNameCaseInsensitive = true };
        var sofa = await ScenarioReplayHelper.WaitForSofaScoreAsync(
-            _fixture.Services, encounterId, TimeSpan.FromSeconds(45),
+            _fixture.Services, encounterId, TimeSpan.FromSeconds(60),
            s =>
            {
                if (string.IsNullOrEmpty(s.StalenessFlags)) return false;
@@ -113,10 +116,11 @@ public class ClinicalRefactorEndToEndTests : IAsyncLifetime
    public async Task UtiSepsisScenario_NowUsesSofa()
    {
        var scenario = ScenarioReplayHelper.Load("uti-sepsis-elderly-01.json");
-        var (_, encounterId) = await ScenarioReplayHelper.ReplayObservationsAsync(_client, scenario);
+        var (_, encounterId) = await ScenarioReplayHelper.ReplayObservationsAsync(
+            _client, scenario, _fixture.Services);

        await ScenarioReplayHelper.WaitForAlertTypeAsync(
-            _fixture.Services, encounterId, AlertType.SofaSepsis, TimeSpan.FromSeconds(45));
+            _fixture.Services, encounterId, AlertType.SofaSepsis, TimeSpan.FromSeconds(60));

        await ScenarioReplayHelper.AssertNoAlertTypeAsync(
            _fixture.Services, encounterId, AlertTypeExtensions.FromDbString("SEPSIS_WARNING"), TimeSpan.FromSeconds(5));
@@ -0,0 +1,38 @@
+using Microsoft.Extensions.Options;
+using RabbitMQ.Client;
+using RabbitMQ.Client.Exceptions;
+
+public static class RabbitMqTestHelper
+{
+    public static void PurgeNotificationQueues(IOptions<RabbitMqOptions> opts)
+    {
+        var o = opts.Value;
+        var factory = new ConnectionFactory
+        {
+            HostName = o.Host,
+            Port = o.Port,
+            UserName = o.Username,
+            Password = o.Password,
+        };
+
+        using var connection = factory.CreateConnection("test-queue-purge");
+        using var channel = connection.CreateModel();
+
+        foreach (var queue in new[]
+        {
+            "alerts.paging.queue",
+            "alerts.paging.dlq",
+            "alerts.escalation.queue",
+        })
+        {
+            try
+            {
+                channel.QueuePurge(queue);
+            }
+            catch (OperationInterruptedException)
+            {
+                // Queue may not exist yet on a cold broker.
+            }
+        }
+    }
+}
@@ -27,7 +27,93 @@ public static class ScenarioReplayHelper
    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;
+    }
+
+    /// <summary>
+    /// 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.
+    /// </summary>
+    public static async Task RunClinicalEnginesForEncounterAsync(
+        IServiceProvider services,
+        Guid encounterId,
+        CancellationToken ct = default)
+    {
+        using var scope = services.CreateScope();
+        var db = scope.ServiceProvider.GetRequiredService<AppDbContext>();
+
+        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<QsofaDetector>();
+        var gcs = scope.ServiceProvider.GetRequiredService<GcsDetector>();
+        var sofa = scope.ServiceProvider.GetRequiredService<SofaDetector>();
+
+        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<AppDbContext>();
+            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
        {
@@ -153,4 +239,4 @@ public static class ScenarioReplayHelper
    private record ApiEnvelope<T>(T Data);
    private record PatientDto(Guid Id);
    private record EncounterDto(Guid Id);
-}
+}
@@ -45,6 +45,9 @@ public class NotificationPipelineTests : IAsyncLifetime
        await redis.GetDatabase(1).StringSetAsync(
            "threshold:POTASSIUM_MEQ_L",
            """{"ObservationCode":"POTASSIUM_MEQ_L","CriticalLow":2.5,"WarningLow":3.5,"WarningHigh":5.0,"CriticalHigh":6.5}""");
+
+        RabbitMqTestHelper.PurgeNotificationQueues(
+            scope.ServiceProvider.GetRequiredService<IOptions<RabbitMqOptions>>());
    }

    public Task DisposeAsync() => Task.CompletedTask;
@@ -88,7 +91,7 @@ public class NotificationPipelineTests : IAsyncLifetime
        // paging worker (≤6s) → DLQ (5s TTL) → escalation worker.
        // Poll instead of a fixed sleep: earlier tests may leave paging jobs queued
        // (prefetch=1), so wall-clock time varies across the full integration suite.
-        var deadline = DateTimeOffset.UtcNow.AddSeconds(35);
+        var deadline = DateTimeOffset.UtcNow.AddSeconds(60);
        AlertStatus status;
        do
        {