vigilcare-clinical/scripts/run-reconciliation-tests.sh

#!/usr/bin/env bash
# Phase 7 verification — reconciliation jobs (see docs/plans/phase-7-plan.md).
#
# Prerequisites:
#   docker compose up -d
#   dotnet run --project VigilCareClinicalAPI
#
# For a practical runtime, set ReconciliationJobs.IntervalMinutes to 1 in appsettings.json
# and restart the API before running this script (default 30 min is too slow).

set -euo pipefail

SCRIPT_DIR="$(cd "$(dirname "${BASH_SOURCE[0]}")" && pwd)"
COMPOSE_FILE="${COMPOSE_FILE:-${SCRIPT_DIR}/../docker-compose.yml}"

BASE_URL="${BASE_URL:-http://localhost:5270}"
RABBITMQ_MGMT_URL="${RABBITMQ_MGMT_URL:-http://localhost:15674}"
RABBITMQ_USER="${RABBITMQ_USER:-guest}"
RABBITMQ_PASS="${RABBITMQ_PASS:-guest}"

PGHOST="${PGHOST:-localhost}"
PGPORT="${PGPORT:-5436}"
PGDATABASE="${PGDATABASE:-vigilcare}"
PGUSER="${PGUSER:-postgres}"
PGPASSWORD="${PGPASSWORD:-password}"

# Scheduler waits: first tick ~15s after API start, then every IntervalMinutes.
RECONCILIATION_WAIT_SECS="${RECONCILIATION_WAIT_SECS:-120}"
RECONCILIATION_CYCLE_WAIT_SECS="${RECONCILIATION_CYCLE_WAIT_SECS:-70}"

SCRIPT_RUN_ID="$(date -u +"%Y%m%d%H%M%S")"
RECORDED_AT="$(date -u +"%Y-%m-%dT%H:%M:%SZ")"

CHECK1="UNACKNOWLEDGED_CRITICAL_ALERT"
CHECK2="PENDING_ORDER_NO_RESULT"
CHECK3="ACTIVE_INPATIENT_NO_OBSERVATION"

EXPECTED_QUEUES=(
  "alerts.paging.queue"
  "alerts.paging.dlq"
  "alerts.escalation.queue"
  "notifications.discharge.queue"
  "notifications.appointment.queue"
  "notifications.reconciliation.queue"
)

TMP_FILES=()

cleanup() {
  local f
  for f in "${TMP_FILES[@]}"; do
    rm -f "${f}" "${f}.status" 2>/dev/null || true
  done
}
trap cleanup EXIT

if ! command -v curl >/dev/null 2>&1; then
  echo "Missing dependency: curl"
  exit 1
fi

if ! command -v jq >/dev/null 2>&1; then
  echo "Missing dependency: jq"
  exit 1
fi

compose() {
  docker compose -f "${COMPOSE_FILE}" "$@"
}

psql_cmd() {
  local sql="$1"
  if command -v psql >/dev/null 2>&1; then
    PGPASSWORD="${PGPASSWORD}" psql -h "${PGHOST}" -p "${PGPORT}" -U "${PGUSER}" -d "${PGDATABASE}" -tAc "${sql}"
  else
    compose exec -T postgres psql -U "${PGUSER}" -d "${PGDATABASE}" -tAc "${sql}"
  fi
}

request() {
  local method="$1"
  local url="$2"
  local body="${3:-}"
  local tmp_body
  tmp_body="$(mktemp)"
  TMP_FILES+=("${tmp_body}")
  local status

  if [[ -n "${body}" ]]; then
    status="$(curl -sS -o "${tmp_body}" -w "%{http_code}" -X "${method}" "${url}" \
      -H "Content-Type: application/json" -d "${body}")"
  else
    status="$(curl -sS -o "${tmp_body}" -w "%{http_code}" -X "${method}" "${url}")"
  fi

  echo "${status}" > "${tmp_body}.status"
  echo "${tmp_body}"
}

