vigilcare-clinical/VigilCareClinicalAPI/Services/ObservationService.cs

using System.Text.Json;
using Microsoft.EntityFrameworkCore;
using Prometheus;
using Serilog.Context;
using StackExchange.Redis;

public class ObservationService : IObservationService
{
    private readonly AppDbContext _db;
    private readonly IConnectionMultiplexer _redis;
    private readonly ILogger<ObservationService> _logger;
    private readonly ClinicalMetrics _metrics;

    public ObservationService(
        AppDbContext db,
        IConnectionMultiplexer redis,
        ILogger<ObservationService> logger,
        ClinicalMetrics metrics)
    {
        _db = db;
        _redis = redis;
        _logger = logger;
        _metrics = metrics;
    }

    public async Task<IngestResult> IngestAsync(Guid encounterId, IngestObservationRequest req)
    {
        // Step 1 — encounter must be active
        var encounter = await _db.Encounters
            .Include(e => e.Patient)
            .AsNoTracking()
            .FirstOrDefaultAsync(e => e.Id == encounterId);

        if (encounter is null)
            throw new NotFoundException("Encounter not found.", "ENCOUNTER_NOT_FOUND");

        if (encounter.Status != EncounterStatus.Active)
            throw new ConflictException(
                $"Cannot record observations for an encounter with status '{encounter.Status}'.",
                "ENCOUNTER_NOT_ACTIVE");

        // Step 2 — idempotency check before entering the transaction
        // The unique partial index is the database safety net for concurrent retries.
        // The pre-check here avoids the exception-and-rollback path for the common retry case.
        if (!string.IsNullOrEmpty(req.IdempotencyKey))
        {
            var existing = await _db.Observations
                .AsNoTracking()
                .FirstOrDefaultAsync(o => o.IdempotencyKey == req.IdempotencyKey);

            if (existing is not null)
            {
                _logger.LogInformation(
                    "Duplicate idempotency key {Key} for encounter {EncounterId} — returning original",
                    req.IdempotencyKey, encounterId);
                return IngestResult.Duplicate(existing);
            }
        }

        // Step 3 — plausibility check
        if (!PlausibilityValidator.IsPlausible(req.ObservationCode, req.Value, out var reason))
            throw new ValidationException(reason!, "OBSERVATION_OUT_OF_PLAUSIBLE_RANGE");

        using (LogContext.PushProperty("EncounterId", encounterId))
        using (LogContext.PushProperty("PatientId", encounter.PatientId))
        {
            using var timer = _metrics.ObservationIngestDuration.NewTimer();

            // Re-use an outer transaction (e.g. FhirBundleProcessor) when one is already active.
            await using var tx = _db.Database.CurrentTransaction is null
                ? await _db.Database.BeginTransactionAsync()
                : null;
            try
            {
                // Step 4 — insert observation
                var observation = new Observation
                {
                    Id = Guid.NewGuid(),
                    EncounterId = encounterId,
                    ObservationCode = req.ObservationCode,
                    Value = req.Value,
                    Unit = req.Unit,
                    Source = req.Source,
                    IdempotencyKey = req.IdempotencyKey,
                    RecordedAt = req.RecordedAt,
                    CreatedAt = DateTimeOffset.UtcNow
                };
                _db.Observations.Add(observation);

                // Step 5 — load threshold from Redis; fall back to PostgreSQL on miss
                var threshold = await LoadThresholdAsync(req.ObservationCode);

                if (threshold is null)
                    throw new ValidationException(
                        $"No alert threshold is configured for observation code '{req.ObservationCode}'. " +
                        "Register a threshold before recording observations for this code.",
                        "UNKNOWN_OBSERVATION_CODE");

                ClinicalAlert? alert = null;

                // Step 6 — critical threshold detection (synchronous)
                if (IsCriticalBreach(req.Value, threshold))
                {
                    alert = new ClinicalAlert
                    {
                        Id = Guid.NewGuid(),
                        EncounterId = encounterId,
                        PatientId = encounter.PatientId,
                        ObservationId = observation.Id,
                        AlertType = AlertTypeExtensions.CriticalFor(req.ObservationCode),
                        Severity = AlertSeverity.Critical,
                        Details = BuildCriticalDetails(req, threshold),
                        Status = AlertStatus.Open,
                        TriggeredAt = DateTimeOffset.UtcNow
                    };
                    _db.ClinicalAlerts.Add(alert);

                    _db.OutboxEvents.Add(BuildOutboxEvent("alert.generated", new
                    {
                        alertId = alert.Id,
                        encounterId,
                        patientId = encounter.PatientId,
                        department = encounter.Department.ToDbString(),
                        alertType = alert.AlertType.ToDbString(),
                        severity = alert.Severity.ToDbString(),
                        details = alert.Details,
                        attendingPhysician = encounter.AttendingPhysician,
                        triggeredAt = alert.TriggeredAt,
                        partitionKey = encounterId.ToString()
                    }, encounterId.ToString()));
                }

                // Step 7 — outbox event for the observation (always; Kafka consumer handles warnings)
                _db.OutboxEvents.Add(BuildOutboxEvent("observation.recorded", new
                {
                    observationId = observation.Id,
                    encounterId,
                    patientId = encounter.PatientId,
                    mrn = encounter.Patient?.Mrn,
                    observationCode = req.ObservationCode,
                    value = req.Value,
                    unit = req.Unit,
                    source = req.Source.ToDbString(),
                    recordedAt = req.RecordedAt,
                    partitionKey = encounterId.ToString()
                }, encounterId.ToString()));

                // Step 8 — COMMIT
                await _db.SaveChangesAsync();
                if (tx is not null) await tx.CommitAsync();

                _metrics.ObservationsIngestedTotal
                    .WithLabels(req.ObservationCode, req.Source.ToDbString())
                    .Inc();

                if (alert is not null)
                {
                    _metrics.ClinicalAlertsTotal
                        .WithLabels(alert.AlertType.ToDbString(), alert.Severity.ToDbString())
                        .Inc();
                    _logger.LogWarning(
                        "Critical threshold breach. Code={Code} Value={Value} AlertId={AlertId}",
                        req.ObservationCode, req.Value, alert.Id);
                }

                _logger.LogInformation(
                    "Observation {ObservationId} ingested for encounter {EncounterId}. AlertCreated={AlertCreated}",
                    observation.Id, encounterId, alert is not null);

                return IngestResult.Created(observation, alert);
            }
            catch (DbUpdateException ex) when (IsUniqueViolation(ex))
            {
                if (tx is not null) await tx.RollbackAsync();
                var existing = await _db.Observations
                    .AsNoTracking()
                    .FirstOrDefaultAsync(o => o.IdempotencyKey == req.IdempotencyKey);
                if (existing is not null)
                    return IngestResult.Duplicate(existing);
                throw new ConflictException("Concurrent duplicate submission.", "CONCURRENT_DUPLICATE");
            }
            catch
            {
                if (tx is not null) await tx.RollbackAsync();
                throw;
            }
        }
    }

