96 KiB
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 and NEWS2 scoring, and clinician notification with automatic escalation.
Implementation status: Nineteen 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, clinical data model expansion (patient demographics, encounter enrichment, 12 observation codes), warning alerts and orders, the NEWS2 composite scoring engine, trend detection with alert suppression, qSOFA scoring with sepsis bundle compliance tracking, medication administration with alert correlation annotations, the console replay simulator, the Vue 3 ward dashboard (virtual ward, patient detail, alert center, vital sign charts, NEWS2 history, replay controls, alert reasoning), and clinician feedback mode (structured alert ratings, feedback summary, JSON/CSV export). See Implemented Phases for the full breakdown. Guides: dashboard-guide.md (technical), clinical-testing-guide.md (doctors & nurses).
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. Composite scoring engines (NEWS2, SIRS/sepsis, qSOFA) aggregate multiple vitals into acuity scores. When sepsis is suspected (via SIRS or qSOFA), a four-element treatment bundle is automatically created with a one-hour compliance deadline. 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, scoring engines, 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, SIRS detection, qSOFA score, or NEWS2 composite score
├── OutboxEvent → Kafka → RabbitMQ → clinician page → escalation
└── SepsisBundle auto-created on SIRS/qSOFA alert → four treatment orders → compliance tracking
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, when the sepsis engine detects two or more concurrent SIRS criteria, when the qSOFA engine detects two or more organ-dysfunction criteria, or when the NEWS2 engine computes a medium/high-risk composite score (or a single-parameter score of 3). 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. SIRS and qSOFA alerts additionally trigger automatic sepsis bundle creation.
SepsisBundle
A SepsisBundle is created automatically when either the SIRS detector or the qSOFA scorer generates a sepsis-related alert. Each bundle contains four mandatory treatment elements (blood cultures, serum lactate, broad-spectrum antibiotics, IV fluid resuscitation) mapped to clinical orders that are created simultaneously. A one-hour compliance deadline is set from the recognition time. As linked orders are resulted, bundle elements transition to COMPLETED; when all four are done, the bundle is marked COMPLIANT (within deadline) or NON_COMPLIANT (past deadline). Only one in-progress bundle can exist per encounter at a time.
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/observationsaccepts single or small batch (up to 10); idempotency viaIdempotency-Keyheader (partial unique index); plausibility validation per observation code; synchronous critical alert creation within the ingest transaction; warning-range breaches evaluated asynchronously byWarningAlertService(Kafka consumer groupwarning-evaluator); outbox event written in the same commit; cursor-paginated history on(encounter_id, observation_code, recorded_at DESC) - Warning Threshold Alerts —
WarningEvaluatorreads thresholds from Redis; createsWARNING-severity alerts for values abovewarningHighor belowwarningLowthat are not also critical breaches; idempotentINSERT WHERE NOT EXISTSper encounter and alert type while status isOPENorACKNOWLEDGED; warning alerts are indexed in Elasticsearch but not published to the RabbitMQ paging queue - Clinical Order Management —
POST /encounters/:id/orderscreate;GET /encounters/:id/orderslist with optional status filter;GET /orders/:iddetail;PATCH /orders/:id/statusstatus transitions;PATCH /orders/:id/resultrecord result and transition toResulted; status machine enforcesPending → InProgress → Resultedand terminalCancelled - 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 —
IHostedServicepolling every 500ms; reads unprocessed outbox rows, publishes to Kafka, marks processed; partitioned byencounterIdfor 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 byKafkaTopicProvisioner - Elasticsearch CQRS Projection —
EsIndexerServiceconsumer group upsertspatient_encountersdocuments, appends to theobservationsindex, incrementsopenAlertCounton alert events, stampsnews2Score/news2RiskLevelwhen a NEWS2 alert is generated, and projectssepsisBundleStatus/sepsisBundleElementsCompleted/sepsisBundleDeadlineAtfrom 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 Early Warning Engine —
SepsisEngineServiceKafka consumer evaluates both SIRS and qSOFA criteria per encounter using Redis keys with a 30-minute TTL sliding window; SIRS evaluates temperature, heart rate, respiratory rate, and WBC; qSOFA evaluates respiratory rate ≥ 22, systolic BP ≤ 100, and altered mentation (AVPU ≥ 1); on ≥ 2 active criteria in either system, inserts aSEPSIS_WARNINGorQSOFA_WARNING / CRITICALalert idempotently (INSERT WHERE NOT EXISTS); both alert types trigger automatic sepsis bundle creation viaSepsisAlertHandler - Sepsis Bundle Compliance —
SepsisBundleServicecreates a four-element treatment bundle (blood cultures, serum lactate, broad-spectrum antibiotics, IV fluid resuscitation) when a SIRS or qSOFA alert fires; each element maps to an auto-created clinical order (orderedBy: sepsis-bundle-engine); one-hour compliance deadline from recognition;OrderService.RecordResultcalls back toOnOrderResultedAsyncto mark elements complete; final element completion sets bundle toCOMPLIANTorNON_COMPLIANT;SepsisBundleMonitorServicescans every 5 minutes for overdue in-progress bundles past their deadline and marks themNON_COMPLIANT; idempotent — only one in-progress bundle per encounter;GET /encounters/:id/sepsis-bundle/currentandGET /sepsis-bundles/:idexpose bundle state; Kafka topicssepsis.bundle.created/sepsis.bundle.updated; Prometheusqsofa_detections_totalandsepsis_bundle_compliance_total - NEWS2 Composite Scoring Engine —
News2ScoringServiceKafka consumer (news2-scoring) evaluates seven vital parameters per encounter (RESP_RATE,SPO2,SYSTOLIC_BP,HEART_RATE,AVPU,TEMP_C,SUPPLEMENTAL_O2) using Redis keys with a 4-hour TTL; when all seven are present, computes the official NEWS2 aggregate score, persists tonews2_scores, and createsNEWS2_WARNING(score 5–6 or single param = 3) orNEWS2_EMERGENCY(score ≥ 7) alerts idempotently;GET /encounters/:id/news2/currentand/historyexpose score history; Prometheusnews2_scores_totalandnews2_scoring_duration_seconds - Trend Detection Engine —
TrendAnalyzerServiceKafka consumer (trend-analyzer) tracks rate-of-change for five vital parameters (HEART_RATE,RESP_RATE,SYSTOLIC_BP,TEMP_C,SPO2) using Redis sliding-window history; when velocity exceeds configured thresholds (e.g. 72→95 bpm in 30 min), creates aRAPID_DETERIORATIONalert even if the current value is below warning thresholds; Prometheustrend_alerts_totalandtrend_analysis_duration_seconds - Alert Suppression Windows — acknowledging a suppressible alert (
WARNING_*,NEWS2_WARNING) sets a Redis keysuppress:{encounterId}:{alertType}with a configurable TTL (default 30 min fromAlertSuppressionconfig; optional per-code override viaalert_thresholds.suppression_window_minutes);WarningEvaluatorandNews2Detectorcheck suppression before creating new warning alerts; critical alerts (CRITICAL_*,NEWS2_EMERGENCY,SEPSIS_WARNING,RAPID_DETERIORATION) are never suppressed; observations and NEWS2 scores continue to persist during suppression; Prometheusalert_suppressions_total - Medication Administration —
POST /encounters/:id/medicationsrecords drug administrations (name, dose, route, timestamp, administered-by);GET /encounters/:id/medicationslists with optionalsincefilter;GET /medications/:iddetail; active-encounter guard; FluentValidation on request DTOs - Medication Correlation Annotations —
MedicationCorrelationHelperappends medication context to warning and NEWS2 alert details when a mapped drug was administered within the correlation window (default 90 min); drug-to-vital mappings inMedicationCorrelationconfig (appsettings.json); annotates rather than suppresses — alerts still fire; sepsis, trend, and critical sync-path alerts are never annotated; design rationale indocs/decisions/medication-correlation-design.md - Ward Dashboard APIs —
GET /encountersreturns paginatedWardEncounterSummaryrows (patient name/MRN, room/bed, department, status, latest NEWS2 score, live qSOFA criteria count from Redis, sepsis bundle status, open alert count); filterable bystatusanddepartment;GET /encounters/:id/qsofa/currentexposes Redis-backed qSOFA state; CORS policyDashboardallows configured origins (defaulthttp://localhost:5173) - Ward Dashboard Frontend — Vue 3 SPA (
vigilcare-dashboard/) with virtual ward table (NEWS2-sorted), patient detail (vitals, scores, alerts, orders, sepsis bundle), alert center (global acknowledge/resolve), vital sign trend charts with local replay scrubbing, NEWS2 history chart, alert reasoning with optional medication context, and clinician feedback on every alert; polls API every 5–10 s; guides indocs/dashboard-guide.mdanddocs/clinical-testing-guide.md - 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 0instant,60= 60× faster); commands:replay,replay-all,validate,dry-run; optional--pollshows alerts, NEWS2, and sepsis bundle state during replay; eight sample scenarios inVigilCare.Simulator/Scenarios/List/; user guide indocs/simulator-guide.md - RabbitMQ Notification Workers —
NotificationPublisherServicereadsalert.generatedfrom Kafka and publishes paging jobs toalerts.paging.queue;PagingWorkerServicesends the page and waits for acknowledgment; if no ack arrives before timeout it NACKs toalerts.paging.dlqwithx-message-ttl = 300000ms; if the host is stopping, in-flight paging messages are NACKed withrequeue=trueso they are retried after restart and do not false-escalate;EscalationWorkerServicepages the on-call backup and sets alert status toescalated;DischargeSummaryWorkerServicereadsencounter.status.changed, generates a discharge summary, and stores it in MinIO under/discharge-summaries/{encounterId}/summary.pdf - Data Lake Writer —
DataLakeWriterService(consumer groupdata-lake-writer) buffersobservation.recorded,alert.generated, andencounter.status.changedevents, 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_partitionandkafka_offsetcolumns 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_alertsrow and publishes to RabbitMQ - Standard Envelope — all responses use a consistent
{ success, statusCode, data, error }wrapper; validation errors use the same shape;ApiBehaviorOptionsoverridden so model validation also produces the standard envelope with field-leveldetails - 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,patientIdon alert paths; Seq sink (http://localhost:5345); Prometheus (http://localhost:9101) scrapesGET /metrics; fifteen application metric families viaClinicalMetricsand three background collectors (AlertsUnacknowledgedCollector,OutboxPendingCollector,KafkaConsumerLagCollector); Grafana clinical dashboard (http://localhost:3101, admin/admin) withalerts_unacknowledged_gaugeas the primary safety panel; per-request correlation IDs in request logs andX-Correlation-Idresponse 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, NEWS2 parameter state, trend history, alert suppression keys)
└── 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 + qSOFA state → PostgreSQL alert → SepsisAlertHandler → SepsisBundleService (consumer group: sepsis-engine)
WarningAlertService → Kafka → WarningEvaluator (+ MedicationCorrelationHelper) → PostgreSQL WARNING alert (consumer group: warning-evaluator)
News2ScoringService → Kafka → News2Detector (+ MedicationCorrelationHelper) → Redis NEWS2 state → PostgreSQL score + alert (consumer group: news2-scoring)
TrendAnalyzerService → Kafka → TrendDetector → Redis trend history → PostgreSQL RAPID_DETERIORATION alert (consumer group: trend-analyzer)
AlertSuppressionService → Redis suppress:{enc}:{type} keys set on acknowledge; read by WarningEvaluator + News2Detector
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
SepsisBundleMonitorService → polls overdue bundles every 5 min → marks NON_COMPLIANT
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, qSOFA, NEWS2, trend, suppression, bundle compliance, 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 & NEWS2 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 |
| Ward dashboard | Vue 3 + Vite + Pinia + Tailwind CSS v4 + Chart.js (vigilcare-dashboard/) |
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 list (ward summary), get, status PATCH, timeline
│ ├── MedicationsController.cs # Medication administration create, list, get
│ ├── QsofaController.cs # Current qSOFA criteria count (Redis-backed)
│ ├── 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
│ ├── News2Controller.cs # Current NEWS2 score and cursor-paginated history
│ ├── SepsisBundlesController.cs # Current bundle per encounter, bundle detail by ID
│ └── 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
│ │ ├── News2Score.cs # Composite score with seven component scores + risk level
│ │ ├── OutboxEvent.cs # topic + payload JSONB + processed_at
│ │ ├── ReconciliationAlert.cs
│ │ ├── SepsisBundle.cs # Four-element treatment bundle with 1-hour compliance deadline
│ │ ├── SepsisBundleElement.cs # Individual bundle element linked to a clinical order
│ │ └── MedicationAdministration.cs # Drug administration record per encounter
│ └── 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, qSOFA, warning*, NEWS2_*, …
│ ├── BloodType.cs # A+, O-, AB-, … with ToDbString/FromDbString
│ ├── ObservationSource.cs # Device, Manual, Lab
│ ├── SepsisBundleComplianceStatus.cs # InProgress, Compliant, NonCompliant
│ ├── SepsisBundleElementStatus.cs # Pending, Completed
│ ├── SepsisBundleElementType.cs # BloodCultures, SerumLactate, BroadSpectrumAntibiotics, IvFluidResuscitation
│ └── 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 (sets suppression), resolve, list
│ ├── AlertSuppressionService.cs # Redis suppress:{enc}:{type} TTL keys
│ ├── OrderService.cs # Order lifecycle; status machine; calls SepsisBundleService.OnOrderResultedAsync on result
│ ├── News2Service.cs # Current score + cursor-paginated history from PostgreSQL
│ ├── SepsisBundleService.cs # Bundle creation, element completion, compliance evaluation
│ ├── MedicationService.cs # Medication CRUD; GetRecentForEncounterAsync for correlation
│ ├── QsofaService.cs # Redis-backed qSOFA criteria count for API/dashboard
│ ├── WarningEvaluator.cs # Warning-range evaluation; suppression + medication annotation; idempotent INSERT
│ ├── AnalyticsService.cs # Elasticsearch query wrappers
│ └── PlausibilityValidator.cs # Per-code numeric range guard
├── Trend/
│ ├── TrendCalculator.cs # Pure static rate-of-change logic
│ └── TrendDetector.cs # Redis history + RAPID_DETERIORATION alert creation
├── Medication/
│ └── MedicationCorrelationHelper.cs # Appends drug context to warning/NEWS2 alert details
├── Validators/ # FluentValidation — RegisterPatient, OpenEncounter, IngestObservation, CreateMedicationAdministration, …
├── Observability/
│ └── Metrics/
│ └── ClinicalMetrics.cs # Fifteen Prometheus metric families (counters, histograms, 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
│ ├── News2ScoringService.cs # consumer group: news2-scoring; observation.recorded → NEWS2 score + alert
│ ├── TrendAnalyzerService.cs # consumer group: trend-analyzer; observation.recorded → RAPID_DETERIORATION alert
│ ├── SepsisBundleMonitorService.cs # Polls every 5 min; marks overdue in-progress bundles NON_COMPLIANT
│ ├── 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
│ ├── MedicationCorrelationOptions.cs # Drug-vital mappings + correlation window
│ └── DashboardOptions.cs # CORS origins for ward dashboard frontend
├── Sepsis/
│ ├── SirsDetector.cs # Redis SIRS state management (SET/DEL/MGET)
│ ├── SirsEvaluator.cs # Per-code criterion evaluation
│ ├── QsofaCalculator.cs # Pure static qSOFA scoring (3 criteria, no I/O)
│ ├── QsofaDetector.cs # Redis qSOFA state, alert creation, SepsisAlertHandler callback
│ └── SepsisAlertHandler.cs # Bridges SIRS/qSOFA alert creation → SepsisBundleService
├── News2/
│ ├── News2Calculator.cs # Pure static NEWS2 scoring tables (no I/O)
│ └── News2Detector.cs # Redis parameter state, score persistence, alert creation
├── 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; graceful shutdown flush
│ └── 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
│ ├── Encounter/WardEncounterSummary.cs # Denormalized row for ward encounter list
│ ├── Medication/CreateMedicationAdministrationRequest.cs
│ ├── Qsofa/QsofaCurrentResponse.cs
│ └── Sepsis/QsofaResult.cs, QsofaOutcome.cs # qSOFA detector result and outcome enum
├── 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
├── News2CalculatorTests.cs # Boundary tests for all seven NEWS2 scoring tables
├── News2DetectorTests.cs # NEWS2 detector — score tiers, alerts, idempotency, incomplete set
├── TrendCalculatorTests.cs # Pure unit tests — rate-of-change, threshold direction, describe
├── TrendDetectorTests.cs # Trend detector — rapid climb, stable, idempotent, non-trend code
├── AlertSuppressionTests.cs # Suppression on acknowledge, read-side skip, TTL expiry
├── QsofaCalculatorTests.cs # Boundary tests for three qSOFA criteria
├── QsofaDetectorTests.cs # qSOFA detector — two-criteria alert, normalization, idempotency
├── SepsisBundleTests.cs # Bundle creation from SIRS/qSOFA, element completion, compliance outcomes
├── MedicationServiceTests.cs # Medication CRUD, discharged encounter rejection, pagination
├── MedicationCorrelationTests.cs # End-to-end warning/NEWS2 annotation with medication context
├── MedicationValidationTests.cs # FluentValidation 400 on invalid medication requests
├── EncountersListTests.cs # Ward encounter list filters and summary fields
└── QsofaCurrentTests.cs # qSOFA current API — Redis state, criteria breakdown
VigilCare.Simulator/ # Phase 16 — console replay simulator (HTTP-only, no direct DB/Kafka)
├── Program.cs # CLI: replay, replay-all, validate, dry-run
├── Commands/ # System.CommandLine command handlers
├── Client/VigilCareApiClient.cs # Typed HTTP client for all API endpoints
├── Engine/ReplayEngine.cs # Scenario replay with speed multiplier + event logging
├── Polling/ApiPoller.cs # Optional post-event alert/score/bundle polling
├── Scenarios/ # schema.json, ScenarioLoader, ScenarioValidator
└── Scenarios/List/ # Eight sample scenarios (sepsis, NEWS2, stable, medication, …)
vigilcare-dashboard/ # Phases 17–19 — Vue 3 ward dashboard SPA
├── src/
│ ├── api/ # HTTP client, encounters, clinical, alerts, normalize
│ ├── components/ # charts, replay, alerts, feedback, patient, ward, layout, ui
│ ├── composables/ # useChartData, useReplayControls, usePolling, useFeedback, chartFormat
│ ├── stores/ # Pinia — ward, alerts, settings, feedback (localStorage)
│ ├── views/ # WardDashboard, PatientDetail, AlertCenter, FeedbackSummary
│ └── __tests__/ # Vitest — 41 tests (store, feedback, replay, charts, alerts, ward)
├── vite.config.js
└── README.md # Dev quick start → docs/dashboard-guide.md
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
├── run-phase12-verification.sh # Phase 12 — NEWS2 end-to-end pipeline, API, ES, Prometheus, integration tests
├── run-phase13-verification.sh # Phase 13 — trend detection, alert suppression, consumer lag, integration tests
├── run-phase14-verification.sh # Phase 14 — qSOFA, sepsis bundle compliance, integration tests
└── run-phase15-verification.sh # Phase 15 — medication administration + correlation annotations
docs/
├── plans/ # Phase implementation and verification guides
├── clinical-testing-guide.md # Doctor/nurse guide — alert review & feedback sessions
├── dashboard-guide.md # VigilCare Dashboard user guide (ward, patient detail, charts)
├── patient-encounter-api-lifecycle.md # Full API walkthrough: registration → active stay → discharge
├── simulator-guide.md # VigilCare.Simulator user guide
├── decisions/
│ ├── data-lake-design.md # Parquet vs JSON, partitioning, replay rationale
│ ├── sepsis-engine-design.md # SIRS sliding window and idempotent alert design
│ └── medication-correlation-design.md # Drug-vital mapping and annotation rationale
├── 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 Five Distinct Purposes
Redis serves five independent roles with different semantics:
-
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.
-
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. -
qSOFA sliding window:
SET qsofa:{encounterId}:{code} EX 1800. Same 30-minute TTL as SIRS, but tracking three organ-dysfunction criteria instead of four inflammatory markers. When a value normalizes, the key is explicitly deleted rather than waiting for TTL — a systolic BP that recovers from 90 to 120 should immediately reduce the qSOFA count. -
NEWS2 parameter state:
SET news2:{encounterId}:{code}with a 4-hour TTL. Each of the seven NEWS2 parameters is scored individually and stored in Redis;MGETacross all seven keys determines completeness. An incomplete set (fewer than seven present keys) does not produce a score — expired parameters must be re-recorded before the aggregate is computed. -
Alert suppression windows:
SET suppress:{encounterId}:{alertType}with a configurable TTL (default 30 min). Set on acknowledge of suppressible alerts; checked byWarningEvaluatorandNews2Detectorbefore creating new warning alerts. Critical alerts are never suppressed.
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.
Dual-Path Sepsis Detection (SIRS + qSOFA)
Both SIRS and qSOFA run within the same SepsisEngineService Kafka consumer against every observation.recorded event. SIRS detects systemic inflammatory response (temperature, heart rate, respiratory rate, WBC — Sepsis-2 criteria). qSOFA detects organ dysfunction (respiratory rate, systolic BP, altered mentation — Sepsis-3 consensus). They fire independently because they measure different clinical dimensions of the same disease process. A patient can trigger SIRS without qSOFA (infection with inflammation but no organ failure) or qSOFA without SIRS (organ dysfunction without classic inflammatory markers). Both paths feed into the same sepsis bundle workflow — the bundle is idempotent, so the second alert for the same encounter does not create a duplicate bundle. This dual-path design reflects current clinical practice where neither scoring system alone captures all sepsis presentations.
Getting Started
Prerequisites
- .NET 8 SDK
- Docker and Docker Compose
Start Infrastructure
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:
- In
docker-compose.ymlunderprometheus:extra_hosts: - "host.docker.internal:host-gateway" - Run the API bound to all interfaces (not only loopback), so containers can reach it:
- Use
http://0.0.0.0:5270(orASPNETCORE_URLS=http://0.0.0.0:5270)
- Use
Without this, Prometheus may show target errors like:
lookup host.docker.internal ... no such host(DNS mapping missing), ordial tcp 172.17.0.1:5270: connect: connection refused(API bound only to127.0.0.1)
For full Docker troubleshooting and recovery steps, see:
docs/docker-compose-usage-and-troubleshooting.md
Install and Run
cd VigilCareClinicalAPI
dotnet restore
dotnet run
On startup the application:
- Runs EF Core migrations
- 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
- Pre-loads all thresholds into Redis
- Provisions Kafka topics and Elasticsearch indices
- Declares the RabbitMQ exchange and queue topology
- Starts all background consumers (outbox relay, ES indexer, sepsis engine with SIRS + qSOFA, warning evaluator, NEWS2 scoring, trend analyzer, notification workers, data lake writer, reconciliation scheduler, sepsis bundle monitor)
- 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 the Simulator
With the API running, replay a scenario from the repository root:
dotnet run --project VigilCare.Simulator -- replay \
VigilCare.Simulator/Scenarios/List/uti-sepsis-elderly-01.json \
--speed 60 --poll
Other commands: validate <file>, dry-run <file>, replay-all <directory>. See docs/simulator-guide.md for the full user guide.
Run the Dashboard
With the API running, start the Vue frontend:
cd vigilcare-dashboard
npm install
npm run dev
Open http://localhost:5173 — Virtual Ward lists active patients sorted by NEWS2 score. Click a row for patient detail (vitals, alerts, charts, replay scrubbing, alert reasoning). Use Alert Center for hospital-wide triage. After reviewing alerts, rate them with the six feedback buttons and export results from Feedback Summary (/feedback).
Replay a simulator scenario in another terminal to watch charts and alerts populate in real time. Run dashboard tests with cd vigilcare-dashboard && npm test. See docs/dashboard-guide.md for technical documentation and docs/clinical-testing-guide.md for structured clinician evaluation sessions.
Run Tests
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 ten /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 |
News2CalculatorTests |
12 | Boundary tests for all seven NEWS2 scoring tables and risk-level determination |
News2DetectorTests |
12 | NEWS2 detector — score tiers, alert creation, incomplete parameters, idempotency |
TrendCalculatorTests |
13 | Pure unit tests — delta/time rate, SPO2/BP decline direction, describe formatting |
TrendDetectorTests |
13 | Trend detector — rapid HR climb alert, stable high HR, idempotency, non-trend code |
AlertSuppressionTests |
13 | Acknowledge sets Redis key, suppressed warning skipped, critical/NEWS2 emergency never suppressed, TTL expiry |
QsofaCalculatorTests |
14 | Boundary tests for three qSOFA criteria (RESP_RATE, SYSTOLIC_BP, AVPU) |
QsofaDetectorTests |
14 | qSOFA detector — two-criteria alert, normalization key delete, idempotent duplicate, non-qSOFA code ignored |
SepsisBundleTests |
14 | Bundle creation from SIRS/qSOFA, four auto-orders, element completion, compliant/non-compliant outcomes, monitor marks overdue bundles, idempotency |
MedicationServiceTests |
15 | Medication create/list on active encounter, discharged encounter 409, pagination, since filter |
MedicationCorrelationTests |
15 | Warning and NEWS2 alert details annotated when correlated drug administered |
MedicationValidationTests |
15 | FluentValidation 400 on empty drug name, zero dose, future administeredAt |
EncountersListTests |
— | Ward encounter list — status/department filters, summary fields |
QsofaCurrentTests |
— | GET /qsofa/current — criteria count and breakdown from Redis |
Verification Scripts
With the API running (dotnet run) and Docker Compose up:
./scripts/run-phase8-verification.sh # Prometheus target UP, ten 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
./scripts/run-phase12-verification.sh # NEWS2 end-to-end pipeline, API, Elasticsearch, Prometheus, Phase 12 integration tests
./scripts/run-phase13-verification.sh # Trend detection, alert suppression, consumer lag, Phase 13 integration tests
./scripts/run-phase14-verification.sh # qSOFA, sepsis bundle compliance, Phase 14 integration tests
./scripts/run-phase15-verification.sh # Medication administration + correlation annotation pipeline
Phase 13 unit/integration tests only:
dotnet test --filter "FullyQualifiedName~Trend|FullyQualifiedName~Suppression"
Phase 15 unit/integration tests only:
dotnet test --filter "FullyQualifiedName~Medication"
Per-phase test runners (subset of dotnet test):
./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):
# 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 through docs/plans/phase-12-plan.md for manual Grafana, Seq, Kafka replay, and DuckDB query examples.
Prometheus Metrics
GET /metrics exposes fifteen application metric families registered in ClinicalMetrics. Three background collectors poll PostgreSQL and Kafka every 30 seconds; counters and histograms are updated inline during request handling and background processing.
| 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, SirsDetector, QsofaDetector, News2Detector, TrendDetector, WarningEvaluator |
sirs_detections_total |
Counter | — | SirsDetector — only on successful idempotent insert |
qsofa_detections_total |
Counter | — | QsofaDetector — only on successful idempotent insert |
sepsis_bundle_compliance_total |
Counter | status |
SepsisBundleService — on bundle completion (COMPLIANT, NON_COMPLIANT) |
news2_scores_total |
Counter | risk_level |
News2Detector — on each persisted score (LOW, MEDIUM, HIGH, …) |
news2_scoring_duration_seconds |
Histogram | — | News2Detector — Redis update through score persistence |
trend_alerts_total |
Counter | observation_code |
TrendDetector — on each RAPID_DETERIORATION alert created |
trend_analysis_duration_seconds |
Histogram | — | TrendDetector — per-observation trend evaluation |
alert_suppressions_total |
Counter | alert_type |
AlertSuppressionService — on each suppression window set after acknowledge |
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:
{ "success": true, "statusCode": 200, "data": {}, "error": null }
Error response:
{
"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 |
|---|---|---|
| GET | /encounters |
Paginated ward list with clinical summaries; optional status, department filters |
| 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 |
| GET | /encounters/{id}/qsofa/current |
Current qSOFA active criteria count (0–3) from Redis |
GET /encounters query params: status (DB literal, e.g. ACTIVE), department (DB literal, e.g. ICU), page, pageSize
Ward summary fields: encounterId, patientId, mrn, firstName, lastName, roomBed, department, status, news2Score, news2RiskLevel, qsofaScore, sepsisActive, sepsisBundleStatus, openAlertCount
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:
- Validate encounter is
active - Check idempotency key
- Validate observation value within plausible range
- Insert observation row
- Load alert threshold from Redis cache (→ PostgreSQL on miss)
- If value breaches
CRITICALthreshold: insertclinical_alert+outbox_event(topic:alert.generated) - Insert
outbox_event(topic:observation.recorded) - 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.
NEWS2 (National Early Warning Score 2)
| Method | Path | Description |
|---|---|---|
| GET | /encounters/{id}/news2/current |
Latest NEWS2 score for an encounter (404 if none computed) |
| GET | /encounters/{id}/news2/history |
Cursor-paginated score history |
GET /news2/current response includes totalScore, riskLevel (LOW, LOW_MEDIUM, MEDIUM, HIGH), all seven component scores (respRateScore, spo2Score, …), hasSingleParamThree, and calculatedAt.
GET /news2/history query params: limit (default 20), cursor (opaque token from previous response).
Scores are computed asynchronously by News2ScoringService after observations are ingested — allow a few seconds for the Kafka consumer to process all seven parameters before querying.
Sepsis Bundles
| Method | Path | Description |
|---|---|---|
| GET | /encounters/{id}/sepsis-bundle/current |
Current (most recent) sepsis bundle for an encounter (404 if none) |
| GET | /sepsis-bundles/{id} |
Bundle detail with all elements and linked orders |
GET /sepsis-bundle/current response includes encounterId, triggeringAlertId, triggeringAlertType (QSOFA_WARNING or SEPSIS_WARNING), recognizedAt, deadlineAt, complianceStatus (IN_PROGRESS, COMPLIANT, NON_COMPLIANT), completedAt, and elements[] — each with elementType, status, orderId, and completedAt.
Bundles are created automatically by SepsisAlertHandler when a SIRS or qSOFA alert fires. Four clinical orders (blood cultures, serum lactate, broad-spectrum antibiotics, IV fluid resuscitation) are auto-created with orderedBy: sepsis-bundle-engine. As orders are resulted via PATCH /orders/{id}/result, the corresponding bundle element is marked complete. When all four elements are done, the bundle transitions to COMPLIANT (within the 1-hour deadline) or NON_COMPLIANT.
Medications
| Method | Path | Description |
|---|---|---|
| POST | /encounters/{id}/medications |
Record a medication administration for an active encounter |
| GET | /encounters/{id}/medications |
List administrations; optional since ISO 8601 filter; paginated |
| GET | /medications/{id} |
Medication administration detail |
POST body:
| Field | Type | Required | Description |
|---|---|---|---|
drugName |
string | yes | Drug name (case-insensitive for correlation lookups) |
dose |
decimal | yes | Must be > 0 |
doseUnit |
string | yes | e.g. mg, g, mcg |
route |
string | yes | e.g. PO, IV |
administeredAt |
DateTimeOffset | no | Defaults to server time if omitted |
administeredBy |
string | yes | Clinician or nurse identifier |
When a correlated drug was given within the MedicationCorrelation.CorrelationWindowMinutes window (default 90), subsequent warning and NEWS2 alerts for affected vitals include an annotation in details — e.g. — note: metoprolol 25mg (PO) administered 45 min ago. See docs/decisions/medication-correlation-design.md.
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 and NEWS2 composite alerts)
alertType string e.g. THRESHOLD_BREACH, SEPSIS_WARNING, NEWS2_WARNING, NEWS2_EMERGENCY
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'
News2Score
id Guid PK
encounterId Guid FK → Encounter
patientId Guid FK → Patient
totalScore int aggregate 0–20+
riskLevel string LOW | LOW_MEDIUM | MEDIUM | HIGH
respRateScore int component 0–3
spo2Score int
systolicBpScore int
heartRateScore int
consciousnessScore int
temperatureScore int
supplementalO2Score int
hasSingleParamThree bool true when any single parameter scored 3
calculatedAt DateTimeOffset
Indexes: (encounter_id, calculated_at DESC), (patient_id, calculated_at DESC)
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')
SepsisBundle
id Guid PK
encounterId Guid FK → Encounter
triggeringAlertId Guid FK → ClinicalAlert
triggeringAlertType string required (max 50) — QSOFA_WARNING | SEPSIS_WARNING
recognizedAt DateTimeOffset
deadlineAt DateTimeOffset — recognizedAt + 1 hour
complianceStatus string IN_PROGRESS | COMPLIANT | NON_COMPLIANT (default: IN_PROGRESS)
completedAt DateTimeOffset?
Indexes: (encounter_id, recognized_at DESC), (compliance_status), (triggering_alert_id)
Check constraint: compliance_status IN ('IN_PROGRESS', 'COMPLIANT', 'NON_COMPLIANT')
SepsisBundleElement
id Guid PK
bundleId Guid FK → SepsisBundle (CASCADE)
elementType string required (max 40) — BLOOD_CULTURES | SERUM_LACTATE | BROAD_SPECTRUM_ANTIBIOTICS | IV_FLUID_RESUSCITATION
status string PENDING | COMPLETED (default: PENDING)
orderId Guid? FK → Order (SET NULL)
completedAt DateTimeOffset?
Indexes: unique (bundle_id, element_type), (order_id)
Check constraints: element_type IN (...), status IN ('PENDING', 'COMPLETED')
MedicationAdministration
id Guid PK
encounterId Guid FK → Encounter (Restrict on delete)
drugName string required (max 200)
dose decimal(10,4) required
doseUnit string required (max 20)
route string required (max 20)
administeredAt DateTimeOffset required
administeredBy string required (max 200)
Indexes: (encounter_id, administered_at), (encounter_id, drug_name)
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
{
"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,
"news2Score": 6,
"news2RiskLevel": "MEDIUM",
"sepsisBundleStatus": "IN_PROGRESS",
"sepsisBundleElementsCompleted": 2,
"sepsisBundleDeadlineAt": "2025-01-01T09:00:00Z",
"lastObservationAt": "2025-01-01T09:45:00Z"
}
news2Score and news2RiskLevel are optional — populated by EsIndexerService when a NEWS2_WARNING or NEWS2_EMERGENCY alert is indexed (payload fields news2Score / news2RiskLevel from News2Detector). LOW-risk scores with no alert are not projected to Elasticsearch.
sepsisBundleStatus, sepsisBundleElementsCompleted, and sepsisBundleDeadlineAt are optional — populated by EsIndexerService from sepsis.bundle.created and sepsis.bundle.updated Kafka events. Enables department dashboards to filter/sort encounters by sepsis bundle compliance in real time.
observations
{
"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
{
"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, news2-scoring, trend-analyzer, data-lake-writer |
alert.generated |
encounterId |
es-indexer, notification-publisher, data-lake-writer |
encounter.status.changed |
encounterId |
es-indexer, data-lake-writer |
sepsis.bundle.created |
encounterId |
es-indexer |
sepsis.bundle.updated |
encounterId |
es-indexer |
All topics use 6 partitions. KAFKA_AUTO_CREATE_TOPICS_ENABLE=false — topics are provisioned explicitly by KafkaTopicProvisioner to guarantee correct partition count.
alert.generated payload (minimum fields for downstream consumers):
| Field | Required by | Notes |
|---|---|---|
alertId, encounterId, patientId |
ES indexer, data lake, paging | UUIDs |
alertType, severity, triggeredAt |
All consumers | DB string literals for type/severity |
details |
Data lake Parquet | Human-readable breach summary; always set on new alerts |
department |
ES indexer | Optional; critical ingest alerts include it |
news2Score, news2RiskLevel |
ES indexer | Optional; set on NEWS2_WARNING / NEWS2_EMERGENCY alerts for encounter document projection |
partitionKey |
Outbox relay | Same as encounterId |
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. A successful SIRS alert triggers sepsis bundle creation via SepsisAlertHandler.
qSOFA Scoring (Sepsis-3)
The qSOFA (quick Sequential Organ Failure Assessment) engine evaluates three organ-dysfunction criteria per encounter, running in parallel with SIRS within the same SepsisEngineService Kafka consumer. Redis keys use a 30-minute TTL sliding window, identical to SIRS:
| Criterion | Observation Code | Trigger |
|---|---|---|
| Tachypnea | RESP_RATE |
≥ 22 breaths/min |
| Hypotension | SYSTOLIC_BP |
≤ 100 mmHg |
| Altered mentation | AVPU |
≥ 1 (any non-Alert state) |
Redis key pattern: qsofa:{encounterId}:{code} with 30-minute TTL. When a criterion normalizes, the key is deleted immediately. When ≥ 2 of 3 criteria are active simultaneously and no open QSOFA_WARNING alert exists, the engine inserts a CRITICAL alert with details formatted as "qSOFA score 2/3: RESP_RATE=24, SYSTOLIC_BP=95". A successful qSOFA alert triggers sepsis bundle creation via SepsisAlertHandler.
API: GET /encounters/{id}/qsofa/current returns activeCriteria (0–3) and per-criterion values from Redis via QsofaService — used by ward dashboards and the simulator poll loop.
Clinical distinction: SIRS detects systemic inflammation (infection response); qSOFA detects organ dysfunction (Sepsis-3 consensus). Both can fire independently for the same encounter. The sepsis bundle is idempotent — if one is already in progress, the second alert does not create a duplicate.
Sepsis Bundle Compliance (SEP-1)
When either a SIRS or qSOFA alert fires, SepsisAlertHandler calls SepsisBundleService.TryCreateBundleAsync() to initiate a four-element treatment bundle with a one-hour compliance deadline:
| Bundle Element | Order Type | Auto-Created Order Description |
|---|---|---|
BLOOD_CULTURES |
Lab | SEP-1: Blood cultures |
SERUM_LACTATE |
Lab | SEP-1: Serum lactate |
BROAD_SPECTRUM_ANTIBIOTICS |
Medication | SEP-1: Broad-spectrum antibiotics |
IV_FLUID_RESUSCITATION |
Procedure | SEP-1: IV fluid bolus |
Lifecycle:
- Alert fires →
SepsisAlertHandler→SepsisBundleService.TryCreateBundleAsync() - Bundle created with
complianceStatus = IN_PROGRESS,deadlineAt = recognizedAt + 1 hour - Four orders created (
orderedBy: sepsis-bundle-engine), each linked to a bundle element - Outbox event → Kafka topic
sepsis.bundle.created→ ES indexer projectssepsisBundleStatuson encounter - Clinicians work through orders;
PATCH /orders/{id}/result→OrderService→SepsisBundleService.OnOrderResultedAsync() - Each resulted order marks its bundle element
COMPLETED; outbox event →sepsis.bundle.updated→ ES update - When all four elements are complete:
complianceStatus = COMPLIANT(within deadline) orNON_COMPLIANT(past deadline); Prometheussepsis_bundle_compliance_total{status}incremented - If the deadline passes with incomplete elements,
SepsisBundleMonitorService(polling every 5 min) marks the bundleNON_COMPLIANT— elements remainPENDINGbut 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.
NEWS2 Scoring
The NEWS2 engine evaluates seven observation codes that overlap with the expanded vital-sign vocabulary from Phase 10:
| Parameter | Observation Code | Score range |
|---|---|---|
| Respiratory rate | RESP_RATE |
0–3 |
| Oxygen saturation (Scale 1) | SPO2 |
0–3 |
| Systolic blood pressure | SYSTOLIC_BP |
0–3 |
| Heart rate | HEART_RATE |
0–3 |
| Consciousness (AVPU) | AVPU |
0 or 3 |
| Temperature | TEMP_C |
0–3 |
| Supplemental oxygen | SUPPLEMENTAL_O2 |
0 or 2 |
Risk levels (aggregate score):
| Total score | Risk level | Alert |
|---|---|---|
| 0–4 (no single param = 3) | LOW |
None |
| Any single param = 3 (total under 5) | LOW_MEDIUM |
NEWS2_WARNING |
| 5–6 | MEDIUM |
NEWS2_WARNING |
| ≥ 7 | HIGH |
NEWS2_EMERGENCY (CRITICAL) |
All seven parameters must be present in Redis (4-hour TTL per key) before a score is computed. Each new observation after completeness triggers a new news2_scores row; alert creation is idempotent per encounter and alert type while an alert remains open.
Elasticsearch Index Replay
If the Elasticsearch indices need to be rebuilt (e.g., after a mapping change or data loss):
# 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:
# 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:
{
"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
Nineteen phases from the project roadmap are implemented and verified. Integration tests (dotnet test — 116 test methods) and per-phase verification scripts cover Phases 8–15. Phases 17–19 add the Vue dashboard and clinician feedback (Vitest in vigilcare-dashboard/).
| 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); ten 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 |
| 12 | NEWS2 composite scoring (News2Calculator, News2Detector, News2ScoringService); news2_scores table; NEWS2_WARNING / NEWS2_EMERGENCY alert types; News2Controller (current + history); ES news2Score / news2RiskLevel projection; Prometheus NEWS2 metrics; News2CalculatorTests, News2DetectorTests; run-phase12-verification.sh |
Done |
| 13 | Trend detection (TrendCalculator, TrendDetector, TrendAnalyzerService); RAPID_DETERIORATION alert type; alert suppression windows (AlertSuppressionService, Redis suppress:{enc}:{type}); TrendCalculatorTests, TrendDetectorTests, AlertSuppressionTests; run-phase13-verification.sh |
Done |
| 14 | qSOFA scoring engine (QsofaCalculator, QsofaDetector); QSOFA_WARNING alert type; sepsis bundle compliance (SepsisBundle, SepsisBundleElement, SepsisBundleService); auto-created treatment orders with 1-hour deadline; SepsisAlertHandler bridge; SepsisBundleMonitorService (5-min overdue scan); SepsisBundlesController API; ES projection of bundle status; Kafka topics sepsis.bundle.created / sepsis.bundle.updated; Prometheus qsofa_detections_total and sepsis_bundle_compliance_total; QsofaCalculatorTests, QsofaDetectorTests, SepsisBundleTests |
Done |
| 15 | Medication administration (MedicationAdministration, MedicationsController, MedicationService); drug-vital correlation config (MedicationCorrelationOptions); MedicationCorrelationHelper annotates WarningEvaluator and News2Detector alert details; medication_administrations table + migration; MedicationServiceTests, MedicationCorrelationTests, MedicationValidationTests; run-phase15-verification.sh; design doc in docs/decisions/medication-correlation-design.md |
Done |
| 16 | Console replay simulator (VigilCare.Simulator); scenario JSON schema; CLI commands replay, replay-all, validate, dry-run; speed multiplier and optional API polling; eight sample scenarios; docs/simulator-guide.md |
Done |
| 17 | Ward dashboard shell — Vue 3 + Vite + Pinia + Tailwind; virtual ward table (NEWS2-sorted, department filter); patient detail (vitals, scores, alerts, orders, sepsis bundle); alert center (global acknowledge/resolve); API polling; CORS-backed GET /encounters ward list |
Done |
| 18 | Clinical review mode — Chart.js vital sign trends (5 charts), NEWS2 history chart, local replay controls, alert reasoning panel, medication context on alerts; fetchNews2History / fetchMedications; Vitest composable and component tests; docs/dashboard-guide.md |
Done |
| 19 | Clinician feedback mode — six rating buttons per alert, optional notes, Feedback Summary with aggregate stats, JSON/CSV export, localStorage persistence; docs/clinical-testing-guide.md for doctor/nurse evaluation sessions |
Done |
Ward dashboard: backend APIs (GET /encounters ward list, GET /qsofa/current, CORS) and frontend SPA — EncountersListTests, QsofaCurrentTests, vigilcare-dashboard Vitest suite (41 tests: replay scrubbing, feedback store, FeedbackButtons, FeedbackSummary, alert components, charts, ward table).
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.