tests: MinIO Data Lake Writer (Parquet, Partitioned) test

README.md

@@ -55,7 +55,7 @@ An `OutboxEvent` is written in the same transaction as any observation or alert,
 - **Elasticsearch CQRS Projection** — `EsIndexerService` consumer group upserts `patient_encounters` documents, appends to the `observations` index, and updates `openAlertCount` on alert events; patient/encounter search; per-encounter observation trend (hourly avg/min/max); alert volume summary by department and severity; population query (numeric range aggregation across all patients)
 - **Sepsis Early Warning Engine** — `SepsisEngineService` Kafka consumer evaluates SIRS criteria (temperature, heart rate, respiratory rate, WBC) per encounter using Redis keys with a 30-minute TTL sliding window; on ≥2 active criteria, inserts a `SEPSIS_WARNING / CRITICAL` alert idempotently (`INSERT WHERE NOT EXISTS`)
 - **RabbitMQ Notification Workers** — `NotificationPublisherService` reads `alert.generated` from Kafka and publishes paging jobs to `alerts.paging.queue`; `PagingWorkerService` sends the page and waits for acknowledgment; if no ack arrives before timeout it NACKs to `alerts.paging.dlq` with `x-message-ttl = 300000ms`; if the host is stopping, in-flight paging messages are NACKed with `requeue=true` so they are retried after restart and do not false-escalate; `EscalationWorkerService` pages the on-call backup and sets alert status to `escalated`; `DischargeSummaryWorkerService` reads `encounter.status.changed`, generates a discharge summary, and stores it in MinIO under `/discharge-summaries/{encounterId}/summary.pdf`
-- **Data Lake Writer (Phase 9 - in progress)** — `DataLakeWriterService` Kafka consumer buffers `observation.recorded`, `alert.generated`, and `encounter.status.changed` events, flushes partitioned Parquet files to MinIO (`/observations/`, `/alerts/`, `/encounters/` by date), and commits offsets after successful uploads
+- **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
@@ -225,6 +225,14 @@ tests/
     ├── 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
+
+scripts/
+├── run-phase8-verification.sh                  # Prometheus + alerts_unacknowledged_gauge checks
+└── run-phase9-verification.sh                  # Data lake integration tests + Kafka + MinIO + DuckDB
+
+docs/
+├── plans/phase-9-plan.md                       # Phase 9 implementation and verification guide
+└── decisions/data-lake-design.md               # Parquet vs JSON, partitioning, replay rationale
 ```
 
 ---
@@ -337,7 +345,8 @@ 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 reconciliation scheduler (three safety checks on a configurable interval)
+6. Starts the data lake writer (`data-lake-writer` → Parquet in MinIO)
+7. Starts the reconciliation scheduler (three safety checks on a configurable interval)
 
 Swagger UI is available at `http://localhost:<port>/swagger` in Development.
 
@@ -349,6 +358,30 @@ dotnet test
 
 Tests use Testcontainers to spin up a real PostgreSQL instance. No manual setup required.
 
+### Verification Scripts
+
+With the API running and Docker Compose up, run phase verification end-to-end:
+
+```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
+```
+
+Phase 9 optional tools (install without sudo):
+
+```bash
+# MinIO client — object listing and download
+mkdir -p ~/.local/bin
+curl -fsSL https://dl.min.io/client/mc/release/linux-amd64/mc -o ~/.local/bin/mc
+chmod +x ~/.local/bin/mc
+
+# DuckDB — query Parquet files locally
+curl https://install.duckdb.org | sh
+export PATH="$HOME/.duckdb/cli/latest:$HOME/.local/bin:$PATH"
+```
+
+See `docs/plans/phase-9-plan.md` for manual Kafka replay and DuckDB query examples.
+
 ---
 
 ## API Reference
@@ -828,4 +861,4 @@ Observation history uses cursor pagination on `(recorded_at DESC, id DESC)`. Off
 | 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 — Kafka consumer group `data-lake-writer`; Parquet flush to MinIO; integration tests (`DataLakePhase9Tests`) | In progress |
+| 9 | Data lake writer — `data-lake-writer` consumer group; Parquet flush to MinIO; `DataLakePhase9Tests`; `run-phase9-verification.sh` | Done |

docs/docker-compose-usage-and-troubleshooting.md

