From 2e80700bf0ae7772cd557af74ba5d9b267159e1e Mon Sep 17 00:00:00 2001 From: voltsrage Date: Wed, 24 Jun 2026 00:46:27 +0800 Subject: [PATCH] fix errors of Degraded Operations Visibility --- .../ClinicalRefactorEndToEndTests.cs | 24 ++--- .../Helpers/RabbitMqTestHelper.cs | 38 ++++++++ .../Helpers/ScenarioReplayHelper.cs | 88 ++++++++++++++++++- .../NotificationPipelineTests.cs | 5 +- .../BackgroundServices/SofaScoringService.cs | 2 +- .../Records/News2/News2ObservationEvent.cs | 3 +- .../Services/OperationsService.cs | 57 ++++++------ 7 files changed, 177 insertions(+), 40 deletions(-) create mode 100644 VigilCareClinicalAPI.Tests/Helpers/RabbitMqTestHelper.cs diff --git a/VigilCareClinicalAPI.Tests/ClinicalRefactorEndToEndTests.cs b/VigilCareClinicalAPI.Tests/ClinicalRefactorEndToEndTests.cs index 25d12f7..d8baa19 100644 --- a/VigilCareClinicalAPI.Tests/ClinicalRefactorEndToEndTests.cs +++ b/VigilCareClinicalAPI.Tests/ClinicalRefactorEndToEndTests.cs @@ -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)); diff --git a/VigilCareClinicalAPI.Tests/Helpers/RabbitMqTestHelper.cs b/VigilCareClinicalAPI.Tests/Helpers/RabbitMqTestHelper.cs new file mode 100644 index 0000000..2a2f965 --- /dev/null +++ b/VigilCareClinicalAPI.Tests/Helpers/RabbitMqTestHelper.cs @@ -0,0 +1,38 @@ +using Microsoft.Extensions.Options; +using RabbitMQ.Client; +using RabbitMQ.Client.Exceptions; + +public static class RabbitMqTestHelper +{ + public static void PurgeNotificationQueues(IOptions 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. + } + } + } +} diff --git a/VigilCareClinicalAPI.Tests/Helpers/ScenarioReplayHelper.cs b/VigilCareClinicalAPI.Tests/Helpers/ScenarioReplayHelper.cs index 00c82ca..2c7debf 100644 --- a/VigilCareClinicalAPI.Tests/Helpers/ScenarioReplayHelper.cs +++ b/VigilCareClinicalAPI.Tests/Helpers/ScenarioReplayHelper.cs @@ -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; + } + + /// + /// 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 { @@ -153,4 +239,4 @@ public static class ScenarioReplayHelper private record ApiEnvelope(T Data); private record PatientDto(Guid Id); private record EncounterDto(Guid Id); -} \ No newline at end of file +} diff --git a/VigilCareClinicalAPI.Tests/NotificationPipelineTests.cs b/VigilCareClinicalAPI.Tests/NotificationPipelineTests.cs index 75ecf98..ec156cb 100644 --- a/VigilCareClinicalAPI.Tests/NotificationPipelineTests.cs +++ b/VigilCareClinicalAPI.Tests/NotificationPipelineTests.cs @@ -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>()); } 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 { diff --git a/VigilCareClinicalAPI/BackgroundServices/SofaScoringService.cs b/VigilCareClinicalAPI/BackgroundServices/SofaScoringService.cs index 2ede97b..8f178a9 100644 --- a/VigilCareClinicalAPI/BackgroundServices/SofaScoringService.cs +++ b/VigilCareClinicalAPI/BackgroundServices/SofaScoringService.cs @@ -80,7 +80,7 @@ public class SofaScoringService : BackgroundService evt.PatientId, evt.ObservationCode, evt.Value, - DateTimeOffset.UtcNow, + evt.RecordedAt == default ? DateTimeOffset.UtcNow : evt.RecordedAt, stoppingToken); if (outcome.Outcome == SofaOutcome.EncounterNotFound) diff --git a/VigilCareClinicalAPI/Models/Records/News2/News2ObservationEvent.cs b/VigilCareClinicalAPI/Models/Records/News2/News2ObservationEvent.cs index 03456b9..3e44752 100644 --- a/VigilCareClinicalAPI/Models/Records/News2/News2ObservationEvent.cs +++ b/VigilCareClinicalAPI/Models/Records/News2/News2ObservationEvent.cs @@ -3,4 +3,5 @@ public record News2ObservationEvent( Guid EncounterId, Guid PatientId, string ObservationCode, - decimal Value); \ No newline at end of file + decimal Value, + DateTimeOffset RecordedAt); \ No newline at end of file diff --git a/VigilCareClinicalAPI/Services/OperationsService.cs b/VigilCareClinicalAPI/Services/OperationsService.cs index cf2302b..cb8999a 100644 --- a/VigilCareClinicalAPI/Services/OperationsService.cs +++ b/VigilCareClinicalAPI/Services/OperationsService.cs @@ -16,34 +16,39 @@ public class OperationsService : IOperationsService public async Task> GetGatewayFleetAsync( GatewayFleetFilter filter, CancellationToken ct) { - await using var conn = new NpgsqlConnection(_db.Database.GetConnectionString()); - await conn.OpenAsync(ct); + var query = _db.WardGateways + .AsNoTracking() + .Include(g => g.Site) + .AsQueryable(); - const string sql = """ - SELECT - g.id, - g.gateway_code AS GatewayCode, - g.department AS Department, - s.site_code AS SiteCode, - s.name AS SiteName, - g.status AS Status, - g.reported_buffer_depth AS ReportedBufferDepth, - g.last_heartbeat_at AS LastHeartbeatAt, - g.last_sync_at AS LastSyncAt, - EXTRACT(EPOCH FROM (NOW() - g.last_heartbeat_at)) / 60 AS MinutesSinceHeartbeat - FROM ward_gateways g - JOIN clinical_sites s ON s.id = g.site_id - WHERE (@Status IS NULL OR g.status = @Status) - AND (@SiteId IS NULL OR g.site_id = @SiteId) - ORDER BY g.status DESC, MinutesSinceHeartbeat DESC NULLS LAST - """; - - var rows = await conn.QueryAsync(sql, new + if (!string.IsNullOrEmpty(filter.Status)) { - Status = filter.Status, - SiteId = filter.SiteId - }); - return rows.ToList(); + var status = GatewayStatusExtensions.FromDbString(filter.Status); + query = query.Where(g => g.Status == status); + } + + if (filter.SiteId is Guid siteId) + query = query.Where(g => g.SiteId == siteId); + + var now = DateTimeOffset.UtcNow; + return await query + .OrderByDescending(g => g.Status) + .ThenByDescending(g => g.LastHeartbeatAt == null) + .ThenByDescending(g => g.LastHeartbeatAt) + .Select(g => new GatewayFleetItem( + g.Id, + g.GatewayCode, + g.Department, + g.Site.SiteCode, + g.Site.Name, + g.Status.ToDbString(), + g.ReportedBufferDepth, + g.LastHeartbeatAt, + g.LastSyncAt, + g.LastHeartbeatAt == null + ? null + : (double?)(now - g.LastHeartbeatAt.Value).TotalMinutes)) + .ToListAsync(ct); } public async Task GetGatewayDetailAsync(Guid gatewayId, CancellationToken ct)