    private async Task<ThresholdCacheEntry?> LoadThresholdAsync(string observationCode)
    {
        var cache = _redis.GetDatabase();
        var cacheKey = $"threshold:{observationCode}";

        var cached = await cache.StringGetAsync(cacheKey);
        if (cached.HasValue)
            return JsonSerializer.Deserialize<ThresholdCacheEntry>(cached!);

        // Cache miss — read from PostgreSQL and write back
        var threshold = await _db.AlertThresholds
            .AsNoTracking()
            .FirstOrDefaultAsync(t => t.ObservationCode == observationCode);

        if (threshold is null) return null;

        var entry = new ThresholdCacheEntry(
            threshold.ObservationCode,
            threshold.CriticalLow,
            threshold.WarningLow,
            threshold.WarningHigh,
            threshold.CriticalHigh);

        await cache.StringSetAsync(cacheKey, JsonSerializer.Serialize(entry));

        _logger.LogDebug("Cache miss for threshold {Code} — loaded from PostgreSQL", observationCode);
        return entry;
    }

    private static bool IsCriticalBreach(decimal value, ThresholdCacheEntry t) =>
        (t.CriticalLow.HasValue && value < t.CriticalLow.Value) ||
        (t.CriticalHigh.HasValue && value > t.CriticalHigh.Value);

    private static string BuildCriticalDetails(IngestObservationRequest req, ThresholdCacheEntry t)
    {
        if (t.CriticalLow.HasValue && req.Value < t.CriticalLow.Value)
            return $"{req.ObservationCode} value {req.Value} {req.Unit} is below critical low of {t.CriticalLow} {req.Unit}.";
        return $"{req.ObservationCode} value {req.Value} {req.Unit} is above critical high of {t.CriticalHigh} {req.Unit}.";
    }

    private static OutboxEvent BuildOutboxEvent(string topic, object payload, string partitionKey) => new()
    {
        Id = Guid.NewGuid(),
        Topic = topic,
        Payload = JsonSerializer.Serialize(payload),
        PartitionKey = partitionKey,
        CreatedAt = DateTimeOffset.UtcNow
    };

    private static bool IsUniqueViolation(DbUpdateException ex) =>
        ex.InnerException is Npgsql.PostgresException pg && pg.SqlState == "23505";
}