98 lines
3.5 KiB
C#
98 lines
3.5 KiB
C#
using System.Text;
|
|
using System.Text.Json;
|
|
using Microsoft.Extensions.Options;
|
|
using RabbitMQ.Client;
|
|
using RabbitMQ.Client.Events;
|
|
using Serilog.Context;
|
|
|
|
public sealed class EscalationWorkerService : BackgroundService
|
|
{
|
|
private readonly IOptions<RabbitMqOptions> _opts;
|
|
private readonly IServiceScopeFactory _scopes;
|
|
private readonly ClinicalMetrics _metrics;
|
|
private readonly ILogger<EscalationWorkerService> _logger;
|
|
|
|
public EscalationWorkerService(
|
|
IOptions<RabbitMqOptions> opts,
|
|
IServiceScopeFactory scopes,
|
|
ClinicalMetrics metrics,
|
|
ILogger<EscalationWorkerService> logger)
|
|
{
|
|
_opts = opts;
|
|
_scopes = scopes;
|
|
_metrics = metrics;
|
|
_logger = logger;
|
|
}
|
|
|
|
protected override async Task ExecuteAsync(CancellationToken stoppingToken)
|
|
{
|
|
await Task.Delay(TimeSpan.FromSeconds(5), stoppingToken);
|
|
|
|
var o = _opts.Value;
|
|
var factory = RabbitMqConnectionFactory.Create(o, 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 = Guid.Parse(doc.RootElement.GetProperty("encounterId").GetString()!);
|
|
|
|
using (LogContext.PushProperty("EncounterId", encounterId))
|
|
using (LogContext.PushProperty("AlertId", alertId))
|
|
{
|
|
_logger.LogCritical("[ESCALATION] Paging on-call backup for alert {AlertId}", alertId);
|
|
}
|
|
|
|
try
|
|
{
|
|
var escalated = await UpdateAlertStatusEscalatedAsync(alertId, ct);
|
|
if (escalated)
|
|
_metrics.EscalationsTotal.Inc();
|
|
|
|
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<bool> 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 false;
|
|
|
|
// Only escalate if still open — if acknowledged between NACK and TTL expiry, leave it.
|
|
if (alert.Status != AlertStatus.Open) return false;
|
|
|
|
alert.Status = AlertStatus.Escalated;
|
|
await db.SaveChangesAsync(ct);
|
|
return true;
|
|
}
|
|
} |