feature: Clinical Data Model Expansion & Observation Vocabulary

2026-06-18 15:00:42 +08:00
parent 3b4c5c524b
commit 7d630fbbd9
33 changed files with 2636 additions and 36 deletions
@@ -2,6 +2,8 @@

 A production-quality clinical backend built with ASP.NET Core 8, PostgreSQL, Apache Kafka, RabbitMQ, Elasticsearch, Redis, and MinIO. The domain models the observe-alert-acknowledge lifecycle at the center of any clinical monitoring system: patient encounters, continuous vital sign and lab result ingest, real-time sepsis detection, and clinician notification with automatic escalation.

+**Implementation status:** All nine planned phases are complete — from schema and CRUD through Kafka, Elasticsearch CQRS, sepsis detection, RabbitMQ paging with DLQ escalation, reconciliation jobs, Prometheus/Grafana observability, and the MinIO Parquet data lake. See [Implemented Phases](#implemented-phases) for the full breakdown.
+
 ## Domain Model — How It Maps to a Real Clinical System

 In a hospital, a patient presents for care and an encounter is opened. Bedside monitors and lab systems post observations continuously against that encounter. A rules engine evaluates each observation against configured thresholds and flags abnormal values as clinical alerts. Clinicians acknowledge and resolve alerts. If a critical alert goes unacknowledged for five minutes, the system escalates to the on-call backup. All events flow through Kafka so the Elasticsearch dashboard, sepsis engine, and data lake writer consume the same stream independently.
@@ -58,7 +60,7 @@ An `OutboxEvent` is written in the same transaction as any observation or alert,
 - **Data Lake Writer** — `DataLakeWriterService` (consumer group `data-lake-writer`) buffers `observation.recorded`, `alert.generated`, and `encounter.status.changed` events, flushes date-partitioned Parquet files to MinIO (`/observations/`, `/alerts/`, `/encounters/`), and commits Kafka offsets only after successful uploads; `kafka_partition` and `kafka_offset` columns provide audit lineage
 - **Reconciliation Jobs** — three scheduled checks: (1) unacknowledged CRITICAL alerts older than 30 minutes, (2) pending orders without results after 4 hours, (3) active inpatients with no observation in 2 hours; each finding creates a `reconciliation_alerts` row and publishes to RabbitMQ
 - **Standard Envelope** — all responses use a consistent `{ success, statusCode, data, error }` wrapper; validation errors use the same shape; `ApiBehaviorOptions` overridden so model validation also produces the standard envelope
- **Observability** — Serilog structured logging enriched with `correlationId`, `encounterId`, `patientId` on alert paths; Seq sink (`http://localhost:5345`); Prometheus (`http://localhost:9101`) scrapes `GET /metrics`; Grafana dashboards (`http://localhost:3101`, admin/admin) for clinical metrics including `alerts_unacknowledged_gauge`; per-request correlation IDs in request logs and response headers
+- **Observability** — Serilog structured logging enriched with `correlationId`, `encounterId`, `patientId` on alert paths; Seq sink (`http://localhost:5345`); Prometheus (`http://localhost:9101`) scrapes `GET /metrics`; eight application metric families via `ClinicalMetrics` and three background collectors (`AlertsUnacknowledgedCollector`, `OutboxPendingCollector`, `KafkaConsumerLagCollector`); Grafana clinical dashboard (`http://localhost:3101`, admin/admin) with `alerts_unacknowledged_gauge` as the primary safety panel; per-request correlation IDs in request logs and `X-Correlation-Id` response headers
 - **Swagger UI** — OpenAPI spec via Swashbuckle (Development only)

 ---
@@ -89,6 +91,10 @@ IHostedServices (background):
  DischargeSummaryWorkerService → RabbitMQ discharge.queue → MinIO PDF
  DataLakeWriterService  → Kafka (data-lake-writer) → Parquet files in MinIO
  ReconciliationScheduler → three scheduled safety checks → reconciliation_alerts + RabbitMQ
+  AlertsUnacknowledgedCollector → polls PostgreSQL every 30s → alerts_unacknowledged_gauge
+  OutboxPendingCollector     → polls outbox every 30s → outbox_pending_events
+  KafkaConsumerLagCollector  → polls four consumer groups every 30s → kafka_consumer_lag
+  ClinicalMetrics (singleton) → inline counters/histogram from ingest, SIRS, escalation paths
 ```

 **Why Kafka and RabbitMQ coexist:** Kafka is an append-only log — the same observation event reaches the Elasticsearch indexer, the sepsis engine, and the data lake independently without coordination. Each consumer holds its own offset and can replay from the beginning. RabbitMQ handles the action side: one message, one worker, one page. A duplicate page at 3am is a patient safety concern, not a minor inconvenience — RabbitMQ's acknowledgment-then-delete model is correct here. The DLQ TTL-based escalation has no equivalent in Kafka.
@@ -107,9 +113,11 @@ IHostedServices (background):
 | Search / analytics | Elasticsearch 8.13 (CQRS read projection) |
 | Data lake | MinIO (Parquet, S3-compatible) |
 | Logging | Serilog + Seq sink |
-| Metrics / dashboards | Prometheus 2.52 + Grafana 10.4 |
+| Metrics | prometheus-net.AspNetCore (`GET /metrics`) |
+| Dashboards | Prometheus 2.52 + Grafana 10.4 |
+| Data lake format | Parquet.Net 4.x |
 | Docs | Swagger / OpenAPI (Swashbuckle) |
-| Testing | xUnit + Testcontainers |
+| Testing | xUnit + Testcontainers + WebApplicationFactory |

 ---

@@ -144,19 +152,30 @@ VigilCareClinicalAPI/
 │       ├── AlertType.cs                        # ThresholdBreach, SepsisWarning, …
 │       ├── ObservationSource.cs                # Device, Manual, Lab
 │       └── OrderType.cs / ReconciliationCheckType.cs / Department.cs / OrderStatus.cs
+│   └── Json/
+│       ├── ObservationSourceJsonConverter.cs
+│       └── DepartmentJsonConverter.cs
 ├── 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
+│   ├── ObservationService.cs                   # Ingest transaction: idempotency → plausibility → threshold → alert → outbox; emits Prometheus counters
 │   ├── ObservationQueryService.cs              # Cursor-paginated history
 │   ├── AlertService.cs                         # Acknowledge, resolve, list
 │   ├── AnalyticsService.cs                     # Elasticsearch query wrappers
 │   └── PlausibilityValidator.cs                # Per-code numeric range guard
+├── Observability/
+│   └── Metrics/
+│       └── ClinicalMetrics.cs                  # Eight Prometheus metric families (counters, histogram, gauges)
 ├── BackgroundServices/
 │   ├── ThresholdCacheLoader.cs                 # Pre-loads all thresholds into Redis on startup
 │   ├── KafkaTopicProvisioner.cs                # Creates topics with NumPartitions from config
 │   ├── OutboxRelayService.cs                   # Polls outbox every 500ms; publishes to Kafka; marks processed
+│   ├── Metrics/
+│   │   ├── AlertsUnacknowledgedCollector.cs    # Polls open CRITICAL alerts > 5 min → alerts_unacknowledged_gauge
+│   │   ├── OutboxPendingCollector.cs           # Polls unprocessed outbox rows → outbox_pending_events
+│   │   └── KafkaConsumerLagCollector.cs        # Lag for es-indexer, sepsis-engine, notification-publisher, data-lake-writer
 │   ├── ElasticsSearch/
 │   │   ├── ElasticIndexProvisioner.cs          # Creates patient_encounters, observations, clinical_alerts indices
 │   │   └── EsIndexerService.cs                 # consumer group: es-indexer; upserts Elasticsearch documents
@@ -172,6 +191,10 @@ VigilCareClinicalAPI/
 │       ├── PendingOrdersCheck.cs               # Pending orders without results > 4 hours
 │       ├── DisconnectedMonitorsCheck.cs      # Active inpatients with no observation > 2 hours
 │       └── ReconciliationPublisher.cs          # Publishes findings to notifications.reconciliation.queue
+├── Configuration/
+│   ├── KafkaOptions.cs / KafkaTopicOptions.cs
+│   ├── RabbitMqOptions.cs / MinioOptions.cs
+│   └── ReconciliationJobOptions.cs
 ├── Sepsis/
 │   ├── SirsDetector.cs                         # Redis SIRS state management (SET/DEL/MGET)
 │   └── SirsEvaluator.cs                        # Per-code criterion evaluation
@@ -193,7 +216,7 @@ VigilCareClinicalAPI/
 │   └── Encounter/EncounterStatusRow.cs         # Parquet row contract for encounter status events
 ├── Data/
 │   ├── AppDbContext.cs                         # EF Core context — entity configs, indexes, constraints
-│   ├── Configurations/                         # IEntityTypeConfiguration per entity
+│   ├── Configurations/                         # IEntityTypeConfiguration per entity; ElasticsearchOptions, ElasticIndexOptions
 │   └── Seed/DataSeeder.cs                      # Seeds patients, encounters, thresholds, observations
 ├── Common/
 │   ├── ApiResponse.cs                          # { success, statusCode, data, error } envelope
@@ -227,12 +250,22 @@ tests/
    └── DataLakePhase9Tests.cs                  # Kafka → MinIO Parquet flow and schema checks

 scripts/
-├── run-phase8-verification.sh                  # Prometheus + alerts_unacknowledged_gauge checks
-└── run-phase9-verification.sh                  # Data lake integration tests + Kafka + MinIO + DuckDB
+├── 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

 docs/
-├── plans/phase-9-plan.md                       # Phase 9 implementation and verification guide
-└── decisions/data-lake-design.md               # Parquet vs JSON, partitioning, replay rationale
+├── plans/                                      # Phase 1–9 implementation and verification guides
+├── decisions/
+│   ├── data-lake-design.md                     # Parquet vs JSON, partitioning, replay rationale
+│   └── sepsis-engine-design.md                 # SIRS sliding window and idempotent alert design
+├── docker-compose-usage-and-troubleshooting.md
+└── vigilcare-clinical-api-prd.md               # Product requirements and phase roadmap
 ```

 ---
@@ -345,10 +378,10 @@ On startup the application:
 3. Pre-loads all thresholds into Redis
 4. Provisions Kafka topics and Elasticsearch indices
 5. Declares the RabbitMQ exchange and queue topology
-6. Starts the data lake writer (`data-lake-writer` → Parquet in MinIO)
-7. Starts the reconciliation scheduler (three safety checks on a configurable interval)
+6. Starts all background consumers (outbox relay, ES indexer, sepsis engine, notification workers, data lake writer, reconciliation scheduler)
+7. Starts Prometheus metric collectors (unacknowledged alerts, outbox pending, Kafka consumer lag)

-Swagger UI is available at `http://localhost:<port>/swagger` in Development.
+Swagger UI is available at `http://localhost:5270/swagger` in Development (API binds to `0.0.0.0:5270` per `launchSettings.json`).

 ### Run Tests

@@ -356,15 +389,36 @@ Swagger UI is available at `http://localhost:<port>/swagger` in Development.
 dotnet test
 ```

-Tests use Testcontainers to spin up a real PostgreSQL instance. No manual setup required.
+Integration tests use `WebApplicationFactory` with a `Testing` environment and Testcontainers where needed (PostgreSQL, Redis, Kafka, RabbitMQ, Elasticsearch, MinIO). No manual infrastructure setup is required for `dotnet test`.
+
+| Test class | Phase | Coverage |
+|---|---|---|
+| `ObservationIngestTests` | 2 | Ingest happy path, critical alert creation, discharged encounter rejection, idempotency |
+| `AlertLifecycleTests` | 2 | Acknowledge, resolve, escalation guard |
+| `SirsDetectorTests` / `SirsEvaluatorTests` | 5 | Redis SIRS state and per-code criterion evaluation |
+| `NotificationPipelineTests` | 6 | RabbitMQ topology, DLQ routing, paging |
+| `ReconciliationTests` | 7 | Three reconciliation checks, deduplication, RabbitMQ publish |
+| `ObservabilityPhase8Tests` | 8 | All eight `/metrics` families, correlation headers, ingest counter increment |
+| `DataLakePhase9Tests` | 9 | Kafka → MinIO Parquet flow and schema checks |

 ### Verification Scripts

-With the API running and Docker Compose up, run phase verification end-to-end:
+With the API running (`dotnet run`) and Docker Compose up:

 ```bash
-./scripts/run-phase8-verification.sh   # Prometheus metrics, alerts_unacknowledged_gauge
-./scripts/run-phase9-verification.sh   # Data lake tests, Kafka offsets, MinIO Parquet, DuckDB schema
+./scripts/run-phase8-verification.sh   # Prometheus target UP, eight metrics, alerts_unacknowledged_gauge live update
+./scripts/run-phase9-verification.sh   # DataLakePhase9Tests, Kafka consumer group, MinIO Parquet, DuckDB schema
+```
+
+Per-phase test runners (subset of `dotnet test`):
+
+```bash
+./scripts/run-api-redis-tests.sh
+./scripts/run-kafka-outbox-tests.sh
+./scripts/run-elasticsearch-analytics-tests.sh
+./scripts/run-sepsis-sirs-tests.sh
+./scripts/run-notification-pipeline-tests.sh
+./scripts/run-reconciliation-tests.sh
 ```

 Phase 9 optional tools (install without sudo):
@@ -380,7 +434,26 @@ curl https://install.duckdb.org | sh
 export PATH="$HOME/.duckdb/cli/latest:$HOME/.local/bin:$PATH"
 ```

-See `docs/plans/phase-9-plan.md` for manual Kafka replay and DuckDB query examples.
+See `docs/plans/phase-8-plan.md` and `docs/plans/phase-9-plan.md` for manual Grafana, Seq, Kafka replay, and DuckDB query examples.
+
+---
+
+## Prometheus Metrics
+
+`GET /metrics` exposes eight application metric families registered in `ClinicalMetrics`. Three background collectors poll PostgreSQL and Kafka every 30 seconds; counters and the ingest histogram are updated inline during request handling.
+
+| Metric | Type | Labels | Source |
+|---|---|---|---|
+| `observations_ingested_total` | Counter | `observation_code`, `source` | `ObservationService` on each committed observation |
+| `observation_ingest_duration_seconds` | Histogram | — | `ObservationService` — full ingest transaction to COMMIT |
+| `clinical_alerts_total` | Counter | `alert_type`, `severity` | `ObservationService` (threshold breach), `SirsDetector` (sepsis) |
+| `sirs_detections_total` | Counter | — | `SirsDetector` — only on successful idempotent insert |
+| `escalations_total` | Counter | — | `EscalationWorkerService` on DLQ escalation |
+| `alerts_unacknowledged_gauge` | Gauge | — | `AlertsUnacknowledgedCollector` — open CRITICAL alerts older than 5 minutes |
+| `outbox_pending_events` | Gauge | — | `OutboxPendingCollector` — unprocessed outbox rows |
+| `kafka_consumer_lag` | Gauge | `consumer_group` | `KafkaConsumerLagCollector` — `es-indexer`, `sepsis-engine`, `notification-publisher`, `data-lake-writer` |
+
+Prometheus scrapes the API via `infra/prometheus/prometheus.yml` (`job: vigilcare_api` → `host.docker.internal:5270`). Grafana loads the clinical dashboard from `infra/grafana/dashboards/vigilcare.json`.

 ---

@@ -824,6 +897,30 @@ The indices rebuild from the full Kafka history. Document count should match Pos

 ---

+## Data Lake Replay
+
+The MinIO Parquet archive is a pure Kafka projection — rebuildable without touching PostgreSQL. See `docs/plans/phase-9-plan.md` for the full procedure. Summary:
+
+```bash
+# Reset data-lake-writer offsets to earliest
+docker compose exec -T kafka /opt/kafka/bin/kafka-consumer-groups.sh \
+  --bootstrap-server localhost:9092 \
+  --group data-lake-writer \
+  --reset-offsets --to-earliest --all-topics --execute
+
+# Clear Parquet prefixes in MinIO (mc alias localvc http://localhost:9005 minioadmin minioadmin)
+mc rm --recursive --force localvc/vigilcare/observations/
+mc rm --recursive --force localvc/vigilcare/alerts/
+mc rm --recursive --force localvc/vigilcare/encounters/
+
+# Restart the API — DataLakeWriterService replays from offset 0
+dotnet run
+```
+
+Verify with `./scripts/run-phase9-verification.sh`.
+
+---
+
 ## Pagination

 List endpoints use offset pagination:
@@ -851,6 +948,8 @@ Observation history uses cursor pagination on `(recorded_at DESC, id DESC)`. Off

 ## Implemented Phases

+All nine phases from the project roadmap are implemented and covered by integration tests and/or verification scripts.
+
 | Phase | Feature | Status |
 |---|---|---|
 | 1 | Schema, EF Core migrations, patient/encounter CRUD, alert threshold CRUD, encounter status machine, Redis threshold pre-load, seed data | Done |
@@ -860,5 +959,7 @@ Observation history uses cursor pagination on `(recorded_at DESC, id DESC)`. Off
 | 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`); Grafana dashboards; eight application metric families | In progress |
-| 9 | Data lake writer — `data-lake-writer` consumer group; Parquet flush to MinIO; `DataLakePhase9Tests`; `run-phase9-verification.sh` | Done |
+| 8 | Prometheus metrics (`GET /metrics`); eight metric families and three collectors; Grafana clinical dashboard; `ObservabilityPhase8Tests`; `run-phase8-verification.sh` | Done |
+| 9 | Data lake writer — `data-lake-writer` consumer group; date-partitioned Parquet flush to MinIO; `DataLakePhase9Tests`; `run-phase9-verification.sh`; design doc in `docs/decisions/data-lake-design.md` | Done |
+
+**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.