feature: Ward Gateway Service (Local-First Clinical Path)
This commit is contained in:
voltsrage
@@ -0,0 +1,44 @@
|
||||
using Microsoft.Extensions.Options;
|
||||
|
||||
public sealed class CentralReachabilityService : BackgroundService
|
||||
{
|
||||
private readonly IHttpClientFactory _http;
|
||||
private readonly CentralApiOptions _central;
|
||||
private readonly GatewayOptions _gateway;
|
||||
private volatile bool _isReachable;
|
||||
|
||||
public bool IsCentralReachable => _isReachable;
|
||||
|
||||
public CentralReachabilityService(
|
||||
IHttpClientFactory http,
|
||||
IOptions<CentralApiOptions> central,
|
||||
IOptions<GatewayOptions> gateway)
|
||||
{
|
||||
_http = http;
|
||||
_central = central.Value;
|
||||
_gateway = gateway.Value;
|
||||
}
|
||||
|
||||
protected override async Task ExecuteAsync(CancellationToken ct)
|
||||
{
|
||||
using var timer = new PeriodicTimer(
|
||||
TimeSpan.FromSeconds(_gateway.CentralReachabilityIntervalSeconds));
|
||||
while (await timer.WaitForNextTickAsync(ct))
|
||||
await ProbeAsync(ct);
|
||||
}
|
||||
|
||||
private async Task ProbeAsync(CancellationToken ct)
|
||||
{
|
||||
try
|
||||
{
|
||||
var client = _http.CreateClient("central");
|
||||
client.BaseAddress = new Uri(_central.BaseUrl);
|
||||
var resp = await client.GetAsync("/health/ready", ct);
|
||||
_isReachable = resp.IsSuccessStatusCode;
|
||||
}
|
||||
catch
|
||||
{
|
||||
_isReachable = false;
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,227 @@
|
||||
using System.Net.Http.Headers;
|
||||
using System.Net.Http.Json;
|
||||
using Microsoft.EntityFrameworkCore;
|
||||
using Microsoft.Extensions.Options;
|
||||
using StackExchange.Redis;
|
||||
|
||||
public sealed class EncounterReplicaSyncService : BackgroundService
|
||||
{
|
||||
private const int EncounterPageSize = 100;
|
||||
|
||||
private readonly IServiceScopeFactory _scopes;
|
||||
private readonly IHttpClientFactory _http;
|
||||
private readonly CentralReachabilityService _reachability;
|
||||
private readonly GatewayOptions _gateway;
|
||||
private readonly ILogger<EncounterReplicaSyncService> _logger;
|
||||
|
||||
public EncounterReplicaSyncService(
|
||||
IServiceScopeFactory scopes,
|
||||
IHttpClientFactory http,
|
||||
CentralReachabilityService reachability,
|
||||
IOptions<GatewayOptions> gateway,
|
||||
ILogger<EncounterReplicaSyncService> logger)
|
||||
{
|
||||
_scopes = scopes;
|
||||
_http = http;
|
||||
_reachability = reachability;
|
||||
_gateway = gateway.Value;
|
||||
_logger = logger;
|
||||
}
|
||||
|
||||
protected override async Task ExecuteAsync(CancellationToken ct)
|
||||
{
|
||||
await SyncOnceAsync(ct);
|
||||
|
||||
using var timer = new PeriodicTimer(
|
||||
TimeSpan.FromMinutes(_gateway.EncounterSyncIntervalMinutes));
|
||||
while (await timer.WaitForNextTickAsync(ct))
|
||||
{
|
||||
if (_reachability.IsCentralReachable)
|
||||
await SyncOnceAsync(ct);
|
||||
}
|
||||
}
|
||||
|
||||
internal async Task SyncOnceAsync(CancellationToken ct)
|
||||
{
|
||||
await using var scope = _scopes.CreateAsyncScope();
|
||||
var db = scope.ServiceProvider.GetRequiredService<GatewayDbContext>();
|
||||
var redis = scope.ServiceProvider.GetRequiredService<IConnectionMultiplexer>();
|
||||
var centralOpts = scope.ServiceProvider.GetRequiredService<IOptions<CentralApiOptions>>().Value;
|
||||
|
||||
var client = _http.CreateClient("central");
|
||||
client.BaseAddress = new Uri(centralOpts.BaseUrl);
|
||||
ConfigureAuth(client);
|
||||
|
||||
if (client.DefaultRequestHeaders.Authorization is null)
|
||||
{
|
||||
_logger.LogWarning(
|
||||
"GATEWAY_SYNC_JWT not set — skipping encounter replica sync (seed local replica manually for dev)");
|
||||
return;
|
||||
}
|
||||
|
||||
try
|
||||
{
|
||||
var syncedCount = await SyncEncountersAsync(client, db, ct);
|
||||
await SyncThresholdsAsync(client, db, redis, ct);
|
||||
|
||||
var state = await db.SyncState.FirstOrDefaultAsync(ct)
|
||||
?? db.SyncState.Add(new GatewaySyncState { Id = Guid.NewGuid() }).Entity;
|
||||
state.LastEncounterSyncAt = DateTimeOffset.UtcNow;
|
||||
state.InitialSyncCompleted = true;
|
||||
await db.SaveChangesAsync(ct);
|
||||
|
||||
_logger.LogInformation("Encounter replica sync completed — {Count} encounters", syncedCount);
|
||||
}
|
||||
catch (HttpRequestException ex)
|
||||
{
|
||||
_logger.LogWarning(ex, "Encounter replica sync failed — central API unreachable");
|
||||
}
|
||||
}
|
||||
|
||||
private async Task<int> SyncEncountersAsync(
|
||||
HttpClient client, GatewayDbContext db, CancellationToken ct)
|
||||
{
|
||||
var dept = Uri.EscapeDataString(_gateway.Department);
|
||||
var summaries = new List<CentralWardEncounterSummary>();
|
||||
var page = 1;
|
||||
|
||||
while (true)
|
||||
{
|
||||
var listResp = await client.GetFromJsonAsync<ApiResponse<CentralEncounterListPage>>(
|
||||
$"/api/v1/encounters?status=ACTIVE&department={dept}&page={page}&pageSize={EncounterPageSize}",
|
||||
CentralApiJson.Options,
|
||||
ct);
|
||||
|
||||
if (listResp?.Data?.Items is null || listResp.Data.Items.Count == 0)
|
||||
break;
|
||||
|
||||
summaries.AddRange(listResp.Data.Items);
|
||||
|
||||
if (page >= listResp.Data.TotalPages)
|
||||
break;
|
||||
|
||||
page++;
|
||||
}
|
||||
|
||||
var synced = 0;
|
||||
foreach (var summary in summaries)
|
||||
{
|
||||
var detailResp = await client.GetFromJsonAsync<ApiResponse<CentralEncounterDetail>>(
|
||||
$"/api/v1/encounters/{summary.EncounterId}",
|
||||
CentralApiJson.Options,
|
||||
ct);
|
||||
|
||||
if (detailResp?.Data is null)
|
||||
{
|
||||
_logger.LogWarning("Central returned no detail for encounter {EncounterId}", summary.EncounterId);
|
||||
continue;
|
||||
}
|
||||
|
||||
UpsertReplica(db, detailResp.Data);
|
||||
synced++;
|
||||
}
|
||||
|
||||
return synced;
|
||||
}
|
||||
|
||||
private static void UpsertReplica(GatewayDbContext db, CentralEncounterDetail detail)
|
||||
{
|
||||
var now = DateTimeOffset.UtcNow;
|
||||
var patient = detail.Patient;
|
||||
|
||||
var replicaPatient = db.Patients.Find(patient.Id);
|
||||
if (replicaPatient is null)
|
||||
{
|
||||
db.Patients.Add(new ReplicaPatient
|
||||
{
|
||||
Id = patient.Id,
|
||||
Mrn = patient.Mrn,
|
||||
FirstName = patient.FirstName,
|
||||
LastName = patient.LastName,
|
||||
DateOfBirth = patient.DateOfBirth,
|
||||
Gender = patient.Gender,
|
||||
SyncedAt = now
|
||||
});
|
||||
}
|
||||
else
|
||||
{
|
||||
replicaPatient.Mrn = patient.Mrn;
|
||||
replicaPatient.FirstName = patient.FirstName;
|
||||
replicaPatient.LastName = patient.LastName;
|
||||
replicaPatient.DateOfBirth = patient.DateOfBirth;
|
||||
replicaPatient.Gender = patient.Gender;
|
||||
replicaPatient.SyncedAt = now;
|
||||
}
|
||||
|
||||
var enc = db.Encounters.Find(detail.Id);
|
||||
if (enc is null)
|
||||
{
|
||||
db.Encounters.Add(new ReplicaEncounter
|
||||
{
|
||||
Id = detail.Id,
|
||||
PatientId = detail.PatientId,
|
||||
EncounterType = detail.EncounterType,
|
||||
Status = detail.Status,
|
||||
Department = detail.Department,
|
||||
AttendingPhysician = detail.AttendingPhysician,
|
||||
RoomBed = detail.RoomBed,
|
||||
AdmittedAt = detail.AdmittedAt,
|
||||
SyncedAt = now
|
||||
});
|
||||
}
|
||||
else
|
||||
{
|
||||
enc.PatientId = detail.PatientId;
|
||||
enc.EncounterType = detail.EncounterType;
|
||||
enc.Status = detail.Status;
|
||||
enc.Department = detail.Department;
|
||||
enc.AttendingPhysician = detail.AttendingPhysician;
|
||||
enc.RoomBed = detail.RoomBed;
|
||||
enc.AdmittedAt = detail.AdmittedAt;
|
||||
enc.SyncedAt = now;
|
||||
}
|
||||
}
|
||||
|
||||
private async Task SyncThresholdsAsync(
|
||||
HttpClient client,
|
||||
GatewayDbContext db,
|
||||
IConnectionMultiplexer redis,
|
||||
CancellationToken ct)
|
||||
{
|
||||
var thrResp = await client.GetFromJsonAsync<ApiResponse<List<CentralAlertThreshold>>>(
|
||||
"/api/v1/alert-thresholds",
|
||||
CentralApiJson.Options,
|
||||
ct);
|
||||
|
||||
if (thrResp?.Data is null)
|
||||
return;
|
||||
|
||||
await db.Database.ExecuteSqlRawAsync("DELETE FROM alert_thresholds", ct);
|
||||
var syncedAt = DateTimeOffset.UtcNow;
|
||||
db.AlertThresholds.AddRange(thrResp.Data.Select(t => new ReplicaAlertThreshold
|
||||
{
|
||||
Id = t.Id,
|
||||
ObservationCode = t.ObservationCode,
|
||||
DisplayName = t.DisplayName,
|
||||
Unit = t.Unit,
|
||||
CriticalLow = t.CriticalLow,
|
||||
WarningLow = t.WarningLow,
|
||||
WarningHigh = t.WarningHigh,
|
||||
CriticalHigh = t.CriticalHigh,
|
||||
SuppressionWindowMinutes = t.SuppressionWindowMinutes,
|
||||
SyncedAt = syncedAt
|
||||
}));
|
||||
|
||||
await db.SaveChangesAsync(ct);
|
||||
await ThresholdCacheLoader.RefreshAsync(db, redis, _logger, ct);
|
||||
}
|
||||
|
||||
private static void ConfigureAuth(HttpClient client)
|
||||
{
|
||||
client.DefaultRequestHeaders.Authorization = null;
|
||||
|
||||
var jwt = Environment.GetEnvironmentVariable("GATEWAY_SYNC_JWT");
|
||||
if (!string.IsNullOrEmpty(jwt))
|
||||
client.DefaultRequestHeaders.Authorization = new AuthenticationHeaderValue("Bearer", jwt);
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,65 @@
|
||||
using Microsoft.EntityFrameworkCore;
|
||||
using Microsoft.Extensions.Options;
|
||||
|
||||
public sealed class GatewayHeartbeatService : BackgroundService
|
||||
{
|
||||
private readonly IServiceScopeFactory _scopes;
|
||||
private readonly IHttpClientFactory _http;
|
||||
private readonly CentralReachabilityService _reachability;
|
||||
private readonly GatewayOptions _gateway;
|
||||
private readonly IConfiguration _config;
|
||||
private readonly ILogger<GatewayHeartbeatService> _logger;
|
||||
|
||||
public GatewayHeartbeatService(
|
||||
IServiceScopeFactory scopes,
|
||||
IHttpClientFactory http,
|
||||
CentralReachabilityService reachability,
|
||||
IOptions<GatewayOptions> gateway,
|
||||
IConfiguration config,
|
||||
ILogger<GatewayHeartbeatService> logger)
|
||||
{
|
||||
_scopes = scopes;
|
||||
_http = http;
|
||||
_reachability = reachability;
|
||||
_gateway = gateway.Value;
|
||||
_config = config;
|
||||
_logger = logger;
|
||||
}
|
||||
|
||||
protected override async Task ExecuteAsync(CancellationToken ct)
|
||||
{
|
||||
using var timer = new PeriodicTimer(TimeSpan.FromSeconds(_gateway.HeartbeatIntervalSeconds));
|
||||
while (await timer.WaitForNextTickAsync(ct))
|
||||
await SendHeartbeatAsync(ct);
|
||||
}
|
||||
|
||||
private async Task SendHeartbeatAsync(CancellationToken ct)
|
||||
{
|
||||
await using var scope = _scopes.CreateAsyncScope();
|
||||
var db = scope.ServiceProvider.GetRequiredService<GatewayDbContext>();
|
||||
var bufferDepth = await db.BufferedSyncItems.CountAsync(b => !b.Synced, ct);
|
||||
|
||||
var status = _reachability.IsCentralReachable ? "ONLINE" : "DEGRADED";
|
||||
var req = new GatewayHeartbeatRequest(status, bufferDepth, DateTimeOffset.UtcNow);
|
||||
|
||||
try
|
||||
{
|
||||
var client = _http.CreateClient("central");
|
||||
client.BaseAddress = new Uri(
|
||||
scope.ServiceProvider.GetRequiredService<IOptions<CentralApiOptions>>().Value.BaseUrl);
|
||||
client.DefaultRequestHeaders.Remove("X-Api-Key");
|
||||
client.DefaultRequestHeaders.Remove("X-Gateway-Id");
|
||||
client.DefaultRequestHeaders.Add("X-Api-Key", _config["ApiKey:Gateway"]);
|
||||
client.DefaultRequestHeaders.Add("X-Gateway-Id", _gateway.GatewayId.ToString());
|
||||
|
||||
var resp = await client.PatchAsJsonAsync(
|
||||
$"/api/v1/gateways/{_gateway.GatewayId}/heartbeat", req, ct);
|
||||
if (!resp.IsSuccessStatusCode)
|
||||
_logger.LogWarning("Heartbeat failed: {Status}", resp.StatusCode);
|
||||
}
|
||||
catch (Exception ex)
|
||||
{
|
||||
_logger.LogWarning(ex, "Heartbeat to central failed — gateway continues in DEGRADED mode");
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -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 LocalEscalationWorkerService : BackgroundService
|
||||
{
|
||||
private readonly IOptions<RabbitMqOptions> _opts;
|
||||
private readonly IServiceScopeFactory _scopes;
|
||||
private readonly ILogger<LocalEscalationWorkerService> _logger;
|
||||
|
||||
public LocalEscalationWorkerService(
|
||||
IOptions<RabbitMqOptions> opts,
|
||||
IServiceScopeFactory scopes,
|
||||
ILogger<LocalEscalationWorkerService> 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("gateway-escalation-worker");
|
||||
using var channel = connection.CreateModel();
|
||||
|
||||
channel.BasicQos(0, prefetchCount: 5, global: false);
|
||||
|
||||
var consumer = new AsyncEventingBasicConsumer(channel);
|
||||
consumer.Received += async (_, ea) =>
|
||||
{
|
||||
await HandleEscalationAsync(channel, ea, stoppingToken);
|
||||
};
|
||||
|
||||
channel.BasicConsume("alerts.escalation.queue", autoAck: false, consumer);
|
||||
|
||||
_logger.LogInformation("LocalEscalationWorkerService 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()!);
|
||||
|
||||
_logger.LogCritical("[ESCALATION] Paging on-call backup for alert {AlertId}", alertId);
|
||||
|
||||
try
|
||||
{
|
||||
var escalated = await UpdateAlertStatusEscalatedAsync(alertId, ct);
|
||||
channel.BasicAck(ea.DeliveryTag, multiple: false);
|
||||
|
||||
if (escalated)
|
||||
{
|
||||
_logger.LogWarning(
|
||||
"[ESCALATION-DONE] Alert {AlertId} status → Escalated in gateway DB", alertId);
|
||||
}
|
||||
}
|
||||
catch (Exception ex)
|
||||
{
|
||||
_logger.LogError(ex, "Local escalation worker 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<GatewayDbContext>();
|
||||
|
||||
var alert = await db.ClinicalAlerts.FindAsync([alertId], ct);
|
||||
if (alert is null) return false;
|
||||
|
||||
if (alert.Status != AlertStatus.Open) return false;
|
||||
|
||||
alert.Status = AlertStatus.Escalated;
|
||||
await db.SaveChangesAsync(ct);
|
||||
return true;
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,122 @@
|
||||
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);
|
||||
|
||||
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("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;
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,235 @@
|
||||
using System.Net.Http.Json;
|
||||
using System.Text.Json;
|
||||
using Microsoft.EntityFrameworkCore;
|
||||
using Microsoft.Extensions.Options;
|
||||
using VigilCare.ClinicalContracts.Sync;
|
||||
|
||||
public sealed class SyncUploaderService : BackgroundService
|
||||
{
|
||||
private static readonly JsonSerializerOptions PayloadJson = new()
|
||||
{
|
||||
PropertyNameCaseInsensitive = true
|
||||
};
|
||||
|
||||
private readonly IServiceScopeFactory _scopes;
|
||||
private readonly CentralReachabilityService _reachability;
|
||||
private readonly GatewayOptions _gateway;
|
||||
private readonly IConfiguration _config;
|
||||
private readonly ILogger<SyncUploaderService> _logger;
|
||||
private int _backoffSeconds = 30;
|
||||
|
||||
public SyncUploaderService(
|
||||
IServiceScopeFactory scopes,
|
||||
CentralReachabilityService reachability,
|
||||
IOptions<GatewayOptions> gateway,
|
||||
IConfiguration config,
|
||||
ILogger<SyncUploaderService> logger)
|
||||
{
|
||||
_scopes = scopes;
|
||||
_reachability = reachability;
|
||||
_gateway = gateway.Value;
|
||||
_config = config;
|
||||
_logger = logger;
|
||||
}
|
||||
|
||||
protected override async Task ExecuteAsync(CancellationToken ct)
|
||||
{
|
||||
using var timer = new PeriodicTimer(TimeSpan.FromSeconds(30));
|
||||
while (await timer.WaitForNextTickAsync(ct))
|
||||
{
|
||||
if (!_reachability.IsCentralReachable) continue;
|
||||
try
|
||||
{
|
||||
await UploadBatchAsync(ct);
|
||||
_backoffSeconds = 30;
|
||||
}
|
||||
catch (Exception ex)
|
||||
{
|
||||
_logger.LogWarning(ex, "Sync upload failed — backoff {Seconds}s", _backoffSeconds);
|
||||
await Task.Delay(TimeSpan.FromSeconds(_backoffSeconds), ct);
|
||||
_backoffSeconds = Math.Min(_backoffSeconds * 2, 600);
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
private async Task UploadBatchAsync(CancellationToken ct)
|
||||
{
|
||||
await using var scope = _scopes.CreateAsyncScope();
|
||||
var db = scope.ServiceProvider.GetRequiredService<GatewayDbContext>();
|
||||
|
||||
var pending = await db.BufferedSyncItems
|
||||
.Where(b => !b.Synced)
|
||||
.OrderBy(b => b.RecordedAt)
|
||||
.Take(_gateway.SyncBatchSize)
|
||||
.ToListAsync(ct);
|
||||
if (pending.Count == 0) return;
|
||||
|
||||
var batchRef = Guid.NewGuid();
|
||||
var request = BuildBatchRequest(batchRef, pending);
|
||||
|
||||
var client = scope.ServiceProvider.GetRequiredService<IHttpClientFactory>().CreateClient("central");
|
||||
client.BaseAddress = new Uri(
|
||||
scope.ServiceProvider.GetRequiredService<IOptions<CentralApiOptions>>().Value.BaseUrl);
|
||||
client.DefaultRequestHeaders.Add("X-Api-Key", _config["ApiKey:Gateway"]!);
|
||||
client.DefaultRequestHeaders.Add("X-Gateway-Id", _gateway.GatewayId.ToString());
|
||||
|
||||
var submit = await client.PostAsJsonAsync("/api/v1/sync/batches", request, ct);
|
||||
submit.EnsureSuccessStatusCode();
|
||||
var batchId = (await submit.Content.ReadFromJsonAsync<ApiResponse<SyncBatchStatus>>(ct))!.Data!.BatchId;
|
||||
|
||||
for (var i = 0; i < 30; i++)
|
||||
{
|
||||
await Task.Delay(2000, ct);
|
||||
var status = await client.GetFromJsonAsync<ApiResponse<SyncBatchStatus>>(
|
||||
$"/api/v1/sync/batches/{batchId}", ct);
|
||||
if (status?.Data?.Status is "APPLIED")
|
||||
{
|
||||
foreach (var item in pending)
|
||||
{
|
||||
item.Synced = true;
|
||||
item.SyncedAt = DateTimeOffset.UtcNow;
|
||||
}
|
||||
await db.SaveChangesAsync(ct);
|
||||
_logger.LogInformation(
|
||||
"Sync batch {BatchId} applied — marked {Count} buffered items synced",
|
||||
batchId, pending.Count);
|
||||
return;
|
||||
}
|
||||
if (status?.Data?.Status is "CONFLICT" or "REJECTED")
|
||||
{
|
||||
_logger.LogError(
|
||||
"Sync batch {BatchId} {Status} — items remain buffered",
|
||||
batchId, status.Data.Status);
|
||||
return;
|
||||
}
|
||||
}
|
||||
|
||||
_logger.LogWarning("Sync batch {BatchId} still processing after poll window", batchId);
|
||||
}
|
||||
|
||||
private ClinicalSyncBatchRequest BuildBatchRequest(Guid batchRef, List<BufferedSyncItem> items)
|
||||
{
|
||||
var observations = new List<SyncedObservation>();
|
||||
var alertEvents = new List<SyncedAlertEvent>();
|
||||
var acks = new List<SyncedAlertAcknowledgment>();
|
||||
var resolves = new List<SyncedAlertResolution>();
|
||||
|
||||
foreach (var item in items)
|
||||
{
|
||||
switch (item.ItemType)
|
||||
{
|
||||
case BufferedSyncItemType.Observation:
|
||||
observations.Add(MapObservation(item));
|
||||
break;
|
||||
case BufferedSyncItemType.Alert:
|
||||
alertEvents.Add(MapAlert(item));
|
||||
break;
|
||||
case BufferedSyncItemType.Ack:
|
||||
acks.Add(MapAck(item));
|
||||
break;
|
||||
case BufferedSyncItemType.Resolve:
|
||||
resolves.Add(MapResolve(item));
|
||||
break;
|
||||
default:
|
||||
throw new InvalidOperationException(
|
||||
$"Unknown buffered sync item type '{item.ItemType}' for item {item.Id}");
|
||||
}
|
||||
}
|
||||
|
||||
return new ClinicalSyncBatchRequest(
|
||||
batchRef,
|
||||
_gateway.GatewayId,
|
||||
_gateway.SiteId,
|
||||
DateTimeOffset.UtcNow,
|
||||
observations,
|
||||
alertEvents,
|
||||
acks,
|
||||
resolves);
|
||||
}
|
||||
|
||||
private static SyncedObservation MapObservation(BufferedSyncItem item)
|
||||
{
|
||||
var p = Deserialize<BufferedObservationPayload>(item.Payload);
|
||||
return new SyncedObservation(
|
||||
p.ClientRef,
|
||||
p.IdempotencyKey ?? item.IdempotencyKey,
|
||||
p.EncounterId,
|
||||
p.ObservationCode,
|
||||
p.Value,
|
||||
p.Unit,
|
||||
p.Source,
|
||||
p.RecordedAt);
|
||||
}
|
||||
|
||||
private static SyncedAlertEvent MapAlert(BufferedSyncItem item)
|
||||
{
|
||||
var p = Deserialize<BufferedAlertPayload>(item.Payload);
|
||||
return new SyncedAlertEvent(
|
||||
p.ClientAlertId,
|
||||
p.EncounterId,
|
||||
p.AlertType,
|
||||
p.Severity,
|
||||
p.Details,
|
||||
p.GeneratedAt);
|
||||
}
|
||||
|
||||
private static SyncedAlertAcknowledgment MapAck(BufferedSyncItem item)
|
||||
{
|
||||
var p = Deserialize<BufferedAckPayload>(item.Payload);
|
||||
return new SyncedAlertAcknowledgment(
|
||||
p.ClientRef,
|
||||
p.ClientAlertId,
|
||||
p.ClinicianId,
|
||||
p.AcknowledgedAt,
|
||||
p.Note);
|
||||
}
|
||||
|
||||
private static SyncedAlertResolution MapResolve(BufferedSyncItem item)
|
||||
{
|
||||
var p = Deserialize<BufferedResolvePayload>(item.Payload);
|
||||
return new SyncedAlertResolution(
|
||||
p.ClientRef,
|
||||
p.ClientAlertId,
|
||||
p.ClinicianId,
|
||||
p.ResolvedAt,
|
||||
p.Note);
|
||||
}
|
||||
|
||||
private static T Deserialize<T>(string json) =>
|
||||
JsonSerializer.Deserialize<T>(json, PayloadJson)
|
||||
?? throw new JsonException($"Failed to deserialize buffered sync payload as {typeof(T).Name}");
|
||||
|
||||
private sealed record BufferedObservationPayload(
|
||||
Guid ClientRef,
|
||||
string? IdempotencyKey,
|
||||
Guid EncounterId,
|
||||
string ObservationCode,
|
||||
decimal Value,
|
||||
string? Unit,
|
||||
string Source,
|
||||
DateTimeOffset RecordedAt);
|
||||
|
||||
private sealed record BufferedAlertPayload(
|
||||
Guid ClientAlertId,
|
||||
Guid EncounterId,
|
||||
string AlertType,
|
||||
string Severity,
|
||||
string Details,
|
||||
DateTimeOffset GeneratedAt);
|
||||
|
||||
private sealed record BufferedAckPayload(
|
||||
Guid ClientRef,
|
||||
Guid ClientAlertId,
|
||||
string ClinicianId,
|
||||
DateTimeOffset AcknowledgedAt,
|
||||
string? Note);
|
||||
|
||||
private sealed record BufferedResolvePayload(
|
||||
Guid ClientRef,
|
||||
Guid ClientAlertId,
|
||||
string ClinicianId,
|
||||
DateTimeOffset ResolvedAt,
|
||||
string? Note);
|
||||
|
||||
private sealed record SyncBatchStatus(Guid BatchId, string Status);
|
||||
}
|
||||
@@ -0,0 +1,94 @@
|
||||
using System.Text.Json;
|
||||
using Microsoft.EntityFrameworkCore;
|
||||
using StackExchange.Redis;
|
||||
|
||||
public class ThresholdCacheLoader : IHostedService
|
||||
{
|
||||
private readonly IServiceProvider _services;
|
||||
private readonly IConnectionMultiplexer _redis;
|
||||
private readonly ILogger<ThresholdCacheLoader> _logger;
|
||||
|
||||
public ThresholdCacheLoader(
|
||||
IServiceProvider services,
|
||||
IConnectionMultiplexer redis,
|
||||
ILogger<ThresholdCacheLoader> logger)
|
||||
{
|
||||
_services = services;
|
||||
_redis = redis;
|
||||
_logger = logger;
|
||||
}
|
||||
|
||||
public async Task StartAsync(CancellationToken cancellationToken)
|
||||
{
|
||||
using var scope = _services.CreateScope();
|
||||
var db = scope.ServiceProvider.GetRequiredService<GatewayDbContext>();
|
||||
await RefreshAsync(db, _redis, _logger, cancellationToken);
|
||||
}
|
||||
|
||||
public Task StopAsync(CancellationToken cancellationToken) => Task.CompletedTask;
|
||||
|
||||
/// <summary>
|
||||
/// Reloads all gateway alert thresholds into Redis. Called on startup and after
|
||||
/// <see cref="EncounterReplicaSyncService"/> pulls thresholds from central.
|
||||
/// </summary>
|
||||
public static async Task RefreshAsync(
|
||||
GatewayDbContext db,
|
||||
IConnectionMultiplexer redis,
|
||||
ILogger logger,
|
||||
CancellationToken cancellationToken = default)
|
||||
{
|
||||
var thresholds = await db.AlertThresholds.ToListAsync(cancellationToken);
|
||||
if (thresholds.Count == 0)
|
||||
{
|
||||
logger.LogWarning("No alert thresholds in gateway DB — Redis cache not updated");
|
||||
return;
|
||||
}
|
||||
|
||||
const int maxAttempts = 3;
|
||||
int[] backoffMs = [2000, 4000, 8000];
|
||||
|
||||
for (var attempt = 0; attempt < maxAttempts; attempt++)
|
||||
{
|
||||
try
|
||||
{
|
||||
WriteThresholds(redis, thresholds);
|
||||
logger.LogInformation(
|
||||
"Loaded {Count} alert thresholds into gateway Redis cache", thresholds.Count);
|
||||
return;
|
||||
}
|
||||
catch (RedisException ex)
|
||||
{
|
||||
logger.LogWarning(ex,
|
||||
"Redis unavailable during gateway threshold cache load — attempt {Attempt}/{Max}",
|
||||
attempt + 1, maxAttempts);
|
||||
|
||||
if (attempt < maxAttempts - 1)
|
||||
await Task.Delay(backoffMs[attempt], cancellationToken);
|
||||
}
|
||||
}
|
||||
|
||||
logger.LogError(
|
||||
"Failed to load thresholds into gateway Redis after {Max} attempts — " +
|
||||
"observation ingest falls back to PostgreSQL for threshold lookups",
|
||||
maxAttempts);
|
||||
}
|
||||
|
||||
private static void WriteThresholds(IConnectionMultiplexer redis, List<ReplicaAlertThreshold> thresholds)
|
||||
{
|
||||
var cache = redis.GetDatabase();
|
||||
var batch = cache.CreateBatch();
|
||||
|
||||
foreach (var t in thresholds)
|
||||
{
|
||||
var json = JsonSerializer.Serialize(new ThresholdCacheEntry(
|
||||
t.ObservationCode,
|
||||
t.CriticalLow,
|
||||
t.WarningLow,
|
||||
t.WarningHigh,
|
||||
t.CriticalHigh));
|
||||
_ = batch.StringSetAsync($"threshold:{t.ObservationCode}", json);
|
||||
}
|
||||
|
||||
batch.Execute();
|
||||
}
|
||||
}
|
||||
Reference in New Issue
Block a user