vigilcare-clinical/VigilCare.WardGateway/Services/LocalObservationService.cs

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

public class LocalObservationService : ILocalObservationService
{
    private readonly GatewayDbContext _db;
    private readonly IConnectionMultiplexer _redis;
    private readonly LocalWarningEvaluator _warnings;
    private readonly LocalPagingPublisher _paging;
    private readonly ILogger<LocalObservationService> _logger;

    public LocalObservationService(
        GatewayDbContext db,
        IConnectionMultiplexer redis,
        LocalWarningEvaluator warnings,
        LocalPagingPublisher paging,
        ILogger<LocalObservationService> logger)
    {
        _db = db;
        _redis = redis;
        _warnings = warnings;
        _paging = paging;
        _logger = logger;
    }

    public async Task<LocalIngestResult> IngestAsync(Guid encounterId, IngestObservationRequest req)
    {
        var encounter = await _db.Encounters
            .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("Encounter is not active.", "ENCOUNTER_NOT_ACTIVE");

        if (!string.IsNullOrEmpty(req.IdempotencyKey))
        {
            var existing = await _db.Observations.AsNoTracking()
                .FirstOrDefaultAsync(o => o.IdempotencyKey == req.IdempotencyKey);
            if (existing is not null)
                return LocalIngestResult.Duplicate(existing);
        }

        if (!PlausibilityValidator.IsPlausible(req.ObservationCode, req.Value, out var reason))
            throw new ValidationException(reason!, "OBSERVATION_OUT_OF_PLAUSIBLE_RANGE");

        await using var tx = await _db.Database.BeginTransactionAsync();
        try
        {
            var observation = new LocalObservation
            {
                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);

            var threshold = await LoadThresholdAsync(req.ObservationCode);
            if (threshold is null)
                throw new ValidationException("Unknown observation code.", "UNKNOWN_OBSERVATION_CODE");

            LocalClinicalAlert? alert = null;

            if (IsCriticalBreach(req.Value, threshold))
            {
                var triggeredAt = DateTimeOffset.UtcNow;
                alert = new LocalClinicalAlert
                {
                    Id = Guid.NewGuid(),
                    ClientAlertId = 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 = triggeredAt
                };
                _db.ClinicalAlerts.Add(alert);

                BufferedSyncWriter.Enqueue(_db, BufferedSyncItemType.Alert, new
                {
                    alert.ClientAlertId,
                    alert.EncounterId,
                    alertType = alert.AlertType.ToDbString(),
                    severity = alert.Severity.ToDbString(),
                    alert.Details,
                    generatedAt = triggeredAt
                }, $"alert:{alert.ClientAlertId}", encounterId, triggeredAt);

                await _paging.PublishCriticalAsync(alert, encounter);
            }

            if (alert is null)
                await _warnings.TryEvaluateAsync(
                    observation.Id, encounterId, encounter.PatientId,
                    req.ObservationCode, req.Value, threshold);

            BufferedSyncWriter.Enqueue(_db, BufferedSyncItemType.Observation, new
            {
                clientRef = observation.Id,
                req.IdempotencyKey,
                encounterId,
                req.ObservationCode,
                req.Value,
                req.Unit,
                source = req.Source.ToDbString(),
                req.RecordedAt
            }, req.IdempotencyKey ?? $"obs:{observation.Id}", encounterId, req.RecordedAt);

            await _db.SaveChangesAsync();
            await tx.CommitAsync();

            return LocalIngestResult.Created(observation, alert);
        }
        catch (DbUpdateException ex) when (IsUniqueViolation(ex))
        {
            await tx.RollbackAsync();
            var existing = await _db.Observations.AsNoTracking()
                .FirstOrDefaultAsync(o => o.IdempotencyKey == req.IdempotencyKey);
            if (existing is not null) return LocalIngestResult.Duplicate(existing);
            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!);

        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 bool IsUniqueViolation(DbUpdateException ex) =>
        ex.InnerException is Npgsql.PostgresException pg && pg.SqlState == "23505";
}