feature: Observation Ingest, Synchronous Alert Detection, and Alert Lifecycle
This commit is contained in:
voltsrage
@@ -0,0 +1,215 @@
|
||||
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
|
||||
.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,
|
||||
alertType = alert.AlertType.ToDbString(),
|
||||
severity = alert.Severity.ToDbString(),
|
||||
triggeredAt = alert.TriggeredAt,
|
||||
partitionKey = 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,
|
||||
observationCode = req.ObservationCode,
|
||||
value = req.Value,
|
||||
unit = req.Unit,
|
||||
source = req.Source.ToDbString(),
|
||||
recordedAt = req.RecordedAt,
|
||||
partitionKey = 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) => new()
|
||||
{
|
||||
Id = Guid.NewGuid(),
|
||||
Topic = topic,
|
||||
Payload = JsonSerializer.Serialize(payload),
|
||||
CreatedAt = DateTimeOffset.UtcNow
|
||||
};
|
||||
|
||||
private static bool IsUniqueViolation(DbUpdateException ex) =>
|
||||
ex.InnerException is Npgsql.PostgresException pg && pg.SqlState == "23505";
|
||||
}
|
||||
Reference in New Issue
Block a user