assert_status() {
  local expected="$1"
  local body_file="$2"
  local status
  status="$(cat "${body_file}.status")"
  if [[ "${status}" != "${expected}" ]]; then
    echo "Expected HTTP ${expected}, got ${status}" >&2
    echo "Response body:" >&2
    cat "${body_file}" >&2
    echo >&2
    exit 1
  fi
}

rabbit_api() {
  curl -sS -u "${RABBITMQ_USER}:${RABBITMQ_PASS}" "${RABBITMQ_MGMT_URL}/api/${1}"
}

queue_field() {
  local queue="$1"
  local field="$2"
  rabbit_api "queues/%2F/${queue}" | jq -r ".${field} // 0"
}

wait_for_queue_increase() {
  local queue="$1"
  local baseline="$2"
  local timeout_secs="${3:-15}"
  local elapsed=0
  while (( elapsed < timeout_secs )); do
    local depth
    depth="$(queue_field "${queue}" "messages")"
    if [[ "${depth}" -gt "${baseline}" ]]; then
      echo "${depth}"
      return 0
    fi
    sleep 1
    elapsed=$((elapsed + 1))
  done
  echo "${depth:-${baseline}}"
  return 1
}

reconciliation_count() {
  local encounter_id="$1"
  local check_type="$2"
  psql_cmd "SELECT COUNT(*) FROM reconciliation_alerts
    WHERE encounter_id = '${encounter_id}'
      AND check_type = '${check_type}'
      AND resolved_at IS NULL"
}

wait_for_reconciliation_row() {
  local encounter_id="$1"
  local check_type="$2"
  local timeout_secs="$3"
  local elapsed=0
  while (( elapsed < timeout_secs )); do
    local count
    count="$(reconciliation_count "${encounter_id}" "${check_type}")"
    if [[ "${count}" -ge 1 ]]; then
      return 0
    fi
    sleep 5
    elapsed=$((elapsed + 5))
  done
  echo "No open reconciliation_alert (${check_type}) for encounter ${encounter_id} within ${timeout_secs}s"
  return 1
}

wait_for_scheduler_cycle() {
  echo "    (waiting ${RECONCILIATION_CYCLE_WAIT_SECS}s for reconciliation scheduler cycle...)"
  sleep "${RECONCILIATION_CYCLE_WAIT_SECS}"
}

create_patient() {
  local suffix="$1"
  local payload
  payload="$(jq -nc \
    --arg first "Recon${suffix}" \
    --arg last "Verify${SCRIPT_RUN_ID}" \
    '{firstName:$first,lastName:$last,dateOfBirth:"1970-06-01",gender:"Other"}')"
  local resp
  resp="$(request POST "${BASE_URL}/api/v1/patients" "${payload}")"
  assert_status "201" "${resp}"
  jq -r '.data.id' "${resp}"
}

create_encounter() {
  local patient_id="$1"
  local encounter_type="$2"
  local department="${3:-ICU}"
  local payload
  payload="$(jq -nc \
    --arg et "${encounter_type}" \
    --arg dept "${department}" \
    --arg physician "Dr. Recon ${SCRIPT_RUN_ID}" \
    '{encounterType:$et,department:$dept,attendingPhysician:$physician}')"
  local resp
  resp="$(request POST "${BASE_URL}/api/v1/patients/${patient_id}/encounters" "${payload}")"
  assert_status "201" "${resp}"
  jq -r '.data.id' "${resp}"
}

