Trent/vigilcare-clinical
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

chore: add technologies to patient lifecycle

Browse Source
This commit is contained in:
voltsrage
2026-06-21 05:35:44 +08:00
parent f93ce1e57b
commit 9f96f54007
1 changed files with 248 additions and 65 deletions
Download Patch File Download Diff File Expand all files Collapse all files
docs/patient-encounter-api-lifecycle.md
+248 -65
View File
@@ -46,6 +46,161 @@ Before we dive in, here are the medical terms you will see throughout this docum
--- ---
## Quick glossary of technology terms
VigilCare uses several specialized technologies behind the scenes. Here is what each one does and why the system needs it:
| Technology | What it is | How VigilCare uses it |
|---|---|---|
| **PostgreSQL** | A relational database — think of it as a giant spreadsheet that stores data permanently and safely | The main database. Stores all patients, encounters, observations, alerts, orders, medications, scores, and sepsis bundles. When you call any API endpoint, the data is read from or written to PostgreSQL |
| **Redis** | An in-memory cache — like a sticky note board that is extremely fast to read but does not survive a full restart without backup | Stores data that needs to be looked up instantly: alert thresholds, NEWS2 parameter windows, qSOFA criteria, SOFA lab values, trend history, and alert suppression timers. Much faster than querying the database every time |
| **Kafka** | A message bus — like a conveyor belt that carries messages from one part of the system to another | When a measurement is recorded, the API puts a message on the Kafka conveyor belt. Multiple background workers pick up that message independently: one calculates NEWS2, another checks for sepsis, another updates the search index, etc. This is why scoring happens in the background, not in the HTTP response |
| **RabbitMQ** | A message queue — like a to-do list for background tasks that need to happen in order and must not be lost | Handles clinician notifications (paging), alert escalation (when nobody responds to a critical alert within 5 minutes), and discharge summary generation. Unlike Kafka (which broadcasts to many listeners), RabbitMQ ensures each task is done exactly once |
| **Elasticsearch** | A search and analytics engine — like a search engine built specifically for this hospital's data | Powers the analytics endpoints (patient search, observation trends, alert summaries). Data flows from Kafka into Elasticsearch so you can run fast searches and aggregations across all patients and encounters without slowing down the main database |
| **MinIO** | An object store — like a file cabinet for large files, compatible with Amazon S3 | Stores two things: (1) discharge summary PDFs generated when patients leave, and (2) Parquet data files (a compact format for large datasets) used for long-term analysis and machine learning |
| **Prometheus** | A monitoring tool — watches the system itself (not the patients) | Collects metrics like "how many observations were recorded this hour," "how many alerts fired," "how long does an ingest take." Helps operators know if the system is healthy |
| **Grafana** | A dashboard tool for Prometheus metrics | Displays charts and graphs about system health — things like observation ingest rate, alert volumes, consumer lag, and response times |
| **Docker** | A container platform — packages each technology into an isolated box that runs the same way everywhere | All the services above (PostgreSQL, Redis, Kafka, etc.) run as Docker containers, orchestrated by a single `docker-compose.yml` file. One command starts the entire system |
---
## How the technology stack works together
Understanding how data flows through these technologies helps explain why some things happen immediately and others take a few seconds.
### The outbox pattern — how data gets from the API to background workers
When a nurse records a heart rate of 95 bpm, here is what happens under the hood:
```
1. API receives POST /observations
2. Inside ONE database transaction (all-or-nothing):
- Save the observation to PostgreSQL
- If critical threshold breached: save the alert to PostgreSQL too
- Write an "outbox event" row to PostgreSQL (a message waiting to be sent)
3. COMMIT — everything saved atomically
4. OutboxRelayService (runs every 500ms):
- Reads unsent outbox events from PostgreSQL
- Sends them to the correct Kafka topic
- Marks them as processed
```
**Why this pattern?** It guarantees that if the observation is saved, the message to Kafka will also be sent — even if Kafka is temporarily down. The database acts as a reliable staging area. This is called the "transactional outbox pattern."
### Kafka topics — the conveyor belts
Each Kafka topic carries a specific type of message. Think of them as labeled conveyor belts in a factory:
| Topic | What flows through it | Who produces it | Who consumes it |
|---|---|---|---|
| `observation.recorded` | Every measurement recorded by any device, nurse, or lab | Outbox relay (from API) | Warning evaluator, NEWS2 scorer, sepsis engine, SOFA scorer, GCS scorer, trend analyzer, ES indexer, data lake writer |
| `alert.generated` | Every alert created by any engine | Outbox relay (from scoring engines) | ES indexer, data lake writer, notification publisher (→ RabbitMQ) |
| `encounter.status.changed` | Admissions, discharges, cancellations | Outbox relay (from API) | ES indexer, data lake writer, notification publisher (→ discharge queue) |
| `sepsis.bundle.created` | New sepsis bundles | Outbox relay (from SOFA engine) | ES indexer |
| `sepsis.bundle.updated` | Bundle element completions and compliance changes | Outbox relay (from order results) | ES indexer |
| `gcs.scored` | GCS score calculations | Outbox relay (from GCS scorer) | ES indexer |
**Key detail:** One message on `observation.recorded` is consumed by up to 8 different workers independently. Each worker has its own "consumer group" — Kafka tracks where each group left off, so every worker gets every message even if they process at different speeds.
### Redis — the fast-lookup layer
Redis stores data that needs to be checked on every observation ingest or scoring calculation. Reading from Redis takes less than 1 millisecond, compared to 5–50ms for a database query.
| What is stored | Redis key pattern | Why it is in Redis | How long it stays (TTL) |
|---|---|---|---|
| Alert thresholds (critical/warning boundaries) | `threshold:{observationCode}` | Checked on every single observation — must be instant | No expiry (loaded at startup from PostgreSQL) |
| Alert suppression timers | `suppress:{encounterId}:{alertType}` | Prevents duplicate warning alerts after a clinician acknowledges one | 30 minutes |
| Trend history (recent vital sign readings) | `trend:{encounterId}:{observationCode}` | The trend analyzer needs the last 10 readings within 30 minutes to calculate rate of change | 2 hours |
| SOFA lab values | `sofa:{encounterId}:{observationCode}` | Lab results arrive infrequently; Redis caches the latest value so SOFA can be recalculated whenever any new observation arrives | 24 hours (considered stale after 12 hours) |
**Example flow:** A heart rate of 105 bpm arrives. The API reads `threshold:HEART_RATE` from Redis (instant) and finds the warning high is 100. Since 105 > 100 but < 150 (critical), it is not a critical alert — but the outbox event goes to Kafka, where the warning evaluator picks it up and checks `suppress:{encounterId}:WARNING_HEART_RATE` in Redis. If no suppression exists, a warning alert is created.
### RabbitMQ — notifications and escalation
RabbitMQ handles tasks where order matters and each task must be completed exactly once. It manages four queues:
| Queue | What it does | How it works |
|---|---|---|
| `alerts.paging.queue` | Pages clinicians when a critical alert fires | A Kafka consumer picks up `alert.generated` events and, if the severity is CRITICAL, drops a message into this queue. The paging worker processes one alert at a time and waits for acknowledgment |
| `alerts.paging.dlq` (Dead Letter Queue) | Catches unacknowledged pages | If a paging message is not acknowledged within 5 minutes, RabbitMQ automatically moves it to this "dead letter" queue. After a timeout, it re-routes to the escalation queue |
| `alerts.escalation.queue` | Escalates alerts that nobody responded to | Picks up messages from the DLQ and marks the alert as "Escalated" in the database. This would trigger an on-call page to a more senior clinician |
| `notifications.discharge.queue` | Generates discharge summary PDFs | When a patient is discharged, a message arrives here. The discharge worker gathers all encounter data (observations, alerts, orders, medications) and builds a PDF summary, then uploads it to MinIO |
**Escalation timeline example:**
```
0:00 Critical alert fires → message enters paging queue
0:00 Paging worker picks it up, simulates clinician page
5:00 Nobody acknowledged → message is NACKed → goes to DLQ
5:00+ DLQ TTL expires → message re-routes to escalation queue
Escalation worker marks alert as "Escalated" in database
```
### Elasticsearch — search and analytics
Elasticsearch is a search-optimized copy of the data. It does not replace PostgreSQL — it mirrors selected data for fast searching and aggregation.
**Three indices (like specialized lookup tables):**
| Index | What it contains | Updated when |
|---|---|---|
| `patient_encounters` | One document per encounter: patient info, department, NEWS2 score, alert count, sepsis bundle status | Encounter created/discharged, alerts generated, bundle updated |
| `observations` | One document per observation: code, value, unit, timestamp, patient/encounter IDs | Every observation recorded |
| `clinical_alerts` | One document per alert: type, severity, status, department, timestamp | Every alert generated |
**Why not just query PostgreSQL?** Some queries are expensive on a relational database — for example, "show me hourly average heart rates for patient X over the last week" or "how many critical alerts fired in the ICU today." Elasticsearch is built for these kinds of aggregations and full-text searches. The trade-off is that data arrives in Elasticsearch 1–3 seconds after it is written to PostgreSQL.
### MinIO — long-term file storage
MinIO stores files organized by path, like a file system in the cloud:
| Path pattern | What is stored | How it gets there |
|---|---|---|
| `discharge-summaries/{encounterId}/summary.pdf` | Discharge summary for each encounter | Generated by the discharge worker (RabbitMQ) when a patient is discharged |
| `observations/{YYYY}/{MM}/{DD}/partition-{p}-offset-{o}.parquet` | Daily observation data in Parquet format | Written by the data lake writer (Kafka consumer) in batches of 1000 events or every 5 minutes |
| `alerts/{YYYY}/{MM}/{DD}/partition-{p}-offset-{o}.parquet` | Daily alert data in Parquet format | Same as above |
| `encounters/{YYYY}/{MM}/{DD}/partition-{p}-offset-{o}.parquet` | Daily encounter status changes in Parquet format | Same as above |
**What is Parquet?** It is a file format designed for large-scale data analysis. Unlike a CSV or JSON file, Parquet files are compressed and organized by column, making them very efficient for queries like "give me all heart rate values from the last month." This data lake is the foundation for future reporting, auditing, and machine learning.
### Prometheus and Grafana — system health monitoring
Prometheus scrapes the `/metrics` endpoint every 15 seconds and records operational metrics:
| Metric | Type | What it tracks |
|---|---|---|
| `observations_ingested_total` | Counter | Total observations recorded, labeled by code and source (DEVICE/MANUAL/LAB) |
| `clinical_alerts_total` | Counter | Total alerts generated, labeled by type and severity |
| `news2_scores_total` | Counter | NEWS2 scores calculated, labeled by risk level |
| `escalations_total` | Counter | Alerts that escalated because nobody acknowledged them |
| `observation_ingest_duration_seconds` | Histogram | How long each observation ingest takes (target: under 50ms) |
| `news2_scoring_duration_seconds` | Histogram | How long each NEWS2 calculation takes |
| `sofa_scoring_duration_seconds` | Histogram | How long each SOFA calculation takes |
| `alerts_unacknowledged_gauge` | Gauge | Current count of critical alerts open for more than 5 minutes |
| `outbox_pending_events` | Gauge | Unprocessed outbox events (if this grows, the relay is falling behind) |
| `kafka_consumer_lag` | Gauge | How far behind each Kafka consumer is (if this grows, scoring is delayed) |
Grafana displays these metrics as charts, and operators can set up alerts (system alerts, not clinical alerts) if things like consumer lag or pending outbox events climb too high.
### Docker Compose — running everything locally
All services run in Docker containers defined in `docker-compose.yml`:
| Service | Port (host) | Purpose |
|---|---|---|
| PostgreSQL 16 | 5436 | Main database |
| Redis 7 | 6382 | Fast cache |
| Kafka 3.7 (KRaft mode) | 9092 | Message bus |
| RabbitMQ 3.13 | 5674 (AMQP), 15674 (management UI) | Task queues |
| Elasticsearch 8.13 | 9200 | Search and analytics |
| MinIO | 9005 (API), 9006 (console UI) | File/object storage |
| Prometheus | 9101 | Metrics collection |
| Grafana | 3101 | Metrics dashboards |
Starting the full stack: `docker compose up -d` brings up all 8 services. The .NET API runs separately (outside Docker) and connects to these services on the ports listed above.
---
## Conventions ## Conventions
| Item | Value | | Item | Value |
@@ -337,13 +492,16 @@ The system looks at changes over a 30-minute sliding window:
## Phase 0 — Platform prerequisites (before any patient) ## Phase 0 — Platform prerequisites (before any patient)
These are typically set up once at deploy time, not called per patient. These are typically set up once at deploy time, not called per patient. This is where the technology stack gets initialized.
| What | How | Why it matters | | What | How | Technology involved | Why it matters |
|---|---|---| |---|---|---|---|
| Alert thresholds | Seeded in PostgreSQL; loaded into Redis on startup; manageable via `POST/GET/PUT /alert-thresholds` | Every measurement is validated against configured thresholds to decide if alerts fire | | Alert thresholds | Seeded in PostgreSQL; `ThresholdCacheLoader` copies them into Redis at startup; manageable via `POST/GET/PUT /alert-thresholds` | PostgreSQL → Redis | Every measurement is validated against thresholds from Redis (fast) to decide if alerts fire. Without this cache, every observation ingest would need a database query |
| Kafka topics | Provisioned by `KafkaTopicProvisioner` | `observation.recorded`, `alert.generated`, `encounter.status.changed` (+ sepsis bundle topics) | | Kafka topics | `KafkaTopicProvisioner` creates 7 topics on startup (6 partitions each, auto-creation disabled) | Kafka | `observation.recorded`, `alert.generated`, `encounter.status.changed`, `sepsis.bundle.created`, `sepsis.bundle.updated`, `gcs.scored`. These must exist before any messages can flow |
| Drug-vital mappings | `MedicationCorrelation` section in `appsettings.json` | Links medications to vital signs so alerts can include context like "heart rate may be elevated due to epinephrine given 20 min ago" | | Elasticsearch indices | `ElasticIndexProvisioner` creates 3 indices with proper field mappings | Elasticsearch | `patient_encounters`, `observations`, `clinical_alerts` — must exist before data can be indexed |
| RabbitMQ topology | `RabbitMqTopologyProvisioner` declares the exchange and 5 queues | RabbitMQ | Sets up the `clinical.notifications.exchange` and all queues (paging, DLQ, escalation, discharge, reconciliation) with proper routing and dead-letter configuration |
| MinIO bucket | Created if it does not exist | MinIO | The `vigilcare` bucket must exist before discharge summaries or data lake files can be written |
| Drug-vital mappings | `MedicationCorrelation` section in `appsettings.json` | Configuration | Links medications to vital signs so alerts can include context like "heart rate may be elevated due to epinephrine given 20 min ago" |
--- ---
@@ -385,9 +543,14 @@ POST /api/v1/patients/{patientId}/encounters
**Response:** `201 Created` — encounter `id`, `status: ACTIVE`, `admittedAt`. **Response:** `201 Created` — encounter `id`, `status: ACTIVE`, `admittedAt`.
**Side effects:** **Side effects (what happens behind the scenes):**
- Event `encounter.status.changed` → Kafka → updates Elasticsearch ward index, data lake, and discharge-summary queue - An `encounter.status.changed` outbox event is written to PostgreSQL in the same transaction as the encounter
- The outbox relay picks it up within 500ms and sends it to the Kafka `encounter.status.changed` topic
- Three Kafka consumers process it independently:
- **ES indexer** → creates a document in the `patient_encounters` Elasticsearch index (so the patient appears in ward searches)
- **Data lake writer** → buffers the event and writes it to a Parquet file in MinIO (for long-term records)
- **Notification publisher** → no action for admissions (only discharges trigger the discharge queue)
- Only **one active encounter per patient per encounter type** — trying to create a duplicate returns **409** (`DUPLICATE_ACTIVE_ENCOUNTER`) - Only **one active encounter per patient per encounter type** — trying to create a duplicate returns **409** (`DUPLICATE_ACTIVE_ENCOUNTER`)
From this point, `encounterId` is the key for all clinical writes. From this point, `encounterId` is the key for all clinical writes.
@@ -453,30 +616,32 @@ Optional header: `Idempotency-Key` — if you accidentally send the same measure
1. Encounter must be `ACTIVE` — otherwise **409** 1. Encounter must be `ACTIVE` — otherwise **409**
2. Idempotency check — skip if already recorded 2. Idempotency check — skip if already recorded
3. Plausibility validation — reject impossible values with **422** 3. Plausibility validation — reject impossible values with **422**
4. Save the measurement 4. Save the observation row to **PostgreSQL**
5. Check against critical thresholds from Redis 5. Read the alert thresholds from **Redis** (sub-millisecond lookup via key `threshold:{observationCode}`)
6. **If a critical threshold is breached:** create an alert immediately in the same database transaction 6. **If a critical threshold is breached:** insert a `clinical_alert` row + an `alert.generated` outbox event into **PostgreSQL** — all in the same transaction (so the alert is never lost)
7. Queue the measurement for background processing 7. Insert an `observation.recorded` outbox event into **PostgreSQL**
8. Commit 8. COMMIT — everything saved atomically. Within 500ms, the **outbox relay** picks up the events and sends them to **Kafka**
**HTTP response (`201`):** for each measurement, you get `observation`, `alertGenerated` (true only for critical alerts), `alertId`, and `duplicate`. **HTTP response (`201`):** for each measurement, you get `observation`, `alertGenerated` (true only for critical alerts), `alertId`, and `duplicate`.
### What happens in the background (not in the HTTP response) ### What happens in the background (not in the HTTP response)
Every recorded measurement is processed independently by these background engines: Once the `observation.recorded` message lands on **Kafka**, multiple consumers pick it up independently. Each consumer belongs to its own consumer group, so Kafka delivers the same message to all of them. This is why one observation can trigger NEWS2, SOFA, qSOFA, trend detection, and search indexing all at the same time.
| Engine | What it does | Alert produced | | Engine (Kafka consumer group) | What it does | Technology used | Alert produced |
|---|---|---| |---|---|---|---|
| **warning-evaluator** | Checks if the value crosses a warning threshold. Optionally adds medication context (e.g. "patient was given morphine 30 min ago") | `WARNING_*` alerts | | **warning-evaluator** | Checks if the value crosses a warning threshold. Reads suppression status from **Redis** (`suppress:{encounterId}:{alertType}`). Optionally adds medication context (e.g. "patient was given morphine 30 min ago") | Redis, PostgreSQL | `WARNING_*` alerts |
| **news2-scoring** | Recalculates NEWS2 if all 7 parameters are available within 4 hours | `NEWS2_WARNING` or `NEWS2_EMERGENCY` | | **news2-scoring** | Reads latest 7 parameters from **Redis** (4-hour window). If all 7 are present, calculates NEWS2 and writes score to **PostgreSQL** | Redis, PostgreSQL | `NEWS2_WARNING` or `NEWS2_EMERGENCY` |
| **sepsis-engine** | Recalculates qSOFA criteria count | `QSOFA_SCREEN` (when 2+ criteria met) | | **sepsis-engine** | Recalculates qSOFA criteria count using latest values from **Redis** | Redis, PostgreSQL | `QSOFA_SCREEN` (when 2+ criteria met) |
| **sofa-scoring** | Recalculates SOFA organ scores using latest vitals and labs | `SOFA_WARNING` or `SOFA_SEPSIS` | | **sofa-scoring** | Reads cached lab values from **Redis** (`sofa:{encounterId}:{code}`), recalculates per-organ SOFA scores, compares to baseline in **PostgreSQL** | Redis, PostgreSQL | `SOFA_WARNING` or `SOFA_SEPSIS` |
| **gcs-scoring** | Recalculates GCS when eye/verbal/motor components are recorded | `GCS_WARNING` or `GCS_CRITICAL` | | **gcs-scoring** | Recalculates GCS when eye/verbal/motor components are recorded, writes score to **PostgreSQL** | PostgreSQL | `GCS_WARNING` or `GCS_CRITICAL` |
| **trend-analyzer** | Checks if the vital sign is changing too fast (rate of change over 30-min window) | `RAPID_DETERIORATION` | | **trend-analyzer** | Reads recent readings from **Redis** (`trend:{encounterId}:{code}`, up to 10 entries), calculates rate of change over 30-minute window | Redis, PostgreSQL | `RAPID_DETERIORATION` |
| **es-indexer** | Updates Elasticsearch for search and analytics | (no alert) | | **es-indexer** | Upserts the observation into the `observations` **Elasticsearch** index. Also updates `patient_encounters` index (e.g. `lastObservationAt`) | Elasticsearch | (no alert) |
| **data-lake-writer** | Writes Parquet files to MinIO for long-term analysis | (no alert) | | **data-lake-writer** | Buffers observations and writes them to **MinIO** as Parquet files (batches of 1000 events or every 5 minutes) | MinIO | (no alert) |
**Important for integrators:** After posting measurements, poll `GET /encounters/{id}/alerts` or score endpoints. Warning and scoring alerts are NOT in the initial HTTP response — they arrive 1–3 seconds later. Any alerts created by these engines are written to **PostgreSQL** with an `alert.generated` outbox event, which flows back through **Kafka** → **Elasticsearch** (for indexing) and **RabbitMQ** (for paging, if critical).
**Important for integrators:** After posting measurements, poll `GET /encounters/{id}/alerts` or score endpoints. Warning and scoring alerts are NOT in the initial HTTP response — they arrive 1–3 seconds later (the time it takes for the message to travel through Kafka and be processed).
### Read measurement history ### Read measurement history
@@ -570,11 +735,15 @@ open → acknowledged → resolved
→ escalated (unacknowledged CRITICAL after 5 min → pages on-call staff) → escalated (unacknowledged CRITICAL after 5 min → pages on-call staff)
``` ```
When a clinician acknowledges a suppressible warning, a 30-minute suppression window starts in Redis. During that window, duplicate warnings of the same type for the same patient will not fire again. Critical and SOFA-sepsis alerts are never suppressed. When a clinician acknowledges a suppressible warning, the system writes a key to **Redis** (`suppress:{encounterId}:{alertType}`) with a 30-minute TTL (time-to-live). During that window, the warning evaluator checks Redis before creating a new alert — if the suppression key exists, the duplicate is silently skipped. After 30 minutes, Redis automatically deletes the key, and new warnings can fire again. Critical and SOFA-sepsis alerts are never suppressed.
### Escalation (no HTTP endpoint) ### Escalation (no HTTP endpoint)
Critical alerts that remain unacknowledged for 5+ minutes are automatically escalated — they are routed to an on-call paging queue. This happens at the infrastructure level; integrators see it through the alert's `status` field changing to `escalated`. When a critical alert fires, the `alert.generated` **Kafka** message is picked up by the notification publisher, which drops a message into the **RabbitMQ** `alerts.paging.queue`. The paging worker processes it and waits for the clinician to acknowledge the alert in the database.
If nobody acknowledges within 5 minutes, the paging worker NACKs (rejects) the message. **RabbitMQ** automatically moves it to the `alerts.paging.dlq` (dead letter queue). After a timeout, the DLQ re-routes it to the `alerts.escalation.queue`, where the escalation worker marks the alert as `Escalated` in **PostgreSQL** and increments the `escalations_total` **Prometheus** metric.
Integrators see escalation through the alert's `status` field changing to `escalated`.
--- ---
@@ -582,35 +751,35 @@ Critical alerts that remain unacknowledged for 5+ minutes are automatically esca
### NEWS2 ### NEWS2
Requires all 7 parameters in Redis within a 4-hour window: `RESP_RATE`, `SPO2`, `SYSTOLIC_BP`, `HEART_RATE`, `AVPU`, `TEMP_C`, `SUPPLEMENTAL_O2`. Requires all 7 parameters within a 4-hour window: `RESP_RATE`, `SPO2`, `SYSTOLIC_BP`, `HEART_RATE`, `AVPU`, `TEMP_C`, `SUPPLEMENTAL_O2`. The NEWS2 scoring consumer reads each parameter's latest value from **Redis** — if all 7 are present and recent enough, it calculates the score, writes the result to **PostgreSQL** (`news2_scores` table), and creates an alert if warranted.
``` ```
GET /api/v1/encounters/{id}/news2/current → latest score (404 if not yet computed) GET /api/v1/encounters/{id}/news2/current → latest score (404 if not yet computed)
GET /api/v1/encounters/{id}/news2/history → score history over time GET /api/v1/encounters/{id}/news2/history → score history over time
``` ```
Scores and NEWS2 alerts are calculated in the background — record the 7th parameter, wait a few seconds, then poll. These endpoints read from **PostgreSQL**. Scores are calculated in the background via **Kafka** — record the 7th parameter, wait a few seconds, then poll.
### SOFA ### SOFA
Requires lab results and vitals across 6 organ systems. Not all systems need data immediately — the score is computed with whatever is available once at least 4 systems have data. Requires lab results and vitals across 6 organ systems. The SOFA scoring consumer reads cached lab values from **Redis** (`sofa:{encounterId}:{code}`) — labs are cached there because they arrive infrequently (hours apart), but SOFA needs to be recalculated every time any new observation arrives. The score is computed once at least 4 systems have data.
``` ```
GET /api/v1/encounters/{id}/sofa/current → latest SOFA score with per-organ breakdown GET /api/v1/encounters/{id}/sofa/current → latest SOFA score with per-organ breakdown
GET /api/v1/encounters/{id}/sofa/history → score history GET /api/v1/encounters/{id}/sofa/history → score history
``` ```
The response includes per-organ scores (respiratory, coagulation, liver, cardiovascular, CNS, renal) so clinicians can see which organ systems are struggling. The response (from **PostgreSQL**) includes per-organ scores (respiratory, coagulation, liver, cardiovascular, CNS, renal) so clinicians can see which organ systems are struggling.
### qSOFA ### qSOFA
Live criteria count from Redis (no stored history — just the current state): Live criteria count from **Redis** (no stored history — just the current state):
``` ```
GET /api/v1/encounters/{id}/qsofa/current GET /api/v1/encounters/{id}/qsofa/current
``` ```
Returns how many of the 3 criteria are currently met (0–3). When the count reaches 2, a `QSOFA_SCREEN` alert fires. Returns how many of the 3 criteria are currently met (0–3). When the count reaches 2, a `QSOFA_SCREEN` alert is written to **PostgreSQL** and flows through **Kafka** to **Elasticsearch**.
### GCS ### GCS
@@ -618,7 +787,7 @@ Returns how many of the 3 criteria are currently met (0–3). When the count rea
GET /api/v1/encounters/{id}/gcs/current → latest GCS with component breakdown GET /api/v1/encounters/{id}/gcs/current → latest GCS with component breakdown
``` ```
Returns the total GCS score, individual component scores (eye, verbal, motor), and severity classification (mild/moderate/severe). Returns the total GCS score (from **PostgreSQL**), individual component scores (eye, verbal, motor), and severity classification (mild/moderate/severe). GCS scores also flow through **Kafka** (`gcs.scored` topic) to **Elasticsearch**.
--- ---
@@ -670,14 +839,14 @@ A combined chronological stream of **observations** and **alerts** — like a sc
### Elasticsearch analytics (read-only, cross-encounter) ### Elasticsearch analytics (read-only, cross-encounter)
| Endpoint | Purpose | These endpoints query **Elasticsearch** — not **PostgreSQL**. They are fast for searches and aggregations across many patients, but the data may be 1–3 seconds behind the live database (because it flows through **Kafka** before landing in Elasticsearch).
|---|---|
| `GET /analytics/patients?q=...&department=...` | Search patients and encounters |
| `GET /analytics/observations/trend?encounterId=...&code=...` | Hourly avg/min/max time series for a measurement |
| `GET /analytics/alerts/summary?severity=...&department=...` | Alert volume by department |
| `GET /analytics/population?code=...&threshold=...` | Find patients above/below a measurement threshold |
These read from a search index fed by Kafka — not from the live database. There may be a few seconds of delay. | Endpoint | Purpose | Elasticsearch index used |
|---|---|---|
| `GET /analytics/patients?q=...&department=...` | Full-text search across patients and encounters | `patient_encounters` |
| `GET /analytics/observations/trend?encounterId=...&code=...` | Hourly avg/min/max time series for a measurement (uses `date_histogram` aggregation) | `observations` |
| `GET /analytics/alerts/summary?severity=...&department=...` | Alert volume by department | `clinical_alerts` |
| `GET /analytics/population?code=...&threshold=...` | Find patients above/below a measurement threshold in a time window | `observations` |
--- ---
@@ -692,10 +861,15 @@ Body: { "status": "DISCHARGED", "dischargeDiagnosis": "optional free text" }
**Response:** `200 OK` — `{ encounterId, newStatus, dischargeDiagnosis }`. **Response:** `200 OK` — `{ encounterId, newStatus, dischargeDiagnosis }`.
**Side effects:** **Side effects (the full technology chain):**
- `dischargedAt` is set to the current time - `dischargedAt` is set to the current time in **PostgreSQL**
- Event fires → Kafka → Elasticsearch status update, data lake, and a **discharge summary worker** that generates a PDF stored at `/discharge-summaries/{encounterId}/summary.pdf` - An `encounter.status.changed` outbox event is saved in **PostgreSQL** and relayed to **Kafka** within 500ms
- **Kafka** consumers pick it up:
- **ES indexer** → updates the encounter's status in the `patient_encounters` **Elasticsearch** index (patient drops off the active ward search)
- **Data lake writer** → writes the discharge event to a Parquet file in **MinIO**
- **Notification publisher** → detects this is a discharge and drops a message into the **RabbitMQ** `notifications.discharge.queue`
- The **discharge summary worker** (RabbitMQ consumer) picks up the message, gathers all encounter data from **PostgreSQL** (observations, alerts, orders, medications, scores), builds a PDF summary, and uploads it to **MinIO** at `discharge-summaries/{encounterId}/summary.pdf`
### What stops working after discharge ### What stops working after discharge
@@ -798,37 +972,46 @@ The Vue ward dashboard (`vigilcare-dashboard/`) implements this lifecycle throug
## Sync vs async quick reference ## Sync vs async quick reference
| Event | In the HTTP response? | Where to check afterward | | Event | In the HTTP response? | Technology path | Where to check afterward |
|---|---|---| |---|---|---|---|
| Measurement saved | Yes (`201`) | `GET .../observations` | | Measurement saved | Yes (`201`) | PostgreSQL (direct write) | `GET .../observations` |
| Critical threshold alert | Yes (`alertGenerated: true`) | `GET .../alerts` | | Critical threshold alert | Yes (`alertGenerated: true`) | PostgreSQL (same transaction) + Redis (threshold lookup) | `GET .../alerts` |
| Warning threshold alert | No — background (Kafka) | `GET .../alerts` (after ~1–3 s) | | Warning threshold alert | No | PostgreSQL → Kafka → warning evaluator → PostgreSQL | `GET .../alerts` (after ~1–3 s) |
| NEWS2 score / alert | No — background (Kafka) | `GET .../news2/current`, `GET .../alerts` | | NEWS2 score / alert | No | PostgreSQL → Kafka → NEWS2 scorer (reads Redis) → PostgreSQL | `GET .../news2/current`, `GET .../alerts` |
| SOFA score / alert | No — background (Kafka) | `GET .../sofa/current`, `GET .../alerts` | | SOFA score / alert | No | PostgreSQL → Kafka → SOFA scorer (reads Redis lab cache) → PostgreSQL | `GET .../sofa/current`, `GET .../alerts` |
| qSOFA screen | No — background (Kafka) | `GET .../alerts`, `GET .../qsofa/current` | | qSOFA screen | No | PostgreSQL → Kafka → sepsis engine (reads Redis) → PostgreSQL | `GET .../alerts`, `GET .../qsofa/current` |
| GCS score / alert | No — background (Kafka) | `GET .../gcs/current`, `GET .../alerts` | | GCS score / alert | No | PostgreSQL → Kafka → GCS scorer → PostgreSQL | `GET .../gcs/current`, `GET .../alerts` |
| Trend alert | No — background (Kafka) | `GET .../alerts` | | Trend alert | No | PostgreSQL → Kafka → trend analyzer (reads Redis history) → PostgreSQL | `GET .../alerts` |
| Medication annotation on alert | No — applied at alert creation | `GET .../alerts` → read `details` field | | Medication annotation on alert | No — applied at alert creation | Kafka consumer reads recent meds from PostgreSQL | `GET .../alerts` → read `details` field |
| Discharge summary PDF | No — background (RabbitMQ) | MinIO `/discharge-summaries/{encounterId}/summary.pdf` | | Search index updated | No | PostgreSQL → Kafka → ES indexer → Elasticsearch | `GET /analytics/...` endpoints |
| Data lake file written | No | PostgreSQL → Kafka → data lake writer → MinIO (Parquet) | MinIO bucket `vigilcare` |
| Clinician paged | No | Kafka → notification publisher → RabbitMQ paging queue | Alert `status` field |
| Alert escalated | No | RabbitMQ paging queue → DLQ → escalation queue → PostgreSQL | Alert `status` = `escalated` |
| Discharge summary PDF | No | Kafka → RabbitMQ discharge queue → worker → MinIO | MinIO `/discharge-summaries/{encounterId}/summary.pdf` |
--- ---
## How it all fits together — a real example ## How it all fits together — a real example
Imagine a patient named Maria arrives at the Emergency Department with a suspected urinary tract infection (UTI): Imagine a patient named Maria arrives at the Emergency Department with a suspected urinary tract infection (UTI). Here is what happens in both the clinical and technology layers:
1. **Admission:** A clerk registers Maria (`POST /patients`) and opens an emergency encounter (`POST /patients/{id}/encounters` with `encounterType: EMERGENCY`). 1. **Admission:** A clerk registers Maria (`POST /patients`) and opens an emergency encounter (`POST /patients/{id}/encounters` with `encounterType: EMERGENCY`). Behind the scenes: patient and encounter rows are saved to **PostgreSQL**, an outbox event goes to **Kafka**, and the **ES indexer** creates a document in the `patient_encounters` **Elasticsearch** index — Maria now appears on the ward board.
2. **First vitals:** A nurse records heart rate 95, respiratory rate 20, BP 115/75, temperature 38.8°C, SpO2 97%, AVPU = Alert, room air. No critical thresholds are crossed, so no immediate alerts. In the background, NEWS2 calculates: total = 3 (temp score 1 + heart rate score 1 + resp rate score 0 + others 0 + temperature gives 1) → LOW risk, no alert. 2. **First vitals:** A nurse records heart rate 95, respiratory rate 20, BP 115/75, temperature 38.8°C, SpO2 97%, AVPU = Alert, room air. The API checks each value against thresholds in **Redis** — no critical thresholds are crossed, so no immediate alerts. The observations are saved to **PostgreSQL** and outbox events flow to **Kafka**. The NEWS2 consumer reads all 7 parameters from **Redis**, calculates total = 3 → LOW risk, no alert. The **trend analyzer** stores these values in **Redis** as the starting point for rate-of-change tracking.
3. **Labs ordered:** The doctor orders blood cultures, lactate, and a CBC (`POST /encounters/{id}/orders`). Broad-spectrum antibiotics are administered (`POST /encounters/{id}/medications`). 3. **Labs ordered:** The doctor orders blood cultures, lactate, and a CBC (`POST /encounters/{id}/orders`). Broad-spectrum antibiotics are administered (`POST /encounters/{id}/medications`). These are saved to **PostgreSQL** — the medication record will be available for correlation if any warning alerts fire later.
4. **Two hours later — getting worse:** New vitals come in: heart rate 118, respiratory rate 24, BP 95/60, temperature 39.5°C, SpO2 94%. The `WARNING_SYSTOLIC_BP` fires (below 100). NEWS2 recalculates to 8 → HIGH risk → `NEWS2_EMERGENCY` alert fires. qSOFA sees respiratory rate ≥22 and systolic BP ≤100 → 2 criteria → `QSOFA_SCREEN` fires. 4. **Two hours later — getting worse:** New vitals come in: heart rate 118, respiratory rate 24, BP 95/60, temperature 39.5°C, SpO2 94%. The API reads `threshold:SYSTOLIC_BP` from **Redis** and sees 95 is below warning (100) but above critical (70) — no sync alert. The observations go to **Kafka**, where multiple consumers react:
- **Warning evaluator** → checks **Redis** suppression keys, finds none → creates `WARNING_SYSTOLIC_BP` in **PostgreSQL**
- **NEWS2 scorer** → reads 7 parameters from **Redis**, calculates total = 8 → HIGH risk → creates `NEWS2_EMERGENCY` in **PostgreSQL**
- **Sepsis engine** → checks qSOFA criteria in **Redis**: respiratory rate ≥22 and systolic BP ≤100 → 2 criteria met → creates `QSOFA_SCREEN` in **PostgreSQL**
- **Trend analyzer** → reads previous heart rate values from **Redis**, calculates 23 bpm rise → `RAPID_DETERIORATION` in **PostgreSQL**
- Each alert generates an outbox event → **Kafka** → **ES indexer** (updates **Elasticsearch**) and **notification publisher** → critical alerts go to **RabbitMQ** `alerts.paging.queue` → clinician is paged
5. **Lab results arrive:** Lactate comes back at 3.2 mmol/L (high — tissue hypoxia). Creatinine is 2.1 mg/dL (elevated — kidneys struggling). SOFA score jumps by 3 points from baseline → `SOFA_SEPSIS` alert fires. A sepsis bundle is automatically created with a 1-hour deadline. 5. **Lab results arrive:** Lactate comes back at 3.2 mmol/L (high — tissue hypoxia). Creatinine is 2.1 mg/dL (elevated — kidneys struggling). These go through **Kafka** to the **SOFA scorer**, which reads all cached lab values from **Redis** (`sofa:{encounterId}:...`) and recalculates. SOFA jumps by 3 points from baseline → `SOFA_SEPSIS` alert fires in **PostgreSQL**. The SOFA engine also creates a sepsis bundle with 4 auto-generated orders in **PostgreSQL**, and a `sepsis.bundle.created` event flows through **Kafka** to **Elasticsearch** (the ward board now shows `sepsisActive: true`).
6. **Sepsis bundle completion:** The team draws blood cultures, gives IV fluids, and confirms antibiotics were already given. Each order is completed via `PATCH /orders/{id}/result`. All 4 elements complete within the deadline → bundle status: `COMPLIANT`. 6. **Sepsis bundle completion:** The team draws blood cultures, gives IV fluids, and confirms antibiotics were already given. Each order is completed via `PATCH /orders/{id}/result` — saved to **PostgreSQL** with a `sepsis.bundle.updated` outbox event → **Kafka** → **Elasticsearch** (bundle status updates in real time on the ward board). All 4 elements complete within the deadline → bundle status: `COMPLIANT`.
7. **Stabilization and discharge:** Over the next 48 hours, vitals normalize. NEWS2 drops back to 2. SOFA returns to baseline. Maria is discharged with `PATCH /encounters/{id}/status`. 7. **Stabilization and discharge:** Over the next 48 hours, vitals normalize. NEWS2 drops back to 2. SOFA returns to baseline. Maria is discharged with `PATCH /encounters/{id}/status`. The discharge event flows through **Kafka** → **Elasticsearch** (Maria drops off the active ward list) → **RabbitMQ** `notifications.discharge.queue` → the **discharge summary worker** gathers all her encounter data from **PostgreSQL**, builds a PDF, and uploads it to **MinIO**. Meanwhile, the **data lake writer** has been steadily writing all her observations, alerts, and encounter events to Parquet files in **MinIO** — ready for future analysis.
Throughout this story, the ward board showed Maria's deterioration in real time — her NEWS2 score rising, qSOFA flagging, SOFA alerting — giving the clinical team the information they needed to act fast. Throughout this story, the ward board showed Maria's deterioration in real time — her NEWS2 score rising, qSOFA flagging, SOFA alerting — giving the clinical team the information they needed to act fast. All of this was powered by the data flowing through PostgreSQL → Kafka → Redis/Elasticsearch/RabbitMQ/MinIO behind the scenes.