@@ -306,10 +306,13 @@ Note: if you accidentally create a typo folder like `dashbpards`, Grafana provis
    docker compose up -d --force-recreate <service>
    ```
 3. Verify logs and health endpoints.
-4. Run project verification scripts (example):
+4. Run project verification scripts (examples):
    ```bash
    ./scripts/run-phase8-verification.sh
+   ./scripts/run-phase9-verification.sh
    ```
 
+   Phase 9 also benefits from the MinIO client (`mc`) and DuckDB CLI for object and Parquet checks. Install without sudo — see `docs/plans/phase-9-plan.md`.
+
 This avoids unnecessary full resets and speeds up local development.
 

scripts/run-phase9-verification.sh

@@ -0,0 +1,173 @@
+#!/usr/bin/env bash
+
+set -euo pipefail
+
+SCRIPT_DIR="$(cd "$(dirname "${BASH_SOURCE[0]}")" && pwd)"
+ROOT_DIR="$(cd "${SCRIPT_DIR}/.." && pwd)"
+COMPOSE_FILE="${COMPOSE_FILE:-${ROOT_DIR}/docker-compose.yml}"
+
+TEST_PROJECT="${TEST_PROJECT:-${ROOT_DIR}/VigilCareClinicalAPI.Tests/VigilCareClinicalAPI.Tests.csproj}"
+TEST_FILTER="${TEST_FILTER:-FullyQualifiedName~DataLakePhase9Tests}"
+BASE_URL="${BASE_URL:-http://localhost:5270}"
+MINIO_ALIAS="${MINIO_ALIAS:-localvc}"
+MINIO_URL="${MINIO_URL:-http://localhost:9005}"
+MINIO_BUCKET="${MINIO_BUCKET:-vigilcare}"
+MINIO_ACCESS_KEY="${MINIO_ACCESS_KEY:-minioadmin}"
+MINIO_SECRET_KEY="${MINIO_SECRET_KEY:-minioadmin}"
+KAFKA_GROUP="${KAFKA_GROUP:-data-lake-writer}"
+KAFKA_SERVICE="${KAFKA_SERVICE:-kafka}"
+KAFKA_BOOTSTRAP="${KAFKA_BOOTSTRAP:-localhost:9092}"
+
+need() {
+  command -v "$1" >/dev/null 2>&1 || {
+    echo "Missing dependency: $1"
+    exit 1
+  }
+}
+
+optional() {
+  command -v "$1" >/dev/null 2>&1
+}
+
+# Prefer MC_BIN, then user-local install, then first executable on PATH.
+resolve_mc() {
+  if [[ -n "${MC_BIN:-}" && -x "${MC_BIN}" ]]; then
+    echo "${MC_BIN}"
+    return 0
+  fi
+  local candidate="${HOME}/.local/bin/mc"
+  if [[ -x "${candidate}" ]]; then
+    echo "${candidate}"
+    return 0
+  fi
+  if command -v mc >/dev/null 2>&1; then
+    local path_mc
+    path_mc="$(command -v mc)"
+    if [[ -x "${path_mc}" ]]; then
+      echo "${path_mc}"
+      return 0
+    fi
+  fi
+  return 1
+}
+
+# Prefer DUCKDB_BIN, then official install location, then PATH.
+resolve_duckdb() {
+  if [[ -n "${DUCKDB_BIN:-}" && -x "${DUCKDB_BIN}" ]]; then
+    echo "${DUCKDB_BIN}"
+    return 0
+  fi
+  local candidate="${HOME}/.duckdb/cli/latest/duckdb"
+  if [[ -x "${candidate}" ]]; then
+    echo "${candidate}"
+    return 0
+  fi
+  if command -v duckdb >/dev/null 2>&1; then
+    local path_duckdb
+    path_duckdb="$(command -v duckdb)"
+    if [[ -x "${path_duckdb}" ]]; then
+      echo "${path_duckdb}"
+      return 0
+    fi
+  fi
+  return 1
+}
+
+need curl
+need dotnet
+need docker
+
+echo "Phase 9 verification starting..."
+echo "Repo root: ${ROOT_DIR}"
+
+echo "[1/6] Preflight API"
+api_status="$(curl -sS -o /dev/null -w "%{http_code}" "${BASE_URL}/metrics" || true)"
+[[ "${api_status}" == "200" ]] || {
+  echo "API /metrics is not reachable at ${BASE_URL} (status=${api_status})"
+  exit 1
+}
+
+echo "[2/6] Run Phase 9 integration tests"
+dotnet test "${TEST_PROJECT}" --filter "${TEST_FILTER}"
+
+echo "[3/6] Verify Kafka consumer group appears"
+group_list="$(docker compose -f "${COMPOSE_FILE}" exec -T "${KAFKA_SERVICE}" \
+  /opt/kafka/bin/kafka-consumer-groups.sh --bootstrap-server "${KAFKA_BOOTSTRAP}" --list)"
+if ! grep -q "^${KAFKA_GROUP}$" <<< "${group_list}"; then
+  echo "Consumer group '${KAFKA_GROUP}' not found."
+  echo "Known groups:"
+  echo "${group_list}"
+  exit 1
+fi
+
+echo "[4/6] Describe Kafka consumer group offsets"
+docker compose -f "${COMPOSE_FILE}" exec -T "${KAFKA_SERVICE}" \
+  /opt/kafka/bin/kafka-consumer-groups.sh \
+  --bootstrap-server "${KAFKA_BOOTSTRAP}" \
+  --group "${KAFKA_GROUP}" \
+  --describe
+
+MC="$(resolve_mc || true)"
+if [[ -z "${MC}" ]]; then
+  echo "[5/6] Skipping MinIO checks: 'mc' not installed."
+  echo "Install without sudo:"
+  echo "  mkdir -p ~/.local/bin"
+  echo "  curl -fsSL https://dl.min.io/client/mc/release/linux-amd64/mc -o ~/.local/bin/mc"
+  echo "  chmod +x ~/.local/bin/mc"
+  echo "  export PATH=\"\$HOME/.local/bin:\$PATH\""
+  echo
+  echo "Phase 9 verification completed (tests + Kafka checks)."
+  exit 0
+fi
+
+echo "[5/6] Verify Parquet objects exist in MinIO (using ${MC})"
+"${MC}" alias set "${MINIO_ALIAS}" "${MINIO_URL}" "${MINIO_ACCESS_KEY}" "${MINIO_SECRET_KEY}" >/dev/null
+
+check_prefix() {
+  local prefix="$1"
+  local count
+  count="$("${MC}" ls --recursive "${MINIO_ALIAS}/${MINIO_BUCKET}/${prefix}" | awk 'END{print NR+0}')"
+  if [[ "${count}" -le 0 ]]; then
+    echo "No objects found under ${prefix}"
+    exit 1
+  fi
+  echo "Found ${count} object(s) under ${prefix}"
+}
+
+check_prefix "observations/"
+check_prefix "alerts/"
+check_prefix "encounters/"
+
+DUCKDB="$(resolve_duckdb || true)"
+if [[ -z "${DUCKDB}" ]]; then
+  echo "[6/6] Skipping DuckDB schema check: 'duckdb' not installed."
+  echo "Install without sudo: curl https://install.duckdb.org | sh"
+  echo "Then add to PATH: export PATH=\"\$HOME/.duckdb/cli/latest:\$PATH\""
+  echo
+  echo "Phase 9 verification completed (tests + Kafka + MinIO checks)."
+  exit 0
+fi
+
+echo "[6/6] Verify Parquet schema with DuckDB (using ${DUCKDB})"
+first_obs_rel="$("${MC}" ls --recursive "${MINIO_ALIAS}/${MINIO_BUCKET}/observations/" | awk '/STANDARD/{print $NF; exit}')"
+[[ -n "${first_obs_rel}" ]] || {
+  echo "Could not resolve an observations Parquet key."
+  exit 1
+}
+first_obs_key="observations/${first_obs_rel}"
+
+tmp_parquet="$(mktemp /tmp/vc-phase9-obs-XXXXXX.parquet)"
+trap 'rm -f "${tmp_parquet}"' EXIT
+"${MC}" cp "${MINIO_ALIAS}/${MINIO_BUCKET}/${first_obs_key}" "${tmp_parquet}" >/dev/null
+
+"${DUCKDB}" -csv -c "DESCRIBE SELECT * FROM '${tmp_parquet}';" | tee /tmp/vc-phase9-duckdb-schema.txt >/dev/null
+
+for col in observation_id encounter_id kafka_partition kafka_offset; do
+  if ! grep -qE "^${col}," /tmp/vc-phase9-duckdb-schema.txt; then
+    echo "Expected column '${col}' not found in Parquet schema."
+    exit 1
+  fi
+done
+
+echo
+echo "Phase 9 verification checks passed."