vigilcare-clinical/VigilCareClinicalAPI/BackgroundServices/Notifications/PagingWorkerService.cs

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 = RabbitMqConnectionFactory.Create(o, 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 (OperationCanceledException) when (stoppingToken.IsCancellationRequested)
            {
                // Host is stopping while we were waiting for ack — requeue so restart does not false-escalate.
                _logger.LogInformation("PagingWorker stopping — requeueing in-flight page message");
                channel.BasicNack(ea.DeliveryTag, multiple: false, requeue: true);
            }
            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;
    }
}