1051 lines
59 KiB
Markdown
1051 lines
59 KiB
Markdown
# VigilCare Clinical API
|
||
|
||
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 detection, and clinician notification with automatic escalation.
|
||
|
||
**Implementation status:** All ten planned phases are complete — 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, and clinical data model expansion (patient demographics, encounter enrichment, 12 observation codes). See [Implemented Phases](#implemented-phases) for the full breakdown.
|
||
|
||
## Domain Model — How It Maps to a Real Clinical System
|
||
|
||
In a hospital, a patient presents for care and an encounter is opened. Bedside monitors and lab systems post observations continuously against that encounter. A rules engine evaluates each observation against configured thresholds and flags abnormal values as clinical alerts. Clinicians acknowledge and resolve alerts. If a critical alert goes unacknowledged for five minutes, the system escalates to the on-call backup. All events flow through Kafka so the Elasticsearch dashboard, sepsis engine, and data lake writer consume the same stream independently.
|
||
|
||
```
|
||
Patient ─────────────────────────── one patient = one MRN, many lifetime encounters
|
||
└── Encounter one clinical episode (inpatient, outpatient, ED)
|
||
├── Observation one measurement: vital sign, lab value, SpO₂
|
||
│ └── OutboxEvent written in the same transaction → relayed to Kafka
|
||
└── ClinicalAlert generated on threshold breach or SIRS detection
|
||
└── OutboxEvent → Kafka → RabbitMQ → clinician page → escalation
|
||
```
|
||
|
||
### Patient
|
||
|
||
A `Patient` is registered with demographic information and assigned a Medical Record Number (MRN) — a stable identifier that never changes across encounters. Optional clinical fields include blood type (`A+`, `O-`, etc.), known allergies, and emergency contact name/phone. Patient search supports both MRN exact match and name partial match (`ILIKE`).
|
||
|
||
### Encounter
|
||
|
||
An `Encounter` is a single clinical episode. Status follows a controlled machine: `scheduled → active → discharged` (or `cancelled` from any pre-discharged state). Optional `roomBed` and `admissionReason` fields support ward assignment and clinical context; `dischargeDiagnosis` is set on discharge. Observations, alerts, and orders belong to an encounter, not directly to a patient — this bounds queries naturally and mirrors real clinical data ownership. Discharge triggers a RabbitMQ job to generate a discharge summary.
|
||
|
||
### AlertThreshold
|
||
|
||
Alert thresholds define the numeric boundaries that trigger a clinical alert for a given observation code. Each threshold has four optional bounds: `criticalLow`, `warningLow`, `warningHigh`, `criticalHigh`. Thresholds are pre-loaded into Redis on startup and invalidated on write — they are read on every observation ingest and must not add database latency to the hot path.
|
||
|
||
### Observation
|
||
|
||
An `Observation` is a single recorded measurement: a vital sign, lab value, or pulse oximetry reading. Observations are append-only — never updated or deleted. Each observation is evaluated against the Redis-cached threshold immediately on ingest. A `CRITICAL` breach synchronously creates a `ClinicalAlert` within the same transaction before the API returns. A `WARNING` breach is deferred to the Kafka consumer. This split is a deliberate patient safety decision.
|
||
|
||
An `idempotencyKey` (partial unique index) prevents duplicate observations when medical devices retry on network failure.
|
||
|
||
### ClinicalAlert
|
||
|
||
A `ClinicalAlert` is generated when an observation breaches a threshold or when the sepsis engine detects two or more concurrent SIRS criteria. Lifecycle: `open → acknowledged → resolved` (or `escalated` after a five-minute NACK cycle through the RabbitMQ dead-letter queue). Alerts carry an audit trail: who acknowledged, when, and with what note.
|
||
|
||
### OutboxEvent
|
||
|
||
An `OutboxEvent` is written in the same transaction as any observation or alert, then relayed to Kafka by a background worker. This decouples Kafka availability from the ingest transaction — observations commit to PostgreSQL while Kafka is down, and the relay catches up on recovery.
|
||
|
||
---
|
||
|
||
## Features
|
||
|
||
- **Patient Registration** — register patients with MRN generation; optional blood type, allergies, and emergency contact; paginated list with name (`ILIKE`) and MRN (exact) search; patient detail with active encounter summary
|
||
- **Encounter Management** — open encounters against a patient with optional room/bed and admission reason; encounter status state machine (`scheduled → active → discharged / cancelled`) with 409 on illegal transitions; optional discharge diagnosis on discharge; encounter timeline as a merged chronological view across status changes, observation summaries, and alerts
|
||
- **Alert Threshold Management** — configure per-observation-code numeric bounds (`criticalLow`, `warningLow`, `warningHigh`, `criticalHigh`) for 12 observation codes; thresholds pre-loaded into Redis on startup; write-through cache invalidation on update
|
||
- **Observation Ingest** — `POST /encounters/:id/observations` accepts single or small batch (up to 10); idempotency via `Idempotency-Key` header (partial unique index); plausibility validation per observation code; synchronous critical alert creation within the ingest transaction; warning-range breaches evaluated asynchronously by `WarningAlertService` (Kafka consumer group `warning-evaluator`); outbox event written in the same commit; cursor-paginated history on `(encounter_id, observation_code, recorded_at DESC)`
|
||
- **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** — three topics (`observation.recorded`, `alert.generated`, `encounter.status.changed`) with six partitions each; KRaft mode, no Zookeeper; `KAFKA_AUTO_CREATE_TOPICS_ENABLE=false` — topics are provisioned explicitly by `KafkaTopicProvisioner`
|
||
- **Elasticsearch CQRS Projection** — `EsIndexerService` consumer group upserts `patient_encounters` documents, appends to the `observations` index, and updates `openAlertCount` on alert 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 Early Warning Engine** — `SepsisEngineService` Kafka consumer evaluates SIRS criteria (temperature, heart rate, respiratory rate, WBC) per encounter using Redis keys with a 30-minute TTL sliding window; on ≥2 active criteria, inserts a `SEPSIS_WARNING / CRITICAL` alert idempotently (`INSERT WHERE NOT EXISTS`)
|
||
- **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 successful uploads; `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
|
||
- **Observability** — Serilog structured logging enriched with `correlationId`, `encounterId`, `patientId` on alert paths; Seq sink (`http://localhost:5345`); Prometheus (`http://localhost:9101`) scrapes `GET /metrics`; eight application metric families via `ClinicalMetrics` and three background collectors (`AlertsUnacknowledgedCollector`, `OutboxPendingCollector`, `KafkaConsumerLagCollector`); Grafana clinical dashboard (`http://localhost:3101`, admin/admin) with `alerts_unacknowledged_gauge` as the primary safety panel; per-request correlation IDs in request logs and `X-Correlation-Id` response headers
|
||
- **Swagger UI** — OpenAPI spec via Swashbuckle (Development only)
|
||
|
||
---
|
||
|
||
## Architecture
|
||
|
||
```
|
||
HTTP request
|
||
→ CorrelationIdMiddleware
|
||
→ ExceptionHandlerMiddleware
|
||
→ Controllers
|
||
→ Services
|
||
├── PostgreSQL (EF Core — writes, keyed reads)
|
||
├── Redis (threshold cache, SIRS state)
|
||
└── OutboxEvent (same transaction as domain write)
|
||
|
||
IHostedServices (background):
|
||
ThresholdCacheLoader → pre-loads Redis on startup
|
||
KafkaTopicProvisioner → creates topics with correct partition count
|
||
RabbitMqTopologyProvisioner → declares exchange, queues, DLQ bindings
|
||
ElasticIndexProvisioner → creates index mappings
|
||
OutboxRelayService → PostgreSQL outbox → Kafka (every 500ms)
|
||
EsIndexerService → Kafka → Elasticsearch (consumer group: es-indexer)
|
||
SepsisEngineService → Kafka → Redis SIRS state → PostgreSQL alert (consumer group: sepsis-engine)
|
||
WarningAlertService → Kafka → WarningEvaluator → PostgreSQL WARNING alert (consumer group: warning-evaluator)
|
||
NotificationPublisherService → Kafka → RabbitMQ paging.queue (consumer group: notification-publisher)
|
||
PagingWorkerService → RabbitMQ paging.queue → log page → NACK on timeout (or requeue on shutdown)
|
||
EscalationWorkerService → RabbitMQ escalation.queue → update alert status
|
||
DischargeSummaryWorkerService → RabbitMQ discharge.queue → MinIO PDF
|
||
DataLakeWriterService → Kafka (data-lake-writer) → Parquet files in MinIO
|
||
ReconciliationScheduler → three scheduled safety checks → reconciliation_alerts + RabbitMQ
|
||
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
|
||
ClinicalMetrics (singleton) → inline counters/histogram from ingest, SIRS, escalation paths
|
||
```
|
||
|
||
**Why Kafka and RabbitMQ coexist:** Kafka is an append-only log — the same observation event reaches the Elasticsearch indexer, the sepsis engine, and the data lake independently without coordination. Each consumer holds its own offset and can replay from the beginning. RabbitMQ handles the action side: one message, one worker, one page. A duplicate page at 3am is a patient safety concern, not a minor inconvenience — RabbitMQ's acknowledgment-then-delete model is correct here. The DLQ TTL-based escalation has no equivalent in Kafka.
|
||
|
||
---
|
||
|
||
## Tech Stack
|
||
|
||
| Layer | Technology |
|
||
|---|---|
|
||
| Server | ASP.NET Core 8 (.NET 8.0) |
|
||
| Database | PostgreSQL 16 with EF Core 8 (code-first migrations) |
|
||
| Cache / SIRS state | Redis 7 |
|
||
| Message log | Apache Kafka 3.7 (KRaft, 6 partitions per topic) |
|
||
| Task queue | RabbitMQ 3.13 (direct exchange, DLQ escalation) |
|
||
| Search / analytics | Elasticsearch 8.13 (CQRS read projection) |
|
||
| Data lake | MinIO (Parquet, S3-compatible) |
|
||
| Logging | Serilog + Seq sink |
|
||
| Metrics | prometheus-net.AspNetCore (`GET /metrics`) |
|
||
| Dashboards | Prometheus 2.52 + Grafana 10.4 |
|
||
| Data lake format | Parquet.Net 4.x |
|
||
| Docs | Swagger / OpenAPI (Swashbuckle) |
|
||
| Validation | FluentValidation.AspNetCore |
|
||
| Testing | xUnit + Testcontainers + WebApplicationFactory |
|
||
|
||
---
|
||
|
||
## Project Structure
|
||
|
||
```
|
||
VigilCareClinicalAPI/
|
||
├── Program.cs # Service registration, middleware, seed on startup
|
||
├── appsettings.json # Connection strings, Kafka, Elasticsearch, RabbitMQ, MinIO, Serilog, ReconciliationJobs
|
||
├── Controllers/
|
||
│ ├── PatientsController.cs # Patient CRUD, search by name/MRN
|
||
│ ├── EncountersController.cs # Encounter open, status PATCH, timeline
|
||
│ ├── ObservationsController.cs # Ingest POST, cursor-paginated GET
|
||
│ ├── AlertThresholdsController.cs # Threshold CRUD + cache invalidation
|
||
│ ├── AlertsController.cs # Alert list (global + per-encounter), acknowledge, resolve
|
||
│ ├── OrdersController.cs # Order create, list, get, status transition, record result
|
||
│ └── AnalyticsController.cs # Elasticsearch-backed patient search, trend, alert summary, population
|
||
├── Domains/
|
||
│ ├── Entities/
|
||
│ │ ├── Patient.cs
|
||
│ │ ├── Encounter.cs # Status machine; SetStatus() enforces transition matrix
|
||
│ │ ├── AlertThreshold.cs
|
||
│ │ ├── Observation.cs # Append-only; IdempotencyKey; partial unique index
|
||
│ │ ├── ClinicalAlert.cs # open → acknowledged → resolved / escalated
|
||
│ │ ├── Order.cs
|
||
│ │ ├── OutboxEvent.cs # topic + payload JSONB + processed_at
|
||
│ │ └── ReconciliationAlert.cs
|
||
│ └── Enums/
|
||
│ ├── EncounterStatus.cs # Scheduled, Active, Discharged, Cancelled
|
||
│ ├── EncounterType.cs # Inpatient, Outpatient, Emergency
|
||
│ ├── AlertSeverity.cs # Warning, Critical
|
||
│ ├── AlertStatus.cs # Open, Acknowledged, Resolved, Escalated
|
||
│ ├── AlertType.cs # Threshold breach, sepsis, warning*, systolic BP, AVPU, glucose, …
|
||
│ ├── BloodType.cs # A+, O-, AB-, … with ToDbString/FromDbString
|
||
│ ├── ObservationSource.cs # Device, Manual, Lab
|
||
│ └── OrderType.cs / ReconciliationCheckType.cs / Department.cs / OrderStatus.cs
|
||
│ └── Json/
|
||
│ ├── ObservationSourceJsonConverter.cs
|
||
│ ├── DepartmentJsonConverter.cs
|
||
│ └── BloodTypeJsonConverter.cs # Clinical notation (A+, AB-) in JSON API
|
||
├── Services/
|
||
│ ├── Interfaces/ # IPatientService, IEncounterService, …
|
||
│ ├── PatientService.cs
|
||
│ ├── EncounterService.cs # Status state machine + ConflictException on invalid transitions
|
||
│ ├── AlertThresholdService.cs # CRUD + Redis write-through invalidation
|
||
│ ├── ObservationService.cs # Ingest transaction: idempotency → plausibility → threshold → alert → outbox; emits Prometheus counters
|
||
│ ├── ObservationQueryService.cs # Cursor-paginated history
|
||
│ ├── AlertService.cs # Acknowledge, resolve, list
|
||
│ ├── OrderService.cs # Order lifecycle; status machine; ConflictException on illegal transitions
|
||
│ ├── WarningEvaluator.cs # Warning-range threshold evaluation; idempotent alert INSERT
|
||
│ ├── AnalyticsService.cs # Elasticsearch query wrappers
|
||
│ └── PlausibilityValidator.cs # Per-code numeric range guard
|
||
├── Validators/ # FluentValidation — RegisterPatient, OpenEncounter, IngestObservation, …
|
||
├── Observability/
|
||
│ └── Metrics/
|
||
│ └── ClinicalMetrics.cs # Eight Prometheus metric families (counters, histogram, gauges)
|
||
├── BackgroundServices/
|
||
│ ├── ThresholdCacheLoader.cs # Pre-loads all thresholds into Redis on startup
|
||
│ ├── KafkaTopicProvisioner.cs # Creates topics with NumPartitions from config
|
||
│ ├── OutboxRelayService.cs # Polls outbox every 500ms; publishes to Kafka; marks processed
|
||
│ ├── Metrics/
|
||
│ │ ├── AlertsUnacknowledgedCollector.cs # Polls open CRITICAL alerts > 5 min → alerts_unacknowledged_gauge
|
||
│ │ ├── OutboxPendingCollector.cs # Polls unprocessed outbox rows → outbox_pending_events
|
||
│ │ └── KafkaConsumerLagCollector.cs # Lag for es-indexer, sepsis-engine, notification-publisher, data-lake-writer
|
||
│ ├── ElasticsSearch/
|
||
│ │ ├── ElasticIndexProvisioner.cs # Creates patient_encounters, observations, clinical_alerts indices
|
||
│ │ └── EsIndexerService.cs # consumer group: es-indexer; upserts Elasticsearch documents
|
||
│ ├── SepsisEngineService.cs # consumer group: sepsis-engine; SIRS eval via Redis TTL keys
|
||
│ ├── WarningAlertService.cs # consumer group: warning-evaluator; observation.recorded → WARNING alerts
|
||
│ ├── Notifications/
|
||
│ │ ├── NotificationPublisherService.cs # consumer group: notification-publisher; alert.generated → RabbitMQ paging.queue
|
||
│ │ ├── PagingWorkerService.cs # RabbitMQ consumer; logs page; NACK on ack timeout → DLQ, requeue on graceful shutdown
|
||
│ │ ├── EscalationWorkerService.cs # RabbitMQ escalation.queue; logs escalation; sets alert.status = escalated
|
||
│ │ └── DischargeSummaryWorkerService.cs # RabbitMQ discharge.queue; generates summary PDF; uploads to MinIO
|
||
│ └── Reconciliation/
|
||
│ ├── ReconciliationScheduler.cs # Runs three safety checks on a configurable interval
|
||
│ ├── UnacknowledgedAlertsCheck.cs # CRITICAL alerts unacknowledged > 30 min
|
||
│ ├── 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
|
||
├── Configuration/
|
||
│ ├── KafkaOptions.cs / KafkaTopicOptions.cs
|
||
│ ├── RabbitMqOptions.cs / MinioOptions.cs
|
||
│ └── ReconciliationJobOptions.cs
|
||
├── Sepsis/
|
||
│ ├── SirsDetector.cs # Redis SIRS state management (SET/DEL/MGET)
|
||
│ └── SirsEvaluator.cs # Per-code criterion evaluation
|
||
├── Elasticsearch/Documents/
|
||
│ ├── PatientEncounterDocument.cs
|
||
│ ├── ObservationDocument.cs
|
||
│ └── ClinicalAlertDocument.cs
|
||
├── Notifications/
|
||
│ └── RabbitMqTopologyProvisioner.cs # Declares exchange, queues, DLQ bindings on startup
|
||
├── Storage/
|
||
│ └── MinioClientFactory.cs
|
||
├── DataLake/
|
||
│ ├── DataLakeOptions.cs # Flush thresholds and bucket settings
|
||
│ ├── DataLakeWriterService.cs # consumer group: data-lake-writer; Kafka → Parquet → MinIO
|
||
│ └── ParquetFileBuilder.cs # Topic row models → Parquet byte arrays
|
||
├── Models/Records/
|
||
│ ├── Observation/ObservationRow.cs # Parquet row contract for observation events
|
||
│ ├── Alert/AlertRow.cs # Parquet row contract for alert events
|
||
│ └── Encounter/EncounterStatusRow.cs # Parquet row contract for encounter status events
|
||
├── Data/
|
||
│ ├── AppDbContext.cs # EF Core context — entity configs, indexes, constraints
|
||
│ ├── Configurations/ # IEntityTypeConfiguration per entity; ElasticsearchOptions, ElasticIndexOptions
|
||
│ └── Seed/DataSeeder.cs # Seeds patients, encounters, thresholds, observations
|
||
├── Common/
|
||
│ ├── ApiResponse.cs # { success, statusCode, data, error } envelope
|
||
│ ├── PagedResult.cs / CursorPage.cs
|
||
│ └── Exceptions/
|
||
│ ├── NotFoundException.cs
|
||
│ ├── ConflictException.cs # Thrown by encounter status machine
|
||
│ ├── DomainException.cs
|
||
│ └── ValidationException.cs
|
||
├── Middlewares/
|
||
│ ├── CorrelationIdMiddleware.cs
|
||
│ └── ExceptionHandlerMiddleware.cs
|
||
└── Migrations/
|
||
|
||
infra/
|
||
├── prometheus/
|
||
│ └── prometheus.yml # Scrape config for vigilcare_api /metrics
|
||
└── grafana/
|
||
├── provisioning/ # Datasource + dashboard provider config
|
||
└── dashboards/ # vigilcare.json clinical dashboard
|
||
|
||
tests/
|
||
└── VigilCareClinicalAPI.Tests/
|
||
├── ObservationIngestTests.cs # Ingest happy path, critical alert creation, discharged encounter rejection, idempotency
|
||
├── AlertLifecycleTests.cs # Acknowledge, resolve, escalation guard
|
||
├── SirsDetectorTests.cs # Redis SIRS state SET/DEL/MGET logic
|
||
├── SirsEvaluatorTests.cs # Per-code criterion evaluation
|
||
├── NotificationPipelineTests.cs # RabbitMQ topology, DLQ routing
|
||
├── ReconciliationTests.cs # Three reconciliation checks, deduplication, RabbitMQ publish
|
||
├── ObservabilityPhase8Tests.cs # /metrics families and correlation header behavior
|
||
├── DataLakePhase9Tests.cs # Kafka → MinIO Parquet flow and schema checks
|
||
├── ClinicalDemographicsAndObservationTests.cs # Patient/encounter enrichment, expanded observation alerts
|
||
├── WarningAlertTests.cs # WarningEvaluator — warning created, normal/critical skipped, idempotent
|
||
├── OrderLifecycleTests.cs # Orders API — create, list, record result, illegal transition 409
|
||
└── ValidationTests.cs # FluentValidation — empty fields, threshold ordering, order description
|
||
|
||
scripts/
|
||
├── run-api-redis-tests.sh # Phase 1 — patient/encounter/threshold + Redis cache
|
||
├── run-kafka-outbox-tests.sh # Phase 3 — outbox relay and Kafka topics
|
||
├── run-elasticsearch-analytics-tests.sh # Phase 4 — Elasticsearch CQRS projection
|
||
├── run-sepsis-sirs-tests.sh # Phase 5 — SIRS detector and sepsis engine
|
||
├── run-notification-pipeline-tests.sh # Phase 6 — RabbitMQ paging, DLQ, discharge summary
|
||
├── run-reconciliation-tests.sh # Phase 7 — reconciliation scheduler checks
|
||
├── run-phase8-verification.sh # Phase 8 — Prometheus metrics, alerts_unacknowledged_gauge, correlation headers
|
||
├── run-phase9-verification.sh # Phase 9 — data lake tests, Kafka offsets, MinIO Parquet, DuckDB schema
|
||
├── run-phase10-verification.sh # Phase 10 — 12 Redis thresholds, clinical enrichment, ES pipeline, integration tests
|
||
└── run-phase11-verification.sh # Phase 11 — warning alerts, orders API, validation, integration tests
|
||
|
||
docs/
|
||
├── plans/ # Phase 1–11 implementation and verification guides
|
||
├── decisions/
|
||
│ ├── data-lake-design.md # Parquet vs JSON, partitioning, replay rationale
|
||
│ └── sepsis-engine-design.md # SIRS sliding window and idempotent alert design
|
||
├── docker-compose-usage-and-troubleshooting.md
|
||
└── vigilcare-clinical-api-prd.md # Product requirements and phase roadmap
|
||
```
|
||
|
||
---
|
||
|
||
## Architecture Decisions
|
||
|
||
### Synchronous vs Asynchronous Alert Detection — The Split
|
||
|
||
Critical threshold breaches are detected synchronously within the ingest transaction. A critical potassium of 2.1 mEq/L is immediately life-threatening. If the API returns `201 Created` before generating the alert and the Kafka consumer lags by 30 seconds, a patient could deteriorate during that window. The synchronous check costs one additional Redis read per observation on the hot path — acceptable for correctness.
|
||
|
||
A warning heart rate of 95 bpm warrants attention but is not an emergency. The additional latency of Kafka consumer processing is clinically acceptable for a warning.
|
||
|
||
This is the architectural decision that separates thinking about healthcare systems from thinking about financial systems. The tradeoff is latency for correctness, and the correctness definition is clinical, not technical.
|
||
|
||
### Encounter as the Aggregate Root (Not Patient)
|
||
|
||
Observations, alerts, and orders belong to an encounter, not directly to a patient. A patient's blood pressure taken during a 2022 admission belongs to that admission. This bounds queries naturally: "show me all observations for this encounter" is a bounded query. "Show me all observations ever recorded for this patient" is a cross-encounter aggregation that belongs in the data lake.
|
||
|
||
### Redis for Two Distinct Purposes
|
||
|
||
Redis serves two independent roles with different semantics:
|
||
|
||
1. **Threshold cache:** write-through invalidation on every threshold update. Staleness here has clinical consequences — a stale threshold could suppress a critical alert. TTL expiry is not sufficient; invalidation must be immediate on write.
|
||
|
||
2. **SIRS sliding window:** `SET sirs:{encounterId}:{code} EX 1800`. The TTL does real work — a heart rate that was abnormal 31 minutes ago stops contributing to the SIRS count without any cleanup job. The 30-minute TTL is a clinical parameter, not an arbitrary cache timeout.
|
||
|
||
### 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.
|
||
|
||
### Kafka Partition Key: `encounterId`
|
||
|
||
All events for the same encounter land on the same partition. The sepsis engine requires this: if observations from the same patient arrive on different partitions, they may be consumed out of order and simultaneous SIRS criteria could be missed. Six partitions balance parallelism against per-encounter ordering guarantees.
|
||
|
||
### Elasticsearch as a CQRS Read Projection
|
||
|
||
PostgreSQL is always the write side and the source of truth. Elasticsearch is a denormalized, queryable projection optimized for the queries clinicians actually run. The `population` endpoint — "how many active patients have a heart rate above 100 in the last hour" — is a numeric range aggregation across potentially millions of observation rows. Running this against PostgreSQL on the operational database would compete with ingest writes. Elasticsearch's aggregation engine is purpose-built for this pattern.
|
||
|
||
**The replay:** stop `EsIndexerService` → delete both indices → reset consumer group offset to 0 → restart → wait for rebuild → verify document count matches PostgreSQL row count. This is the proof that Elasticsearch is a projection and not a source of truth, and the clearest demonstration of why Kafka retains events after consumption.
|
||
|
||
### DLQ as a Clinical Escalation Protocol
|
||
|
||
The five-minute escalation is not a retry — it is a clinical workflow. When an alert is created, the paging worker sends a page to the attending physician. If no acknowledgment arrives within five minutes, the message NACKs to `alerts.paging.dlq` with `x-message-ttl = 300000ms`. After TTL expires, the DLQ re-routes to `alerts.escalation.queue` and the on-call backup is paged. During graceful shutdown, cancellation of the in-flight wait loop is treated as non-failure and the message is NACKed with `requeue=true`, preventing false escalation during deploy/restart windows. This pattern has no equivalent in Kafka — Kafka has no concept of per-message TTL or conditional re-routing based on consumer acknowledgment.
|
||
|
||
### Why Not a Time-Series Database for Observations?
|
||
|
||
A medium hospital with 200 concurrent inpatients at five observations per patient per minute produces approximately 17 observations per second at steady state. PostgreSQL with the composite index `(encounter_id, observation_code, recorded_at DESC)` handles this volume with headroom. TimescaleDB would be the correct next step at 10,000+ observations/second — it is PostgreSQL with automatic time partitioning, meaning the query layer would not change. The Parquet data lake handles the analytics workload that would otherwise stress the operational database over a 10-year horizon.
|
||
|
||
---
|
||
|
||
## Getting Started
|
||
|
||
### Prerequisites
|
||
|
||
- .NET 8 SDK
|
||
- Docker and Docker Compose
|
||
|
||
### Start Infrastructure
|
||
|
||
```bash
|
||
docker compose up -d
|
||
```
|
||
|
||
All services join the `vigilcare_net` bridge network so containers can reach each other by service name (e.g. Grafana → `http://prometheus:9090`). Connection strings in `appsettings.json` use **host** ports when running `dotnet run` on your machine.
|
||
|
||
| Service | Host Port | Notes |
|
||
|---|---|---|
|
||
| PostgreSQL 16 | 5436 | Database: `vigilcare`, user: `postgres`, password: `password` |
|
||
| Redis 7 | 6382 | No auth |
|
||
| Seq | 5345 | UI at `http://localhost:5345` — login: `admin` / `admin` |
|
||
| Kafka 3.7 | 9092 | KRaft mode, no Zookeeper |
|
||
| Elasticsearch 8.13 | 9200 | Security disabled for development |
|
||
| RabbitMQ 3.13 | 5674 (AMQP), 15674 (UI) | login: `guest` / `guest` |
|
||
| MinIO | 9005 (S3 API), 9006 (console) | login: `minioadmin` / `minioadmin` |
|
||
| Prometheus 2.52 | 9101 | UI at `http://localhost:9101` — scrapes `GET /metrics` on the API |
|
||
| Grafana 10.4 | 3101 | UI at `http://localhost:3101` — login: `admin` / `admin` |
|
||
|
||
**Seq first-run:** `SEQ_FIRSTRUN_ADMINPASSWORD=admin` is set in `docker-compose.yml`. This password is only applied on the very first container start (when the `/data` volume is empty). After initialization, the password is stored in the volume and this env var is ignored.
|
||
|
||
### Docker notes for Linux
|
||
|
||
Prometheus scrapes the API using `host.docker.internal:5270`. On Linux, two things are required:
|
||
|
||
1. In `docker-compose.yml` under `prometheus`:
|
||
```yaml
|
||
extra_hosts:
|
||
- "host.docker.internal:host-gateway"
|
||
```
|
||
2. Run the API bound to all interfaces (not only loopback), so containers can reach it:
|
||
- Use `http://0.0.0.0:5270` (or `ASPNETCORE_URLS=http://0.0.0.0:5270`)
|
||
|
||
Without this, Prometheus may show target errors like:
|
||
- `lookup host.docker.internal ... no such host` (DNS mapping missing), or
|
||
- `dial tcp 172.17.0.1:5270: connect: connection refused` (API bound only to `127.0.0.1`)
|
||
|
||
For full Docker troubleshooting and recovery steps, see:
|
||
- `docs/docker-compose-usage-and-troubleshooting.md`
|
||
|
||
### Install and Run
|
||
|
||
```bash
|
||
cd VigilCareClinicalAPI
|
||
dotnet restore
|
||
dotnet run
|
||
```
|
||
|
||
On startup the application:
|
||
1. Runs EF Core migrations
|
||
2. Seeds two patients (with blood type, allergies, emergency contact), one active inpatient encounter each (with room/bed and admission reason), twelve alert thresholds, and sample observations
|
||
3. Pre-loads all thresholds into Redis
|
||
4. Provisions Kafka topics and Elasticsearch indices
|
||
5. Declares the RabbitMQ exchange and queue topology
|
||
6. Starts all background consumers (outbox relay, ES indexer, sepsis engine, notification workers, data lake writer, reconciliation scheduler)
|
||
7. Starts Prometheus metric collectors (unacknowledged alerts, outbox pending, Kafka consumer lag)
|
||
|
||
Swagger UI is available at `http://localhost:5270/swagger` in Development (API binds to `0.0.0.0:5270` per `launchSettings.json`).
|
||
|
||
### Run Tests
|
||
|
||
```bash
|
||
dotnet test
|
||
```
|
||
|
||
Integration tests use `WebApplicationFactory` with a `Testing` environment and Testcontainers where needed (PostgreSQL, Redis, Kafka, RabbitMQ, Elasticsearch, MinIO). No manual infrastructure setup is required for `dotnet test`.
|
||
|
||
| Test class | Phase | Coverage |
|
||
|---|---|---|
|
||
| `ObservationIngestTests` | 2 | Ingest happy path, critical alert creation, discharged encounter rejection, idempotency |
|
||
| `AlertLifecycleTests` | 2 | Acknowledge, resolve, escalation guard |
|
||
| `SirsDetectorTests` / `SirsEvaluatorTests` | 5 | Redis SIRS state and per-code criterion evaluation |
|
||
| `NotificationPipelineTests` | 6 | RabbitMQ topology, DLQ routing, paging |
|
||
| `ReconciliationTests` | 7 | Three reconciliation checks, deduplication, RabbitMQ publish |
|
||
| `ObservabilityPhase8Tests` | 8 | All eight `/metrics` families, correlation headers, ingest counter increment |
|
||
| `DataLakePhase9Tests` | 9 | Kafka → MinIO Parquet flow and schema checks |
|
||
| `ClinicalDemographicsAndObservationTests` | 10 | Patient clinical fields, encounter enrichment, expanded observation codes, critical glucose alert |
|
||
| `WarningAlertTests` | 11 | WarningEvaluator — warning HR alert, normal/critical skipped, duplicate idempotent |
|
||
| `OrderLifecycleTests` | 11 | Orders API create, list, record result, cancel-resulted 409 |
|
||
| `ValidationTests` | 11 | FluentValidation 400 on empty first name, invalid threshold order, empty order description |
|
||
|
||
### Verification Scripts
|
||
|
||
With the API running (`dotnet run`) and Docker Compose up:
|
||
|
||
```bash
|
||
./scripts/run-phase8-verification.sh # Prometheus target UP, eight metrics, alerts_unacknowledged_gauge live update
|
||
./scripts/run-phase9-verification.sh # DataLakePhase9Tests, Kafka consumer group, MinIO Parquet, DuckDB schema
|
||
./scripts/run-phase10-verification.sh # 12 Redis thresholds, clinical enrichment, ES pipeline, Phase 10 integration tests
|
||
./scripts/run-phase11-verification.sh # Warning alert pipeline, orders API, FluentValidation, Phase 11 integration tests
|
||
```
|
||
|
||
Per-phase test runners (subset of `dotnet test`):
|
||
|
||
```bash
|
||
./scripts/run-api-redis-tests.sh
|
||
./scripts/run-kafka-outbox-tests.sh
|
||
./scripts/run-elasticsearch-analytics-tests.sh
|
||
./scripts/run-sepsis-sirs-tests.sh
|
||
./scripts/run-notification-pipeline-tests.sh
|
||
./scripts/run-reconciliation-tests.sh
|
||
```
|
||
|
||
Phase 9 optional tools (install without sudo):
|
||
|
||
```bash
|
||
# MinIO client — object listing and download
|
||
mkdir -p ~/.local/bin
|
||
curl -fsSL https://dl.min.io/client/mc/release/linux-amd64/mc -o ~/.local/bin/mc
|
||
chmod +x ~/.local/bin/mc
|
||
|
||
# DuckDB — query Parquet files locally
|
||
curl https://install.duckdb.org | sh
|
||
export PATH="$HOME/.duckdb/cli/latest:$HOME/.local/bin:$PATH"
|
||
```
|
||
|
||
See `docs/plans/phase-8-plan.md`, `docs/plans/phase-9-plan.md`, and `docs/plans/phase-10-plan.md` for manual Grafana, Seq, Kafka replay, and DuckDB query examples.
|
||
|
||
---
|
||
|
||
## Prometheus Metrics
|
||
|
||
`GET /metrics` exposes eight application metric families registered in `ClinicalMetrics`. Three background collectors poll PostgreSQL and Kafka every 30 seconds; counters and the ingest histogram are updated inline during request handling.
|
||
|
||
| Metric | Type | Labels | Source |
|
||
|---|---|---|---|
|
||
| `observations_ingested_total` | Counter | `observation_code`, `source` | `ObservationService` on each committed observation |
|
||
| `observation_ingest_duration_seconds` | Histogram | — | `ObservationService` — full ingest transaction to COMMIT |
|
||
| `clinical_alerts_total` | Counter | `alert_type`, `severity` | `ObservationService` (threshold breach), `SirsDetector` (sepsis) |
|
||
| `sirs_detections_total` | Counter | — | `SirsDetector` — only on successful idempotent insert |
|
||
| `escalations_total` | Counter | — | `EscalationWorkerService` on DLQ escalation |
|
||
| `alerts_unacknowledged_gauge` | Gauge | — | `AlertsUnacknowledgedCollector` — open CRITICAL alerts older than 5 minutes |
|
||
| `outbox_pending_events` | Gauge | — | `OutboxPendingCollector` — unprocessed outbox rows |
|
||
| `kafka_consumer_lag` | Gauge | `consumer_group` | `KafkaConsumerLagCollector` — `es-indexer`, `sepsis-engine`, `notification-publisher`, `data-lake-writer` |
|
||
|
||
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`.
|
||
|
||
---
|
||
|
||
## API Reference
|
||
|
||
All endpoints are prefixed `/api/v1`. Responses follow the standard envelope:
|
||
|
||
```json
|
||
{ "success": true, "statusCode": 200, "data": {}, "error": null }
|
||
```
|
||
|
||
Error response:
|
||
|
||
```json
|
||
{
|
||
"success": false,
|
||
"statusCode": 422,
|
||
"data": null,
|
||
"error": {
|
||
"message": "Observation value exceeds plausible range for this code.",
|
||
"code": "OBSERVATION_OUT_OF_PLAUSIBLE_RANGE"
|
||
}
|
||
}
|
||
```
|
||
|
||
### Patients
|
||
|
||
| Method | Path | Description |
|
||
|---|---|---|
|
||
| POST | `/patients` | Register a patient; generates MRN |
|
||
| GET | `/patients` | Paginated list; optional `q` search by name (`ILIKE`) or MRN (exact) |
|
||
| GET | `/patients/{id}` | Patient detail with active encounter summary |
|
||
|
||
**POST body:**
|
||
|
||
| Field | Type | Required | Description |
|
||
|---|---|---|---|
|
||
| `firstName` | string | yes | |
|
||
| `lastName` | string | yes | |
|
||
| `dateOfBirth` | date | yes | |
|
||
| `gender` | string | yes | |
|
||
| `bloodType` | string | no | Clinical notation: `A+`, `O-`, `AB-`, etc. |
|
||
| `allergies` | string | no | Free-text allergy list |
|
||
| `emergencyContactName` | string | no | |
|
||
| `emergencyContactPhone` | string | no | |
|
||
|
||
### Encounters
|
||
|
||
| Method | Path | Description |
|
||
|---|---|---|
|
||
| POST | `/patients/{id}/encounters` | Open an encounter |
|
||
| GET | `/encounters/{id}` | Encounter detail with recent observations and open alerts |
|
||
| PATCH | `/encounters/{id}/status` | Advance encounter status |
|
||
| GET | `/encounters/{id}/timeline` | Merged chronological view: status changes, observations, alerts |
|
||
|
||
**Encounter status machine:**
|
||
|
||
```
|
||
scheduled → active → discharged
|
||
→ cancelled
|
||
```
|
||
|
||
`PATCH /encounters/{id}/status` returns **409** on illegal transitions.
|
||
|
||
**POST body:**
|
||
|
||
| Field | Type | Required | Description |
|
||
|---|---|---|---|
|
||
| `encounterType` | string | yes | `INPATIENT`, `OUTPATIENT`, `EMERGENCY` |
|
||
| `department` | string | yes | |
|
||
| `attendingPhysician` | string | yes | |
|
||
| `roomBed` | string | no | Ward and bed assignment (e.g. `ICU-1A`) |
|
||
| `admissionReason` | string | no | Clinical reason for admission |
|
||
|
||
**PATCH `/encounters/{id}/status` body** — optional `dischargeDiagnosis` when transitioning to `DISCHARGED`.
|
||
|
||
### Alert Thresholds
|
||
|
||
| Method | Path | Description |
|
||
|---|---|---|
|
||
| POST | `/alert-thresholds` | Create a threshold |
|
||
| GET | `/alert-thresholds` | List all thresholds (paginated) |
|
||
| GET | `/alert-thresholds/{id}` | Get a threshold by ID |
|
||
| PUT | `/alert-thresholds/{id}` | Update a threshold; invalidates Redis cache |
|
||
|
||
**Body:**
|
||
|
||
| Field | Type | Required | Description |
|
||
|---|---|---|---|
|
||
| `observationCode` | string | yes | e.g. `HEART_RATE`, `SYSTOLIC_BP`, `AVPU`, `GLUCOSE_MG_DL` |
|
||
| `displayName` | string | yes | Human-readable label |
|
||
| `unit` | string | yes | e.g. `bpm`, `°C`, `mEq/L` |
|
||
| `criticalLow` | decimal | no | |
|
||
| `warningLow` | decimal | no | |
|
||
| `warningHigh` | decimal | no | |
|
||
| `criticalHigh` | decimal | no | |
|
||
|
||
Seeded thresholds (12 codes):
|
||
|
||
| Code | Display | Unit | Critical Low | Warning Low | Warning High | Critical High |
|
||
|---|---|---|---|---|---|---|
|
||
| `HEART_RATE` | Heart Rate | bpm | 30 | 50 | 100 | 150 |
|
||
| `TEMP_C` | Temperature | °C | 35.0 | 36.0 | 38.3 | 40.0 |
|
||
| `POTASSIUM_MEQ_L` | Serum Potassium | mEq/L | 2.5 | 3.5 | 5.0 | 6.5 |
|
||
| `SPO2` | Oxygen Saturation | % | 88 | 92 | — | — |
|
||
| `RESP_RATE` | Respiratory Rate | breaths/min | — | 12 | 20 | 30 |
|
||
| `WBC_K_UL` | White Blood Cell Count | k/µL | 2.0 | 4.0 | 12.0 | 20.0 |
|
||
| `SYSTOLIC_BP` | Systolic Blood Pressure | mmHg | 70 | 90 | 160 | 180 |
|
||
| `DIASTOLIC_BP` | Diastolic Blood Pressure | mmHg | 40 | 60 | 90 | 110 |
|
||
| `LACTATE_MMOL_L` | Serum Lactate | mmol/L | — | — | 2.0 | 4.0 |
|
||
| `AVPU` | AVPU Consciousness | score | — | — | — | 2 |
|
||
| `SUPPLEMENTAL_O2` | Supplemental Oxygen | flag | — | — | — | — |
|
||
| `GLUCOSE_MG_DL` | Blood Glucose | mg/dL | 40 | 70 | 180 | 400 |
|
||
|
||
### Observations
|
||
|
||
| Method | Path | Description |
|
||
|---|---|---|
|
||
| POST | `/encounters/{id}/observations` | Ingest one or more observations (max 10 per call) |
|
||
| GET | `/encounters/{id}/observations` | Cursor-paginated observation history |
|
||
|
||
**POST body:**
|
||
|
||
| Field | Type | Required | Description |
|
||
|---|---|---|---|
|
||
| `observations` | array | yes | One to ten observation objects |
|
||
|
||
**Observation object:**
|
||
|
||
| Field | Type | Required | Description |
|
||
|---|---|---|---|
|
||
| `observationCode` | string | yes | Must match a configured alert threshold |
|
||
| `value` | decimal | yes | Numeric measurement |
|
||
| `unit` | string | yes | Unit of measure |
|
||
| `source` | string | no | `DEVICE` (default), `MANUAL`, `LAB` |
|
||
| `recordedAt` | DateTimeOffset | yes | When the measurement was taken |
|
||
|
||
**Idempotency:** Pass an `Idempotency-Key` header. Same key → original `201` response, no duplicate row.
|
||
|
||
**Ingest transaction sequence:**
|
||
|
||
1. Validate encounter is `active`
|
||
2. Check idempotency key
|
||
3. Validate observation value within plausible range
|
||
4. Insert observation row
|
||
5. Load alert threshold from Redis cache (→ PostgreSQL on miss)
|
||
6. If value breaches `CRITICAL` threshold: insert `clinical_alert` + `outbox_event` (topic: `alert.generated`)
|
||
7. Insert `outbox_event` (topic: `observation.recorded`)
|
||
8. COMMIT
|
||
|
||
**Status codes:**
|
||
|
||
| Code | Meaning |
|
||
|---|---|
|
||
| 201 | Observation(s) recorded |
|
||
| 200 | `Idempotency-Key` matched existing observation |
|
||
| 404 | Encounter not found |
|
||
| 409 | Encounter is not active (discharged or cancelled) |
|
||
| 422 | Value outside plausible range or body invalid |
|
||
|
||
**GET query params:**
|
||
|
||
| Param | Description |
|
||
|---|---|
|
||
| `code` | Filter by observation code |
|
||
| `from` | Inclusive start (DateTimeOffset) |
|
||
| `to` | Inclusive end (DateTimeOffset) |
|
||
| `limit` | Page size (default 20) |
|
||
| `cursor` | Opaque cursor from previous response for next page |
|
||
|
||
Uses cursor pagination on `(recorded_at DESC, id DESC)` — offset pagination would shift results as new observations arrive in a continuously growing table.
|
||
|
||
### Clinical Alerts
|
||
|
||
| Method | Path | Description |
|
||
|---|---|---|
|
||
| GET | `/encounters/{id}/alerts` | Paginated alert list for an encounter |
|
||
| GET | `/alerts` | Global alert list; optional `status`, `severity`, `department` filter |
|
||
| GET | `/alerts/{id}` | Alert detail |
|
||
| POST | `/alerts/{id}/acknowledge` | Acknowledge with clinician ID and optional note |
|
||
| POST | `/alerts/{id}/resolve` | Resolve (must be acknowledged first) |
|
||
|
||
**Alert lifecycle:**
|
||
|
||
```
|
||
open → acknowledged → resolved
|
||
→ escalated (RabbitMQ DLQ after 5 min unacknowledged)
|
||
```
|
||
|
||
**POST `/alerts/{id}/acknowledge` body:**
|
||
|
||
| Field | Type | Required | Description |
|
||
|---|---|---|---|
|
||
| `clinicianId` | string | yes | Clinician identifier |
|
||
| `note` | string | no | Optional acknowledgment note |
|
||
|
||
### Orders
|
||
|
||
| Method | Path | Description |
|
||
|---|---|---|
|
||
| POST | `/encounters/{id}/orders` | Create a clinical order for an active encounter |
|
||
| GET | `/encounters/{id}/orders` | List orders for an encounter; optional `status`, `page`, `pageSize` |
|
||
| GET | `/orders/{id}` | Order detail with encounter |
|
||
| PATCH | `/orders/{id}/status` | Transition order status |
|
||
| PATCH | `/orders/{id}/result` | Record a result; transitions to `Resulted` |
|
||
|
||
**Order status machine:**
|
||
|
||
```
|
||
pending → in_progress → resulted
|
||
→ cancelled
|
||
```
|
||
|
||
`PATCH /orders/{id}/status` and `PATCH /orders/{id}/result` return **409** on illegal transitions (e.g. cancelling a resulted order).
|
||
|
||
**POST body:**
|
||
|
||
| Field | Type | Required | Description |
|
||
|---|---|---|---|
|
||
| `orderType` | string | yes | `Lab`, `Imaging`, `Medication`, `Procedure` |
|
||
| `description` | string | yes | Order description |
|
||
| `orderedBy` | string | yes | Ordering clinician |
|
||
|
||
**PATCH `/orders/{id}/result` body:**
|
||
|
||
| Field | Type | Required | Description |
|
||
|---|---|---|---|
|
||
| `resultSummary` | string | no | Free-text result summary |
|
||
|
||
### Analytics (Elasticsearch)
|
||
|
||
| Method | Path | Description |
|
||
|---|---|---|
|
||
| GET | `/analytics/patients` | Patient/encounter search across MRN, name, department |
|
||
| GET | `/analytics/observations/trend` | Time-series aggregation (hourly avg/min/max) for a specific observation code per encounter |
|
||
| GET | `/analytics/alerts/summary` | Alert volume by department and severity over a time window |
|
||
| GET | `/analytics/population` | Count of patients with a value above or below a threshold in a time window |
|
||
|
||
**GET `/analytics/patients` query params:** `q` (free text), `department`, `status`
|
||
|
||
**GET `/analytics/observations/trend` query params:** `encounterId` (required), `code` (required), `from`, `to`
|
||
|
||
**GET `/analytics/alerts/summary` query params:** `severity`, `from`, `to`, `department`
|
||
|
||
**GET `/analytics/population` query params:** `code` (required), `threshold` (required), `from`, `to`
|
||
|
||
The `population` query uses Elasticsearch's numeric range aggregation engine — no full-text search. Running this against PostgreSQL on the operational database would compete with ingest writes under load.
|
||
|
||
---
|
||
|
||
## Data Models
|
||
|
||
### Patient
|
||
|
||
```
|
||
id Guid PK
|
||
mrn string required, unique — auto-generated on registration (e.g. MRN-000001)
|
||
firstName string required (max 100)
|
||
lastName string required (max 100)
|
||
dateOfBirth Date required
|
||
gender string required (max 10)
|
||
bloodType string? A+ | A- | B+ | B- | AB+ | AB- | O+ | O-
|
||
allergies string? free text
|
||
emergencyContactName string? (max 200)
|
||
emergencyContactPhone string? (max 30)
|
||
status string active | inactive (default: active)
|
||
createdAt DateTimeOffset
|
||
```
|
||
|
||
### Encounter
|
||
|
||
```
|
||
id Guid PK
|
||
patientId Guid FK → Patient
|
||
encounterType string INPATIENT | OUTPATIENT | EMERGENCY
|
||
status string scheduled | active | discharged | cancelled (default: scheduled)
|
||
department string required (max 100)
|
||
attendingPhysician string required (max 200)
|
||
roomBed string? ward/bed assignment (max 50)
|
||
admissionReason string? clinical reason for admission
|
||
dischargeDiagnosis string? set on discharge
|
||
admittedAt DateTimeOffset
|
||
dischargedAt DateTimeOffset?
|
||
createdAt DateTimeOffset
|
||
```
|
||
|
||
Indexes: `(patient_id, admitted_at DESC)`, partial `(status, admitted_at DESC) WHERE status = 'active'`
|
||
|
||
### AlertThreshold
|
||
|
||
```
|
||
id Guid PK
|
||
observationCode string required, unique (max 50)
|
||
displayName string required (max 200)
|
||
unit string required (max 20)
|
||
criticalLow decimal(10,3)?
|
||
warningLow decimal(10,3)?
|
||
warningHigh decimal(10,3)?
|
||
criticalHigh decimal(10,3)?
|
||
createdAt DateTimeOffset
|
||
```
|
||
|
||
### Observation
|
||
|
||
```
|
||
id Guid PK
|
||
encounterId UUID FK → Encounter
|
||
observationCode string required (max 50)
|
||
value decimal(10,3) required
|
||
unit string required (max 20)
|
||
source string DEVICE | MANUAL | LAB (default: DEVICE)
|
||
idempotencyKey string? optional, partial unique index
|
||
recordedAt DateTimeOffset required
|
||
createdAt DateTimeOffset
|
||
```
|
||
|
||
Indexes: partial unique `(idempotency_key) WHERE idempotency_key IS NOT NULL`, `(encounter_id, observation_code, recorded_at DESC)`
|
||
|
||
### ClinicalAlert
|
||
|
||
```
|
||
id Guid PK
|
||
encounterId Guid FK → Encounter
|
||
patientId Guid FK → Patient
|
||
observationId Guid? FK → Observation (null for SIRS alerts)
|
||
alertType string e.g. THRESHOLD_BREACH, SEPSIS_WARNING
|
||
severity string WARNING | CRITICAL
|
||
details text required
|
||
status string open | acknowledged | resolved | escalated (default: open)
|
||
acknowledgedAt DateTimeOffset?
|
||
acknowledgedBy string?
|
||
resolvedAt DateTimeOffset?
|
||
triggeredAt DateTimeOffset
|
||
```
|
||
|
||
Indexes: `(encounter_id, triggered_at DESC)`, `(patient_id, triggered_at DESC)`, partial `(severity, triggered_at DESC) WHERE status = 'open'`
|
||
|
||
### Order
|
||
|
||
```
|
||
id Guid PK
|
||
encounterId Guid FK → Encounter
|
||
orderType string LAB | MEDICATION | IMAGING
|
||
description string required (max 500)
|
||
orderedBy string required (max 200)
|
||
status string pending | in_progress | resulted | cancelled (default: pending)
|
||
orderedAt DateTimeOffset
|
||
resultedAt DateTimeOffset?
|
||
resultSummary string? Free-text result summary (set on record result)
|
||
```
|
||
|
||
Indexes: `(encounter_id, ordered_at DESC)`, partial `(status, ordered_at) WHERE status IN ('pending', 'in_progress')`
|
||
|
||
### OutboxEvent
|
||
|
||
```
|
||
id Guid PK
|
||
topic string required (max 200)
|
||
partitionKey string? — encounterId for per-encounter ordering
|
||
payload JSONB required
|
||
createdAt DateTimeOffset
|
||
processedAt DateTimeOffset?
|
||
```
|
||
|
||
Partial index: `(created_at) WHERE processed_at IS NULL`
|
||
|
||
### ReconciliationAlert
|
||
|
||
```
|
||
id Guid PK
|
||
checkType string UNACKNOWLEDGED_CRITICAL_ALERT | PENDING_ORDER_NO_RESULT | ACTIVE_INPATIENT_NO_OBSERVATION
|
||
encounterId Guid? FK → Encounter
|
||
patientId Guid? FK → Patient
|
||
details text required
|
||
resolvedAt DateTimeOffset?
|
||
createdAt DateTimeOffset
|
||
```
|
||
|
||
---
|
||
|
||
## Elasticsearch Index Shapes
|
||
|
||
### `patient_encounters`
|
||
|
||
```json
|
||
{
|
||
"encounterId": "uuid",
|
||
"patientId": "uuid",
|
||
"mrn": "MRN-000001",
|
||
"patientName": "Jane Smith",
|
||
"department": "ICU",
|
||
"status": "active",
|
||
"attendingPhysician": "Dr. Osei",
|
||
"roomBed": "ICU-4B",
|
||
"admissionReason": "Chest pain, rule out MI",
|
||
"admittedAt": "2025-01-01T08:00:00Z",
|
||
"openAlertCount": 2,
|
||
"lastObservationAt": "2025-01-01T09:45:00Z"
|
||
}
|
||
```
|
||
|
||
### `observations`
|
||
|
||
```json
|
||
{
|
||
"observationId": "uuid",
|
||
"encounterId": "uuid",
|
||
"patientId": "uuid",
|
||
"mrn": "MRN-000001",
|
||
"observationCode": "HEART_RATE",
|
||
"value": 118.0,
|
||
"unit": "bpm",
|
||
"source": "DEVICE",
|
||
"recordedAt": "2025-01-01T09:45:00Z"
|
||
}
|
||
```
|
||
|
||
### `clinical_alerts`
|
||
|
||
```json
|
||
{
|
||
"alertId": "uuid",
|
||
"encounterId": "uuid",
|
||
"patientId": "uuid",
|
||
"department": "ICU",
|
||
"alertType": "THRESHOLD_BREACH",
|
||
"severity": "CRITICAL",
|
||
"status": "open",
|
||
"triggeredAt": "2025-01-01T09:45:00Z"
|
||
}
|
||
```
|
||
|
||
---
|
||
|
||
## RabbitMQ Exchange Topology
|
||
|
||
Exchange: `clinical.notifications.exchange` (direct)
|
||
|
||
| Queue | Purpose | DLQ |
|
||
|---|---|---|
|
||
| `alerts.paging.queue` | Physician paging jobs; prefetch=3 | → `alerts.paging.dlq` on NACK |
|
||
| `alerts.paging.dlq` | Dead-letter queue; `x-message-ttl = 300000ms` | → `alerts.escalation.queue` on TTL expiry |
|
||
| `alerts.escalation.queue` | On-call backup paging | — |
|
||
| `notifications.discharge.queue` | Discharge summary PDF generation + MinIO upload | — |
|
||
| `notifications.reconciliation.queue` | Reconciliation safety findings from scheduled checks | — |
|
||
| `notifications.appointment.queue` | Appointment reminder SMS | — |
|
||
|
||
---
|
||
|
||
## Kafka Topics
|
||
|
||
| Topic | Partition key | Consumer groups |
|
||
|---|---|---|
|
||
| `observation.recorded` | `encounterId` | `es-indexer`, `sepsis-engine`, `warning-evaluator`, `data-lake-writer` |
|
||
| `alert.generated` | `encounterId` | `es-indexer`, `notification-publisher`, `data-lake-writer` |
|
||
| `encounter.status.changed` | `encounterId` | `es-indexer`, `data-lake-writer` |
|
||
|
||
All topics use 6 partitions. `KAFKA_AUTO_CREATE_TOPICS_ENABLE=false` — topics are provisioned explicitly by `KafkaTopicProvisioner` to guarantee correct partition count.
|
||
|
||
---
|
||
|
||
## SIRS Criteria
|
||
|
||
The sepsis engine evaluates four SIRS (Systemic Inflammatory Response Syndrome) criteria per encounter using Redis keys with a 30-minute TTL:
|
||
|
||
| Criterion | Observation Code | Trigger |
|
||
|---|---|---|
|
||
| Fever or hypothermia | `TEMP_C` | > 38.3°C or < 36.0°C |
|
||
| Tachycardia | `HEART_RATE` | > 90 bpm |
|
||
| Tachypnea | `RESP_RATE` | > 20 breaths/min |
|
||
| Abnormal WBC | `WBC_K_UL` | > 12.0 or < 4.0 k/µL |
|
||
|
||
When ≥ 2 criteria are active simultaneously (keys present in Redis) for the same encounter and no open `SEPSIS_WARNING` alert already exists, the engine inserts a `CRITICAL` alert and outbox event. The 30-minute TTL is a clinical parameter — it bounds the window within which simultaneous SIRS criteria must co-occur.
|
||
|
||
---
|
||
|
||
## Elasticsearch Index Replay
|
||
|
||
If the Elasticsearch indices need to be rebuilt (e.g., after a mapping change or data loss):
|
||
|
||
```bash
|
||
# 1. Stop the indexer consumer group (set consumer group to a known-good offset, or reset to beginning)
|
||
# 2. Delete existing indices
|
||
curl -X DELETE http://localhost:9200/patient_encounters
|
||
curl -X DELETE http://localhost:9200/observations
|
||
curl -X DELETE http://localhost:9200/clinical_alerts
|
||
|
||
# 3. Reset the es-indexer consumer group offset to the beginning
|
||
docker exec -it <kafka-container> kafka-consumer-groups.sh \
|
||
--bootstrap-server localhost:9092 \
|
||
--group es-indexer \
|
||
--topic observation.recorded \
|
||
--reset-offsets --to-earliest --execute
|
||
|
||
# 4. Restart the application — EsIndexerService will replay all events from offset 0
|
||
dotnet run
|
||
```
|
||
|
||
The indices rebuild from the full Kafka history. Document count should match PostgreSQL row count when complete. This is only possible because Kafka retains events after consumption.
|
||
|
||
---
|
||
|
||
## Data Lake Replay
|
||
|
||
The MinIO Parquet archive is a pure Kafka projection — rebuildable without touching PostgreSQL. See `docs/plans/phase-9-plan.md` for the full procedure. Summary:
|
||
|
||
```bash
|
||
# Reset data-lake-writer offsets to earliest
|
||
docker compose exec -T kafka /opt/kafka/bin/kafka-consumer-groups.sh \
|
||
--bootstrap-server localhost:9092 \
|
||
--group data-lake-writer \
|
||
--reset-offsets --to-earliest --all-topics --execute
|
||
|
||
# Clear Parquet prefixes in MinIO (mc alias localvc http://localhost:9005 minioadmin minioadmin)
|
||
mc rm --recursive --force localvc/vigilcare/observations/
|
||
mc rm --recursive --force localvc/vigilcare/alerts/
|
||
mc rm --recursive --force localvc/vigilcare/encounters/
|
||
|
||
# Restart the API — DataLakeWriterService replays from offset 0
|
||
dotnet run
|
||
```
|
||
|
||
Verify with `./scripts/run-phase9-verification.sh`.
|
||
|
||
---
|
||
|
||
## Pagination
|
||
|
||
List endpoints use offset pagination:
|
||
|
||
| Param | Default | Description |
|
||
|---|---|---|
|
||
| `page` | 1 | Page number (1-based) |
|
||
| `pageSize` | 20 | Items per page (max 100) |
|
||
|
||
Response shape:
|
||
|
||
```json
|
||
{
|
||
"items": [],
|
||
"page": 1,
|
||
"pageSize": 20,
|
||
"totalCount": 42,
|
||
"totalPages": 3
|
||
}
|
||
```
|
||
|
||
Observation history uses cursor pagination on `(recorded_at DESC, id DESC)`. Offset pagination would shift results as new observations arrive continuously. The cursor is opaque and returned in the response; pass it as `?cursor=` on the next request.
|
||
|
||
---
|
||
|
||
## Implemented Phases
|
||
|
||
Eleven phases from the project roadmap are implemented. Phases 1–10 and Step 4 of Phase 11 are covered by integration tests (`dotnet test` — 64 passing). Phase 11 Step 5 manual verification (docker compose end-to-end) is documented in `docs/plans/phase-11-plan.md`.
|
||
|
||
| Phase | Feature | Status |
|
||
|---|---|---|
|
||
| 1 | Schema, EF Core migrations, patient/encounter CRUD, alert threshold CRUD, encounter status machine, Redis threshold pre-load, seed data | Done |
|
||
| 2 | Observation ingest — idempotency, plausibility validation, synchronous critical alert creation, outbox event, cursor-paginated history; alert lifecycle (acknowledge, resolve); integration tests | Done |
|
||
| 3 | Outbox relay (`IHostedService`, 500ms poll); Kafka topics with 6 partitions; `encounterId` partition key; relay survives Kafka restart | Done |
|
||
| 4 | Elasticsearch CQRS projection (`EsIndexerService`); patient search; observation trend; alert summary; population aggregation; replay procedure | Done |
|
||
| 5 | Sepsis detection engine (`SepsisEngineService`); Redis SIRS state with 30-min TTL; idempotent alert creation; integration tests | Done |
|
||
| 6 | RabbitMQ exchange and queue topology; `NotificationPublisherService`; `PagingWorkerService`; DLQ escalation (`EscalationWorkerService`); discharge summary (`DischargeSummaryWorkerService` → MinIO); integration tests | Done |
|
||
| 7 | Reconciliation scheduler — unacknowledged critical alerts, stale pending orders, disconnected monitors; `reconciliation_alerts` table; RabbitMQ publish; integration tests | Done |
|
||
| 8 | Prometheus metrics (`GET /metrics`); eight metric families and three collectors; Grafana clinical dashboard; `ObservabilityPhase8Tests`; `run-phase8-verification.sh` | Done |
|
||
| 9 | Data lake writer — `data-lake-writer` consumer group; date-partitioned Parquet flush to MinIO; `DataLakePhase9Tests`; `run-phase9-verification.sh`; design doc in `docs/decisions/data-lake-design.md` | Done |
|
||
| 10 | Clinical data model expansion — `BloodType`, patient allergies/emergency contact, encounter room/bed/admission/discharge fields; five new observation codes (`SYSTOLIC_BP`, `DIASTOLIC_BP`, `LACTATE_MMOL_L`, `AVPU`, `SUPPLEMENTAL_O2`); `GLUCOSE_MG_DL` threshold fix; 12 seeded thresholds; `ClinicalDemographicsAndObservationTests`; `run-phase10-verification.sh` | Done |
|
||
| 11 | Warning alert consumer (`WarningAlertService` / `warning-evaluator`); 10 `Warning*` alert types; Orders API (`OrdersController`, `OrderService`); FluentValidation on all request DTOs; `WarningAlertTests`, `OrderLifecycleTests`, `ValidationTests`; `run-phase11-verification.sh` | Done (Step 5 E2E verification via script) |
|
||
|
||
**Optional follow-up:** execute and document the Kafka replay demonstration for the data lake (reset `data-lake-writer` offsets, clear MinIO prefixes, restart API, confirm Parquet rebuild). See `docs/plans/phase-9-plan.md` § Replay demonstration.
|