vigilcare-clinical/VigilCare.WardGateway/BackgroundService/LocalPagingWorkerService.cs

using System.Text;
using System.Text.Json;
using Microsoft.EntityFrameworkCore;
using Microsoft.Extensions.Options;
using RabbitMQ.Client;
using RabbitMQ.Client.Events;

public sealed class LocalPagingWorkerService : BackgroundService
{
    private readonly IOptions<RabbitMqOptions> _opts;
    private readonly IServiceScopeFactory _scopes;
    private readonly ILogger<LocalPagingWorkerService> _logger;

    public LocalPagingWorkerService(
        IOptions<RabbitMqOptions> opts,
        IServiceScopeFactory scopes,
        ILogger<LocalPagingWorkerService> logger)
    {
        _opts = opts;
        _scopes = scopes;
        _logger = logger;
    }

    protected override async Task ExecuteAsync(CancellationToken stoppingToken)
    {
        await Task.Delay(TimeSpan.FromSeconds(5), stoppingToken);

        while (!stoppingToken.IsCancellationRequested)
        {
            try
            {
                await RunConsumerAsync(stoppingToken);
                return;
            }
            catch (OperationCanceledException) when (stoppingToken.IsCancellationRequested)
            {
                return;
            }
            catch (Exception ex)
            {
                _logger.LogError(ex,
                    "LocalPagingWorkerService disconnected — retrying in 5s");
                await Task.Delay(TimeSpan.FromSeconds(5), stoppingToken);
            }
        }
    }

    private async Task RunConsumerAsync(CancellationToken stoppingToken)
    {
        var o = _opts.Value;
        var factory = RabbitMqConnectionFactory.Create(o, dispatchConsumersAsync: true);

        using var connection = factory.CreateConnection("gateway-paging-worker");
        using var channel = connection.CreateModel();

        channel.BasicQos(prefetchSize: 0, prefetchCount: 1, global: false);

        var consumer = new AsyncEventingBasicConsumer(channel);
        consumer.Received += async (_, ea) =>
        {
            try
            {
                await HandlePageAsync(channel, ea, stoppingToken);
            }
            catch (OperationCanceledException) when (stoppingToken.IsCancellationRequested)
            {
                _logger.LogInformation("Local paging worker stopping — requeueing in-flight page message");
                channel.BasicNack(ea.DeliveryTag, multiple: false, requeue: true);
            }
            catch (Exception ex)
            {
                _logger.LogError(ex, "Local paging worker failed — NACKing to DLQ");
                channel.BasicNack(ea.DeliveryTag, multiple: false, requeue: false);
            }
        };

        channel.BasicConsume("alerts.paging.queue", autoAck: false, consumer);

        _logger.LogInformation(
            "LocalPagingWorkerService consuming alerts.paging.queue (prefetch=1, timeout={Timeout}ms)",
            o.PagingAckTimeoutMs);

        await Task.Delay(Timeout.Infinite, stoppingToken);
    }

    private async Task HandlePageAsync(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);

            if (await IsAlertAcknowledgedAsync(alertId, ct))
            {
                _logger.LogInformation(
                    "[PAGE-ACK] Alert {AlertId} acknowledged — ACKing RabbitMQ message", alertId);
                channel.BasicAck(ea.DeliveryTag, multiple: false);
                return;
            }
        }

        _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<GatewayDbContext>();

        var status = await db.ClinicalAlerts
            .Where(a => a.Id == alertId)
            .Select(a => a.Status)
            .FirstOrDefaultAsync(ct);

        return status is AlertStatus.Acknowledged or AlertStatus.Resolved;
    }
}