Clinical Sync Batch Engine: first commit

2026-06-23 03:02:30 +08:00
parent 90b8baa2a1
commit c9994b1ba2
60 changed files with 3547 additions and 31 deletions
@@ -200,4 +200,48 @@ public class AlertService : IAlertService

        return alert;
    }
+
+    public async Task ApplySyncedAcknowledgmentAsync(
+        Guid alertId, SyncedAlertAcknowledgment ack, CancellationToken ct)
+    {
+        var alert = await _db.ClinicalAlerts.FindAsync([alertId], ct)
+            ?? throw new NotFoundException("Alert not found.", "ALERT_NOT_FOUND");
+
+        alert.Status = AlertStatus.Acknowledged;
+        alert.AcknowledgedAt = ack.AcknowledgedAt;
+        alert.AcknowledgedBy = ack.ClinicianId;
+
+        _db.OutboxEvents.Add(new OutboxEvent
+        {
+            Id = Guid.NewGuid(),
+            Topic = "alert.acknowledged",
+            Payload = JsonSerializer.Serialize(new
+            {
+                alertId = alert.Id,
+                encounterId = alert.EncounterId,
+                acknowledgedBy = ack.ClinicianId,
+                acknowledgedAt = ack.AcknowledgedAt,
+                note = ack.Note,
+                syncedFromGateway = true
+            }),
+            PartitionKey = alert.EncounterId.ToString(),
+            CreatedAt = DateTimeOffset.UtcNow
+        });
+
+        await _db.SaveChangesAsync(ct);
+    }
+
+    public async Task ApplySyncedResolutionAsync(
+        Guid alertId, SyncedAlertResolution resolve, CancellationToken ct)
+    {
+        var alert = await _db.ClinicalAlerts.FindAsync([alertId], ct)
+            ?? throw new NotFoundException("Alert not found.", "ALERT_NOT_FOUND");
+
+        if (alert.Status != AlertStatus.Acknowledged)
+            alert.Status = AlertStatus.Acknowledged;
+
+        alert.Status = AlertStatus.Resolved;
+        alert.ResolvedAt = resolve.ResolvedAt;
+        await _db.SaveChangesAsync(ct);
+    }
 }
