feature: RabbitMQ Notification Workers and DLQ Escalation

2026-06-17 02:37:30 +08:00
parent ec780144c6
commit 0246143815
17 changed files with 1410 additions and 2 deletions
@@ -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<RabbitMqOptions> _rabbitOpts;
+    private readonly IOptions<MinioOptions> _minioOpts;
+    private readonly IServiceScopeFactory _scopes;
+    private readonly ILogger<DischargeSummaryWorkerService> _logger;
+
+    public DischargeSummaryWorkerService(
+        IOptions<RabbitMqOptions> rabbitOpts,
+        IOptions<MinioOptions> minioOpts,
+        IServiceScopeFactory scopes,
+        ILogger<DischargeSummaryWorkerService> 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<string> BuildSummaryAsync(Guid encounterId, CancellationToken ct)
+    {
+        await using var scope = _scopes.CreateAsyncScope();
+        var db = scope.ServiceProvider.GetRequiredService<AppDbContext>();
+
+        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);
+    }
+}
@@ -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<RabbitMqOptions> _opts;
+    private readonly IServiceScopeFactory _scopes;
+    private readonly ILogger<EscalationWorkerService> _logger;
+
+    public EscalationWorkerService(
+        IOptions<RabbitMqOptions> opts,
+        IServiceScopeFactory scopes,
+        ILogger<EscalationWorkerService> 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<AppDbContext>();
+
+        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);
+    }
+}
@@ -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<RabbitMqOptions> _rabbitOpts;
+    private readonly KafkaOptions _kafkaOptions;
+    private readonly ILogger<NotificationPublisherService> _logger;
+
+    public NotificationPublisherService(
+        IOptions<RabbitMqOptions> rabbitOpts,
+        IOptions<KafkaOptions> kafkaOptions,
+        ILogger<NotificationPublisherService> 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<string, string>(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<string, string>? 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,
+        };
+    }
+}
@@ -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<RabbitMqOptions> _opts;
+    private readonly IServiceScopeFactory _scopes;
+    private readonly ILogger<PagingWorkerService> _logger;
+
+    public PagingWorkerService(
+        IOptions<RabbitMqOptions> opts,
+        IServiceScopeFactory scopes,
+        ILogger<PagingWorkerService> 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<bool> IsAlertAcknowledgedAsync(Guid alertId, CancellationToken ct)
+    {
+        await using var scope = _scopes.CreateAsyncScope();
+        var db = scope.ServiceProvider.GetRequiredService<AppDbContext>();
+
+        var status = await db.ClinicalAlerts
+            .Where(a => a.Id == alertId)
+            .Select(a => a.Status)
+            .FirstOrDefaultAsync(ct);
+
+        return status is AlertStatus.Acknowledged or AlertStatus.Resolved;
+    }
+}