ingest_critical_potassium() {
  local encounter_id="$1"
  local idempotency_key="$2"
  local payload
  payload="$(jq -nc \
    --arg recordedAt "${RECORDED_AT}" \
    --arg key "${idempotency_key}" \
    '{observations:[{observationCode:"POTASSIUM_MEQ_L",value:2.1,unit:"mEq/L",source:"LAB",recordedAt:$recordedAt,idempotencyKey:$key}]}')"
  local resp
  resp="$(request POST "${BASE_URL}/api/v1/encounters/${encounter_id}/observations" "${payload}")"
  assert_status "201" "${resp}"
  if [[ "$(jq -r '.data.alertGenerated' "${resp}")" != "true" ]]; then
    echo "Expected critical potassium ingest to generate an alert"
    exit 1
  fi
  jq -r '.data.alertId' "${resp}"
}

ingest_heart_rate() {
  local encounter_id="$1"
  local recorded_at="$2"
  local idempotency_key="$3"
  local payload
  payload="$(jq -nc \
    --arg recordedAt "${recorded_at}" \
    --arg key "${idempotency_key}" \
    '{observations:[{observationCode:"HEART_RATE",value:72,unit:"bpm",source:"DEVICE",recordedAt:$recordedAt,idempotencyKey:$key}]}')"
  local resp
  resp="$(request POST "${BASE_URL}/api/v1/encounters/${encounter_id}/observations" "${payload}")"
  assert_status "201" "${resp}"
}

insert_stale_pending_order() {
  local encounter_id="$1"
  local description="$2"
  psql_cmd "INSERT INTO orders (id, encounter_id, order_type, description, ordered_by, status, ordered_at, resulted_at)
    VALUES (gen_random_uuid(), '${encounter_id}', 'LAB', '${description}', 'Dr. Osei', 'PENDING',
            NOW() - INTERVAL '5 hours', NULL)"
}

insert_resulted_order() {
  local encounter_id="$1"
  psql_cmd "INSERT INTO orders (id, encounter_id, order_type, description, ordered_by, status, ordered_at, resulted_at)
    VALUES (gen_random_uuid(), '${encounter_id}', 'LAB', 'CBC', 'Dr. Osei', 'RESULTED',
            NOW() - INTERVAL '5 hours', NOW() - INTERVAL '4 hours')"
}

backdate_alert_triggered_at() {
  local alert_id="$1"
  local minutes_ago="$2"
  psql_cmd "UPDATE clinical_alerts
    SET triggered_at = NOW() - INTERVAL '${minutes_ago} minutes'
    WHERE id = '${alert_id}'"
}

TOTAL_STEPS=11

echo "Running Phase 7 reconciliation verification against ${BASE_URL}"
echo "Script run id: ${SCRIPT_RUN_ID}"
echo "Hint: set ReconciliationJobs.IntervalMinutes=1 and restart the API for ~${RECONCILIATION_CYCLE_WAIT_SECS}s waits per cycle."

echo ""
echo "[0/${TOTAL_STEPS}] Preflight — API, Postgres, and RabbitMQ reachable"
preflight_status="$(curl -sS -o /dev/null -w "%{http_code}" "${BASE_URL}/api/v1/alert-thresholds" || true)"
if [[ "${preflight_status}" != "200" ]]; then
  echo "API not reachable at ${BASE_URL} (HTTP ${preflight_status})."
  echo "Start the API with: dotnet run --project VigilCareClinicalAPI"
  exit 1
fi

if ! psql_cmd "SELECT 1" >/dev/null 2>&1; then
  echo "Postgres not reachable on ${PGHOST}:${PGPORT}."
  echo "Start the stack with: docker compose up -d"
  exit 1
fi

rabbit_health="$(rabbit_api "health/checks/alarms" 2>/dev/null | jq -r '.status // empty' || true)"
if [[ -z "${rabbit_health}" ]]; then
  echo "RabbitMQ management API not reachable at ${RABBITMQ_MGMT_URL}."
  exit 1
fi
potassium_threshold="$(psql_cmd "SELECT COUNT(*) FROM alert_thresholds WHERE observation_code = 'POTASSIUM_MEQ_L'")"
heart_rate_threshold="$(psql_cmd "SELECT COUNT(*) FROM alert_thresholds WHERE observation_code = 'HEART_RATE'")"
if [[ "${potassium_threshold}" == "0" || "${heart_rate_threshold}" == "0" ]]; then
  echo "Missing alert thresholds (POTASSIUM_MEQ_L and/or HEART_RATE)."
  echo "Start the API once against an empty database so DataSeeder runs, or register thresholds via the API."
  exit 1
