vigilcare-clinical/VigilCareClinicalAPI/Services/ObservationService.cs

using System.Text.Json;
using Microsoft.EntityFrameworkCore;
using StackExchange.Redis;

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

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

    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");

        // Steps 4–8 are one atomic transaction
        await using var tx = await _db.Database.BeginTransactionAsync();
        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)
            // WARNING detection is intentionally deferred to the Kafka consumer.
            // A critical potassium of 2.1 mEq/L is immediately life-threatening — the alert
            // must exist before this API call returns. A warning heart rate of 95 bpm warrants
            // attention but not an emergency page; the additional Kafka latency is clinically safe.
            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);

                // Step 6b — outbox event for the alert (relay picks this up in Phase 3)
                _db.OutboxEvents.Add(BuildOutboxEvent("alert.generated", new
                {
                    alertId = alert.Id,
                    encounterId,
                    patientId = encounter.PatientId,
                    department = encounter.Department,
                    alertType = alert.AlertType.ToDbString(),
                    severity = alert.Severity.ToDbString(),
                    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();
            await tx.CommitAsync();

            _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))
        {
            // Race condition: two concurrent retries both passed the pre-check above.
            // The unique partial index caught it. Roll back and return the existing row.
            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
        {
            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";
}