@@ -0,0 +1,234 @@
+using System.Text.Json;
+using Microsoft.EntityFrameworkCore;
+using Npgsql;
+using Prometheus;
+using VigilCare.ClinicalContracts.Sync;
+
+public class ClinicalSyncBatchProcessor
+{
+    private readonly AppDbContext _db;
+    private readonly IObservationService _observations;
+    private readonly IAlertService _alerts;
+    private readonly ClinicalMetrics _metrics;
+    private readonly ILogger<ClinicalSyncBatchProcessor> _logger;
+
+    public ClinicalSyncBatchProcessor(
+        AppDbContext db,
+        IObservationService observations,
+        IAlertService alerts,
+        ClinicalMetrics metrics,
+        ILogger<ClinicalSyncBatchProcessor> logger)
+    {
+        _db = db;
+        _observations = observations;
+        _alerts = alerts;
+        _metrics = metrics;
+        _logger = logger;
+    }
+
+    public async Task ProcessBatchAsync(Guid batchId, CancellationToken ct)
+    {
+        using var timer = _metrics.ClinicalSyncBatchDuration.NewTimer();
+
+        // Phase 1 — lock batch row
+        ClinicalSyncBatch? batch;
+        try
+        {
+            await using var lockTx = await _db.Database.BeginTransactionAsync(ct);
+            batch = await _db.ClinicalSyncBatches
+                .FromSqlInterpolated($"""
+                    SELECT * FROM clinical_sync_batches
+                    WHERE id = {batchId}
+                    FOR UPDATE NOWAIT
+                    """)
+                .FirstOrDefaultAsync(ct);
+
+            if (batch is null || batch.Status != ClinicalSyncBatchStatus.Received)
+                return;
+
+            batch.MarkProcessing();
+            await _db.SaveChangesAsync(ct);
+            await lockTx.CommitAsync(ct);
+        }
+        catch (PostgresException ex) when (ex.SqlState == "55P03") // lock_not_available
+        {
+            _logger.LogInformation("Batch {BatchId} already locked by another consumer", batchId);
+            return;
+        }
+
+        // Phase 2 — deserialize payload
+        var request = JsonSerializer.Deserialize<ClinicalSyncBatchRequest>(batch!.Payload)!;
+        var hasConflict = false;
+
+        // Phase 3 — replay order: observations → alerts → acks → resolutions
+        foreach (var obs in request.Observations.OrderBy(o => o.RecordedAt))
+        {
+            try
+            {
+                await ApplyObservationAsync(obs, batch, ct);
+            }
+            catch (Exception ex)
+            {
+                _db.ChangeTracker.Clear();
+                await RecordConflictAsync(batch.Id, obs.ClientRef, "OBSERVATION", ex.Message, ct);
+                hasConflict = true;
+            }
+        }
+
+        foreach (var alert in request.AlertEvents.OrderBy(a => a.GeneratedAt))
+        {
+            try
+            {
+                await ApplyAlertEventAsync(alert, batch, ct);
+            }
+            catch (Exception ex)
+            {
+                _db.ChangeTracker.Clear();
+                await RecordConflictAsync(batch.Id, alert.ClientAlertId, "ALERT", ex.Message, ct);
+                hasConflict = true;
+            }
+        }
+
+        foreach (var ack in request.AlertAcknowledgments.OrderBy(a => a.AcknowledgedAt))
+        {
+            try
+            {
+                if (await ApplyAckAsync(ack, batch, ct))
+                    hasConflict = true;
+            }
+            catch (Exception ex)
+            {
+                _db.ChangeTracker.Clear();
+                await RecordConflictAsync(batch.Id, ack.ClientRef, "ACK", ex.Message, ct);
+                hasConflict = true;
+            }
+        }
+
+        foreach (var resolve in request.AlertResolutions.OrderBy(r => r.ResolvedAt))
+        {
+            try
+            {
+                if (await ApplyResolveAsync(resolve, batch, ct))
+                    hasConflict = true;
+            }
+            catch (Exception ex)
+            {
+                _db.ChangeTracker.Clear();
+                await RecordConflictAsync(batch.Id, resolve.ClientRef, "RESOLVE", ex.Message, ct);
+                hasConflict = true;
+            }
+        }
+
+        // Phase 4 — finalize batch
+        batch = await _db.ClinicalSyncBatches.FindAsync([batchId], ct);
+        if (batch is null) return;
+
+        if (hasConflict) batch.MarkConflict();
+        else batch.MarkApplied();
+
+        var gateway = await _db.WardGateways.FindAsync([batch.GatewayId], ct);
+        gateway?.MarkSynced(DateTimeOffset.UtcNow);
+
+        await _db.SaveChangesAsync(ct);
+
+        _metrics.ClinicalSyncBatchesTotal.WithLabels(batch.Status.ToDbString()).Inc();
+        _logger.LogInformation("Batch {BatchId} finalized as {Status}", batchId, batch.Status);
+    }
+
+    private async Task ApplyObservationAsync(
+        SyncedObservation obs, ClinicalSyncBatch batch, CancellationToken ct)
+    {
+        if (await _db.Observations.AnyAsync(o => o.IdempotencyKey == obs.IdempotencyKey, ct))
+            return;
+
+        await _observations.ApplySyncedObservationAsync(obs, ct);
+    }
+
+    private async Task ApplyAlertEventAsync(
+        SyncedAlertEvent alert, ClinicalSyncBatch batch, CancellationToken ct)
+    {
+        if (await _db.ClinicalAlerts.AnyAsync(a => a.ClientAlertId == alert.ClientAlertId, ct))
+            return;
+
+        var encounter = await _db.Encounters.FindAsync([alert.EncounterId], ct)
+            ?? throw new ValidationException("Encounter not found.", "ENCOUNTER_NOT_FOUND");
+
+        var clinicalAlert = new ClinicalAlert
+        {
+            Id = Guid.NewGuid(),
+            ClientAlertId = alert.ClientAlertId,
+            SyncedFromGateway = true,
+            EncounterId = alert.EncounterId,
+            PatientId = encounter.PatientId,
+            AlertType = AlertTypeExtensions.FromDbString(alert.AlertType),
+            Severity = AlertSeverityExtensions.FromDbString(alert.Severity),
+            Details = alert.Details,
+            Status = AlertStatus.Open,
+            TriggeredAt = alert.GeneratedAt
+        };
+        _db.ClinicalAlerts.Add(clinicalAlert);
+
+        _db.OutboxEvents.Add(new OutboxEvent
+        {
+            Id = Guid.NewGuid(),
+            Topic = "alert.generated",
+            Payload = JsonSerializer.Serialize(new
+            {
+                alertId = clinicalAlert.Id,
+                encounterId = alert.EncounterId,
+                patientId = encounter.PatientId,
+                alertType = alert.AlertType,
+                severity = alert.Severity,
+                details = alert.Details,
+                syncedFromGateway = true,
+                triggeredAt = alert.GeneratedAt,
+                partitionKey = alert.EncounterId.ToString()
+            }),
+            PartitionKey = alert.EncounterId.ToString(),
+            CreatedAt = DateTimeOffset.UtcNow
+        });
+
+        await _db.SaveChangesAsync(ct);
+    }
+
+    private async Task<bool> ApplyAckAsync(
+        SyncedAlertAcknowledgment ack, ClinicalSyncBatch batch, CancellationToken ct)
+    {
+        var alert = await _db.ClinicalAlerts
+            .FirstOrDefaultAsync(a => a.ClientAlertId == ack.ClientAlertId, ct);
+        if (alert is null)
+        {
+            await RecordConflictAsync(batch.Id, ack.ClientRef, "ACK", "ALERT_NOT_YET_SYNCED", ct);
+            return true;
+        }
+        if (alert.Status != AlertStatus.Open && alert.Status != AlertStatus.Escalated)
+            return false;
+
+        await _alerts.ApplySyncedAcknowledgmentAsync(alert.Id, ack, ct);
+        return false;
+    }
+
+    private async Task<bool> ApplyResolveAsync(
+        SyncedAlertResolution resolve, ClinicalSyncBatch batch, CancellationToken ct)
+    {
+        var alert = await _db.ClinicalAlerts
+            .FirstOrDefaultAsync(a => a.ClientAlertId == resolve.ClientAlertId, ct);
+        if (alert is null)
+        {
+            await RecordConflictAsync(batch.Id, resolve.ClientRef, "RESOLVE", "ALERT_NOT_YET_SYNCED", ct);
+            return true;
+        }
+        if (alert.Status == AlertStatus.Resolved)
+            return false;
+
+        await _alerts.ApplySyncedResolutionAsync(alert.Id, resolve, ct);
+        return false;
+    }
+
+    private async Task RecordConflictAsync(
+        Guid batchId, Guid clientRef, string itemType, string reason, CancellationToken ct)
+    {
+        _db.ClinicalSyncConflicts.Add(new ClinicalSyncConflict(batchId, clientRef, itemType, reason));
+        await _db.SaveChangesAsync(ct);
+    }
+}
@@ -0,0 +1,94 @@
+using System.Text.Json;
+using Microsoft.EntityFrameworkCore;
+using VigilCare.ClinicalContracts.Sync;
+
+public class ClinicalSyncService : IClinicalSyncService
+{
+    private readonly AppDbContext _db;
+    private readonly ILogger<ClinicalSyncService> _logger;
+
+    public ClinicalSyncService(AppDbContext db, ILogger<ClinicalSyncService> logger)
+    {
+        _db = db;
+        _logger = logger;
+    }
+
+    public async Task<ClinicalBatchUploadResponse> UploadBatchAsync(
+        ClinicalSyncBatchRequest request, CancellationToken ct)
+    {
+        var existing = await _db.ClinicalSyncBatches
+            .AsNoTracking()
+            .FirstOrDefaultAsync(b => b.BatchReference == request.BatchReference, ct);
+        if (existing is not null)
+            return new ClinicalBatchUploadResponse(existing.Id, existing.Status.ToDbString());
+
+        var gateway = await _db.WardGateways
+            .FirstOrDefaultAsync(g => g.Id == request.GatewayId && g.SiteId == request.SiteId, ct)
+            ?? throw new NotFoundException("Gateway not found for site.", "GATEWAY_NOT_FOUND");
+
+        var payload = JsonSerializer.Serialize(request);
+        var batch = new ClinicalSyncBatch(gateway.Id, request.SiteId, request.BatchReference, payload);
+        var batchId = Guid.NewGuid();
+        _db.Entry(batch).Property(nameof(ClinicalSyncBatch.Id)).CurrentValue = batchId;
+
+        await using var tx = await _db.Database.BeginTransactionAsync(ct);
+        _db.ClinicalSyncBatches.Add(batch);
+        _db.OutboxEvents.Add(new OutboxEvent
+        {
+            Id = Guid.NewGuid(),
+            Topic = ClinicalSyncOptions.BatchReceivedOutboxTopic,
+            Payload = JsonSerializer.Serialize(new
+            {
+                batchId,
+                gatewayId = gateway.Id,
+                siteId = request.SiteId
+            }),
+            PartitionKey = gateway.Id.ToString(),
+            CreatedAt = DateTimeOffset.UtcNow
+        });
+        await _db.SaveChangesAsync(ct);
+        await tx.CommitAsync(ct);
+
+        _logger.LogInformation(
+            "Sync batch {BatchId} received from gateway {GatewayId} — {ObsCount} observations",
+            batch.Id, gateway.Id, request.Observations.Count);
+
+        return new ClinicalBatchUploadResponse(batch.Id, "RECEIVED");
+    }
+
+    public async Task<ClinicalBatchStatusResponse> GetBatchStatusAsync(Guid batchId, CancellationToken ct)
+    {
+        var batch = await _db.ClinicalSyncBatches
+            .AsNoTracking()
+            .Include(b => b.Conflicts)
+            .FirstOrDefaultAsync(b => b.Id == batchId, ct)
+            ?? throw new NotFoundException("Sync batch not found.", "BATCH_NOT_FOUND");
+
+        var conflicts = batch.Conflicts.Select(c =>
+            new ClinicalConflictDetail(c.ClientRef, c.ItemType, c.ConflictReason)).ToList();
+
+        return new ClinicalBatchStatusResponse(batch.Id, batch.Status.ToDbString(), conflicts);
+    }
+
+    public async Task<IReadOnlyList<ClinicalSyncHistoryItem>> GetSyncHistoryAsync(
+        Guid siteId, Guid gatewayId, int limit, CancellationToken ct)
+    {
+        var gatewayExists = await _db.WardGateways
+            .AnyAsync(g => g.Id == gatewayId && g.SiteId == siteId, ct);
+        if (!gatewayExists)
+            throw new NotFoundException("Gateway not found for site.", "GATEWAY_NOT_FOUND");
+
+        return await _db.ClinicalSyncBatches
+            .AsNoTracking()
+            .Where(b => b.GatewayId == gatewayId && b.SiteId == siteId)
+            .OrderByDescending(b => b.SubmittedAt)
+            .Take(limit)
+            .Select(b => new ClinicalSyncHistoryItem(
+                b.Id,
+                b.BatchReference,
+                b.Status.ToDbString(),
+                b.Conflicts.Count,
+                b.SubmittedAt))
+            .ToListAsync(ct);
+    }
+}
@@ -11,4 +11,6 @@ public interface IAlertService
    Task<ClinicalAlert> AcknowledgeAsync(Guid id, AcknowledgeAlertRequest req);

    Task<ClinicalAlert> ResolveAsync(Guid id);
