From 0246143815bf0dc0d42fc06b3da69460207d6681 Mon Sep 17 00:00:00 2001 From: voltsrage Date: Wed, 17 Jun 2026 02:37:30 +0800 Subject: [PATCH] feature: RabbitMQ Notification Workers and DLQ Escalation --- .../Fixtures/ApiFixture.cs | 9 +- .../NotificationPipelineTests.cs | 166 ++++++ .../DischargeSummaryWorkerService.cs | 158 ++++++ .../Notifications/EscalationWorkerService.cs | 95 ++++ .../NotificationPublisherService.cs | 151 ++++++ .../Notifications/PagingWorkerService.cs | 121 +++++ .../Configuration/MinioOptions.cs | 9 + .../Configuration/RabbitMqOptions.cs | 11 + .../RabbitMqTopologyProvisioner.cs | 139 +++++ VigilCareClinicalAPI/Program.cs | 12 + .../Services/ObservationService.cs | 2 + .../Storage/MinioClientFactory.cs | 15 + .../VigilCareClinicalAPI.csproj | 2 + VigilCareClinicalAPI/appsettings.Testing.json | 3 + VigilCareClinicalAPI/appsettings.json | 14 + docker-compose.yml | 31 +- scripts/run-notification-pipeline-tests.sh | 474 ++++++++++++++++++ 17 files changed, 1410 insertions(+), 2 deletions(-) create mode 100644 VigilCareClinicalAPI.Tests/NotificationPipelineTests.cs create mode 100644 VigilCareClinicalAPI/BackgroundServices/Notifications/DischargeSummaryWorkerService.cs create mode 100644 VigilCareClinicalAPI/BackgroundServices/Notifications/EscalationWorkerService.cs create mode 100644 VigilCareClinicalAPI/BackgroundServices/Notifications/NotificationPublisherService.cs create mode 100644 VigilCareClinicalAPI/BackgroundServices/Notifications/PagingWorkerService.cs create mode 100644 VigilCareClinicalAPI/Configuration/MinioOptions.cs create mode 100644 VigilCareClinicalAPI/Configuration/RabbitMqOptions.cs create mode 100644 VigilCareClinicalAPI/Notifications/RabbitMqTopologyProvisioner.cs create mode 100644 VigilCareClinicalAPI/Storage/MinioClientFactory.cs create mode 100755 scripts/run-notification-pipeline-tests.sh diff --git a/VigilCareClinicalAPI.Tests/Fixtures/ApiFixture.cs b/VigilCareClinicalAPI.Tests/Fixtures/ApiFixture.cs index a0f2440..c8367c3 100644 --- a/VigilCareClinicalAPI.Tests/Fixtures/ApiFixture.cs +++ b/VigilCareClinicalAPI.Tests/Fixtures/ApiFixture.cs @@ -18,8 +18,15 @@ public class ApiFixture : WebApplicationFactory, IAsyncLifetime { ["ConnectionStrings:DefaultConnection"] = "Host=localhost;Port=5436;Database=vigilcare_test;Username=postgres;Password=password", - ["Redis:ConnectionString"] = "localhost:6382,defaultDatabase=1,allowAdmin=true" + ["Redis:ConnectionString"] = "localhost:6382,defaultDatabase=1,allowAdmin=true", + ["RabbitMq:Host"] = "localhost", + ["RabbitMq:Port"] = "5674", + ["RabbitMq:Username"] = "guest", + ["RabbitMq:Password"] = "guest", + ["RabbitMq:PagingAckTimeoutMs"] = "5000", }); + + config.AddJsonFile("appsettings.Testing.json", optional: true, reloadOnChange: false); }); } diff --git a/VigilCareClinicalAPI.Tests/NotificationPipelineTests.cs b/VigilCareClinicalAPI.Tests/NotificationPipelineTests.cs new file mode 100644 index 0000000..d63983c --- /dev/null +++ b/VigilCareClinicalAPI.Tests/NotificationPipelineTests.cs @@ -0,0 +1,166 @@ +using System.Net.Http.Json; +using System.Text.Json; +using FluentAssertions; +using Microsoft.EntityFrameworkCore; +using Microsoft.Extensions.DependencyInjection; +using Microsoft.Extensions.Options; +using Minio.DataModel.Args; +using Minio.Exceptions; +using StackExchange.Redis; + +[Collection("Integration")] +public class NotificationPipelineTests : IAsyncLifetime +{ + private readonly ApiFixture _fixture; + private readonly HttpClient _http; + + public NotificationPipelineTests(ApiFixture fixture) + { + _fixture = fixture; + _http = fixture.CreateClient(); + } + + public async Task InitializeAsync() + { + using var scope = _fixture.Services.CreateScope(); + var db = scope.ServiceProvider.GetRequiredService(); + await DbResetHelper.ResetAsync(db); + + db.AlertThresholds.Add(new AlertThreshold + { + Id = Guid.NewGuid(), + ObservationCode = "POTASSIUM_MEQ_L", + DisplayName = "Serum Potassium", + Unit = "mEq/L", + CriticalLow = 2.5m, + WarningLow = 3.5m, + WarningHigh = 5.0m, + CriticalHigh = 6.5m, + CreatedAt = DateTimeOffset.UtcNow, + }); + await db.SaveChangesAsync(); + + var redis = scope.ServiceProvider.GetRequiredService(); + 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}"""); + } + + public Task DisposeAsync() => Task.CompletedTask; + + [Fact] + public async Task DischargeEncounter_UploadsSummaryToMinIO() + { + var encounterId = await CreateActiveEncounterAsync(); + + var resp = await _http.PatchAsJsonAsync( + $"/api/v1/encounters/{encounterId}/status", + new { status = "Discharged" }); + resp.EnsureSuccessStatusCode(); + + // outbox relay → Kafka → notification-publisher → RabbitMQ → discharge worker → MinIO + await Task.Delay(TimeSpan.FromSeconds(8)); + + var mo = _fixture.Services.GetRequiredService>().Value; + var client = MinioClientFactory.Build(mo); + var objectKey = $"discharge-summaries/{encounterId}/summary.pdf"; + + var exists = false; + try + { + await client.StatObjectAsync(new StatObjectArgs() + .WithBucket(mo.BucketName) + .WithObject(objectKey)); + exists = true; + } + catch (ObjectNotFoundException) { } + + exists.Should().BeTrue($"MinIO object not found: {objectKey}"); + } + + [Fact] + public async Task CriticalAlert_Unacknowledged_EscalatesAfterTimeout() + { + var encounterId = await CreateActiveEncounterAsync(); + var alertId = await IngestCriticalPotassiumAsync(encounterId); + + // 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); + AlertStatus status; + do + { + await Task.Delay(500); + using var scope = _fixture.Services.CreateScope(); + var db = scope.ServiceProvider.GetRequiredService(); + status = (await db.ClinicalAlerts.FindAsync(alertId))!.Status; + } + while (status == AlertStatus.Open && DateTimeOffset.UtcNow < deadline); + + status.Should().Be(AlertStatus.Escalated); + } + + [Fact] + public async Task CriticalAlert_AcknowledgedBeforeTimeout_DoesNotEscalate() + { + var encounterId = await CreateActiveEncounterAsync(); + var alertId = await IngestCriticalPotassiumAsync(encounterId); + + await Task.Delay(TimeSpan.FromSeconds(3)); + + var ackResp = await _http.PostAsJsonAsync( + $"/api/v1/alerts/{alertId}/acknowledge", + new AcknowledgeAlertRequest("Dr. Kwame Mensah", "Reviewed — will adjust potassium replacement.")); + ackResp.EnsureSuccessStatusCode(); + + await Task.Delay(TimeSpan.FromSeconds(8)); + + using var scope = _fixture.Services.CreateScope(); + var db = scope.ServiceProvider.GetRequiredService(); + + var alert = await db.ClinicalAlerts.FindAsync(alertId); + alert.Should().NotBeNull(); + alert!.Status.Should().Be(AlertStatus.Acknowledged); + } + + private async Task CreateActiveEncounterAsync() + { + var patientId = await CreatePatientAsync(); + + var resp = await _http.PostAsJsonAsync( + $"/api/v1/patients/{patientId}/encounters", + new OpenEncounterRequest(EncounterType.Inpatient, "ICU", "Dr. Osei")); + resp.EnsureSuccessStatusCode(); + + var body = await resp.Content.ReadFromJsonAsync(); + return body!.RootElement.GetProperty("data").GetProperty("id").GetGuid(); + } + + private async Task CreatePatientAsync() + { + var resp = await _http.PostAsJsonAsync( + "/api/v1/patients", + new RegisterPatientRequest("Eleanor", "Vance", new DateOnly(1962, 9, 14), "Female")); + resp.EnsureSuccessStatusCode(); + + var body = await resp.Content.ReadFromJsonAsync(); + return body!.RootElement.GetProperty("data").GetProperty("id").GetGuid(); + } + + private async Task IngestCriticalPotassiumAsync(Guid encounterId) + { + var ingestResp = await _http.PostAsJsonAsync( + $"/api/v1/encounters/{encounterId}/observations", + new BatchIngestRequest(new List + { + new("POTASSIUM_MEQ_L", 2.1m, "mEq/L", ObservationSource.Device, DateTimeOffset.UtcNow, null), + })); + ingestResp.EnsureSuccessStatusCode(); + + var body = await ingestResp.Content.ReadFromJsonAsync(); + var alertId = body!.RootElement.GetProperty("data").GetProperty("alertId").GetGuid(); + alertId.Should().NotBe(Guid.Empty); + return alertId; + } +} diff --git a/VigilCareClinicalAPI/BackgroundServices/Notifications/DischargeSummaryWorkerService.cs b/VigilCareClinicalAPI/BackgroundServices/Notifications/DischargeSummaryWorkerService.cs new file mode 100644 index 0000000..1307f21 --- /dev/null +++ b/VigilCareClinicalAPI/BackgroundServices/Notifications/DischargeSummaryWorkerService.cs @@ -0,0 +1,158 @@ +using System.Text; +using System.Text.Json; +using Microsoft.EntityFrameworkCore; +using Microsoft.Extensions.Options; +using Minio.DataModel.Args; +using RabbitMQ.Client; +using RabbitMQ.Client.Events; + +public sealed class DischargeSummaryWorkerService : BackgroundService +{ + private readonly IOptions _rabbitOpts; + private readonly IOptions _minioOpts; + private readonly IServiceScopeFactory _scopes; + private readonly ILogger _logger; + + public DischargeSummaryWorkerService( + IOptions rabbitOpts, + IOptions minioOpts, + IServiceScopeFactory scopes, + ILogger logger) + { + _rabbitOpts = rabbitOpts; + _minioOpts = minioOpts; + _scopes = scopes; + _logger = logger; + } + + protected override async Task ExecuteAsync(CancellationToken stoppingToken) + { + await Task.Delay(TimeSpan.FromSeconds(5), stoppingToken); + + var o = _rabbitOpts.Value; + var factory = new ConnectionFactory + { + HostName = o.Host, + Port = o.Port, + UserName = o.Username, + Password = o.Password, + DispatchConsumersAsync = true, + }; + + using var connection = factory.CreateConnection("discharge-summary-worker"); + using var channel = connection.CreateModel(); + + channel.BasicQos(0, prefetchCount: 3, global: false); + + var consumer = new AsyncEventingBasicConsumer(channel); + consumer.Received += async (sender, ea) => + { + await HandleDischargeSummaryAsync(channel, ea, stoppingToken); + }; + + channel.BasicConsume("notifications.discharge.queue", autoAck: false, consumer); + + _logger.LogInformation("DischargeSummaryWorkerService consuming notifications.discharge.queue"); + + await Task.Delay(Timeout.Infinite, stoppingToken); + } + + private async Task HandleDischargeSummaryAsync(IModel channel, BasicDeliverEventArgs ea, CancellationToken ct) + { + var payload = Encoding.UTF8.GetString(ea.Body.Span); + var doc = JsonDocument.Parse(payload); + var encounterId = Guid.Parse(doc.RootElement.GetProperty("encounterId").GetString()!); + + try + { + var summary = await BuildSummaryAsync(encounterId, ct); + await UploadToMinioAsync(encounterId, summary, ct); + + channel.BasicAck(ea.DeliveryTag, multiple: false); + + _logger.LogInformation( + "[DISCHARGE-SUMMARY] Uploaded for encounter {EncounterId}", encounterId); + } + catch (Exception ex) + { + _logger.LogError(ex, "DischargeSummaryWorker failed for encounter {EncounterId}", encounterId); + channel.BasicNack(ea.DeliveryTag, multiple: false, requeue: true); + } + } + + private async Task BuildSummaryAsync(Guid encounterId, CancellationToken ct) + { + await using var scope = _scopes.CreateAsyncScope(); + var db = scope.ServiceProvider.GetRequiredService(); + + var encounter = await db.Encounters + .Include(e => e.Patient) + .FirstOrDefaultAsync(e => e.Id == encounterId, ct); + + if (encounter is null) + return $"DISCHARGE SUMMARY\nEncounter {encounterId} not found in database."; + + var obsCount = await db.Observations.CountAsync(o => o.EncounterId == encounterId, ct); + var alertCount = await db.ClinicalAlerts.CountAsync(a => a.EncounterId == encounterId, ct); + var orders = await db.Orders + .Where(o => o.EncounterId == encounterId) + .Select(o => new { o.OrderType, o.Description, o.Status, o.OrderedAt, o.ResultedAt }) + .ToListAsync(ct); + + var sb = new StringBuilder(); + sb.AppendLine("=== VIGILCARE DISCHARGE SUMMARY ==="); + sb.AppendLine($"Generated: {DateTimeOffset.UtcNow:u}"); + sb.AppendLine(); + sb.AppendLine($"Patient: {encounter.Patient.FirstName} {encounter.Patient.LastName}"); + sb.AppendLine($"MRN: {encounter.Patient.Mrn}"); + sb.AppendLine($"Date of Birth: {encounter.Patient.DateOfBirth:yyyy-MM-dd}"); + sb.AppendLine(); + sb.AppendLine($"Encounter ID: {encounter.Id}"); + sb.AppendLine($"Type: {encounter.EncounterType}"); + sb.AppendLine($"Department: {encounter.Department}"); + sb.AppendLine($"Attending: {encounter.AttendingPhysician}"); + sb.AppendLine($"Admitted: {encounter.AdmittedAt:u}"); + sb.AppendLine($"Discharged: {encounter.DischargedAt:u}"); + sb.AppendLine(); + sb.AppendLine($"Observations Recorded: {obsCount}"); + sb.AppendLine($"Clinical Alerts: {alertCount}"); + sb.AppendLine(); + + if (orders.Any()) + { + sb.AppendLine("--- Orders ---"); + foreach (var o in orders) + sb.AppendLine($" [{o.Status}] {o.OrderType.ToDbString()}: {o.Description} (ordered: {o.OrderedAt:u})"); + } + + sb.AppendLine(); + sb.AppendLine("NOTE: This summary is generated automatically. Clinician review required."); + sb.AppendLine("NOTE: HIPAA Notice — This document contains protected health information."); + + return sb.ToString(); + } + + private async Task UploadToMinioAsync(Guid encounterId, string content, CancellationToken ct) + { + var mo = _minioOpts.Value; + var client = MinioClientFactory.Build(mo); + var bucket = mo.BucketName; + + var exists = await client.BucketExistsAsync( + new BucketExistsArgs().WithBucket(bucket), ct); + + if (!exists) + await client.MakeBucketAsync(new MakeBucketArgs().WithBucket(bucket), ct); + + var objectKey = $"discharge-summaries/{encounterId}/summary.pdf"; + var bytes = Encoding.UTF8.GetBytes(content); + + await client.PutObjectAsync(new PutObjectArgs() + .WithBucket(bucket) + .WithObject(objectKey) + .WithStreamData(new MemoryStream(bytes)) + .WithObjectSize(bytes.Length) + .WithContentType("text/plain"), + ct); + } +} \ No newline at end of file diff --git a/VigilCareClinicalAPI/BackgroundServices/Notifications/EscalationWorkerService.cs b/VigilCareClinicalAPI/BackgroundServices/Notifications/EscalationWorkerService.cs new file mode 100644 index 0000000..41b8eda --- /dev/null +++ b/VigilCareClinicalAPI/BackgroundServices/Notifications/EscalationWorkerService.cs @@ -0,0 +1,95 @@ +using System.Text; +using System.Text.Json; +using Microsoft.Extensions.Options; +using RabbitMQ.Client; +using RabbitMQ.Client.Events; + +public sealed class EscalationWorkerService : BackgroundService +{ + private readonly IOptions _opts; + private readonly IServiceScopeFactory _scopes; + private readonly ILogger _logger; + + public EscalationWorkerService( + IOptions opts, + IServiceScopeFactory scopes, + ILogger logger) + { + _opts = opts; + _scopes = scopes; + _logger = logger; + } + + protected override async Task ExecuteAsync(CancellationToken stoppingToken) + { + await Task.Delay(TimeSpan.FromSeconds(5), stoppingToken); + + var o = _opts.Value; + var factory = new ConnectionFactory + { + HostName = o.Host, + Port = o.Port, + UserName = o.Username, + Password = o.Password, + DispatchConsumersAsync = true, + }; + + using var connection = factory.CreateConnection("escalation-worker"); + using var channel = connection.CreateModel(); + + channel.BasicQos(0, prefetchCount: 5, global: false); + + var consumer = new AsyncEventingBasicConsumer(channel); + consumer.Received += async (sender, ea) => + { + await HandleEscalationAsync(channel, ea, stoppingToken); + }; + + channel.BasicConsume("alerts.escalation.queue", autoAck: false, consumer); + + _logger.LogInformation("EscalationWorkerService consuming alerts.escalation.queue"); + + await Task.Delay(Timeout.Infinite, stoppingToken); + } + + private async Task HandleEscalationAsync(IModel channel, BasicDeliverEventArgs ea, CancellationToken ct) + { + var payload = Encoding.UTF8.GetString(ea.Body.Span); + var doc = JsonDocument.Parse(payload); + var alertId = Guid.Parse(doc.RootElement.GetProperty("alertId").GetString()!); + var encounterId = doc.RootElement.GetProperty("encounterId").GetString(); + + _logger.LogCritical( + "[ESCALATION] Paging on-call backup — AlertId={AlertId} EncounterId={EncounterId}", + alertId, encounterId); + + try + { + await UpdateAlertStatusEscalatedAsync(alertId, ct); + channel.BasicAck(ea.DeliveryTag, multiple: false); + + _logger.LogWarning( + "[ESCALATION-DONE] Alert {AlertId} status → Escalated in PostgreSQL", alertId); + } + catch (Exception ex) + { + _logger.LogError(ex, "EscalationWorker failed for alert {AlertId}", alertId); + channel.BasicNack(ea.DeliveryTag, multiple: false, requeue: true); + } + } + + private async Task UpdateAlertStatusEscalatedAsync(Guid alertId, CancellationToken ct) + { + await using var scope = _scopes.CreateAsyncScope(); + var db = scope.ServiceProvider.GetRequiredService(); + + var alert = await db.ClinicalAlerts.FindAsync(new object[] { alertId }, ct); + if (alert is null) return; + + // Only escalate if still open — if acknowledged between NACK and TTL expiry, leave it. + if (alert.Status != AlertStatus.Open) return; + + alert.Status = AlertStatus.Escalated; + await db.SaveChangesAsync(ct); + } +} \ No newline at end of file diff --git a/VigilCareClinicalAPI/BackgroundServices/Notifications/NotificationPublisherService.cs b/VigilCareClinicalAPI/BackgroundServices/Notifications/NotificationPublisherService.cs new file mode 100644 index 0000000..b6eb4ef --- /dev/null +++ b/VigilCareClinicalAPI/BackgroundServices/Notifications/NotificationPublisherService.cs @@ -0,0 +1,151 @@ +using System.Text; +using System.Text.Json; +using Confluent.Kafka; +using Microsoft.Extensions.Options; +using RabbitMQ.Client; + +public sealed class NotificationPublisherService : BackgroundService +{ + private readonly IOptions _rabbitOpts; + private readonly KafkaOptions _kafkaOptions; + private readonly ILogger _logger; + + public NotificationPublisherService( + IOptions rabbitOpts, + IOptions kafkaOptions, + ILogger logger) + { + _rabbitOpts = rabbitOpts; + _kafkaOptions = kafkaOptions.Value; + _logger = logger; + } + + protected override async Task ExecuteAsync(CancellationToken stoppingToken) + { + // Short delay so topology provisioner finishes before first publish. + await Task.Delay(TimeSpan.FromSeconds(3), stoppingToken); + + var factory = BuildRabbitFactory(); + using var conn = factory.CreateConnection("notification-publisher"); + using var chan = conn.CreateModel(); + + var props = chan.CreateBasicProperties(); + props.Persistent = true; // messages survive broker restart + + var consumerConfig = new ConsumerConfig + { + BootstrapServers = _kafkaOptions.BootstrapServers, + GroupId = "notification-publisher", + AutoOffsetReset = AutoOffsetReset.Earliest, + EnableAutoCommit = false, + }; + + using var consumer = new ConsumerBuilder(consumerConfig).Build(); + consumer.Subscribe(new[] + { + _kafkaOptions.Topics.AlertGenerated, + _kafkaOptions.Topics.EncounterStatusChanged, + }); + + _logger.LogInformation("NotificationPublisherService started — consuming {Topics}", + string.Join(", ", _kafkaOptions.Topics.AlertGenerated, _kafkaOptions.Topics.EncounterStatusChanged)); + + try + { + while (!stoppingToken.IsCancellationRequested) + { + ConsumeResult? result; + try + { + result = consumer.Consume(TimeSpan.FromMilliseconds(500)); + } + catch (ConsumeException ex) + { + _logger.LogError(ex, "NotificationPublisher consume error"); + continue; + } + + if (result is null) continue; + + try + { + if (result.Topic == _kafkaOptions.Topics.AlertGenerated) + await HandleAlertGeneratedAsync(chan, props, result.Message.Value, stoppingToken); + else if (result.Topic == _kafkaOptions.Topics.EncounterStatusChanged) + await HandleEncounterStatusChangedAsync(chan, props, result.Message.Value, stoppingToken); + + consumer.Commit(result); + } + catch (Exception ex) + { + _logger.LogError(ex, "NotificationPublisher failed to process message from {Topic}", result.Topic); + // Do not commit — message will be reprocessed after consumer restart. + } + } + } + finally + { + consumer.Close(); + } + } + + private Task HandleAlertGeneratedAsync( + IModel chan, IBasicProperties props, string payload, CancellationToken ct) + { + var doc = JsonDocument.Parse(payload); + var severity = doc.RootElement.GetProperty("severity").GetString(); + + if (!string.Equals(severity, "Critical", StringComparison.OrdinalIgnoreCase)) + { + // Warning-level alerts are not paged — they appear in the clinician dashboard only. + return Task.CompletedTask; + } + + var body = Encoding.UTF8.GetBytes(payload); + chan.BasicPublish( + exchange: RabbitMqTopologyProvisioner.Exchange, + routingKey: RabbitMqTopologyProvisioner.PagingKey, + basicProperties: props, + body: body); + + var alertId = doc.RootElement.GetProperty("alertId").GetString(); + _logger.LogInformation("Published paging job to alerts.paging.queue for alert {AlertId}", alertId); + + return Task.CompletedTask; + } + + private Task HandleEncounterStatusChangedAsync( + IModel chan, IBasicProperties props, string payload, CancellationToken ct) + { + var doc = JsonDocument.Parse(payload); + var newStatus = doc.RootElement.GetProperty("newStatus").GetString(); + + if (!string.Equals(newStatus, "Discharged", StringComparison.OrdinalIgnoreCase)) + return Task.CompletedTask; + + var body = Encoding.UTF8.GetBytes(payload); + chan.BasicPublish( + exchange: RabbitMqTopologyProvisioner.Exchange, + routingKey: RabbitMqTopologyProvisioner.DischargeKey, + basicProperties: props, + body: body); + + var encounterId = doc.RootElement.GetProperty("encounterId").GetString(); + _logger.LogInformation("Published discharge summary job for encounter {EncounterId}", encounterId); + + return Task.CompletedTask; + } + + private IConnectionFactory BuildRabbitFactory() + { + var o = _rabbitOpts.Value; + return new ConnectionFactory + { + HostName = o.Host, + Port = o.Port, + UserName = o.Username, + Password = o.Password, + DispatchConsumersAsync = true, + }; + } +} \ No newline at end of file diff --git a/VigilCareClinicalAPI/BackgroundServices/Notifications/PagingWorkerService.cs b/VigilCareClinicalAPI/BackgroundServices/Notifications/PagingWorkerService.cs new file mode 100644 index 0000000..6c95966 --- /dev/null +++ b/VigilCareClinicalAPI/BackgroundServices/Notifications/PagingWorkerService.cs @@ -0,0 +1,121 @@ +using System.Text; +using System.Text.Json; +using Microsoft.EntityFrameworkCore; +using Microsoft.EntityFrameworkCore.Metadata; +using Microsoft.Extensions.Options; +using RabbitMQ.Client; +using RabbitMQ.Client.Events; + +public sealed class PagingWorkerService : BackgroundService +{ + private readonly IOptions _opts; + private readonly IServiceScopeFactory _scopes; + private readonly ILogger _logger; + + public PagingWorkerService( + IOptions opts, + IServiceScopeFactory scopes, + ILogger logger) + { + _opts = opts; + _scopes = scopes; + _logger = logger; + } + + protected override async Task ExecuteAsync(CancellationToken stoppingToken) + { + await Task.Delay(TimeSpan.FromSeconds(5), stoppingToken); // wait for topology + + var o = _opts.Value; + var factory = new ConnectionFactory + { + HostName = o.Host, + Port = o.Port, + UserName = o.Username, + Password = o.Password, + DispatchConsumersAsync = true, + }; + + using var connection = factory.CreateConnection("paging-worker"); + using var channel = connection.CreateModel(); + + // Prefetch 1: hold exactly one page at a time. + // Releasing the next message only after ACK or NACK keeps the worker + // from consuming more alerts than it can actively page on. + channel.BasicQos(prefetchSize: 0, prefetchCount: 1, global: false); + + var consumer = new AsyncEventingBasicConsumer(channel); + consumer.Received += async (sender, ea) => + { + try + { + await HandlePageAsync(channel, ea, stoppingToken); + } + catch (Exception ex) + { + _logger.LogError(ex, "PagingWorker failed — NACKing to DLQ"); + channel.BasicNack(ea.DeliveryTag, multiple: false, requeue: false); + } + }; + + channel.BasicConsume("alerts.paging.queue", autoAck: false, consumer); + + _logger.LogInformation("PagingWorkerService consuming alerts.paging.queue (prefetch=1, timeout={Timeout}ms)", + o.PagingAckTimeoutMs); + + await Task.Delay(Timeout.Infinite, stoppingToken); + } + + private async Task HandlePageAsync(RabbitMQ.Client.IModel channel, BasicDeliverEventArgs ea, CancellationToken ct) + { + var o = _opts.Value; + var payload = Encoding.UTF8.GetString(ea.Body.Span); + var doc = JsonDocument.Parse(payload); + var alertId = Guid.Parse(doc.RootElement.GetProperty("alertId").GetString()!); + var encounterId = doc.RootElement.GetProperty("encounterId").GetString(); + var details = doc.RootElement.TryGetProperty("details", out var d) ? d.GetString() : null; + var physician = doc.RootElement.TryGetProperty("attendingPhysician", out var p) ? p.GetString() : "unknown"; + + _logger.LogWarning( + "[PAGE] Paging attending physician '{Physician}' for encounter {EncounterId} — " + + "AlertId={AlertId} Details={Details}", + physician, encounterId, alertId, details); + + var deadline = DateTimeOffset.UtcNow.AddMilliseconds(o.PagingAckTimeoutMs); + + while (DateTimeOffset.UtcNow < deadline && !ct.IsCancellationRequested) + { + await Task.Delay(2_000, ct); + + var acknowledged = await IsAlertAcknowledgedAsync(alertId, ct); + if (acknowledged) + { + _logger.LogInformation( + "[PAGE-ACK] Alert {AlertId} acknowledged — ACKing RabbitMQ message", alertId); + channel.BasicAck(ea.DeliveryTag, multiple: false); + return; + } + } + + // Timeout: NACK with requeue=false → message routes to alerts.paging.dlq. + // After x-message-ttl expires on the DLQ, it re-routes to alerts.escalation.queue. + _logger.LogWarning( + "[PAGE-TIMEOUT] No acknowledgment within {TimeoutMs}ms for alert {AlertId} — " + + "NACKing to DLQ for escalation", + o.PagingAckTimeoutMs, alertId); + channel.BasicNack(ea.DeliveryTag, multiple: false, requeue: false); + } + + private async Task IsAlertAcknowledgedAsync(Guid alertId, CancellationToken ct) + { + await using var scope = _scopes.CreateAsyncScope(); + var db = scope.ServiceProvider.GetRequiredService(); + + var status = await db.ClinicalAlerts + .Where(a => a.Id == alertId) + .Select(a => a.Status) + .FirstOrDefaultAsync(ct); + + return status is AlertStatus.Acknowledged or AlertStatus.Resolved; + } +} \ No newline at end of file diff --git a/VigilCareClinicalAPI/Configuration/MinioOptions.cs b/VigilCareClinicalAPI/Configuration/MinioOptions.cs new file mode 100644 index 0000000..31df5b0 --- /dev/null +++ b/VigilCareClinicalAPI/Configuration/MinioOptions.cs @@ -0,0 +1,9 @@ +public sealed class MinioOptions +{ + public const string Section = "Minio"; + public string Endpoint { get; init; } = "localhost:9005"; + public string AccessKey { get; init; } = "minioadmin"; + public string SecretKey { get; init; } = "minioadmin"; + public string BucketName { get; init; } = "vigilcare"; + public bool UseSSL { get; init; } = false; +} \ No newline at end of file diff --git a/VigilCareClinicalAPI/Configuration/RabbitMqOptions.cs b/VigilCareClinicalAPI/Configuration/RabbitMqOptions.cs new file mode 100644 index 0000000..487a384 --- /dev/null +++ b/VigilCareClinicalAPI/Configuration/RabbitMqOptions.cs @@ -0,0 +1,11 @@ +public sealed class RabbitMqOptions +{ + public const string Section = "RabbitMq"; + public string Host { get; init; } = "localhost"; + public int Port { get; init; } = 5674; + public string Username { get; init; } = "guest"; + public string Password { get; init; } = "guest"; + // Drives both the paging worker poll timeout and the DLQ x-message-ttl. + // In production: 300000 (5 min). In tests: 5000 (5 sec). + public int PagingAckTimeoutMs { get; init; } = 300000; +} \ No newline at end of file diff --git a/VigilCareClinicalAPI/Notifications/RabbitMqTopologyProvisioner.cs b/VigilCareClinicalAPI/Notifications/RabbitMqTopologyProvisioner.cs new file mode 100644 index 0000000..daa4752 --- /dev/null +++ b/VigilCareClinicalAPI/Notifications/RabbitMqTopologyProvisioner.cs @@ -0,0 +1,139 @@ +using Microsoft.Extensions.Hosting; +using Microsoft.Extensions.Options; +using RabbitMQ.Client; +using RabbitMQ.Client.Exceptions; + +public sealed class RabbitMqTopologyProvisioner : IHostedService +{ + public const string Exchange = "clinical.notifications.exchange"; + public const string PagingKey = "alerts.paging"; + public const string EscalKey = "alerts.escalation"; + public const string DischargeKey = "notifications.discharge"; + + private readonly RabbitMqOptions _opts; + private readonly IHostEnvironment _env; + private readonly ILogger _logger; + + public RabbitMqTopologyProvisioner( + IOptions opts, + IHostEnvironment env, + ILogger logger) + { + _opts = opts.Value; + _env = env; + _logger = logger; + } + + public Task StartAsync(CancellationToken ct) + { + var factory = BuildFactory(); + using var connection = factory.CreateConnection(); + using var channel = connection.CreateModel(); + + if (_env.IsEnvironment("Testing")) + { + // Use a throwaway channel — a failed purge on a missing queue closes the + // channel, which would break the declare calls below. + using var cleanup = connection.CreateModel(); + + try + { + // Dev runs provision the DLQ with a 5-minute TTL; delete it so the + // test timeout (5 s) is applied when the queue is re-declared below. + cleanup.QueueDelete("alerts.paging.dlq", ifUnused: false, ifEmpty: false); + } + catch (OperationInterruptedException ex) + { + _logger.LogDebug(ex, "DLQ delete skipped — queue may not exist yet"); + } + + foreach (var queue in new[] { "alerts.paging.queue", "alerts.escalation.queue" }) + { + try { cleanup.QueuePurge(queue); } + catch (OperationInterruptedException ex) + { + _logger.LogDebug(ex, "Queue purge skipped for {Queue}", queue); + } + } + } + + // Direct exchange — routing key determines destination queue. + channel.ExchangeDeclare(Exchange, ExchangeType.Direct, durable: true); + + // --- alerts.paging.queue --- + // Dead-letters to the default exchange with routing key = alerts.paging.dlq. + // Prefetch is set per-consumer, not here. + channel.QueueDeclare( + queue: "alerts.paging.queue", + durable: true, + exclusive: false, + autoDelete: false, + arguments: new Dictionary + { + ["x-dead-letter-exchange"] = "", // default exchange + ["x-dead-letter-routing-key"] = "alerts.paging.dlq", + }); + channel.QueueBind("alerts.paging.queue", Exchange, PagingKey); + + // --- alerts.paging.dlq --- + // Messages land here after NACK from the paging worker. + // After x-message-ttl expires, re-routes to clinical.notifications.exchange + // with routing key alerts.escalation → reaches alerts.escalation.queue. + channel.QueueDeclare( + queue: "alerts.paging.dlq", + durable: true, + exclusive: false, + autoDelete: false, + arguments: new Dictionary + { + ["x-message-ttl"] = (int)_opts.PagingAckTimeoutMs, + ["x-dead-letter-exchange"] = Exchange, + ["x-dead-letter-routing-key"] = EscalKey, + }); + // DLQ is reached via the default exchange — no binding to the direct exchange needed. + + // --- alerts.escalation.queue --- + channel.QueueDeclare( + queue: "alerts.escalation.queue", + durable: true, + exclusive: false, + autoDelete: false, + arguments: null); + channel.QueueBind("alerts.escalation.queue", Exchange, EscalKey); + + // --- notifications.discharge.queue --- + channel.QueueDeclare( + queue: "notifications.discharge.queue", + durable: true, + exclusive: false, + autoDelete: false, + arguments: null); + channel.QueueBind("notifications.discharge.queue", Exchange, DischargeKey); + + // --- notifications.appointment.queue (placeholder) --- + channel.QueueDeclare( + queue: "notifications.appointment.queue", + durable: true, + exclusive: false, + autoDelete: false, + arguments: null); + channel.QueueBind("notifications.appointment.queue", Exchange, "notifications.appointment"); + + _logger.LogInformation( + "RabbitMQ topology provisioned. Exchange={Exchange} PagingDlqTtlMs={Ttl}", + Exchange, _opts.PagingAckTimeoutMs); + + return Task.CompletedTask; + } + + public Task StopAsync(CancellationToken ct) => Task.CompletedTask; + + public IConnectionFactory BuildFactory() => new ConnectionFactory + { + HostName = _opts.Host, + Port = _opts.Port, + UserName = _opts.Username, + Password = _opts.Password, + DispatchConsumersAsync = true, + }; +} \ No newline at end of file diff --git a/VigilCareClinicalAPI/Program.cs b/VigilCareClinicalAPI/Program.cs index e79f8ec..9353a7e 100644 --- a/VigilCareClinicalAPI/Program.cs +++ b/VigilCareClinicalAPI/Program.cs @@ -41,6 +41,14 @@ try builder.Services.AddSingleton( new ElasticsearchClient(new Uri(esOptions.Uri))); + builder.Services.Configure( + builder.Configuration.GetSection(RabbitMqOptions.Section)); + builder.Services.Configure( + builder.Configuration.GetSection(MinioOptions.Section)); + + builder.Services.AddSingleton(); + builder.Services.AddHostedService(sp => sp.GetRequiredService()); + builder.Services.AddScoped(); builder.Services.AddScoped(); builder.Services.AddScoped(); @@ -56,6 +64,10 @@ try builder.Services.AddHostedService(); builder.Services.AddHostedService(); builder.Services.AddHostedService(); + builder.Services.AddHostedService(); + builder.Services.AddHostedService(); + builder.Services.AddHostedService(); + builder.Services.AddHostedService(); builder.Services.AddControllers() .AddJsonOptions(opts => diff --git a/VigilCareClinicalAPI/Services/ObservationService.cs b/VigilCareClinicalAPI/Services/ObservationService.cs index 93ad13b..c8a334d 100644 --- a/VigilCareClinicalAPI/Services/ObservationService.cs +++ b/VigilCareClinicalAPI/Services/ObservationService.cs @@ -116,6 +116,8 @@ public class ObservationService : IObservationService department = encounter.Department, alertType = alert.AlertType.ToDbString(), severity = alert.Severity.ToDbString(), + details = alert.Details, + attendingPhysician = encounter.AttendingPhysician, triggeredAt = alert.TriggeredAt, partitionKey = encounterId.ToString() }, encounterId.ToString())); diff --git a/VigilCareClinicalAPI/Storage/MinioClientFactory.cs b/VigilCareClinicalAPI/Storage/MinioClientFactory.cs new file mode 100644 index 0000000..aa90efd --- /dev/null +++ b/VigilCareClinicalAPI/Storage/MinioClientFactory.cs @@ -0,0 +1,15 @@ +using Minio; + +public static class MinioClientFactory +{ + public static IMinioClient Build(MinioOptions opts) + { + var client = new MinioClient() + .WithEndpoint(opts.Endpoint) + .WithCredentials(opts.AccessKey, opts.SecretKey); + + if (opts.UseSSL) client = client.WithSSL(); + + return client.Build(); + } +} \ No newline at end of file diff --git a/VigilCareClinicalAPI/VigilCareClinicalAPI.csproj b/VigilCareClinicalAPI/VigilCareClinicalAPI.csproj index a643762..b7669c9 100644 --- a/VigilCareClinicalAPI/VigilCareClinicalAPI.csproj +++ b/VigilCareClinicalAPI/VigilCareClinicalAPI.csproj @@ -16,7 +16,9 @@ runtime; build; native; contentfiles; analyzers; buildtransitive all + + diff --git a/VigilCareClinicalAPI/appsettings.Testing.json b/VigilCareClinicalAPI/appsettings.Testing.json index 27d4508..a6617e6 100644 --- a/VigilCareClinicalAPI/appsettings.Testing.json +++ b/VigilCareClinicalAPI/appsettings.Testing.json @@ -10,5 +10,8 @@ "WriteTo": [ { "Name": "Console" } ] + }, + "RabbitMq": { + "PagingAckTimeoutMs": 5000 } } diff --git a/VigilCareClinicalAPI/appsettings.json b/VigilCareClinicalAPI/appsettings.json index ad29fdf..c033486 100644 --- a/VigilCareClinicalAPI/appsettings.json +++ b/VigilCareClinicalAPI/appsettings.json @@ -54,5 +54,19 @@ "Observations": "observations", "ClinicalAlerts": "clinical_alerts" } + }, + "RabbitMq": { + "Host": "localhost", + "Port": 5674, + "Username": "guest", + "Password": "guest", + "PagingAckTimeoutMs": 300000 + }, + "Minio": { + "Endpoint": "localhost:9005", + "AccessKey": "minioadmin", + "SecretKey": "minioadmin", + "BucketName": "vigilcare", + "UseSSL": false } } diff --git a/docker-compose.yml b/docker-compose.yml index c09c153..695c52d 100644 --- a/docker-compose.yml +++ b/docker-compose.yml @@ -69,8 +69,37 @@ services: timeout: 5s retries: 10 + rabbitmq: + image: rabbitmq:3.13-management-alpine + container_name: vigilcare_rabbitmq + ports: + - "5674:5672" # AMQP + - "15674:15672" # management UI + environment: + RABBITMQ_DEFAULT_USER: guest + RABBITMQ_DEFAULT_PASS: guest + healthcheck: + test: ["CMD", "rabbitmq-diagnostics", "ping"] + interval: 10s + timeout: 5s + retries: 5 + + minio: + image: minio/minio:RELEASE.2024-07-04T14-25-45Z + container_name: vigilcare_minio + command: server /data --console-address ":9001" + ports: + - "9005:9000" # S3 API + - "9006:9001" # console + environment: + MINIO_ROOT_USER: minioadmin + MINIO_ROOT_PASSWORD: minioadmin + volumes: + - minio_data:/data + volumes: pg_data: seq_data: kafka_data: - es_data: \ No newline at end of file + es_data: + minio_data: \ No newline at end of file diff --git a/scripts/run-notification-pipeline-tests.sh b/scripts/run-notification-pipeline-tests.sh new file mode 100755 index 0000000..72fd518 --- /dev/null +++ b/scripts/run-notification-pipeline-tests.sh @@ -0,0 +1,474 @@ +#!/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}" +RABBITMQ_MGMT_URL="${RABBITMQ_MGMT_URL:-http://localhost:15674}" +RABBITMQ_USER="${RABBITMQ_USER:-guest}" +RABBITMQ_PASS="${RABBITMQ_PASS:-guest}" +MINIO_ENDPOINT="${MINIO_ENDPOINT:-localhost:9005}" +MINIO_ACCESS_KEY="${MINIO_ACCESS_KEY:-minioadmin}" +MINIO_SECRET_KEY="${MINIO_SECRET_KEY:-minioadmin}" +MINIO_BUCKET="${MINIO_BUCKET:-vigilcare}" + +PGHOST="${PGHOST:-localhost}" +PGPORT="${PGPORT:-5436}" +PGDATABASE="${PGDATABASE:-vigilcare}" +PGUSER="${PGUSER:-postgres}" +PGPASSWORD="${PGPASSWORD:-password}" + +RELAY_WAIT_SECS="${RELAY_WAIT_SECS:-45}" +KAFKA_READY_WAIT_SECS="${KAFKA_READY_WAIT_SECS:-30}" +PAGING_PIPELINE_WAIT_SECS="${PAGING_PIPELINE_WAIT_SECS:-30}" +MINIO_WAIT_SECS="${MINIO_WAIT_SECS:-15}" +ACK_PIPELINE_WAIT_SECS="${ACK_PIPELINE_WAIT_SECS:-10}" + +SCRIPT_RUN_ID="$(date -u +"%Y%m%d%H%M%S")" +RECORDED_AT="$(date -u +"%Y-%m-%dT%H:%M:%SZ")" + +EXPECTED_QUEUES=( + "alerts.paging.queue" + "alerts.paging.dlq" + "alerts.escalation.queue" + "notifications.discharge.queue" + "notifications.appointment.queue" +) + +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 "$@" +} + +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 +} + +rabbit_api() { + curl -sS -u "${RABBITMQ_USER}:${RABBITMQ_PASS}" "${RABBITMQ_MGMT_URL}/api/${1}" +} + +queue_field() { + local queue="$1" + local field="$2" + rabbit_api "queues/%2F/${queue}" | jq -r ".${field} // 0" +} + +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 +} + +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 +} + +wait_for_alert_outbox_relayed() { + local encounter_id="$1" + local elapsed=0 + while (( elapsed < RELAY_WAIT_SECS )); do + local outbox_id + 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")" + if [[ -n "${outbox_id}" ]]; then + 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 + fi + sleep 1 + elapsed=$((elapsed + 1)) + done + echo "alert.generated outbox for encounter ${encounter_id} was not relayed within ${RELAY_WAIT_SECS}s" + return 1 +} + +wait_for_paging_activity() { + local elapsed=0 + while (( elapsed < PAGING_PIPELINE_WAIT_SECS )); do + local unacked ready total + unacked="$(queue_field "alerts.paging.queue" "messages_unacknowledged")" + ready="$(queue_field "alerts.paging.queue" "messages_ready")" + total="$(queue_field "alerts.paging.queue" "messages")" + if [[ "${unacked}" -ge 1 || "${ready}" -ge 1 || "${total}" -ge 1 ]]; then + return 0 + fi + sleep 1 + elapsed=$((elapsed + 1)) + done + echo "No paging activity on alerts.paging.queue within ${PAGING_PIPELINE_WAIT_SECS}s" + return 1 +} + +wait_for_alert_status() { + local alert_id="$1" + local expected="$2" + local timeout_secs="$3" + local elapsed=0 + while (( elapsed < timeout_secs )); do + local status + status="$(psql_cmd "SELECT status FROM clinical_alerts WHERE id = '${alert_id}'")" + if [[ "${status}" == "${expected}" ]]; then + return 0 + fi + sleep 1 + elapsed=$((elapsed + 1)) + done + local final + final="$(psql_cmd "SELECT status FROM clinical_alerts WHERE id = '${alert_id}'")" + echo "Alert ${alert_id} did not reach status ${expected} within ${timeout_secs}s (last status=${final:-unknown})" + return 1 +} + +minio_object_exists() { + local object_key="$1" + if command -v mc >/dev/null 2>&1; then + mc alias set local "http://${MINIO_ENDPOINT}" "${MINIO_ACCESS_KEY}" "${MINIO_SECRET_KEY}" --api S3v4 >/dev/null 2>&1 || true + mc stat "local/${MINIO_BUCKET}/${object_key}" >/dev/null 2>&1 + return $? + fi + + docker run --rm --network host --entrypoint /bin/sh minio/mc:latest \ + -c "mc alias set local http://${MINIO_ENDPOINT} ${MINIO_ACCESS_KEY} ${MINIO_SECRET_KEY} >/dev/null 2>&1 && mc stat local/${MINIO_BUCKET}/${object_key}" \ + >/dev/null 2>&1 +} + +create_patient() { + local payload='{"firstName":"Notify","lastName":"Verifier","dateOfBirth":"1972-03-18","gender":"Female"}' + local resp + resp="$(request POST "${BASE_URL}/api/v1/patients" "${payload}")" + assert_status "201" "${resp}" + jq -r '.data.id' "${resp}" +} + +create_encounter() { + local patient_id="$1" + local suffix="$2" + local payload + payload="$(jq -nc \ + --arg physician "Dr. Notify ${suffix}" \ + '{encounterType:"Inpatient",department:"ICU",attendingPhysician:$physician}')" + local resp + resp="$(request POST "${BASE_URL}/api/v1/patients/${patient_id}/encounters" "${payload}")" + assert_status "201" "${resp}" + jq -r '.data.id' "${resp}" +} + +ingest_critical_potassium() { + local encounter_id="$1" + local idempotency_key="$2" + local payload + payload="$(jq -nc \ + --arg recordedAt "${RECORDED_AT}" \ + --arg key "${idempotency_key}" \ + '{observations:[{observationCode:"POTASSIUM_MEQ_L",value:2.1,unit:"mEq/L",source:"LAB",recordedAt:$recordedAt,idempotencyKey:$key}]}')" + local resp + resp="$(request POST "${BASE_URL}/api/v1/encounters/${encounter_id}/observations" "${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 + jq -r '.data.alertId' "${resp}" +} + +TOTAL_STEPS=8 + +echo "Running RabbitMQ + MinIO notification pipeline verification against ${BASE_URL}" +echo "Script run id: ${SCRIPT_RUN_ID}" + +echo "" +echo "[0/${TOTAL_STEPS}] Preflight — API, Postgres, Kafka, RabbitMQ, and MinIO 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 ! wait_for_kafka; then + exit 1 +fi + +rabbit_health="$(rabbit_api "health/checks/alarms" 2>/dev/null | jq -r '.status // empty' || true)" +if [[ -z "${rabbit_health}" ]]; then + echo "RabbitMQ management API not reachable at ${RABBITMQ_MGMT_URL}." + echo "Start RabbitMQ with: docker compose up -d rabbitmq" + exit 1 +fi + +if ! curl -sS -o /dev/null -w "%{http_code}" "http://${MINIO_ENDPOINT}/minio/health/live" | grep -q '^200$'; then + echo "MinIO not reachable at http://${MINIO_ENDPOINT}." + echo "Start MinIO with: docker compose up -d minio" + exit 1 +fi +echo "OK: API, Postgres, Kafka, RabbitMQ, and MinIO are up" + +echo "" +echo "[1/${TOTAL_STEPS}] Verifying RabbitMQ topology (exchange, five queues, DLQ arguments)" +exchange_name="$(rabbit_api "exchanges/%2F/clinical.notifications.exchange" | jq -r '.name // empty')" +if [[ "${exchange_name}" != "clinical.notifications.exchange" ]]; then + echo "Exchange clinical.notifications.exchange not found. Start the API so RabbitMqTopologyProvisioner runs." + exit 1 +fi + +for queue in "${EXPECTED_QUEUES[@]}"; do + queue_name="$(rabbit_api "queues/%2F/${queue}" | jq -r '.name // empty')" + if [[ "${queue_name}" != "${queue}" ]]; then + echo "Queue ${queue} not found (got '${queue_name}')." + exit 1 + fi +done + +dlq_ttl_ms="$(rabbit_api "queues/%2F/alerts.paging.dlq" | jq -r '.arguments["x-message-ttl"] // empty')" +dlq_dlx="$(rabbit_api "queues/%2F/alerts.paging.dlq" | jq -r '.arguments["x-dead-letter-exchange"] // empty')" +dlq_dlk="$(rabbit_api "queues/%2F/alerts.paging.dlq" | jq -r '.arguments["x-dead-letter-routing-key"] // empty')" +paging_dlk="$(rabbit_api "queues/%2F/alerts.paging.queue" | jq -r '.arguments["x-dead-letter-routing-key"] // empty')" + +if [[ -z "${dlq_ttl_ms}" ]]; then + echo "alerts.paging.dlq is missing x-message-ttl" + exit 1 +fi +if [[ "${dlq_dlx}" != "clinical.notifications.exchange" ]]; then + echo "Expected DLQ x-dead-letter-exchange=clinical.notifications.exchange, got '${dlq_dlx}'" + exit 1 +fi +if [[ "${dlq_dlk}" != "alerts.escalation" ]]; then + echo "Expected DLQ x-dead-letter-routing-key=alerts.escalation, got '${dlq_dlk}'" + exit 1 +fi +if [[ "${paging_dlk}" != "alerts.paging.dlq" ]]; then + echo "Expected paging queue DLK=alerts.paging.dlq, got '${paging_dlk}'" + exit 1 +fi + +ttl_sec=$(( dlq_ttl_ms / 1000 )) +ESCALATION_WAIT_SECS="${ESCALATION_WAIT_SECS:-$(( ttl_sec * 2 + 25 ))}" +echo "OK: exchange + 5 queues provisioned; DLQ TTL=${dlq_ttl_ms}ms (escalation wait budget=${ESCALATION_WAIT_SECS}s)" + +echo "" +echo "[2/${TOTAL_STEPS}] Verifying non-alert observation does not enqueue paging jobs" +patient_no_page="$(create_patient)" +encounter_no_page="$(create_encounter "${patient_no_page}" "NoPage")" + +paging_before="$(queue_field "alerts.paging.queue" "messages")" +normal_payload="$(jq -nc \ + --arg recordedAt "${RECORDED_AT}" \ + --arg key "notify-normal-${SCRIPT_RUN_ID}" \ + '{observations:[{observationCode:"POTASSIUM_MEQ_L",value:3.8,unit:"mEq/L",source:"LAB",recordedAt:$recordedAt,idempotencyKey:$key}]}')" +resp="$(request POST "${BASE_URL}/api/v1/encounters/${encounter_no_page}/observations" "${normal_payload}")" +assert_status "201" "${resp}" +if [[ "$(jq -r '.data.alertGenerated' "${resp}")" != "false" ]]; then + echo "Expected normal potassium ingest to skip alert generation" + exit 1 +fi + +sleep 3 +paging_after="$(queue_field "alerts.paging.queue" "messages")" +if [[ "${paging_after}" -gt "${paging_before}" ]]; then + echo "Paging queue grew after non-alert ingest (${paging_before} -> ${paging_after})" + exit 1 +fi +echo "OK: normal observation produced no paging queue traffic" + +echo "" +echo "[3/${TOTAL_STEPS}] Verifying critical alert reaches alerts.paging.queue via Kafka bridge" +patient_page="$(create_patient)" +encounter_page="$(create_encounter "${patient_page}" "Page")" +alert_page_id="$(ingest_critical_potassium "${encounter_page}" "notify-critical-page-${SCRIPT_RUN_ID}")" + +if ! wait_for_alert_outbox_relayed "${encounter_page}"; then + exit 1 +fi +if ! wait_for_paging_activity; then + echo "Hint: confirm NotificationPublisherService is running and alert.generated severity is Critical." + exit 1 +fi +echo "OK: critical alert ${alert_page_id} reached alerts.paging.queue (worker holding or ready message present)" + +echo "" +echo "[4/${TOTAL_STEPS}] Verifying acknowledge-before-timeout path (no escalation)" +ack_payload='{"clinicianId":"DR-SCRIPT","note":"Acknowledged from notification verification script."}' +resp="$(request POST "${BASE_URL}/api/v1/alerts/${alert_page_id}/acknowledge" "${ack_payload}")" +assert_status "200" "${resp}" + +sleep "${ACK_PIPELINE_WAIT_SECS}" +if ! wait_for_alert_status "${alert_page_id}" "ACKNOWLEDGED" 5; then + exit 1 +fi +echo "OK: alert ${alert_page_id} is ACKNOWLEDGED (not escalated)" + +echo "" +echo "[5/${TOTAL_STEPS}] Verifying full DLQ escalation path (unacknowledged critical alert)" +patient_escalate="$(create_patient)" +encounter_escalate="$(create_encounter "${patient_escalate}" "Escalate")" +alert_escalate_id="$(ingest_critical_potassium "${encounter_escalate}" "notify-critical-escalate-${SCRIPT_RUN_ID}")" + +if ! wait_for_alert_outbox_relayed "${encounter_escalate}"; then + exit 1 +fi +if ! wait_for_paging_activity; then + exit 1 +fi + +echo "Waiting up to ${ESCALATION_WAIT_SECS}s for paging timeout + DLQ TTL + escalation worker..." +if ! wait_for_alert_status "${alert_escalate_id}" "ESCALATED" "${ESCALATION_WAIT_SECS}"; then + echo "Hint: default DLQ TTL is 300000ms (~10 min total). Set PagingAckTimeoutMs=5000 in appsettings.json and restart the API for a faster run, or raise ESCALATION_WAIT_SECS." + exit 1 +fi +echo "OK: alert ${alert_escalate_id} escalated to ESCALATED in PostgreSQL" + +echo "" +echo "[6/${TOTAL_STEPS}] Verifying discharge summary upload to MinIO" +patient_discharge="$(create_patient)" +encounter_discharge="$(create_encounter "${patient_discharge}" "Discharge")" +discharge_payload='{"status":"Discharged"}' +resp="$(request PATCH "${BASE_URL}/api/v1/encounters/${encounter_discharge}/status" "${discharge_payload}")" +assert_status "200" "${resp}" + +discharge_outbox_id="$(psql_cmd "SELECT id FROM outbox_events WHERE topic = 'encounter.status.changed' AND partition_key = '${encounter_discharge}' ORDER BY created_at DESC LIMIT 1")" +if [[ -z "${discharge_outbox_id}" ]]; then + echo "No encounter.status.changed outbox row for encounter ${encounter_discharge}" + exit 1 +fi +wait_for_outbox_processed "${discharge_outbox_id}" + +object_key="discharge-summaries/${encounter_discharge}/summary.pdf" +elapsed=0 +while (( elapsed < MINIO_WAIT_SECS )); do + if minio_object_exists "${object_key}"; then + break + fi + sleep 1 + elapsed=$((elapsed + 1)) +done + +if ! minio_object_exists "${object_key}"; then + echo "MinIO object not found: ${MINIO_BUCKET}/${object_key}" + echo "Hint: confirm DischargeSummaryWorkerService is running and notifications.discharge.queue is draining." + exit 1 +fi +echo "OK: discharge summary object exists at ${MINIO_BUCKET}/${object_key}" + +echo "" +echo "[7/${TOTAL_STEPS}] Verifying discharge queue drained after summary upload" +elapsed=0 +while (( elapsed < MINIO_WAIT_SECS )); do + discharge_ready="$(queue_field "notifications.discharge.queue" "messages_ready")" + discharge_unacked="$(queue_field "notifications.discharge.queue" "messages_unacknowledged")" + if [[ "${discharge_ready}" == "0" && "${discharge_unacked}" == "0" ]]; then + break + fi + sleep 1 + elapsed=$((elapsed + 1)) +done + +discharge_ready="$(queue_field "notifications.discharge.queue" "messages_ready")" +discharge_unacked="$(queue_field "notifications.discharge.queue" "messages_unacknowledged")" +if [[ "${discharge_ready}" != "0" || "${discharge_unacked}" != "0" ]]; then + echo "Expected notifications.discharge.queue to be empty (ready=${discharge_ready}, unacked=${discharge_unacked})" + exit 1 +fi +echo "OK: notifications.discharge.queue is drained" + +echo "" +echo "All ${TOTAL_STEPS} notification pipeline checks passed." +echo "" +echo "Prerequisites: docker compose up -d && dotnet run --project VigilCareClinicalAPI" +echo "Optional: set PagingAckTimeoutMs=5000 in appsettings.json (restart API) for ~15s escalation instead of ~10 min." +echo "Optional: ESCALATION_WAIT_SECS=330 when using the production 300000ms DLQ TTL."