fi
echo "OK: API, Postgres, RabbitMQ, and alert thresholds are ready"

echo ""
echo "[1/${TOTAL_STEPS}] Verifying reconciliation_alerts migration"
table_exists="$(psql_cmd "SELECT EXISTS (
  SELECT 1 FROM information_schema.tables
  WHERE table_schema = 'public' AND table_name = 'reconciliation_alerts'
)")"
if [[ "${table_exists}" != "t" ]]; then
  echo "Table reconciliation_alerts not found. Run: dotnet ef database update --project VigilCareClinicalAPI"
  exit 1
fi

columns="$(psql_cmd "SELECT string_agg(column_name, ',' ORDER BY ordinal_position)
  FROM information_schema.columns
  WHERE table_name = 'reconciliation_alerts'")"
for col in check_type encounter_id patient_id details resolved_at created_at; do
  if [[ "${columns}" != *"${col}"* ]]; then
    echo "reconciliation_alerts missing column: ${col}"
    exit 1
  fi
done
echo "OK: reconciliation_alerts table and columns present"

echo ""
echo "[2/${TOTAL_STEPS}] Verifying RabbitMQ topology includes notifications.reconciliation.queue"
exchange_name="$(rabbit_api "exchanges/%2F/clinical.notifications.exchange" | jq -r '.name // empty')"
if [[ "${exchange_name}" != "clinical.notifications.exchange" ]]; then
  echo "Exchange clinical.notifications.exchange not found. Start the API so RabbitMqTopologyProvisioner runs."
  exit 1
fi

for queue in "${EXPECTED_QUEUES[@]}"; do
  queue_name="$(rabbit_api "queues/%2F/${queue}" | jq -r '.name // empty')"
  if [[ "${queue_name}" != "${queue}" ]]; then
    echo "Queue ${queue} not found (got '${queue_name}')."
    exit 1
  fi
done
recon_queue_before="$(queue_field "notifications.reconciliation.queue" "messages")"
echo "OK: 6 queues provisioned (including notifications.reconciliation.queue); messages=${recon_queue_before}"

echo ""
echo "[3/${TOTAL_STEPS}] Check 1 — stale critical alert creates reconciliation_alert"
patient_c1="$(create_patient "C1")"
encounter_c1="$(create_encounter "${patient_c1}" "Inpatient")"
alert_c1="$(ingest_critical_potassium "${encounter_c1}" "recon-c1-${SCRIPT_RUN_ID}")"
backdate_alert_triggered_at "${alert_c1}" 31

if ! wait_for_reconciliation_row "${encounter_c1}" "${CHECK1}" "${RECONCILIATION_WAIT_SECS}"; then
  echo "Hint: set ReconciliationJobs.IntervalMinutes=1 in appsettings.json and restart the API."
  exit 1
fi

details_c1="$(psql_cmd "SELECT details FROM reconciliation_alerts
  WHERE encounter_id = '${encounter_c1}' AND check_type = '${CHECK1}' AND resolved_at IS NULL
  ORDER BY created_at DESC LIMIT 1")"
if [[ "${details_c1}" != *"${alert_c1}"* ]]; then
  echo "Check 1 details should reference alert id ${alert_c1}"
  echo "Got: ${details_c1}"
  exit 1
fi
if ! recon_queue_after_c1="$(wait_for_queue_increase "notifications.reconciliation.queue" "${recon_queue_before}" 15)"; then
  echo "Expected messages on notifications.reconciliation.queue to increase after Check 1"
  echo "Before=${recon_queue_before}, after=${recon_queue_after_c1}"
  exit 1
