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 _logger; public LocalObservationService( GatewayDbContext db, IConnectionMultiplexer redis, LocalWarningEvaluator warnings, LocalPagingPublisher paging, ILogger logger) { _db = db; _redis = redis; _warnings = warnings; _paging = paging; _logger = logger; } public async Task 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 LoadThresholdAsync(string observationCode) { var cache = _redis.GetDatabase(); var cacheKey = $"threshold:{observationCode}"; var cached = await cache.StringGetAsync(cacheKey); if (cached.HasValue) return JsonSerializer.Deserialize(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"; }