feature: Warning Alert Consumer, Orders API & Input Validation
This commit is contained in:
voltsrage
@@ -0,0 +1,8 @@
|
||||
public interface IOrderService
|
||||
{
|
||||
Task<Order> CreateAsync(Guid encounterId, CreateOrderRequest req);
|
||||
Task<PagedResult<Order>> ListByEncounterAsync(Guid encounterId, OrderStatus? status, int page, int pageSize);
|
||||
Task<Order> GetByIdAsync(Guid id);
|
||||
Task<Order> TransitionStatusAsync(Guid id, OrderStatus targetStatus);
|
||||
Task<Order> RecordResultAsync(Guid id, RecordOrderResultRequest req);
|
||||
}
|
||||
@@ -0,0 +1,114 @@
|
||||
using Microsoft.EntityFrameworkCore;
|
||||
|
||||
public class OrderService : IOrderService
|
||||
{
|
||||
private static readonly Dictionary<OrderStatus, HashSet<OrderStatus>> _allowedTransitions = new()
|
||||
{
|
||||
[OrderStatus.Pending] = new() { OrderStatus.InProgress, OrderStatus.Cancelled },
|
||||
[OrderStatus.InProgress] = new() { OrderStatus.Resulted, OrderStatus.Cancelled },
|
||||
[OrderStatus.Resulted] = new(),
|
||||
[OrderStatus.Cancelled] = new(),
|
||||
};
|
||||
|
||||
private readonly AppDbContext _db;
|
||||
|
||||
public OrderService(AppDbContext db) => _db = db;
|
||||
|
||||
public async Task<Order> CreateAsync(Guid encounterId, CreateOrderRequest req)
|
||||
{
|
||||
var encounter = await _db.Encounters.FindAsync(encounterId);
|
||||
if (encounter is null)
|
||||
throw new NotFoundException("Encounter not found.", "ENCOUNTER_NOT_FOUND");
|
||||
|
||||
if (encounter.Status != EncounterStatus.Active)
|
||||
throw new ConflictException(
|
||||
"Cannot create orders for a non-active encounter.",
|
||||
"ENCOUNTER_NOT_ACTIVE");
|
||||
|
||||
var order = new Order
|
||||
{
|
||||
Id = Guid.NewGuid(),
|
||||
EncounterId = encounterId,
|
||||
OrderType = req.OrderType,
|
||||
Description = req.Description,
|
||||
OrderedBy = req.OrderedBy,
|
||||
Status = OrderStatus.Pending,
|
||||
OrderedAt = DateTimeOffset.UtcNow
|
||||
};
|
||||
_db.Orders.Add(order);
|
||||
await _db.SaveChangesAsync();
|
||||
return order;
|
||||
}
|
||||
|
||||
public async Task<PagedResult<Order>> ListByEncounterAsync(
|
||||
Guid encounterId, OrderStatus? status, int page, int pageSize)
|
||||
{
|
||||
var query = _db.Orders
|
||||
.AsNoTracking()
|
||||
.Where(o => o.EncounterId == encounterId);
|
||||
|
||||
if (status.HasValue)
|
||||
query = query.Where(o => o.Status == status.Value);
|
||||
|
||||
var total = await query.CountAsync();
|
||||
var orders = await query
|
||||
.OrderByDescending(o => o.OrderedAt)
|
||||
.Skip((page - 1) * pageSize)
|
||||
.Take(pageSize)
|
||||
.ToListAsync();
|
||||
|
||||
return new PagedResult<Order>(orders, page, pageSize, total);
|
||||
}
|
||||
|
||||
public async Task<Order> GetByIdAsync(Guid id)
|
||||
{
|
||||
var order = await _db.Orders
|
||||
.AsNoTracking()
|
||||
.Include(o => o.Encounter)
|
||||
.FirstOrDefaultAsync(o => o.Id == id);
|
||||
|
||||
if (order is null)
|
||||
throw new NotFoundException("Order not found.", "ORDER_NOT_FOUND");
|
||||
|
||||
return order;
|
||||
}
|
||||
|
||||
public async Task<Order> TransitionStatusAsync(Guid id, OrderStatus targetStatus)
|
||||
{
|
||||
var order = await _db.Orders.FindAsync(id);
|
||||
if (order is null)
|
||||
throw new NotFoundException("Order not found.", "ORDER_NOT_FOUND");
|
||||
|
||||
if (!_allowedTransitions[order.Status].Contains(targetStatus))
|
||||
throw new ConflictException(
|
||||
$"Transition to '{targetStatus}' is not permitted from status '{order.Status}'.",
|
||||
"ILLEGAL_ORDER_STATUS_TRANSITION");
|
||||
|
||||
order.Status = targetStatus;
|
||||
if (targetStatus == OrderStatus.Resulted)
|
||||
order.ResultedAt = DateTimeOffset.UtcNow;
|
||||
|
||||
await _db.SaveChangesAsync();
|
||||
return order;
|
||||
}
|
||||
|
||||
public async Task<Order> RecordResultAsync(Guid id, RecordOrderResultRequest req)
|
||||
{
|
||||
var order = await _db.Orders.FindAsync(id);
|
||||
if (order is null)
|
||||
throw new NotFoundException("Order not found.", "ORDER_NOT_FOUND");
|
||||
|
||||
if (order.Status == OrderStatus.Resulted)
|
||||
throw new ConflictException("Order already resulted.", "ORDER_ALREADY_RESULTED");
|
||||
|
||||
if (order.Status == OrderStatus.Cancelled)
|
||||
throw new ConflictException("Cannot result a cancelled order.", "ORDER_CANCELLED");
|
||||
|
||||
order.Status = OrderStatus.Resulted;
|
||||
order.ResultedAt = DateTimeOffset.UtcNow;
|
||||
order.ResultSummary = req.ResultSummary;
|
||||
|
||||
await _db.SaveChangesAsync();
|
||||
return order;
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,138 @@
|
||||
using System.Text.Json;
|
||||
using Microsoft.EntityFrameworkCore;
|
||||
using StackExchange.Redis;
|
||||
|
||||
public class WarningEvaluator
|
||||
{
|
||||
private readonly IConnectionMultiplexer _redis;
|
||||
private readonly IServiceProvider _services;
|
||||
private readonly ILogger<WarningEvaluator> _logger;
|
||||
|
||||
public WarningEvaluator(
|
||||
IConnectionMultiplexer redis,
|
||||
IServiceProvider services,
|
||||
ILogger<WarningEvaluator> logger)
|
||||
{
|
||||
_redis = redis;
|
||||
_services = services;
|
||||
_logger = logger;
|
||||
}
|
||||
|
||||
public async Task<bool> EvaluateAsync(
|
||||
Guid observationId,
|
||||
Guid encounterId,
|
||||
Guid patientId,
|
||||
string observationCode,
|
||||
decimal value,
|
||||
CancellationToken ct = default)
|
||||
{
|
||||
var threshold = await LoadThresholdAsync(observationCode);
|
||||
if (threshold is null) return false;
|
||||
|
||||
if (!IsWarningBreach(value, threshold)) return false;
|
||||
|
||||
// Do not create a warning if the value is also a critical breach —
|
||||
// critical alerts are created synchronously by the ingest path.
|
||||
if (IsCriticalBreach(value, threshold)) return false;
|
||||
|
||||
return await TryCreateWarningAlertAsync(
|
||||
observationId, encounterId, patientId, observationCode, value, threshold, ct);
|
||||
}
|
||||
|
||||
private static bool IsWarningBreach(decimal value, ThresholdCacheEntry t) =>
|
||||
(t.WarningHigh.HasValue && value > t.WarningHigh.Value) ||
|
||||
(t.WarningLow.HasValue && value < t.WarningLow.Value);
|
||||
|
||||
private static bool IsCriticalBreach(decimal value, ThresholdCacheEntry t) =>
|
||||
(t.CriticalLow.HasValue && value < t.CriticalLow.Value) ||
|
||||
(t.CriticalHigh.HasValue && value > t.CriticalHigh.Value);
|
||||
|
||||
private async Task<ThresholdCacheEntry?> LoadThresholdAsync(string observationCode)
|
||||
{
|
||||
var cache = _redis.GetDatabase();
|
||||
var cached = await cache.StringGetAsync($"threshold:{observationCode}");
|
||||
if (cached.HasValue)
|
||||
return JsonSerializer.Deserialize<ThresholdCacheEntry>(cached!);
|
||||
return null;
|
||||
}
|
||||
|
||||
// Idempotent INSERT: prevents duplicate warning alerts for the same observation.
|
||||
// The WHERE NOT EXISTS checks for an open warning alert of the same type for the
|
||||
// same encounter. Unlike critical alerts (one per encounter), warning alerts are
|
||||
// expected to recur — but not for every single observation in a series. If the
|
||||
// patient's heart rate stays at 105 bpm for an hour, one WARNING_HEART_RATE is
|
||||
// sufficient until acknowledged or resolved.
|
||||
private async Task<bool> TryCreateWarningAlertAsync(
|
||||
Guid observationId,
|
||||
Guid encounterId,
|
||||
Guid patientId,
|
||||
string observationCode,
|
||||
decimal value,
|
||||
ThresholdCacheEntry threshold,
|
||||
CancellationToken ct)
|
||||
{
|
||||
using var scope = _services.CreateScope();
|
||||
var db = scope.ServiceProvider.GetRequiredService<AppDbContext>();
|
||||
|
||||
await using var tx = await db.Database.BeginTransactionAsync(ct);
|
||||
|
||||
var alertType = AlertTypeExtensions.WarningFor(observationCode);
|
||||
var alertId = Guid.NewGuid();
|
||||
var triggeredAt = DateTimeOffset.UtcNow;
|
||||
var details = BuildWarningDetails(observationCode, value, threshold);
|
||||
|
||||
var affected = await db.Database.ExecuteSqlInterpolatedAsync($"""
|
||||
INSERT INTO clinical_alerts
|
||||
(id, encounter_id, patient_id, observation_id, alert_type, severity, details, status, triggered_at)
|
||||
SELECT {alertId}, {encounterId}, {patientId}, {observationId},
|
||||
{alertType.ToDbString()}, 'WARNING', {details}, 'OPEN', {triggeredAt}
|
||||
WHERE NOT EXISTS (
|
||||
SELECT 1 FROM clinical_alerts
|
||||
WHERE encounter_id = {encounterId}
|
||||
AND alert_type = {alertType.ToDbString()}
|
||||
AND status IN ('OPEN', 'ACKNOWLEDGED')
|
||||
)
|
||||
""", ct);
|
||||
|
||||
if (affected == 0)
|
||||
{
|
||||
await tx.RollbackAsync(ct);
|
||||
return false;
|
||||
}
|
||||
|
||||
db.OutboxEvents.Add(new OutboxEvent
|
||||
{
|
||||
Id = Guid.NewGuid(),
|
||||
Topic = "alert.generated",
|
||||
Payload = JsonSerializer.Serialize(new
|
||||
{
|
||||
alertId,
|
||||
encounterId,
|
||||
patientId,
|
||||
alertType = alertType.ToDbString(),
|
||||
severity = "Warning",
|
||||
triggeredAt,
|
||||
partitionKey = encounterId.ToString()
|
||||
}),
|
||||
PartitionKey = encounterId.ToString(),
|
||||
CreatedAt = DateTimeOffset.UtcNow
|
||||
});
|
||||
|
||||
await db.SaveChangesAsync(ct);
|
||||
await tx.CommitAsync(ct);
|
||||
|
||||
_logger.LogInformation(
|
||||
"WARNING alert {AlertId} created for encounter {EncounterId} — {Code}={Value}",
|
||||
alertId, encounterId, observationCode, value);
|
||||
|
||||
return true;
|
||||
}
|
||||
|
||||
private static string BuildWarningDetails(
|
||||
string code, decimal value, ThresholdCacheEntry t)
|
||||
{
|
||||
if (t.WarningHigh.HasValue && value > t.WarningHigh.Value)
|
||||
return $"{code} value {value} is above warning high of {t.WarningHigh}.";
|
||||
return $"{code} value {value} is below warning low of {t.WarningLow}.";
|
||||
}
|
||||
}
|
||||
Reference in New Issue
Block a user