fi
echo "OK: Check 1 reconciliation_alert created; queue depth ${recon_queue_before} -> ${recon_queue_after_c1}"

echo ""
echo "[4/${TOTAL_STEPS}] Check 1 — second scheduler cycle does not duplicate"
count_before_dup="$(reconciliation_count "${encounter_c1}" "${CHECK1}")"
wait_for_scheduler_cycle
count_after_dup="$(reconciliation_count "${encounter_c1}" "${CHECK1}")"
if [[ "${count_after_dup}" != "${count_before_dup}" ]]; then
  echo "Expected deduplication to keep count at ${count_before_dup}, got ${count_after_dup}"
  exit 1
fi
echo "OK: open reconciliation_alert count unchanged (${count_after_dup})"

echo ""
echo "[5/${TOTAL_STEPS}] Check 1 — acknowledged critical alert is not flagged"
patient_c1a="$(create_patient "C1A")"
encounter_c1a="$(create_encounter "${patient_c1a}" "Inpatient")"
alert_c1a="$(ingest_critical_potassium "${encounter_c1a}" "recon-c1a-${SCRIPT_RUN_ID}")"

ack_payload='{"clinicianId":"DR-RECON-SCRIPT","note":"Acknowledged during reconciliation verification."}'
resp="$(request POST "${BASE_URL}/api/v1/alerts/${alert_c1a}/acknowledge" "${ack_payload}")"
assert_status "200" "${resp}"

backdate_alert_triggered_at "${alert_c1a}" 31
wait_for_scheduler_cycle

if [[ "$(reconciliation_count "${encounter_c1a}" "${CHECK1}")" != "0" ]]; then
  echo "Acknowledged alert should not produce a Check 1 reconciliation_alert"
  exit 1
fi
echo "OK: acknowledged alert produced no Check 1 row"

echo ""
echo "[6/${TOTAL_STEPS}] Check 2 — stale pending order creates reconciliation_alert"
patient_c2="$(create_patient "C2")"
encounter_c2="$(create_encounter "${patient_c2}" "Inpatient")"
insert_stale_pending_order "${encounter_c2}" "Comprehensive metabolic panel"

if ! wait_for_reconciliation_row "${encounter_c2}" "${CHECK2}" "${RECONCILIATION_WAIT_SECS}"; then
  exit 1
fi
details_c2="$(psql_cmd "SELECT details FROM reconciliation_alerts
  WHERE encounter_id = '${encounter_c2}' AND check_type = '${CHECK2}' AND resolved_at IS NULL
  ORDER BY created_at DESC LIMIT 1")"
if [[ "${details_c2}" != *"Comprehensive metabolic panel"* ]]; then
  echo "Check 2 details should mention the stale order"
  echo "Got: ${details_c2}"
  exit 1
fi
echo "OK: Check 2 reconciliation_alert created for stale pending order"

echo ""
echo "[7/${TOTAL_STEPS}] Check 2 — resulted order is not flagged"
patient_c2r="$(create_patient "C2R")"
encounter_c2r="$(create_encounter "${patient_c2r}" "Inpatient")"
insert_resulted_order "${encounter_c2r}"
wait_for_scheduler_cycle

if [[ "$(reconciliation_count "${encounter_c2r}" "${CHECK2}")" != "0" ]]; then
  echo "Resulted order should not produce a Check 2 reconciliation_alert"
  exit 1
fi
echo "OK: resulted order produced no Check 2 row"

echo ""
echo "[8/${TOTAL_STEPS}] Check 3 — active inpatient with no observations is flagged"
patient_c3="$(create_patient "C3")"
encounter_c3="$(create_encounter "${patient_c3}" "Inpatient")"

if ! wait_for_reconciliation_row "${encounter_c3}" "${CHECK3}" "${RECONCILIATION_WAIT_SECS}"; then
  exit 1
