245 lines
10 KiB
C#
245 lines
10 KiB
C#
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();
|
|
|
|
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.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();
|
|
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))
|
|
{
|
|
// 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";
|
|
} |