fix: Missing input validators + No patient update endpoint + Pagination inconsistencies + Missing list/get endpoints
This commit is contained in:
voltsrage
@@ -2,7 +2,7 @@
|
||||
|
||||
A production-quality clinical backend built with ASP.NET Core 8, PostgreSQL, Apache Kafka, RabbitMQ, Elasticsearch, Redis, and MinIO. The domain models the observe-alert-acknowledge lifecycle at the center of any clinical monitoring system: patient encounters, continuous vital sign and lab result ingest, real-time sepsis and NEWS2 scoring, and clinician notification with automatic escalation.
|
||||
|
||||
**Implementation status:** Twenty-six planned phases are complete through Phase 31 — from schema and CRUD through Kafka, Elasticsearch CQRS, sepsis detection, RabbitMQ paging with DLQ escalation, reconciliation jobs, Prometheus/Grafana observability, the MinIO Parquet data lake, clinical data model expansion, warning alerts and orders, the NEWS2 composite scoring engine, trend detection with alert suppression, qSOFA bedside screening, medication administration with alert correlation annotations, the console replay simulator, the **Vue 3 ward dashboard**, clinician feedback mode, **Glasgow Coma Scale (GCS) scoring**, **SOFA organ-dysfunction scoring with baseline tracking and delta sepsis alerts**, the **Sepsis-3 clinical refactor** (SIRS removed, qSOFA repositioned as screening, SOFA delta ≥ 2 triggers bundles), **frontend GCS entry and SOFA display**, **expanded simulator scenarios with clinical validation**, the **FHIR R4 Inbound Facade** for EHR integration, and **Role-Based Access Control (RBAC) with clinical audit logging**. See [Implemented Phases](#implemented-phases) for the full breakdown. Guides: [dashboard-guide.md](docs/dashboard-guide.md) (technical), [clinical-testing-guide.md](docs/clinical-testing-guide.md) (doctors & nurses).
|
||||
**Implementation status:** Twenty-six planned phases are complete through Phase 31 — from schema and CRUD through Kafka, Elasticsearch CQRS, sepsis detection, RabbitMQ paging with DLQ escalation, reconciliation jobs, Prometheus/Grafana observability, the MinIO Parquet data lake, clinical data model expansion, warning alerts and orders, the NEWS2 composite scoring engine, trend detection with alert suppression, qSOFA bedside screening, medication administration with alert correlation annotations, the console replay simulator, the **Vue 3 ward dashboard**, clinician feedback mode, **Glasgow Coma Scale (GCS) scoring**, **SOFA organ-dysfunction scoring with baseline tracking and delta sepsis alerts**, the **Sepsis-3 clinical refactor** (SIRS removed, qSOFA repositioned as screening, SOFA delta ≥ 2 triggers bundles), **frontend GCS entry and SOFA display**, **expanded simulator scenarios with clinical validation**, the **FHIR R4 Inbound Facade** for EHR integration, and **Role-Based Access Control (RBAC) with clinical audit logging**. Post-phase hardening includes health check endpoints, Kafka poison pill protection, outbox dead-letter with retry tracking, data lake partial-commit safety, MRN sequence-based generation, and FHIR bundle transaction rollback. See [Implemented Phases](#implemented-phases) for the full breakdown. Guides: [dashboard-guide.md](docs/dashboard-guide.md) (technical), [clinical-testing-guide.md](docs/clinical-testing-guide.md) (doctors & nurses).
|
||||
|
||||
## Domain Model — How It Maps to a Real Clinical System
|
||||
|
||||
@@ -59,8 +59,8 @@ An `OutboxEvent` is written in the same transaction as any observation or alert,
|
||||
- **Warning Threshold Alerts** — `WarningEvaluator` reads thresholds from Redis; creates `WARNING`-severity alerts for values above `warningHigh` or below `warningLow` that are not also critical breaches; idempotent `INSERT WHERE NOT EXISTS` per encounter and alert type while status is `OPEN` or `ACKNOWLEDGED`; warning alerts are indexed in Elasticsearch but not published to the RabbitMQ paging queue
|
||||
- **Clinical Order Management** — `POST /encounters/:id/orders` create; `GET /encounters/:id/orders` list with optional status filter; `GET /orders/:id` detail; `PATCH /orders/:id/status` status transitions; `PATCH /orders/:id/result` record result and transition to `Resulted`; status machine enforces `Pending → InProgress → Resulted` and terminal `Cancelled`
|
||||
- **Clinical Alert Lifecycle** — paginated alert list per encounter and globally; acknowledge with clinician ID and optional note; resolve (must be acknowledged first); global list filterable by status, severity, and department
|
||||
- **Outbox Relay** — `IHostedService` polling every 500ms; reads unprocessed outbox rows, publishes to Kafka, marks processed; partitioned by `encounterId` for per-encounter ordering
|
||||
- **Kafka Pipeline** — core topics (`observation.recorded`, `alert.generated`, `encounter.status.changed`, `gcs.scored`, sepsis bundle topics) with six partitions each; KRaft mode, no Zookeeper; `KAFKA_AUTO_CREATE_TOPICS_ENABLE=false` — topics are provisioned explicitly by `KafkaTopicProvisioner` (including `gcs.scored` for SOFA CNS re-scoring and `sofa.scored` for downstream consumers)
|
||||
- **Outbox Relay** — `IHostedService` polling every 500ms; reads unprocessed outbox rows with `FOR UPDATE SKIP LOCKED` (safe for concurrent instances), publishes to Kafka via idempotent producer (`EnableIdempotence = true`), marks processed; per-event retry tracking (`RetryCount`, `LastError`); events exceeding `OutboxMaxRetries` (default 10) are marked permanently failed (`FailedAt`) and excluded from future polls; partitioned by `encounterId` for per-encounter ordering
|
||||
- **Kafka Pipeline** — core topics (`observation.recorded`, `alert.generated`, `encounter.status.changed`, `gcs.scored`, sepsis bundle topics) with six partitions each; configurable `ReplicationFactor` (default 3); KRaft mode, no Zookeeper; `KAFKA_AUTO_CREATE_TOPICS_ENABLE=false` — topics are provisioned explicitly by `KafkaTopicProvisioner` (including `gcs.scored` for SOFA CNS re-scoring and `sofa.scored` for downstream consumers); all consumers protected by `PoisonPillGuard` (permanent errors skipped, transient errors retried up to `MaxPoisonRetries`)
|
||||
- **Elasticsearch CQRS Projection** — `EsIndexerService` consumer group upserts `patient_encounters` documents, appends to the `observations` index, increments `openAlertCount` on alert events, stamps `news2Score` / `news2RiskLevel` when a NEWS2 alert is generated, and projects `sepsisBundleStatus` / `sepsisBundleElementsCompleted` / `sepsisBundleDeadlineAt` from SOFA-triggered sepsis bundle events; patient/encounter search; per-encounter observation trend (hourly avg/min/max); alert volume summary by department and severity; population query (numeric range aggregation across all patients)
|
||||
- **Sepsis Screening Engine** — `SepsisEngineService` Kafka consumer evaluates qSOFA criteria per encounter using Redis keys with a 30-minute TTL sliding window; qSOFA evaluates respiratory rate ≥ 22, systolic BP ≤ 100, and altered mentation (GCS < 15 or AVPU ≥ 1); on ≥ 2 active criteria and no open screening alert, inserts a `QSOFA_SCREEN` (WARNING-level) alert idempotently (`INSERT WHERE NOT EXISTS`); qSOFA screening recommends ordering SOFA labs — definitive sepsis detection and bundle triggering are handled by `SofaScoringService` via SOFA delta ≥ 2
|
||||
- **Sepsis Bundle Compliance** — `SepsisBundleService` creates a four-element treatment bundle (blood cultures, serum lactate, broad-spectrum antibiotics, IV fluid resuscitation) when a `SOFA_SEPSIS` alert fires (delta ≥ 2 from baseline); each element maps to an auto-created clinical order (`orderedBy: sepsis-bundle-engine`); one-hour compliance deadline from recognition; `OrderService.RecordResult` calls back to `OnOrderResultedAsync` to mark elements complete; final element completion sets bundle to `COMPLIANT` or `NON_COMPLIANT`; `SepsisBundleMonitorService` scans every 5 minutes for overdue in-progress bundles past their deadline and marks them `NON_COMPLIANT`; idempotent — only one in-progress bundle per encounter; `GET /encounters/:id/sepsis-bundle/current` and `GET /sepsis-bundles/:id` expose bundle state; Kafka topics `sepsis.bundle.created` / `sepsis.bundle.updated`; Prometheus `qsofa_detections_total` and `sepsis_bundle_compliance_total`
|
||||
@@ -76,10 +76,17 @@ An `OutboxEvent` is written in the same transaction as any observation or alert,
|
||||
- **FHIR R4 Inbound Facade** — `POST /fhir/R4/{Patient,Encounter,Observation,MedicationAdministration}` accepts FHIR R4 JSON resources (`application/fhir+json`); `POST /fhir/R4` processes transaction Bundles (Patient → Encounter → Observation in dependency order); `GET /fhir/R4/metadata` returns a CapabilityStatement; LOINC-to-internal code mapping (19 observation codes + SNOMED CT fallbacks); Fahrenheit-to-Celsius unit conversion; `ExternalResourceIdentifier` table links hospital MRNs and visit numbers to internal UUIDs for idempotent upserts; `FhirApiKeyMiddleware` authenticates via `X-Api-Key` header; `FhirExceptionFilter` returns FHIR `OperationOutcome` on errors; configurable identifier systems, department codes, and encounter class mappings via `Fhir` config section; Prometheus `fhir_ingest_total` and `fhir_mapping_errors_total`; integration guide for Mirth Connect HL7v2→FHIR channels in `docs/integration/mirth-fhir-channels.md`
|
||||
- **Role-Based Access Control (RBAC)** — JWT bearer authentication (`POST /auth/login`); four clinical roles (`Nurse`, `Physician`, `Admin`, `Integration`) with 16 granular permissions (`patients:read`, `alerts:acknowledge`, `thresholds:write`, `fhir:ingest`, `audit:read`, etc.); `AuthorizePermission` attribute on every controller action; `PermissionAuthorizationHandler` resolves role → permission at runtime from `ClinicalRolePermissionMap`; `CurrentUserService` exposes authenticated identity (user ID, display name, role, IP address) to services; nurses and physicians get clinical read/write permissions; admins additionally get `thresholds:write`, `audit:read`, and `users:admin`; integration accounts get write-only access for FHIR ingest; FHIR endpoints accept both JWT and `X-Api-Key` authentication via `FhirApiKeyOrJwtMiddleware`; alert `acknowledgedBy` is set from the authenticated user identity, not the request body; four seeded demo users (`nurse.demo`, `physician.demo`, `admin.demo`, `integration.mirth`); frontend login page with `localStorage` token persistence and automatic `Authorization: Bearer` header injection
|
||||
- **Clinical Audit Logging** — append-only `clinical_audit_logs` table records clinical write actions with user identity, entity type/ID, before/after state (JSONB), reason, IP address, and correlation ID; eight audit actions (`THRESHOLD_CREATED`, `THRESHOLD_UPDATED`, `ALERT_ACKNOWLEDGED`, `ALERT_RESOLVED`, `ENCOUNTER_STATUS_CHANGED`, `PATIENT_REGISTERED`, `SUPPRESSION_WINDOW_SET`, `USER_LOGIN`); `AuditService` writes log entries inline with domain operations; `GET /audit-logs` admin-only query endpoint with filters by entity type, entity ID, user ID, action, and time range; indexed on entity type, entity ID, user ID, and timestamp
|
||||
- **Health Check Endpoints** — `GET /health/live` (liveness — always returns 200 if the process is running) and `GET /health/ready` (readiness — checks PostgreSQL, Redis, Kafka, RabbitMQ, and Elasticsearch connectivity); both return structured JSON with per-check status and duration; anonymous access; suitable for Kubernetes probes and load balancer health checks
|
||||
- **Kafka Poison Pill Protection** — `PoisonPillGuard` prevents a single un-processable message from blocking a consumer partition forever; permanent errors (malformed JSON, bad format) are skipped immediately; transient errors are retried up to `MaxPoisonRetries` (default 5) before the offset is committed and the message is abandoned; all eight Kafka consumers use the guard; skipped messages are logged at CRITICAL with full payload and tracked by Prometheus `kafka_poison_pills_skipped_total` (labeled by consumer group and topic)
|
||||
- **Outbox Dead-Letter with Retry Tracking** — `OutboxRelayService` tracks `RetryCount`, `LastError`, and `FailedAt` per event; events that fail `OutboxMaxRetries` (default 10) Kafka produce attempts are marked permanently failed (`FailedAt` set) and excluded from future relay polls; uses `FOR UPDATE SKIP LOCKED` for safe concurrent relay instances; idempotent Kafka producer (`EnableIdempotence = true`) prevents duplicate messages from network-level retries
|
||||
- **Data Lake Partial-Commit Safety** — `DataLakeWriterService` commits Kafka offsets only for topic-partitions where all MinIO uploads succeeded; failed partition buffers are retained in memory and retried on the next flush cycle; prevents data loss from partial upload failures
|
||||
- **ThresholdCacheLoader Resilience** — retries Redis connection up to 3 times with exponential backoff (2s, 4s, 8s); if Redis remains unavailable, the application starts without the cache and observation ingest falls back to PostgreSQL queries for threshold lookups
|
||||
- **MRN Sequence Generation** — MRN numbers are generated via a PostgreSQL sequence (`mrn_seq`) instead of MAX+1 queries; eliminates race conditions under concurrent patient registration; configurable prefix and digit count via `PatientOptions`
|
||||
- **FHIR Bundle Transaction Rollback** — `FhirBundleProcessor` wraps all bundle entry processing in a database transaction; on any entry failure, the transaction is rolled back and the response includes the `OperationOutcome` for the failed entry; prevents partial state from orphaned Patient/Encounter records
|
||||
- **Clinician Feedback Mode** — six quick ratings per alert (useful, too early, too late, false positive, missing context, would act); optional notes; Feedback Summary with aggregate stats and JSON/CSV export; client-side persistence for product research
|
||||
- **Console Replay Simulator** — standalone `VigilCare.Simulator` .NET console app replays JSON scenario files against the live API with configurable speed (`--speed 0` instant, `60` = 60× faster); commands: `replay`, `replay-all`, `validate`, `dry-run`; optional `--poll` shows alerts, NEWS2, GCS, SOFA, and sepsis bundle state during replay; eleven sample scenarios in `VigilCare.Simulator/Scenarios/List/` (including GCS neurological decline, SOFA sepsis progression, and SpO₂/FiO₂ fallback); user guide in `docs/simulator-guide.md`
|
||||
- **RabbitMQ Notification Workers** — `NotificationPublisherService` reads `alert.generated` from Kafka and publishes paging jobs to `alerts.paging.queue`; `PagingWorkerService` sends the page and waits for acknowledgment; if no ack arrives before timeout it NACKs to `alerts.paging.dlq` with `x-message-ttl = 300000ms`; if the host is stopping, in-flight paging messages are NACKed with `requeue=true` so they are retried after restart and do not false-escalate; `EscalationWorkerService` pages the on-call backup and sets alert status to `escalated`; `DischargeSummaryWorkerService` reads `encounter.status.changed`, generates a discharge summary, and stores it in MinIO under `/discharge-summaries/{encounterId}/summary.pdf`
|
||||
- **Data Lake Writer** — `DataLakeWriterService` (consumer group `data-lake-writer`) buffers `observation.recorded`, `alert.generated`, and `encounter.status.changed` events, flushes date-partitioned Parquet files to MinIO (`/observations/`, `/alerts/`, `/encounters/`), and commits Kafka offsets only after at least one successful upload; shutdown flush uses an uncanceled token so MinIO writes complete on Ctrl+C; `kafka_partition` and `kafka_offset` columns provide audit lineage
|
||||
- **Data Lake Writer** — `DataLakeWriterService` (consumer group `data-lake-writer`) buffers `observation.recorded`, `alert.generated`, and `encounter.status.changed` events, flushes date-partitioned Parquet files to MinIO (`/observations/`, `/alerts/`, `/encounters/`), and commits Kafka offsets only for topic-partitions where all uploads succeeded; failed partition buffers are retained in memory and retried on the next flush cycle (prevents data loss from partial upload failures); shutdown flush uses an uncanceled token so MinIO writes complete on Ctrl+C; `kafka_partition` and `kafka_offset` columns provide audit lineage
|
||||
- **Reconciliation Jobs** — three scheduled checks: (1) unacknowledged CRITICAL alerts older than 30 minutes, (2) pending orders without results after 4 hours, (3) active inpatients with no observation in 2 hours; each finding creates a `reconciliation_alerts` row and publishes to RabbitMQ
|
||||
- **Standard Envelope** — all responses use a consistent `{ success, statusCode, data, error }` wrapper; validation errors use the same shape; `ApiBehaviorOptions` overridden so model validation also produces the standard envelope with field-level `details`
|
||||
- **Input Validation** — FluentValidation validators on all request DTOs (patient registration, encounter open, observation ingest, alert acknowledge, alert thresholds, orders); invalid requests return 400 before reaching the service layer
|
||||
@@ -125,6 +132,7 @@ IHostedServices (background):
|
||||
DataLakeWriterService → Kafka (data-lake-writer) → Parquet files in MinIO
|
||||
ReconciliationScheduler → three scheduled safety checks → reconciliation_alerts + RabbitMQ
|
||||
SepsisBundleMonitorService → polls overdue bundles every 5 min → marks NON_COMPLIANT
|
||||
PoisonPillGuard (per consumer) → skips permanently malformed messages after MaxPoisonRetries
|
||||
AlertsUnacknowledgedCollector → polls PostgreSQL every 30s → alerts_unacknowledged_gauge
|
||||
OutboxPendingCollector → polls outbox every 30s → outbox_pending_events
|
||||
KafkaConsumerLagCollector → polls four consumer groups every 30s → kafka_consumer_lag
|
||||
@@ -306,11 +314,19 @@ VigilCareClinicalAPI/
|
||||
│ ├── PendingOrdersCheck.cs # Pending orders without results > 4 hours
|
||||
│ ├── DisconnectedMonitorsCheck.cs # Active inpatients with no observation > 2 hours
|
||||
│ └── ReconciliationPublisher.cs # Publishes findings to notifications.reconciliation.queue
|
||||
├── Infrastructure/
|
||||
│ ├── PoisonPillGuard.cs # Kafka consumer protection — skips permanent errors, retries transient up to MaxPoisonRetries
|
||||
│ ├── ElasticsearchHealthCheck.cs # Readiness check for Elasticsearch connectivity
|
||||
│ ├── HealthCheckResponseWriter.cs # Structured JSON response for /health/* endpoints
|
||||
│ ├── KafkaHealthCheck.cs # Readiness check for Kafka broker connectivity
|
||||
│ ├── RabbitMqHealthCheck.cs # Readiness check for RabbitMQ connectivity
|
||||
│ └── RedisHealthCheck.cs # Readiness check for Redis connectivity
|
||||
├── Configuration/
|
||||
│ ├── KafkaOptions.cs / KafkaTopicOptions.cs
|
||||
│ ├── KafkaOptions.cs / KafkaTopicOptions.cs # Includes ReplicationFactor, OutboxMaxRetries, MaxPoisonRetries
|
||||
│ ├── RabbitMqOptions.cs / MinioOptions.cs
|
||||
│ ├── ReconciliationJobOptions.cs
|
||||
│ ├── MedicationCorrelationOptions.cs # Drug-vital mappings + correlation window
|
||||
│ ├── PatientOptions.cs # MRN prefix + digit count for sequence-based generation
|
||||
│ ├── FhirOptions.cs # API key, identifier systems, department/class maps, defaults
|
||||
│ ├── JwtOptions.cs # Issuer, audience, signing key, expiration (default 8 hours)
|
||||
│ └── DashboardOptions.cs # CORS origins for ward dashboard frontend
|
||||
@@ -510,7 +526,7 @@ Redis serves seven independent roles with different semantics:
|
||||
|
||||
### Outbox Pattern
|
||||
|
||||
Observation and alert writes use the transactional outbox: the `outbox_events` row is inserted in the same transaction as the domain record. The relay publishes to Kafka asynchronously. This prevents message loss when Kafka is temporarily unavailable and prevents phantom messages when the transaction rolls back. The relay is idempotent — re-publishing an already-processed event is safe because all downstream consumers check for duplicates.
|
||||
Observation and alert writes use the transactional outbox: the `outbox_events` row is inserted in the same transaction as the domain record. The relay publishes to Kafka asynchronously using an idempotent producer (`EnableIdempotence = true`) — the broker deduplicates in-flight retries using producer ID and sequence number. This prevents message loss when Kafka is temporarily unavailable and prevents phantom messages when the transaction rolls back. Events that repeatedly fail Kafka delivery (network partition, broker-level rejection) are retried up to `OutboxMaxRetries` (default 10) before being marked permanently failed — acting as a dead-letter mechanism that prevents a single poisoned event from blocking the entire relay. The relay uses `FOR UPDATE SKIP LOCKED` so multiple instances can run concurrently without contention.
|
||||
|
||||
### Kafka Partition Key: `encounterId`
|
||||
|
||||
@@ -609,6 +625,10 @@ On startup the application:
|
||||
|
||||
Swagger UI is available at `http://localhost:5270/swagger` in Development (API binds to `0.0.0.0:5270` per `launchSettings.json`).
|
||||
|
||||
Health checks (anonymous, no JWT required):
|
||||
- `GET /health/live` — liveness probe (always 200 if process is running)
|
||||
- `GET /health/ready` — readiness probe (checks PostgreSQL, Redis, Kafka, RabbitMQ, Elasticsearch)
|
||||
|
||||
### Run the Simulator
|
||||
|
||||
With the API running, replay a scenario from the repository root:
|
||||
@@ -781,6 +801,7 @@ See `docs/plans/phase-8-plan.md` through `docs/plans/phase-12-plan.md` for manua
|
||||
| `kafka_consumer_lag` | Gauge | `consumer_group` | `KafkaConsumerLagCollector` — `es-indexer`, `sepsis-engine`, `notification-publisher`, `data-lake-writer` |
|
||||
| `fhir_ingest_total` | Counter | `resource_type`, `outcome` | `FhirIngestController` — per resource type (`Patient`, `Encounter`, `Observation`, `MedicationAdministration`, `Bundle`) with `success` / `error` outcome |
|
||||
| `fhir_mapping_errors_total` | Counter | `resource_type` | `FhirExceptionFilter` — mapping/validation failures by resource type |
|
||||
| `kafka_poison_pills_skipped_total` | Counter | `consumer_group`, `topic` | `PoisonPillGuard` — messages skipped as permanently un-processable (malformed JSON, format errors, or transient failures exceeding MaxPoisonRetries) |
|
||||
|
||||
Prometheus scrapes the API via `infra/prometheus/prometheus.yml` (`job: vigilcare_api` → `host.docker.internal:5270`). Grafana loads the clinical dashboard from `infra/grafana/dashboards/vigilcare.json`.
|
||||
|
||||
@@ -1273,6 +1294,7 @@ observationId Guid? FK → Observation (null for NEWS2, GCS, SOFA composite
|
||||
alertType string e.g. CRITICAL_HEART_RATE, QSOFA_SCREEN, SOFA_SEPSIS, NEWS2_WARNING, NEWS2_EMERGENCY, GCS_CRITICAL
|
||||
severity string WARNING | CRITICAL
|
||||
details text required
|
||||
observationCode string? observation code that triggered this alert (e.g. HEART_RATE) — enables direct lookups without LIKE pattern matching
|
||||
status string open | acknowledged | resolved | escalated (default: open)
|
||||
acknowledgedAt DateTimeOffset?
|
||||
acknowledgedBy string?
|
||||
@@ -1280,7 +1302,7 @@ resolvedAt DateTimeOffset?
|
||||
triggeredAt DateTimeOffset
|
||||
```
|
||||
|
||||
Indexes: `(encounter_id, triggered_at DESC)`, `(patient_id, triggered_at DESC)`, partial `(severity, triggered_at DESC) WHERE status = 'open'`
|
||||
Indexes: `(encounter_id, triggered_at DESC)`, `(patient_id, triggered_at DESC)`, partial `(severity, triggered_at DESC) WHERE status = 'open'`, `(encounter_id, observation_code, status)`
|
||||
|
||||
### News2Score
|
||||
|
||||
@@ -1369,7 +1391,7 @@ complianceStatus string IN_PROGRESS | COMPLIANT | NON_COMPLIANT (default:
|
||||
completedAt DateTimeOffset?
|
||||
```
|
||||
|
||||
Indexes: `(encounter_id, recognized_at DESC)`, `(compliance_status)`, `(triggering_alert_id)`
|
||||
Indexes: `(encounter_id, recognized_at DESC)`, `(compliance_status)`, `(triggering_alert_id)`, partial unique `(encounter_id) WHERE compliance_status = 'IN_PROGRESS'` (prevents TOCTOU race on concurrent bundle creation)
|
||||
|
||||
Check constraint: `compliance_status IN ('IN_PROGRESS', 'COMPLIANT', 'NON_COMPLIANT')`
|
||||
|
||||
@@ -1412,9 +1434,12 @@ partitionKey string? — encounterId for per-encounter ordering
|
||||
payload JSONB required
|
||||
createdAt DateTimeOffset
|
||||
processedAt DateTimeOffset?
|
||||
retryCount int default 0 — Kafka produce attempt counter
|
||||
lastError string? — last Kafka produce failure reason
|
||||
failedAt DateTimeOffset? — set when retryCount exceeds OutboxMaxRetries (permanently failed)
|
||||
```
|
||||
|
||||
Partial index: `(created_at) WHERE processed_at IS NULL`
|
||||
Partial index: `(created_at) WHERE processed_at IS NULL AND failed_at IS NULL`
|
||||
|
||||
### ExternalResourceIdentifier
|
||||
|
||||
@@ -1623,7 +1648,7 @@ When a `SOFA_SEPSIS` alert fires (delta ≥ 2 from baseline), `SepsisAlertHandle
|
||||
7. When all four elements are complete: `complianceStatus = COMPLIANT` (within deadline) or `NON_COMPLIANT` (past deadline); Prometheus `sepsis_bundle_compliance_total{status}` incremented
|
||||
8. If the deadline passes with incomplete elements, `SepsisBundleMonitorService` (polling every 5 min) marks the bundle `NON_COMPLIANT` — elements remain `PENDING` but the bundle status reflects the missed deadline
|
||||
|
||||
**Idempotency:** Only one in-progress bundle can exist per encounter. A second alert for the same encounter returns early without creating a duplicate.
|
||||
**Idempotency:** Only one in-progress bundle can exist per encounter, enforced by a partial unique index `(encounter_id) WHERE compliance_status = 'IN_PROGRESS'`. A second alert for the same encounter returns early without creating a duplicate — the database constraint prevents TOCTOU race conditions even under concurrent SOFA scoring events.
|
||||
|
||||
---
|
||||
|
||||
|
||||
@@ -0,0 +1,534 @@
|
||||
using System.Net;
|
||||
using System.Net.Http.Json;
|
||||
using System.Text.Json;
|
||||
using FluentAssertions;
|
||||
using Microsoft.EntityFrameworkCore;
|
||||
using Microsoft.Extensions.DependencyInjection;
|
||||
using StackExchange.Redis;
|
||||
|
||||
/// <summary>
|
||||
/// Tests for gap-analysis fixes: MRN sequence, sepsis bundle idempotency,
|
||||
/// trend alert exact match, order→bundle transaction, FHIR bundle rollback,
|
||||
/// patient update endpoint, new list endpoints, and new validators.
|
||||
/// </summary>
|
||||
[Collection("Integration")]
|
||||
public class GapAnalysisFixTests : IAsyncLifetime
|
||||
{
|
||||
private readonly ApiFixture _fixture;
|
||||
private readonly HttpClient _client;
|
||||
|
||||
public GapAnalysisFixTests(ApiFixture fixture)
|
||||
{
|
||||
_fixture = fixture;
|
||||
_client = fixture.CreateClient();
|
||||
}
|
||||
|
||||
public async Task InitializeAsync()
|
||||
{
|
||||
using var scope = _fixture.Services.CreateScope();
|
||||
var db = scope.ServiceProvider.GetRequiredService<AppDbContext>();
|
||||
await DbResetHelper.ResetAsync(db);
|
||||
|
||||
var redis = scope.ServiceProvider.GetRequiredService<IConnectionMultiplexer>();
|
||||
await DataSeeder.SeedThresholdsOnlyAsync(db, redis);
|
||||
}
|
||||
|
||||
public Task DisposeAsync() => Task.CompletedTask;
|
||||
|
||||
// -------------------------------------------------------------------------
|
||||
// P0 — MRN generation uses sequence (no duplicates on concurrent registration)
|
||||
// -------------------------------------------------------------------------
|
||||
|
||||
[Fact]
|
||||
public async Task MrnGeneration_SequentialRegistrations_ProduceUniqueMrns()
|
||||
{
|
||||
var resp1 = await _client.PostAsJsonAsync("/api/v1/patients",
|
||||
new RegisterPatientRequest("Alice", "One", new DateOnly(1990, 1, 1), "F"));
|
||||
var resp2 = await _client.PostAsJsonAsync("/api/v1/patients",
|
||||
new RegisterPatientRequest("Bob", "Two", new DateOnly(1991, 2, 2), "M"));
|
||||
|
||||
resp1.StatusCode.Should().Be(HttpStatusCode.Created);
|
||||
resp2.StatusCode.Should().Be(HttpStatusCode.Created);
|
||||
|
||||
var p1 = (await resp1.Content.ReadFromJsonAsync<JsonElement>())
|
||||
.GetProperty("data").GetProperty("mrn").GetString();
|
||||
var p2 = (await resp2.Content.ReadFromJsonAsync<JsonElement>())
|
||||
.GetProperty("data").GetProperty("mrn").GetString();
|
||||
|
||||
p1.Should().StartWith("MRN-");
|
||||
p2.Should().StartWith("MRN-");
|
||||
p1.Should().NotBe(p2);
|
||||
}
|
||||
|
||||
[Fact]
|
||||
public async Task MrnGeneration_FormatMatchesConfiguredPattern()
|
||||
{
|
||||
var resp = await _client.PostAsJsonAsync("/api/v1/patients",
|
||||
new RegisterPatientRequest("Format", "Test", new DateOnly(1985, 5, 5), "F"));
|
||||
resp.StatusCode.Should().Be(HttpStatusCode.Created);
|
||||
|
||||
var mrn = (await resp.Content.ReadFromJsonAsync<JsonElement>())
|
||||
.GetProperty("data").GetProperty("mrn").GetString()!;
|
||||
|
||||
mrn.Should().MatchRegex(@"^MRN-\d{6}$");
|
||||
}
|
||||
|
||||
// -------------------------------------------------------------------------
|
||||
// P0 — Trend alert uses exact observation_code match
|
||||
// -------------------------------------------------------------------------
|
||||
|
||||
[Fact]
|
||||
public async Task TrendAlert_StoresObservationCode()
|
||||
{
|
||||
using var scope = _fixture.Services.CreateScope();
|
||||
var db = scope.ServiceProvider.GetRequiredService<AppDbContext>();
|
||||
var detector = scope.ServiceProvider.GetRequiredService<TrendDetector>();
|
||||
|
||||
var patient = new Patient
|
||||
{
|
||||
Id = Guid.NewGuid(), Mrn = "MRN-TREND-OC", FirstName = "OC", LastName = "Test",
|
||||
DateOfBirth = new DateOnly(1970, 1, 1), Gender = "M",
|
||||
CreatedAt = DateTimeOffset.UtcNow
|
||||
};
|
||||
var encounter = new Encounter
|
||||
{
|
||||
Id = Guid.NewGuid(), PatientId = patient.Id, EncounterType = EncounterType.Inpatient,
|
||||
Status = EncounterStatus.Active, Department = Department.Icu,
|
||||
AttendingPhysician = "Dr. OC", AdmittedAt = DateTimeOffset.UtcNow,
|
||||
CreatedAt = DateTimeOffset.UtcNow
|
||||
};
|
||||
db.Patients.Add(patient);
|
||||
db.Encounters.Add(encounter);
|
||||
await db.SaveChangesAsync();
|
||||
|
||||
var redis = scope.ServiceProvider.GetRequiredService<IConnectionMultiplexer>();
|
||||
foreach (var key in TrendCalculator.AllHistoryKeys(encounter.Id))
|
||||
await redis.GetDatabase().KeyDeleteAsync(key);
|
||||
|
||||
var baseTime = new DateTimeOffset(2026, 6, 20, 10, 0, 0, TimeSpan.Zero);
|
||||
await detector.ProcessObservationAsync(
|
||||
encounter.Id, patient.Id, "HEART_RATE", 72m, baseTime);
|
||||
var result = await detector.ProcessObservationAsync(
|
||||
encounter.Id, patient.Id, "HEART_RATE", 95m, baseTime.AddMinutes(10));
|
||||
|
||||
result.Outcome.Should().Be(TrendOutcome.RapidDeterioration);
|
||||
|
||||
var alert = await db.ClinicalAlerts.SingleAsync(a =>
|
||||
a.EncounterId == encounter.Id && a.AlertType == AlertType.RapidDeterioration);
|
||||
alert.ObservationCode.Should().Be("HEART_RATE");
|
||||
}
|
||||
|
||||
// -------------------------------------------------------------------------
|
||||
// P1 — Order→Bundle spanning transaction
|
||||
// -------------------------------------------------------------------------
|
||||
|
||||
[Fact]
|
||||
public async Task OrderResult_UpdatesBundleElement_InSameTransaction()
|
||||
{
|
||||
using var scope = _fixture.Services.CreateScope();
|
||||
var db = scope.ServiceProvider.GetRequiredService<AppDbContext>();
|
||||
var handler = scope.ServiceProvider.GetRequiredService<SepsisAlertHandler>();
|
||||
var bundleService = scope.ServiceProvider.GetRequiredService<ISepsisBundleService>();
|
||||
|
||||
var patient = new Patient
|
||||
{
|
||||
Id = Guid.NewGuid(), Mrn = "MRN-TX-001", FirstName = "Tx", LastName = "Test",
|
||||
DateOfBirth = new DateOnly(1960, 1, 1), Gender = "M",
|
||||
CreatedAt = DateTimeOffset.UtcNow
|
||||
};
|
||||
var encounter = new Encounter
|
||||
{
|
||||
Id = Guid.NewGuid(), PatientId = patient.Id, EncounterType = EncounterType.Inpatient,
|
||||
Status = EncounterStatus.Active, Department = Department.Icu,
|
||||
AttendingPhysician = "Dr. Tx", AdmittedAt = DateTimeOffset.UtcNow,
|
||||
CreatedAt = DateTimeOffset.UtcNow
|
||||
};
|
||||
db.Patients.Add(patient);
|
||||
db.Encounters.Add(encounter);
|
||||
|
||||
var alertId = Guid.NewGuid();
|
||||
db.ClinicalAlerts.Add(new ClinicalAlert
|
||||
{
|
||||
Id = alertId, EncounterId = encounter.Id, PatientId = patient.Id,
|
||||
AlertType = AlertType.SofaSepsis, Severity = AlertSeverity.Critical,
|
||||
Details = "SOFA delta +2", Status = AlertStatus.Open,
|
||||
TriggeredAt = DateTimeOffset.UtcNow
|
||||
});
|
||||
await db.SaveChangesAsync();
|
||||
|
||||
await handler.OnSepsisAlertCreatedAsync(
|
||||
encounter.Id, alertId, AlertType.SofaSepsis, CancellationToken.None);
|
||||
|
||||
var element = await db.SepsisBundleElements
|
||||
.Include(e => e.Order)
|
||||
.FirstAsync(e => e.ElementType == SepsisBundleElementType.SerumLactate);
|
||||
|
||||
var orderService = scope.ServiceProvider.GetRequiredService<IOrderService>();
|
||||
var order = element.Order!;
|
||||
order.Status = OrderStatus.InProgress;
|
||||
await db.SaveChangesAsync();
|
||||
|
||||
await orderService.RecordResultAsync(order.Id,
|
||||
new RecordOrderResultRequest("Lactate 1.5 mmol/L"));
|
||||
|
||||
var updatedOrder = await db.Orders.AsNoTracking().FirstAsync(o => o.Id == order.Id);
|
||||
var updatedElement = await db.SepsisBundleElements.AsNoTracking()
|
||||
.FirstAsync(e => e.Id == element.Id);
|
||||
|
||||
updatedOrder.Status.Should().Be(OrderStatus.Resulted);
|
||||
updatedElement.Status.Should().Be(SepsisBundleElementStatus.Completed);
|
||||
}
|
||||
|
||||
// -------------------------------------------------------------------------
|
||||
// P4 — Patient update endpoint (PATCH)
|
||||
// -------------------------------------------------------------------------
|
||||
|
||||
[Fact]
|
||||
public async Task PatientUpdate_PartialFields_UpdatesOnlyProvided()
|
||||
{
|
||||
var createResp = await _client.PostAsJsonAsync("/api/v1/patients",
|
||||
new RegisterPatientRequest("Original", "Name", new DateOnly(1990, 1, 1), "F"));
|
||||
createResp.StatusCode.Should().Be(HttpStatusCode.Created);
|
||||
var patientId = (await createResp.Content.ReadFromJsonAsync<JsonElement>())
|
||||
.GetProperty("data").GetProperty("id").GetGuid();
|
||||
|
||||
var patchResp = await _client.PatchAsJsonAsync($"/api/v1/patients/{patientId}",
|
||||
new { lastName = "Updated", allergies = "Penicillin" });
|
||||
patchResp.StatusCode.Should().Be(HttpStatusCode.OK);
|
||||
|
||||
var getResp = await _client.GetFromJsonAsync<JsonElement>($"/api/v1/patients/{patientId}");
|
||||
var data = getResp.GetProperty("data");
|
||||
data.GetProperty("firstName").GetString().Should().Be("Original");
|
||||
data.GetProperty("lastName").GetString().Should().Be("Updated");
|
||||
data.GetProperty("allergies").GetString().Should().Be("Penicillin");
|
||||
}
|
||||
|
||||
[Fact]
|
||||
public async Task PatientUpdate_NonExistent_Returns404()
|
||||
{
|
||||
var resp = await _client.PatchAsJsonAsync($"/api/v1/patients/{Guid.NewGuid()}",
|
||||
new { firstName = "Ghost" });
|
||||
resp.StatusCode.Should().Be(HttpStatusCode.NotFound);
|
||||
}
|
||||
|
||||
[Fact]
|
||||
public async Task PatientUpdate_CreatesAuditLog()
|
||||
{
|
||||
var createResp = await _client.PostAsJsonAsync("/api/v1/patients",
|
||||
new RegisterPatientRequest("Audit", "Before", new DateOnly(1985, 3, 3), "M"));
|
||||
var patientId = (await createResp.Content.ReadFromJsonAsync<JsonElement>())
|
||||
.GetProperty("data").GetProperty("id").GetGuid();
|
||||
|
||||
await _client.PatchAsJsonAsync($"/api/v1/patients/{patientId}",
|
||||
new { lastName = "After" });
|
||||
|
||||
var auditResp = await _client.GetFromJsonAsync<JsonElement>(
|
||||
$"/api/v1/audit-logs?entityType=Patient&entityId={patientId}");
|
||||
var logs = auditResp.GetProperty("data").GetProperty("items");
|
||||
logs.EnumerateArray().Should().Contain(log =>
|
||||
log.GetProperty("action").GetString() == "PATIENT_UPDATED");
|
||||
}
|
||||
|
||||
// -------------------------------------------------------------------------
|
||||
// P4 — Sepsis bundle list endpoint
|
||||
// -------------------------------------------------------------------------
|
||||
|
||||
[Fact]
|
||||
public async Task SepsisBundleList_ReturnsPagedResults()
|
||||
{
|
||||
using var scope = _fixture.Services.CreateScope();
|
||||
var db = scope.ServiceProvider.GetRequiredService<AppDbContext>();
|
||||
var handler = scope.ServiceProvider.GetRequiredService<SepsisAlertHandler>();
|
||||
|
||||
var patient = new Patient
|
||||
{
|
||||
Id = Guid.NewGuid(), Mrn = "MRN-LIST-001", FirstName = "List", LastName = "Test",
|
||||
DateOfBirth = new DateOnly(1970, 1, 1), Gender = "F",
|
||||
CreatedAt = DateTimeOffset.UtcNow
|
||||
};
|
||||
var encounter = new Encounter
|
||||
{
|
||||
Id = Guid.NewGuid(), PatientId = patient.Id, EncounterType = EncounterType.Inpatient,
|
||||
Status = EncounterStatus.Active, Department = Department.Icu,
|
||||
AttendingPhysician = "Dr. List", AdmittedAt = DateTimeOffset.UtcNow,
|
||||
CreatedAt = DateTimeOffset.UtcNow
|
||||
};
|
||||
db.Patients.Add(patient);
|
||||
db.Encounters.Add(encounter);
|
||||
|
||||
var alertId = Guid.NewGuid();
|
||||
db.ClinicalAlerts.Add(new ClinicalAlert
|
||||
{
|
||||
Id = alertId, EncounterId = encounter.Id, PatientId = patient.Id,
|
||||
AlertType = AlertType.SofaSepsis, Severity = AlertSeverity.Critical,
|
||||
Details = "SOFA delta +2", Status = AlertStatus.Open,
|
||||
TriggeredAt = DateTimeOffset.UtcNow
|
||||
});
|
||||
await db.SaveChangesAsync();
|
||||
|
||||
await handler.OnSepsisAlertCreatedAsync(
|
||||
encounter.Id, alertId, AlertType.SofaSepsis, CancellationToken.None);
|
||||
|
||||
var resp = await _client.GetFromJsonAsync<JsonElement>(
|
||||
"/api/v1/sepsis-bundles?status=IN_PROGRESS&page=1&pageSize=10");
|
||||
|
||||
var data = resp.GetProperty("data");
|
||||
data.GetProperty("totalCount").GetInt32().Should().BeGreaterThanOrEqualTo(1);
|
||||
data.GetProperty("items").EnumerateArray().Should().Contain(b =>
|
||||
b.GetProperty("complianceStatus").GetString() == "IN_PROGRESS");
|
||||
}
|
||||
|
||||
// -------------------------------------------------------------------------
|
||||
// P4 — qSOFA history endpoint
|
||||
// -------------------------------------------------------------------------
|
||||
|
||||
[Fact]
|
||||
public async Task QsofaHistory_ReturnsAlertRecords()
|
||||
{
|
||||
using var scope = _fixture.Services.CreateScope();
|
||||
var db = scope.ServiceProvider.GetRequiredService<AppDbContext>();
|
||||
|
||||
var patient = new Patient
|
||||
{
|
||||
Id = Guid.NewGuid(), Mrn = "MRN-QH-001", FirstName = "QH", LastName = "Test",
|
||||
DateOfBirth = new DateOnly(1975, 1, 1), Gender = "F",
|
||||
CreatedAt = DateTimeOffset.UtcNow
|
||||
};
|
||||
var encounter = new Encounter
|
||||
{
|
||||
Id = Guid.NewGuid(), PatientId = patient.Id, EncounterType = EncounterType.Inpatient,
|
||||
Status = EncounterStatus.Active, Department = Department.Icu,
|
||||
AttendingPhysician = "Dr. QH", AdmittedAt = DateTimeOffset.UtcNow,
|
||||
CreatedAt = DateTimeOffset.UtcNow
|
||||
};
|
||||
db.Patients.Add(patient);
|
||||
db.Encounters.Add(encounter);
|
||||
|
||||
db.ClinicalAlerts.Add(new ClinicalAlert
|
||||
{
|
||||
Id = Guid.NewGuid(), EncounterId = encounter.Id, PatientId = patient.Id,
|
||||
AlertType = AlertType.QsofaScreen, Severity = AlertSeverity.Warning,
|
||||
Details = "qSOFA >= 2", Status = AlertStatus.Open,
|
||||
TriggeredAt = DateTimeOffset.UtcNow.AddMinutes(-30)
|
||||
});
|
||||
db.ClinicalAlerts.Add(new ClinicalAlert
|
||||
{
|
||||
Id = Guid.NewGuid(), EncounterId = encounter.Id, PatientId = patient.Id,
|
||||
AlertType = AlertType.QsofaScreen, Severity = AlertSeverity.Warning,
|
||||
Details = "qSOFA >= 2", Status = AlertStatus.Resolved,
|
||||
TriggeredAt = DateTimeOffset.UtcNow.AddMinutes(-10),
|
||||
ResolvedAt = DateTimeOffset.UtcNow
|
||||
});
|
||||
await db.SaveChangesAsync();
|
||||
|
||||
var resp = await _client.GetFromJsonAsync<JsonElement>(
|
||||
$"/api/v1/encounters/{encounter.Id}/qsofa/history?limit=10");
|
||||
|
||||
var data = resp.GetProperty("data");
|
||||
data.GetProperty("items").GetArrayLength().Should().Be(2);
|
||||
}
|
||||
|
||||
// -------------------------------------------------------------------------
|
||||
// P4 — Reconciliation alerts endpoint
|
||||
// -------------------------------------------------------------------------
|
||||
|
||||
[Fact]
|
||||
public async Task ReconciliationAlerts_ListWithFilter()
|
||||
{
|
||||
using var scope = _fixture.Services.CreateScope();
|
||||
var db = scope.ServiceProvider.GetRequiredService<AppDbContext>();
|
||||
|
||||
var patient = new Patient
|
||||
{
|
||||
Id = Guid.NewGuid(), Mrn = "MRN-RECON-001", FirstName = "Recon", LastName = "Test",
|
||||
DateOfBirth = new DateOnly(1980, 1, 1), Gender = "F",
|
||||
CreatedAt = DateTimeOffset.UtcNow
|
||||
};
|
||||
var encounter = new Encounter
|
||||
{
|
||||
Id = Guid.NewGuid(), PatientId = patient.Id, EncounterType = EncounterType.Inpatient,
|
||||
Status = EncounterStatus.Active, Department = Department.Icu,
|
||||
AttendingPhysician = "Dr. Recon", AdmittedAt = DateTimeOffset.UtcNow,
|
||||
CreatedAt = DateTimeOffset.UtcNow
|
||||
};
|
||||
db.Patients.Add(patient);
|
||||
db.Encounters.Add(encounter);
|
||||
db.ReconciliationAlerts.Add(new ReconciliationAlert
|
||||
{
|
||||
Id = Guid.NewGuid(),
|
||||
CheckType = ReconciliationCheckType.UnacknowledgedCriticalAlert,
|
||||
EncounterId = encounter.Id,
|
||||
PatientId = patient.Id,
|
||||
Details = "Test reconciliation alert",
|
||||
CreatedAt = DateTimeOffset.UtcNow
|
||||
});
|
||||
await db.SaveChangesAsync();
|
||||
|
||||
var resp = await _client.GetFromJsonAsync<JsonElement>(
|
||||
"/api/v1/reconciliation-alerts?resolved=false&page=1&pageSize=10");
|
||||
|
||||
var data = resp.GetProperty("data");
|
||||
data.GetProperty("totalCount").GetInt32().Should().BeGreaterThanOrEqualTo(1);
|
||||
data.GetProperty("items").EnumerateArray().Should().Contain(a =>
|
||||
a.GetProperty("details").GetString()!.Contains("Test reconciliation alert"));
|
||||
}
|
||||
|
||||
[Fact]
|
||||
public async Task ReconciliationAlerts_FilterByCheckType()
|
||||
{
|
||||
using var scope = _fixture.Services.CreateScope();
|
||||
var db = scope.ServiceProvider.GetRequiredService<AppDbContext>();
|
||||
|
||||
var patient = new Patient
|
||||
{
|
||||
Id = Guid.NewGuid(), Mrn = "MRN-RECON-002", FirstName = "Recon2", LastName = "Test",
|
||||
DateOfBirth = new DateOnly(1980, 1, 1), Gender = "M",
|
||||
CreatedAt = DateTimeOffset.UtcNow
|
||||
};
|
||||
var encounter = new Encounter
|
||||
{
|
||||
Id = Guid.NewGuid(), PatientId = patient.Id, EncounterType = EncounterType.Inpatient,
|
||||
Status = EncounterStatus.Active, Department = Department.Icu,
|
||||
AttendingPhysician = "Dr. Recon2", AdmittedAt = DateTimeOffset.UtcNow,
|
||||
CreatedAt = DateTimeOffset.UtcNow
|
||||
};
|
||||
db.Patients.Add(patient);
|
||||
db.Encounters.Add(encounter);
|
||||
db.ReconciliationAlerts.Add(new ReconciliationAlert
|
||||
{
|
||||
Id = Guid.NewGuid(),
|
||||
CheckType = ReconciliationCheckType.PendingOrderNoResult,
|
||||
EncounterId = encounter.Id,
|
||||
Details = "Pending order check",
|
||||
CreatedAt = DateTimeOffset.UtcNow
|
||||
});
|
||||
await db.SaveChangesAsync();
|
||||
|
||||
var resp = await _client.GetFromJsonAsync<JsonElement>(
|
||||
"/api/v1/reconciliation-alerts?checkType=PENDING_ORDER_NO_RESULT");
|
||||
var items = resp.GetProperty("data").GetProperty("items");
|
||||
items.GetArrayLength().Should().BeGreaterThanOrEqualTo(1);
|
||||
}
|
||||
|
||||
// -------------------------------------------------------------------------
|
||||
// P2 — New validators return 400 for invalid input
|
||||
// -------------------------------------------------------------------------
|
||||
|
||||
[Fact]
|
||||
public async Task TransitionStatusValidator_LongDiagnosis_Returns400()
|
||||
{
|
||||
var patientId = await CreatePatientAsync();
|
||||
var encounterId = await CreateEncounterAsync(patientId);
|
||||
|
||||
var resp = await _client.PatchAsJsonAsync(
|
||||
$"/api/v1/encounters/{encounterId}/status",
|
||||
new { status = "Discharged", dischargeDiagnosis = new string('x', 501) });
|
||||
|
||||
resp.StatusCode.Should().Be(HttpStatusCode.BadRequest);
|
||||
}
|
||||
|
||||
[Fact]
|
||||
public async Task UpdatePatientValidator_LongFirstName_Returns400()
|
||||
{
|
||||
var patientId = await CreatePatientAsync();
|
||||
|
||||
var resp = await _client.PatchAsJsonAsync(
|
||||
$"/api/v1/patients/{patientId}",
|
||||
new { firstName = new string('x', 101) });
|
||||
|
||||
resp.StatusCode.Should().Be(HttpStatusCode.BadRequest);
|
||||
}
|
||||
|
||||
// -------------------------------------------------------------------------
|
||||
// P2 — Outbox retry columns exist
|
||||
// -------------------------------------------------------------------------
|
||||
|
||||
[Fact]
|
||||
public async Task OutboxEvent_HasRetryColumns()
|
||||
{
|
||||
using var scope = _fixture.Services.CreateScope();
|
||||
var db = scope.ServiceProvider.GetRequiredService<AppDbContext>();
|
||||
|
||||
var ev = new OutboxEvent
|
||||
{
|
||||
Id = Guid.NewGuid(),
|
||||
Topic = "test.retry",
|
||||
Payload = "{}",
|
||||
PartitionKey = "test",
|
||||
CreatedAt = DateTimeOffset.UtcNow,
|
||||
RetryCount = 3,
|
||||
LastError = "Connection refused",
|
||||
FailedAt = DateTimeOffset.UtcNow
|
||||
};
|
||||
db.OutboxEvents.Add(ev);
|
||||
await db.SaveChangesAsync();
|
||||
|
||||
var loaded = await db.OutboxEvents.AsNoTracking().FirstAsync(e => e.Id == ev.Id);
|
||||
loaded.RetryCount.Should().Be(3);
|
||||
loaded.LastError.Should().Be("Connection refused");
|
||||
loaded.FailedAt.Should().NotBeNull();
|
||||
}
|
||||
|
||||
// -------------------------------------------------------------------------
|
||||
// P0 — ClinicalAlert.ObservationCode column persists
|
||||
// -------------------------------------------------------------------------
|
||||
|
||||
[Fact]
|
||||
public async Task ClinicalAlert_ObservationCode_Persists()
|
||||
{
|
||||
using var scope = _fixture.Services.CreateScope();
|
||||
var db = scope.ServiceProvider.GetRequiredService<AppDbContext>();
|
||||
|
||||
var patient = new Patient
|
||||
{
|
||||
Id = Guid.NewGuid(), Mrn = "MRN-OC-001", FirstName = "OC", LastName = "Persist",
|
||||
DateOfBirth = new DateOnly(1980, 1, 1), Gender = "F",
|
||||
CreatedAt = DateTimeOffset.UtcNow
|
||||
};
|
||||
var encounter = new Encounter
|
||||
{
|
||||
Id = Guid.NewGuid(), PatientId = patient.Id, EncounterType = EncounterType.Inpatient,
|
||||
Status = EncounterStatus.Active, Department = Department.Icu,
|
||||
AttendingPhysician = "Dr. OC", AdmittedAt = DateTimeOffset.UtcNow,
|
||||
CreatedAt = DateTimeOffset.UtcNow
|
||||
};
|
||||
db.Patients.Add(patient);
|
||||
db.Encounters.Add(encounter);
|
||||
|
||||
var alert = new ClinicalAlert
|
||||
{
|
||||
Id = Guid.NewGuid(), EncounterId = encounter.Id, PatientId = patient.Id,
|
||||
AlertType = AlertType.RapidDeterioration, Severity = AlertSeverity.Warning,
|
||||
Details = "Test", ObservationCode = "SPO2", Status = AlertStatus.Open,
|
||||
TriggeredAt = DateTimeOffset.UtcNow
|
||||
};
|
||||
db.ClinicalAlerts.Add(alert);
|
||||
await db.SaveChangesAsync();
|
||||
|
||||
var loaded = await db.ClinicalAlerts.AsNoTracking().FirstAsync(a => a.Id == alert.Id);
|
||||
loaded.ObservationCode.Should().Be("SPO2");
|
||||
}
|
||||
|
||||
// -------------------------------------------------------------------------
|
||||
// Helpers
|
||||
// -------------------------------------------------------------------------
|
||||
|
||||
private async Task<Guid> CreatePatientAsync()
|
||||
{
|
||||
var resp = await _client.PostAsJsonAsync("/api/v1/patients",
|
||||
new RegisterPatientRequest("Gap", "Test", new DateOnly(1990, 1, 1), "F"));
|
||||
resp.EnsureSuccessStatusCode();
|
||||
var body = await resp.Content.ReadFromJsonAsync<JsonElement>();
|
||||
return body.GetProperty("data").GetProperty("id").GetGuid();
|
||||
}
|
||||
|
||||
private async Task<Guid> CreateEncounterAsync(Guid patientId)
|
||||
{
|
||||
var resp = await _client.PostAsJsonAsync(
|
||||
$"/api/v1/patients/{patientId}/encounters",
|
||||
new OpenEncounterRequest(EncounterType.Inpatient, Department.Icu, "Dr. Gap"));
|
||||
resp.EnsureSuccessStatusCode();
|
||||
var body = await resp.Content.ReadFromJsonAsync<JsonElement>();
|
||||
return body.GetProperty("data").GetProperty("id").GetGuid();
|
||||
}
|
||||
}
|
||||
@@ -100,7 +100,7 @@ public class AnalyticsController : ControllerBase
|
||||
[FromQuery] string? q,
|
||||
[FromQuery] string? department,
|
||||
[FromQuery] string? status,
|
||||
[FromQuery] int page = 0,
|
||||
[FromQuery] int page = 1,
|
||||
[FromQuery] int pageSize = 20)
|
||||
{
|
||||
var result = await _analytics.SearchPatientsAsync(q, department, status, page, pageSize);
|
||||
|
||||
@@ -77,7 +77,7 @@ public class ObservationsController : ControllerBase
|
||||
[FromQuery] string? code,
|
||||
[FromQuery] DateTimeOffset? from,
|
||||
[FromQuery] DateTimeOffset? to,
|
||||
[FromQuery] int limit = 50,
|
||||
[FromQuery] int limit = 20,
|
||||
[FromQuery] string? cursor = null)
|
||||
{
|
||||
var page = await _query.GetHistoryAsync(encounterId, code, from, to, limit, cursor);
|
||||
|
||||
@@ -67,6 +67,22 @@ public class PatientsController : ControllerBase
|
||||
return Ok(ApiResponse<Patient>.Ok(patient));
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// Updates patient demographics. Only non-null fields are applied.
|
||||
/// </summary>
|
||||
/// <param name="id">Patient id.</param>
|
||||
/// <param name="req">Fields to update.</param>
|
||||
/// <returns>The updated patient record.</returns>
|
||||
[HttpPatch("{id:guid}")]
|
||||
[AuthorizePermission(ClinicalPermissions.PatientsWrite)]
|
||||
[ProducesResponseType(typeof(ApiResponse<Patient>), StatusCodes.Status200OK)]
|
||||
[ProducesResponseType(typeof(ApiResponse<object>), StatusCodes.Status404NotFound)]
|
||||
public async Task<IActionResult> Update(Guid id, [FromBody] UpdatePatientRequest req)
|
||||
{
|
||||
var patient = await _patients.UpdateAsync(id, req);
|
||||
return Ok(ApiResponse<Patient>.Ok(patient));
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// Opens a new active encounter for the patient.
|
||||
/// </summary>
|
||||
|
||||
@@ -25,4 +25,23 @@ public class QsofaController : ControllerBase
|
||||
var score = await _qsofa.GetCurrentAsync(encounterId);
|
||||
return Ok(ApiResponse<QsofaCurrentResponse>.Ok(score));
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// Returns cursor-paginated qSOFA alert history for an encounter.
|
||||
/// </summary>
|
||||
[HttpGet("history")]
|
||||
[ProducesResponseType(typeof(ApiResponse<object>), StatusCodes.Status200OK)]
|
||||
public async Task<IActionResult> History(
|
||||
Guid encounterId,
|
||||
[FromQuery] int limit = 20,
|
||||
[FromQuery] string? cursor = null)
|
||||
{
|
||||
var page = await _qsofa.GetHistoryAsync(encounterId, limit, cursor);
|
||||
return Ok(ApiResponse<object>.Ok(new
|
||||
{
|
||||
items = page.Items,
|
||||
nextCursor = page.NextCursor,
|
||||
hasMore = page.HasMore
|
||||
}));
|
||||
}
|
||||
}
|
||||
|
||||
@@ -0,0 +1,62 @@
|
||||
using Microsoft.AspNetCore.Mvc;
|
||||
using Microsoft.EntityFrameworkCore;
|
||||
|
||||
/// <summary>
|
||||
/// Data quality reconciliation alerts: unacknowledged critical alerts, pending orders without results,
|
||||
/// and active inpatients without recent observations.
|
||||
/// </summary>
|
||||
[ApiController]
|
||||
[Route("api/v1/reconciliation-alerts")]
|
||||
[Produces("application/json")]
|
||||
[AuthorizePermission(ClinicalPermissions.AlertsRead)]
|
||||
public class ReconciliationAlertsController : ControllerBase
|
||||
{
|
||||
private readonly AppDbContext _db;
|
||||
|
||||
public ReconciliationAlertsController(AppDbContext db) => _db = db;
|
||||
|
||||
/// <summary>
|
||||
/// Lists reconciliation alerts with optional filters.
|
||||
/// </summary>
|
||||
[HttpGet]
|
||||
[ProducesResponseType(typeof(ApiResponse<object>), StatusCodes.Status200OK)]
|
||||
public async Task<IActionResult> List(
|
||||
[FromQuery] string? checkType,
|
||||
[FromQuery] bool? resolved,
|
||||
[FromQuery] int page = 1,
|
||||
[FromQuery] int pageSize = 20)
|
||||
{
|
||||
pageSize = Math.Clamp(pageSize, 1, 100);
|
||||
|
||||
var query = _db.ReconciliationAlerts
|
||||
.AsNoTracking()
|
||||
.AsQueryable();
|
||||
|
||||
if (checkType is not null)
|
||||
{
|
||||
var parsed = ReconciliationCheckTypeExtensions.FromDbString(checkType);
|
||||
query = query.Where(a => a.CheckType == parsed);
|
||||
}
|
||||
|
||||
if (resolved == true)
|
||||
query = query.Where(a => a.ResolvedAt != null);
|
||||
else if (resolved == false)
|
||||
query = query.Where(a => a.ResolvedAt == null);
|
||||
|
||||
var total = await query.CountAsync();
|
||||
var items = await query
|
||||
.OrderByDescending(a => a.CreatedAt)
|
||||
.Skip((page - 1) * pageSize)
|
||||
.Take(pageSize)
|
||||
.ToListAsync();
|
||||
|
||||
return Ok(ApiResponse<object>.Ok(new
|
||||
{
|
||||
items,
|
||||
page,
|
||||
pageSize,
|
||||
totalCount = total,
|
||||
totalPages = (int)Math.Ceiling((double)total / pageSize)
|
||||
}));
|
||||
}
|
||||
}
|
||||
@@ -13,6 +13,31 @@ public class SepsisBundlesController : ControllerBase
|
||||
|
||||
public SepsisBundlesController(ISepsisBundleService bundles) => _bundles = bundles;
|
||||
|
||||
/// <summary>
|
||||
/// Lists sepsis bundles across all encounters with optional status filter.
|
||||
/// </summary>
|
||||
[HttpGet("api/v1/sepsis-bundles")]
|
||||
[ProducesResponseType(typeof(ApiResponse<object>), StatusCodes.Status200OK)]
|
||||
public async Task<IActionResult> List(
|
||||
[FromQuery] string? status,
|
||||
[FromQuery] int page = 1,
|
||||
[FromQuery] int pageSize = 20)
|
||||
{
|
||||
SepsisBundleComplianceStatus? parsed = status is not null
|
||||
? SepsisBundleComplianceStatusExtensions.FromDbString(status)
|
||||
: null;
|
||||
|
||||
var result = await _bundles.ListAsync(parsed, page, pageSize);
|
||||
return Ok(ApiResponse<object>.Ok(new
|
||||
{
|
||||
items = result.Items,
|
||||
page = result.Page,
|
||||
pageSize = result.PageSize,
|
||||
totalCount = result.TotalCount,
|
||||
totalPages = result.TotalPages
|
||||
}));
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// Returns the current (most recent) sepsis bundle for an encounter with elements and linked orders.
|
||||
/// </summary>
|
||||
|
||||
@@ -6,6 +6,7 @@ public enum AuditAction
|
||||
AlertResolved,
|
||||
EncounterStatusChanged,
|
||||
PatientRegistered,
|
||||
PatientUpdated,
|
||||
SuppressionWindowSet,
|
||||
UserLogin
|
||||
}
|
||||
@@ -20,6 +21,7 @@ public static class AuditActionExtensions
|
||||
AuditAction.AlertResolved => "ALERT_RESOLVED",
|
||||
AuditAction.EncounterStatusChanged => "ENCOUNTER_STATUS_CHANGED",
|
||||
AuditAction.PatientRegistered => "PATIENT_REGISTERED",
|
||||
AuditAction.PatientUpdated => "PATIENT_UPDATED",
|
||||
AuditAction.SuppressionWindowSet => "SUPPRESSION_WINDOW_SET",
|
||||
AuditAction.UserLogin => "USER_LOGIN",
|
||||
_ => throw new ArgumentOutOfRangeException(nameof(a))
|
||||
@@ -33,6 +35,7 @@ public static class AuditActionExtensions
|
||||
"ALERT_RESOLVED" => AuditAction.AlertResolved,
|
||||
"ENCOUNTER_STATUS_CHANGED" => AuditAction.EncounterStatusChanged,
|
||||
"PATIENT_REGISTERED" => AuditAction.PatientRegistered,
|
||||
"PATIENT_UPDATED" => AuditAction.PatientUpdated,
|
||||
"SUPPRESSION_WINDOW_SET" => AuditAction.SuppressionWindowSet,
|
||||
"USER_LOGIN" => AuditAction.UserLogin,
|
||||
_ => throw new ArgumentOutOfRangeException(nameof(v))
|
||||
|
||||
@@ -0,0 +1,11 @@
|
||||
using System.Text.Json;
|
||||
using System.Text.Json.Serialization;
|
||||
|
||||
public sealed class AuditActionJsonConverter : JsonConverter<AuditAction>
|
||||
{
|
||||
public override AuditAction Read(ref Utf8JsonReader reader, Type typeToConvert, JsonSerializerOptions options)
|
||||
=> AuditActionExtensions.FromDbString(reader.GetString()!);
|
||||
|
||||
public override void Write(Utf8JsonWriter writer, AuditAction value, JsonSerializerOptions options)
|
||||
=> writer.WriteStringValue(value.ToDbString());
|
||||
}
|
||||
@@ -0,0 +1,11 @@
|
||||
using System.Text.Json;
|
||||
using System.Text.Json.Serialization;
|
||||
|
||||
public sealed class SepsisBundleComplianceStatusJsonConverter : JsonConverter<SepsisBundleComplianceStatus>
|
||||
{
|
||||
public override SepsisBundleComplianceStatus Read(ref Utf8JsonReader reader, Type typeToConvert, JsonSerializerOptions options)
|
||||
=> SepsisBundleComplianceStatusExtensions.FromDbString(reader.GetString()!);
|
||||
|
||||
public override void Write(Utf8JsonWriter writer, SepsisBundleComplianceStatus value, JsonSerializerOptions options)
|
||||
=> writer.WriteStringValue(value.ToDbString());
|
||||
}
|
||||
@@ -0,0 +1,9 @@
|
||||
public record UpdatePatientRequest(
|
||||
string? FirstName = null,
|
||||
string? LastName = null,
|
||||
DateOnly? DateOfBirth = null,
|
||||
string? Gender = null,
|
||||
BloodType? BloodType = null,
|
||||
string? Allergies = null,
|
||||
string? EmergencyContactName = null,
|
||||
string? EmergencyContactPhone = null);
|
||||
@@ -211,6 +211,8 @@ try
|
||||
{
|
||||
opts.JsonSerializerOptions.ReferenceHandler = ReferenceHandler.IgnoreCycles;
|
||||
opts.JsonSerializerOptions.Converters.Add(new BloodTypeJsonConverter());
|
||||
opts.JsonSerializerOptions.Converters.Add(new AuditActionJsonConverter());
|
||||
opts.JsonSerializerOptions.Converters.Add(new SepsisBundleComplianceStatusJsonConverter());
|
||||
opts.JsonSerializerOptions.Converters.Add(new JsonStringEnumConverter());
|
||||
opts.JsonSerializerOptions.Converters.Add(new ObservationSourceJsonConverter());
|
||||
opts.JsonSerializerOptions.Converters.Add(new DepartmentJsonConverter());
|
||||
|
||||
@@ -195,7 +195,7 @@ public class AnalyticsService : IAnalyticsService
|
||||
b.Filter(filters.ToArray());
|
||||
})
|
||||
)
|
||||
.From(page * pageSize)
|
||||
.From((page - 1) * pageSize)
|
||||
.Size(pageSize));
|
||||
|
||||
if (!resp.IsValidResponse)
|
||||
|
||||
@@ -4,5 +4,6 @@ public interface IPatientService
|
||||
Task<PagedResult<Patient>> ListAsync(string? q, int page, int pageSize);
|
||||
Task<Patient> GetByIdAsync(Guid id);
|
||||
Task<Encounter> OpenEncounterAsync(Guid patientId, OpenEncounterRequest req);
|
||||
Task<Patient> UpdateAsync(Guid id, UpdatePatientRequest req);
|
||||
Task<Patient> RegisterOrUpdateByIdentifierAsync(FhirPatientUpsertRequest req);
|
||||
}
|
||||
@@ -2,4 +2,5 @@ public interface IQsofaService
|
||||
{
|
||||
Task<QsofaCurrentResponse> GetCurrentAsync(Guid encounterId);
|
||||
Task<int> GetActiveCriteriaCountAsync(Guid encounterId);
|
||||
Task<CursorPage<ClinicalAlert>> GetHistoryAsync(Guid encounterId, int limit, string? cursor);
|
||||
}
|
||||
|
||||
@@ -4,5 +4,6 @@ public interface ISepsisBundleService
|
||||
Guid encounterId, Guid triggeringAlertId, AlertType alertType, CancellationToken ct = default);
|
||||
Task<SepsisBundle?> GetCurrentByEncounterAsync(Guid encounterId);
|
||||
Task<SepsisBundle> GetByIdAsync(Guid id);
|
||||
Task<PagedResult<SepsisBundle>> ListAsync(SepsisBundleComplianceStatus? status, int page, int pageSize);
|
||||
Task OnOrderResultedAsync(Guid orderId, CancellationToken ct = default);
|
||||
}
|
||||
@@ -90,6 +90,44 @@ public class PatientService : IPatientService
|
||||
return new PagedResult<Patient>(patients, page, pageSize, total);
|
||||
}
|
||||
|
||||
public async Task<Patient> UpdateAsync(Guid id, UpdatePatientRequest req)
|
||||
{
|
||||
var patient = await _db.Patients.FindAsync(id)
|
||||
?? throw new NotFoundException("Patient not found.", "PATIENT_NOT_FOUND");
|
||||
|
||||
var before = new
|
||||
{
|
||||
patient.FirstName, patient.LastName, patient.DateOfBirth,
|
||||
patient.Gender, patient.BloodType, patient.Allergies,
|
||||
patient.EmergencyContactName, patient.EmergencyContactPhone
|
||||
};
|
||||
|
||||
if (req.FirstName is not null) patient.FirstName = req.FirstName;
|
||||
if (req.LastName is not null) patient.LastName = req.LastName;
|
||||
if (req.DateOfBirth is not null) patient.DateOfBirth = req.DateOfBirth.Value;
|
||||
if (req.Gender is not null) patient.Gender = req.Gender;
|
||||
if (req.BloodType is not null) patient.BloodType = req.BloodType;
|
||||
if (req.Allergies is not null) patient.Allergies = req.Allergies;
|
||||
if (req.EmergencyContactName is not null) patient.EmergencyContactName = req.EmergencyContactName;
|
||||
if (req.EmergencyContactPhone is not null) patient.EmergencyContactPhone = req.EmergencyContactPhone;
|
||||
|
||||
await _db.SaveChangesAsync();
|
||||
|
||||
await _audit.WriteAsync(
|
||||
AuditAction.PatientUpdated,
|
||||
"Patient",
|
||||
patient.Id,
|
||||
previousValue: before,
|
||||
newValue: new
|
||||
{
|
||||
patient.FirstName, patient.LastName, patient.DateOfBirth,
|
||||
patient.Gender, patient.BloodType, patient.Allergies,
|
||||
patient.EmergencyContactName, patient.EmergencyContactPhone
|
||||
});
|
||||
|
||||
return patient;
|
||||
}
|
||||
|
||||
public async Task<Patient> GetByIdAsync(Guid id)
|
||||
{
|
||||
var patient = await _db.Patients
|
||||
|
||||
@@ -26,6 +26,28 @@ public class QsofaService : IQsofaService
|
||||
return QsofaCalculator.CountActiveCriteria(values);
|
||||
}
|
||||
|
||||
public async Task<CursorPage<ClinicalAlert>> GetHistoryAsync(
|
||||
Guid encounterId, int limit, string? cursor)
|
||||
{
|
||||
limit = Math.Clamp(limit, 1, 100);
|
||||
|
||||
var query = _db.ClinicalAlerts
|
||||
.AsNoTracking()
|
||||
.Where(a => a.EncounterId == encounterId
|
||||
&& (a.AlertType == AlertType.QsofaScreen || a.AlertType == AlertType.QsofaWarning));
|
||||
|
||||
var items = await query
|
||||
.OrderByDescending(a => a.TriggeredAt)
|
||||
.ThenByDescending(a => a.Id)
|
||||
.Take(limit + 1)
|
||||
.ToListAsync();
|
||||
|
||||
var hasMore = items.Count > limit;
|
||||
if (hasMore) items.RemoveAt(limit);
|
||||
|
||||
return new CursorPage<ClinicalAlert>(items, null, hasMore);
|
||||
}
|
||||
|
||||
private async Task EnsureEncounterExistsAsync(Guid encounterId)
|
||||
{
|
||||
var exists = await _db.Encounters.AnyAsync(e => e.Id == encounterId);
|
||||
|
||||
@@ -122,6 +122,29 @@ public class SepsisBundleService : ISepsisBundleService
|
||||
return bundle;
|
||||
}
|
||||
|
||||
public async Task<PagedResult<SepsisBundle>> ListAsync(
|
||||
SepsisBundleComplianceStatus? status, int page, int pageSize)
|
||||
{
|
||||
pageSize = Math.Clamp(pageSize, 1, 100);
|
||||
|
||||
var query = _db.SepsisBundles
|
||||
.AsNoTracking()
|
||||
.Include(b => b.Elements)
|
||||
.AsQueryable();
|
||||
|
||||
if (status.HasValue)
|
||||
query = query.Where(b => b.ComplianceStatus == status.Value);
|
||||
|
||||
var total = await query.CountAsync();
|
||||
var items = await query
|
||||
.OrderByDescending(b => b.RecognizedAt)
|
||||
.Skip((page - 1) * pageSize)
|
||||
.Take(pageSize)
|
||||
.ToListAsync();
|
||||
|
||||
return new PagedResult<SepsisBundle>(items, page, pageSize, total);
|
||||
}
|
||||
|
||||
public async Task OnOrderResultedAsync(Guid orderId, CancellationToken ct = default)
|
||||
{
|
||||
var element = await _db.SepsisBundleElements
|
||||
|
||||
@@ -0,0 +1,21 @@
|
||||
using FluentValidation;
|
||||
|
||||
public class FhirEncounterUpsertRequestValidator : AbstractValidator<FhirEncounterUpsertRequest>
|
||||
{
|
||||
public FhirEncounterUpsertRequestValidator()
|
||||
{
|
||||
RuleFor(x => x.IdentifierSystem).NotEmpty().MaximumLength(500);
|
||||
RuleFor(x => x.IdentifierValue).NotEmpty().MaximumLength(200);
|
||||
RuleFor(x => x.PatientId).NotEmpty();
|
||||
RuleFor(x => x.EncounterType).IsInEnum();
|
||||
RuleFor(x => x.Department).IsInEnum();
|
||||
RuleFor(x => x.AttendingPhysician).NotEmpty().MaximumLength(200);
|
||||
RuleFor(x => x.TargetStatus).IsInEnum();
|
||||
RuleFor(x => x.RoomBed).MaximumLength(20)
|
||||
.When(x => x.RoomBed is not null);
|
||||
RuleFor(x => x.AdmissionReason).MaximumLength(500)
|
||||
.When(x => x.AdmissionReason is not null);
|
||||
RuleFor(x => x.DischargeDiagnosis).MaximumLength(500)
|
||||
.When(x => x.DischargeDiagnosis is not null);
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,27 @@
|
||||
using FluentValidation;
|
||||
|
||||
public class FhirPatientUpsertRequestValidator : AbstractValidator<FhirPatientUpsertRequest>
|
||||
{
|
||||
private static readonly HashSet<string> ValidGenders = new(StringComparer.OrdinalIgnoreCase)
|
||||
{ "male", "female", "other", "unknown" };
|
||||
|
||||
public FhirPatientUpsertRequestValidator()
|
||||
{
|
||||
RuleFor(x => x.IdentifierSystem).NotEmpty().MaximumLength(500);
|
||||
RuleFor(x => x.IdentifierValue).NotEmpty().MaximumLength(200);
|
||||
RuleFor(x => x.FirstName).NotEmpty().MaximumLength(100);
|
||||
RuleFor(x => x.LastName).NotEmpty().MaximumLength(100);
|
||||
RuleFor(x => x.DateOfBirth).NotEmpty()
|
||||
.LessThanOrEqualTo(DateOnly.FromDateTime(DateTime.UtcNow))
|
||||
.WithMessage("Date of birth cannot be in the future.");
|
||||
RuleFor(x => x.Gender).NotEmpty()
|
||||
.Must(g => ValidGenders.Contains(g))
|
||||
.WithMessage("Gender must be one of: male, female, other, unknown.");
|
||||
RuleFor(x => x.BloodType).IsInEnum()
|
||||
.When(x => x.BloodType is not null);
|
||||
RuleFor(x => x.EmergencyContactName).MaximumLength(200)
|
||||
.When(x => x.EmergencyContactName is not null);
|
||||
RuleFor(x => x.EmergencyContactPhone).MaximumLength(20)
|
||||
.When(x => x.EmergencyContactPhone is not null);
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,10 @@
|
||||
using FluentValidation;
|
||||
|
||||
public class RecordOrderResultRequestValidator : AbstractValidator<RecordOrderResultRequest>
|
||||
{
|
||||
public RecordOrderResultRequestValidator()
|
||||
{
|
||||
RuleFor(x => x.ResultSummary).MaximumLength(2000)
|
||||
.When(x => x.ResultSummary is not null);
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,11 @@
|
||||
using FluentValidation;
|
||||
|
||||
public class TransitionStatusRequestValidator : AbstractValidator<TransitionStatusRequest>
|
||||
{
|
||||
public TransitionStatusRequestValidator()
|
||||
{
|
||||
RuleFor(x => x.Status).IsInEnum();
|
||||
RuleFor(x => x.DischargeDiagnosis).MaximumLength(500)
|
||||
.When(x => x.DischargeDiagnosis is not null);
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,24 @@
|
||||
using FluentValidation;
|
||||
|
||||
public class UpdatePatientRequestValidator : AbstractValidator<UpdatePatientRequest>
|
||||
{
|
||||
public UpdatePatientRequestValidator()
|
||||
{
|
||||
RuleFor(x => x.FirstName).MaximumLength(100)
|
||||
.When(x => x.FirstName is not null);
|
||||
RuleFor(x => x.LastName).MaximumLength(100)
|
||||
.When(x => x.LastName is not null);
|
||||
RuleFor(x => x.DateOfBirth)
|
||||
.LessThanOrEqualTo(DateOnly.FromDateTime(DateTime.UtcNow))
|
||||
.WithMessage("Date of birth cannot be in the future.")
|
||||
.When(x => x.DateOfBirth is not null);
|
||||
RuleFor(x => x.Gender).MaximumLength(10)
|
||||
.When(x => x.Gender is not null);
|
||||
RuleFor(x => x.BloodType).IsInEnum()
|
||||
.When(x => x.BloodType is not null);
|
||||
RuleFor(x => x.EmergencyContactName).MaximumLength(200)
|
||||
.When(x => x.EmergencyContactName is not null);
|
||||
RuleFor(x => x.EmergencyContactPhone).MaximumLength(20)
|
||||
.When(x => x.EmergencyContactPhone is not null);
|
||||
}
|
||||
}
|
||||
@@ -10,6 +10,7 @@
|
||||
- [Medication correlation design](decisions/medication-correlation-design.md) — how drug context annotates warning/NEWS2 alerts
|
||||
- [Simulator guide](simulator-guide.md) — replay JSON scenarios against this lifecycle
|
||||
- [Dashboard guide](dashboard-guide.md) — which endpoints the Vue ward UI polls
|
||||
- [Mirth FHIR integration](integration/mirth-fhir-channels.md) — HL7v2 → FHIR R4 ingest via integration engines
|
||||
|
||||
---
|
||||
|
||||
@@ -81,13 +82,17 @@ When a nurse records a heart rate of 95 bpm, here is what happens under the hood
|
||||
3. COMMIT — everything saved atomically
|
||||
|
||||
4. OutboxRelayService (runs every 500ms):
|
||||
- Reads unsent outbox events from PostgreSQL
|
||||
- Sends them to the correct Kafka topic
|
||||
- Reads unsent outbox events from PostgreSQL (using FOR UPDATE SKIP LOCKED for safe concurrency)
|
||||
- Sends them to the correct Kafka topic via an idempotent producer (broker deduplicates retries)
|
||||
- Marks them as processed
|
||||
- If Kafka produce fails: increments retry counter and tries again on next poll
|
||||
- After 10 failed attempts: marks the event as permanently failed (dead-lettered) — stops blocking other events
|
||||
```
|
||||
|
||||
**Why this pattern?** It guarantees that if the observation is saved, the message to Kafka will also be sent — even if Kafka is temporarily down. The database acts as a reliable staging area. This is called the "transactional outbox pattern."
|
||||
|
||||
**What about permanently failed events?** If a message truly cannot be delivered after 10 attempts (corrupted payload, topic deleted, etc.), it is marked `FailedAt` and excluded from future relay polls. This prevents a single poisoned event from blocking all subsequent messages. Operators can query failed events via `SELECT * FROM outbox_events WHERE failed_at IS NOT NULL` for manual investigation.
|
||||
|
||||
### Kafka topics — the conveyor belts
|
||||
|
||||
Each Kafka topic carries a specific type of message. Think of them as labeled conveyor belts in a factory:
|
||||
@@ -103,6 +108,8 @@ Each Kafka topic carries a specific type of message. Think of them as labeled co
|
||||
|
||||
**Key detail:** One message on `observation.recorded` is consumed by up to 8 different workers independently. Each worker has its own "consumer group" — Kafka tracks where each group left off, so every worker gets every message even if they process at different speeds.
|
||||
|
||||
**Poison pill protection:** If a consumer receives a malformed or un-processable message (corrupted JSON, invalid format), the system does not get stuck retrying it forever. A "poison pill guard" detects permanent errors and skips them immediately. For transient errors (temporary network issues), it retries up to 5 times before giving up on that message. Skipped messages are logged and tracked by a Prometheus metric (`kafka_poison_pills_skipped_total`) so operators know if something is wrong with the data flowing through the system.
|
||||
|
||||
### Redis — the fast-lookup layer
|
||||
|
||||
Redis stores data that needs to be checked on every observation ingest or scoring calculation. Reading from Redis takes less than 1 millisecond, compared to 5–50ms for a database query.
|
||||
@@ -157,7 +164,7 @@ MinIO stores files organized by path, like a file system in the cloud:
|
||||
| Path pattern | What is stored | How it gets there |
|
||||
|---|---|---|
|
||||
| `discharge-summaries/{encounterId}/summary.pdf` | Discharge summary for each encounter | Generated by the discharge worker (RabbitMQ) when a patient is discharged |
|
||||
| `observations/{YYYY}/{MM}/{DD}/partition-{p}-offset-{o}.parquet` | Daily observation data in Parquet format | Written by the data lake writer (Kafka consumer) in batches of 1000 events or every 5 minutes |
|
||||
| `observations/{YYYY}/{MM}/{DD}/partition-{p}-offset-{o}.parquet` | Daily observation data in Parquet format | Written by the data lake writer (Kafka consumer) in batches of 1000 events or every 5 minutes; Kafka offsets are only committed for partitions where the upload succeeded — failed partitions are retried on the next flush |
|
||||
| `alerts/{YYYY}/{MM}/{DD}/partition-{p}-offset-{o}.parquet` | Daily alert data in Parquet format | Same as above |
|
||||
| `encounters/{YYYY}/{MM}/{DD}/partition-{p}-offset-{o}.parquet` | Daily encounter status changes in Parquet format | Same as above |
|
||||
|
||||
@@ -179,6 +186,7 @@ Prometheus scrapes the `/metrics` endpoint every 15 seconds and records operatio
|
||||
| `alerts_unacknowledged_gauge` | Gauge | Current count of critical alerts open for more than 5 minutes |
|
||||
| `outbox_pending_events` | Gauge | Unprocessed outbox events (if this grows, the relay is falling behind) |
|
||||
| `kafka_consumer_lag` | Gauge | How far behind each Kafka consumer is (if this grows, scoring is delayed) |
|
||||
| `kafka_poison_pills_skipped_total` | Counter | Messages skipped as un-processable (labeled by consumer group and topic). If this grows, investigate the source of malformed messages |
|
||||
|
||||
Grafana displays these metrics as charts, and operators can set up alerts (system alerts, not clinical alerts) if things like consumer lag or pending outbox events climb too high.
|
||||
|
||||
@@ -199,6 +207,10 @@ All services run in Docker containers defined in `docker-compose.yml`:
|
||||
|
||||
Starting the full stack: `docker compose up -d` brings up all 8 services. The .NET API runs separately (outside Docker) and connects to these services on the ports listed above.
|
||||
|
||||
**Health check endpoints:** The API exposes two health endpoints (no authentication required):
|
||||
- `GET /health/live` — liveness probe. Returns 200 if the process is running. Use this for Kubernetes liveness probes or load balancer checks.
|
||||
- `GET /health/ready` — readiness probe. Checks connectivity to all five infrastructure services (PostgreSQL, Redis, Kafka, RabbitMQ, Elasticsearch). Returns 200 only when all services are reachable. Use this for readiness gates — if it returns unhealthy, the API cannot process requests properly.
|
||||
|
||||
---
|
||||
|
||||
## Conventions
|
||||
@@ -208,6 +220,8 @@ Starting the full stack: `docker compose up -d` brings up all 8 services. The .N
|
||||
| Base path | `/api/v1` |
|
||||
| Response envelope | `{ success, statusCode, data, error }` |
|
||||
| Correlation | Optional `X-Correlation-Id` request header; echoed on the response |
|
||||
| Authentication | JWT bearer token required on all clinical endpoints; obtain via `POST /api/v1/auth/login` |
|
||||
| Health checks | `GET /health/live` (liveness) and `GET /health/ready` (readiness) — no authentication required |
|
||||
| Active encounter guard | `POST` observations, medications, and orders return **409** (`ENCOUNTER_NOT_ACTIVE`) when the encounter is discharged or cancelled |
|
||||
| Async latency | Warning alerts, NEWS2 scores, qSOFA, SOFA, trend alerts, and sepsis bundles are created by background workers — allow a few seconds after recording a measurement before polling for results |
|
||||
|
||||
@@ -496,8 +510,8 @@ These are typically set up once at deploy time, not called per patient. This is
|
||||
|
||||
| What | How | Technology involved | Why it matters |
|
||||
|---|---|---|---|
|
||||
| Alert thresholds | Seeded in PostgreSQL; `ThresholdCacheLoader` copies them into Redis at startup; manageable via `POST/GET/PUT /alert-thresholds` | PostgreSQL → Redis | Every measurement is validated against thresholds from Redis (fast) to decide if alerts fire. Without this cache, every observation ingest would need a database query |
|
||||
| Kafka topics | `KafkaTopicProvisioner` creates 7 topics on startup (6 partitions each, auto-creation disabled) | Kafka | `observation.recorded`, `alert.generated`, `encounter.status.changed`, `sepsis.bundle.created`, `sepsis.bundle.updated`, `gcs.scored`. These must exist before any messages can flow |
|
||||
| Alert thresholds | Seeded in PostgreSQL; `ThresholdCacheLoader` copies them into Redis at startup (retries up to 3 times with exponential backoff if Redis is unavailable; application starts without cache if Redis remains down — falls back to PostgreSQL queries); manageable via `POST/GET/PUT /alert-thresholds` | PostgreSQL → Redis | Every measurement is validated against thresholds from Redis (fast) to decide if alerts fire. Without this cache, every observation ingest would need a database query |
|
||||
| Kafka topics | `KafkaTopicProvisioner` creates 7 topics on startup (6 partitions each, configurable replication factor — default 3, auto-creation disabled) | Kafka | `observation.recorded`, `alert.generated`, `encounter.status.changed`, `sepsis.bundle.created`, `sepsis.bundle.updated`, `gcs.scored`. These must exist before any messages can flow |
|
||||
| Elasticsearch indices | `ElasticIndexProvisioner` creates 3 indices with proper field mappings | Elasticsearch | `patient_encounters`, `observations`, `clinical_alerts` — must exist before data can be indexed |
|
||||
| RabbitMQ topology | `RabbitMqTopologyProvisioner` declares the exchange and 5 queues | RabbitMQ | Sets up the `clinical.notifications.exchange` and all queues (paging, DLQ, escalation, discharge, reconciliation) with proper routing and dead-letter configuration |
|
||||
| MinIO bucket | Created if it does not exist | MinIO | The `vigilcare` bucket must exist before discharge summaries or data lake files can be written |
|
||||
@@ -518,7 +532,7 @@ POST /api/v1/patients
|
||||
| `firstName`, `lastName`, `dateOfBirth`, `gender` | yes | |
|
||||
| `bloodType`, `allergies`, `emergencyContactName`, `emergencyContactPhone` | no | Stored for ward context |
|
||||
|
||||
**Response:** `201 Created` — `data` includes a system-generated `mrn` (e.g. `MRN-000042`) and `id` (UUID). Save both; the MRN is what nurses see on wristbands, the UUID is used in all API paths.
|
||||
**Response:** `201 Created` — `data` includes a system-generated `mrn` (e.g. `MRN-000042`) and `id` (UUID). Save both; the MRN is what nurses see on wristbands, the UUID is used in all API paths. MRNs are generated from a PostgreSQL sequence (`mrn_seq`), guaranteeing uniqueness even under concurrent registration.
|
||||
|
||||
**Later lookups:**
|
||||
|
||||
@@ -811,7 +825,7 @@ GET /api/v1/sepsis-bundles/{id}
|
||||
|
||||
**Compliance flow:**
|
||||
|
||||
1. Bundle is created with `complianceStatus: IN_PROGRESS` and `deadlineAt` set to 1 hour from detection
|
||||
1. Bundle is created with `complianceStatus: IN_PROGRESS` and `deadlineAt` set to 1 hour from detection. A database constraint ensures only one in-progress bundle can exist per encounter at any time — preventing duplicate bundles from race conditions.
|
||||
2. Clinicians complete orders via `PATCH /orders/{id}/result`
|
||||
3. Each completed order marks its bundle element `COMPLETED`
|
||||
4. When all 4 elements are done:
|
||||
@@ -912,7 +926,7 @@ A patient may later return and receive a new encounter via `POST /patients/{id}/
|
||||
| | `GET` | `/analytics/patients`, `/analytics/observations/trend`, `/analytics/alerts/summary`, `/analytics/population` |
|
||||
| **9 — Discharge** | `PATCH` | `/encounters/{id}/status` |
|
||||
|
||||
**Operational (not encounter-scoped):** `GET /metrics` (Prometheus), Swagger UI (development only).
|
||||
**Operational (not encounter-scoped):** `GET /metrics` (Prometheus), `GET /health/live` (liveness probe), `GET /health/ready` (readiness probe — checks PostgreSQL, Redis, Kafka, RabbitMQ, Elasticsearch), Swagger UI (development only).
|
||||
|
||||
---
|
||||
|
||||
@@ -984,7 +998,7 @@ The Vue ward dashboard (`vigilcare-dashboard/`) implements this lifecycle throug
|
||||
| Trend alert | No | PostgreSQL → Kafka → trend analyzer (reads Redis history) → PostgreSQL | `GET .../alerts` |
|
||||
| Medication annotation on alert | No — applied at alert creation | Kafka consumer reads recent meds from PostgreSQL | `GET .../alerts` → read `details` field |
|
||||
| Search index updated | No | PostgreSQL → Kafka → ES indexer → Elasticsearch | `GET /analytics/...` endpoints |
|
||||
| Data lake file written | No | PostgreSQL → Kafka → data lake writer → MinIO (Parquet) | MinIO bucket `vigilcare` |
|
||||
| Data lake file written | No | PostgreSQL → Kafka → data lake writer → MinIO (Parquet); offsets only committed for successful partition uploads | MinIO bucket `vigilcare` |
|
||||
| Clinician paged | No | Kafka → notification publisher → RabbitMQ paging queue | Alert `status` field |
|
||||
| Alert escalated | No | RabbitMQ paging queue → DLQ → escalation queue → PostgreSQL | Alert `status` = `escalated` |
|
||||
| Discharge summary PDF | No | Kafka → RabbitMQ discharge queue → worker → MinIO | MinIO `/discharge-summaries/{encounterId}/summary.pdf` |
|
||||
|
||||
Reference in New Issue
Block a user