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 _logger; private int _backoffSeconds = 30; public SyncUploaderService( IServiceScopeFactory scopes, CentralReachabilityService reachability, IOptions gateway, IConfiguration config, ILogger 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(); 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().CreateClient("central"); client.BaseAddress = new Uri( scope.ServiceProvider.GetRequiredService>().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>(ct))!.Data!.BatchId; for (var i = 0; i < 30; i++) { await Task.Delay(2000, ct); var status = await client.GetFromJsonAsync>( $"/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 items) { var observations = new List(); var alertEvents = new List(); var acks = new List(); var resolves = new List(); 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(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(item.Payload); return new SyncedAlertEvent( p.ClientAlertId, p.EncounterId, p.AlertType, p.Severity, p.Details, p.GeneratedAt, p.ExplanationJson); } private static SyncedAlertAcknowledgment MapAck(BufferedSyncItem item) { var p = Deserialize(item.Payload); return new SyncedAlertAcknowledgment( p.ClientRef, p.ClientAlertId, p.ClinicianId, p.AcknowledgedAt, p.Note); } private static SyncedAlertResolution MapResolve(BufferedSyncItem item) { var p = Deserialize(item.Payload); return new SyncedAlertResolution( p.ClientRef, p.ClientAlertId, p.ClinicianId, p.ResolvedAt, p.Note); } private static T Deserialize(string json) => JsonSerializer.Deserialize(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, string? ExplanationJson = null); 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); }