fi
details_c3="$(psql_cmd "SELECT details FROM reconciliation_alerts
  WHERE encounter_id = '${encounter_c3}' AND check_type = '${CHECK3}' AND resolved_at IS NULL
  ORDER BY created_at DESC LIMIT 1")"
if [[ "${details_c3}" != *"no observations ever recorded"* ]]; then
  echo "Check 3 details should mention no observations ever recorded"
  echo "Got: ${details_c3}"
  exit 1
fi
echo "OK: Check 3 reconciliation_alert created for inpatient without observations"

echo ""
echo "[9/${TOTAL_STEPS}] Check 3 — stale observation (3 hours ago) is flagged"
patient_c3s="$(create_patient "C3S")"
encounter_c3s="$(create_encounter "${patient_c3s}" "Inpatient")"
stale_recorded_at="$(date -u -d '3 hours ago' +"%Y-%m-%dT%H:%M:%SZ" 2>/dev/null || date -u -v-3H +"%Y-%m-%dT%H:%M:%SZ")"
ingest_heart_rate "${encounter_c3s}" "${stale_recorded_at}" "recon-c3s-${SCRIPT_RUN_ID}"

if ! wait_for_reconciliation_row "${encounter_c3s}" "${CHECK3}" "${RECONCILIATION_WAIT_SECS}"; then
  exit 1
fi
echo "OK: Check 3 reconciliation_alert created for stale observation"

echo ""
echo "[10/${TOTAL_STEPS}] Check 3 — recent observation (30 minutes ago) is not flagged"
patient_c3r="$(create_patient "C3R")"
encounter_c3r="$(create_encounter "${patient_c3r}" "Inpatient")"
recent_recorded_at="$(date -u -d '30 minutes ago' +"%Y-%m-%dT%H:%M:%SZ" 2>/dev/null || date -u -v-30M +"%Y-%m-%dT%H:%M:%SZ")"
ingest_heart_rate "${encounter_c3r}" "${recent_recorded_at}" "recon-c3r-${SCRIPT_RUN_ID}"
wait_for_scheduler_cycle

if [[ "$(reconciliation_count "${encounter_c3r}" "${CHECK3}")" != "0" ]]; then
  echo "Recent observation should not produce a Check 3 reconciliation_alert"
  exit 1
fi
echo "OK: recent observation produced no Check 3 row"

echo ""
echo "[11/${TOTAL_STEPS}] Check 3 — outpatient encounter without observations is ignored"
patient_c3o="$(create_patient "C3O")"
encounter_c3o="$(create_encounter "${patient_c3o}" "Outpatient" "GeneralMedicine")"
wait_for_scheduler_cycle

if [[ "$(reconciliation_count "${encounter_c3o}" "${CHECK3}")" != "0" ]]; then
  echo "Outpatient encounter should not produce a Check 3 reconciliation_alert"
  exit 1
fi

# Plan verification query — candidate inpatients with no recent observations
candidate_count="$(psql_cmd "
  SELECT COUNT(*) FROM (
    SELECT e.id
    FROM encounters e
    LEFT JOIN observations o ON o.encounter_id = e.id
    WHERE e.status = 'ACTIVE' AND e.encounter_type = 'INPATIENT'
    GROUP BY e.id
    HAVING MAX(o.recorded_at) < NOW() - INTERVAL '2 hours'
        OR MAX(o.recorded_at) IS NULL
  ) t")"
echo "OK: outpatient not flagged; ${candidate_count} active inpatient(s) match Check 3 candidate query"

echo ""
echo "All ${TOTAL_STEPS} Phase 7 reconciliation checks passed."
echo ""
echo "Prerequisites: docker compose up -d && dotnet run --project VigilCareClinicalAPI"
echo "Recommended: ReconciliationJobs.IntervalMinutes=1 in appsettings.json (restart API before running)."
echo "Optional: RECONCILIATION_WAIT_SECS=180 RECONCILIATION_CYCLE_WAIT_SECS=75"