+    Task ApplySyncedAcknowledgmentAsync(Guid alertId, SyncedAlertAcknowledgment ack, CancellationToken ct);
+    Task ApplySyncedResolutionAsync(Guid alertId, SyncedAlertResolution resolve, CancellationToken ct);
 }
@@ -0,0 +1,9 @@
+using VigilCare.ClinicalContracts.Sync;
+
+public interface IClinicalSyncService
+{
+    Task<ClinicalBatchUploadResponse> UploadBatchAsync(ClinicalSyncBatchRequest request, CancellationToken ct);
+    Task<ClinicalBatchStatusResponse> GetBatchStatusAsync(Guid batchId, CancellationToken ct);
+    Task<IReadOnlyList<ClinicalSyncHistoryItem>> GetSyncHistoryAsync(
+        Guid siteId, Guid gatewayId, int limit, CancellationToken ct);
+}
@@ -1,4 +1,5 @@
 public interface IObservationService
 {
    Task<IngestResult> IngestAsync(Guid encounterId, IngestObservationRequest req);
+    Task ApplySyncedObservationAsync(SyncedObservation obs, CancellationToken ct);
 }
@@ -187,6 +187,95 @@ public class ObservationService : IObservationService
        }
    }

+    public async Task ApplySyncedObservationAsync(SyncedObservation obs, CancellationToken ct)
+    {
+        var encounter = await _db.Encounters
+            .Include(e => e.Patient)
+            .FirstOrDefaultAsync(e => e.Id == obs.EncounterId, ct)
+            ?? throw new ValidationException("Encounter not found.", "ENCOUNTER_NOT_FOUND");
+
+        if (encounter.Status != EncounterStatus.Active)
+            throw new ValidationException("Encounter not active.", "ENCOUNTER_NOT_ACTIVE");
+
+        await using var tx = await _db.Database.BeginTransactionAsync(ct);
+        try
+        {
+            var observation = new Observation
+            {
+                Id = obs.ClientRef,
+                EncounterId = obs.EncounterId,
+                ObservationCode = obs.ObservationCode,
+                Value = obs.Value,
+                Unit = obs.Unit ?? "",
+                Source = ObservationSourceExtensions.FromDbString(obs.Source),
+                IdempotencyKey = obs.IdempotencyKey,
+                RecordedAt = obs.RecordedAt,
+                CreatedAt = DateTimeOffset.UtcNow
+            };
+            _db.Observations.Add(observation);
+
+            var threshold = await LoadThresholdAsync(obs.ObservationCode);
+            if (threshold is not null && IsCriticalBreach(obs.Value, threshold))
+            {
+                var hasOpenAlert = await _db.ClinicalAlerts.AnyAsync(a =>
+                    a.EncounterId == obs.EncounterId
+                    && a.ObservationId == observation.Id, ct);
+
+                if (!hasOpenAlert)
+                {
+                    var alert = new ClinicalAlert
+                    {
+                        Id = Guid.NewGuid(),
+                        EncounterId = obs.EncounterId,
+                        PatientId = encounter.PatientId,
+                        ObservationId = observation.Id,
+                        AlertType = AlertTypeExtensions.CriticalFor(obs.ObservationCode),
+                        Severity = AlertSeverity.Critical,
+                        Details = BuildCriticalDetails(
+                            new IngestObservationRequest(obs.ObservationCode, obs.Value, obs.Unit ?? "",
+                                observation.Source, obs.RecordedAt, obs.IdempotencyKey),
+                            threshold),
+                        Status = AlertStatus.Open,
+                        TriggeredAt = obs.RecordedAt
+                    };
+                    _db.ClinicalAlerts.Add(alert);
+                    _db.OutboxEvents.Add(BuildOutboxEvent("alert.generated", new
+                    {
+                        alertId = alert.Id,
+                        encounterId = obs.EncounterId,
+                        patientId = encounter.PatientId,
+                        alertType = alert.AlertType.ToDbString(),
+                        severity = alert.Severity.ToDbString(),
+                        details = alert.Details,
+                        triggeredAt = alert.TriggeredAt,
+                        partitionKey = obs.EncounterId.ToString()
+                    }, obs.EncounterId.ToString()));
+                }
+            }
+
+            _db.OutboxEvents.Add(BuildOutboxEvent("observation.recorded", new
+            {
+                observationId = observation.Id,
+                encounterId = obs.EncounterId,
+                patientId = encounter.PatientId,
+                observationCode = obs.ObservationCode,
+                value = obs.Value,
+                unit = obs.Unit,
+                source = obs.Source,
+                recordedAt = obs.RecordedAt,
+                partitionKey = obs.EncounterId.ToString()
+            }, obs.EncounterId.ToString()));
+
+            await _db.SaveChangesAsync(ct);
+            await tx.CommitAsync(ct);
+        }
+        catch (DbUpdateException ex) when (IsUniqueViolation(ex))
+        {
+            await tx.RollbackAsync(ct);
+            // silent skip — idempotency key already applied
+        }
+    }
+
    private async Task<ThresholdCacheEntry?> LoadThresholdAsync(string observationCode)
    {
        var cache = _redis.GetDatabase();
@@ -34,7 +34,9 @@ public class QsofaService : IQsofaService
        var query = _db.ClinicalAlerts
            .AsNoTracking()
            .Where(a => a.EncounterId == encounterId
+#pragma warning disable CS0618 // Include legacy QSOFA_WARNING rows in history
                && (a.AlertType == AlertType.QsofaScreen || a.AlertType == AlertType.QsofaWarning));
+#pragma warning restore CS0618

        var items = await query
            .OrderByDescending(a => a.TriggeredAt)