vigilcare-clinical/scripts/run-sepsis-sirs-tests.sh

#!/usr/bin/env bash

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}"
REDIS_PORT="${REDIS_PORT:-6382}"
ES_URL="${ES_URL:-http://localhost:9200}"
PGHOST="${PGHOST:-localhost}"
PGPORT="${PGPORT:-5436}"
PGDATABASE="${PGDATABASE:-vigilcare}"
PGUSER="${PGUSER:-postgres}"
PGPASSWORD="${PGPASSWORD:-password}"

SEPSIS_CONSUMER_GROUP="${SEPSIS_CONSUMER_GROUP:-sepsis-engine}"
ES_CONSUMER_GROUP="${ES_CONSUMER_GROUP:-es-indexer}"
SEPSIS_WAIT_SECS="${SEPSIS_WAIT_SECS:-45}"
INDEX_WAIT_SECS="${INDEX_WAIT_SECS:-60}"
RELAY_WAIT_SECS="${RELAY_WAIT_SECS:-45}"
KAFKA_READY_WAIT_SECS="${KAFKA_READY_WAIT_SECS:-30}"
TTL_DECAY_WAIT_SECS="${TTL_DECAY_WAIT_SECS:-5}"

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

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

if ! command -v docker >/dev/null 2>&1 || [[ ! -f "${COMPOSE_FILE}" ]]; then
  echo "Missing dependency: docker compose (${COMPOSE_FILE})"
  exit 1
fi

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

kafka_exec() {
  compose exec -T kafka "$@"
}

redis_cmd() {
  if command -v redis-cli >/dev/null 2>&1; then
    redis-cli -p "${REDIS_PORT}" "$@"
  else
    compose exec -T redis redis-cli "$@"
  fi
}

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}"
    echo "Response body:"
    cat "${body_file}"
    echo
    return 1
  fi
}

ingest_observation() {
  local encounter_id="$1"
  local code="$2"
  local value="$3"
  local unit="$4"
  local source="$5"
  local recorded_at="$6"
  local idempotency_key="$7"

  local payload
  payload="$(jq -nc \
    --arg code "${code}" \
    --argjson value "${value}" \
    --arg unit "${unit}" \
    --arg source "${source}" \
    --arg recordedAt "${recorded_at}" \
    --arg key "${idempotency_key}" \
    '{observations:[{observationCode:$code,value:$value,unit:$unit,source:$source,recordedAt:$recordedAt,idempotencyKey:$key}]}')"

  local resp
  resp="$(request POST "${BASE_URL}/api/v1/encounters/${encounter_id}/observations" "${payload}")"
  assert_status "201" "${resp}"
}

wait_for_kafka() {
  local elapsed=0
  while (( elapsed < KAFKA_READY_WAIT_SECS )); do
    if kafka_exec /opt/kafka/bin/kafka-broker-api-versions.sh --bootstrap-server localhost:9092 >/dev/null 2>&1; then
      return 0
    fi
    sleep 2
    elapsed=$((elapsed + 2))
  done
  echo "Kafka did not become ready within ${KAFKA_READY_WAIT_SECS}s"
  return 1
}

consumer_group_lag() {
  local group="$1"
  kafka_exec /opt/kafka/bin/kafka-consumer-groups.sh \
    --bootstrap-server localhost:9092 \
    --describe \
    --group "${group}" 2>/dev/null | \
    awk '/observation.recorded/ { sum += $6 } END { print sum + 0 }'
}

wait_for_consumer_lag_zero() {
  local group="$1"
  local elapsed=0
  while (( elapsed < SEPSIS_WAIT_SECS )); do
    local lag
    lag="$(consumer_group_lag "${group}")"
    if [[ "${lag}" == "0" ]]; then
      return 0
    fi
    sleep 1
    elapsed=$((elapsed + 1))
  done
  echo "Consumer group ${group} lag did not reach zero within ${SEPSIS_WAIT_SECS}s (lag=${lag:-unknown})"
  return 1
}

wait_for_sepsis_alert() {
  local encounter_id="$1"
  local elapsed=0
  while (( elapsed < SEPSIS_WAIT_SECS )); do
    local count
    count="$(psql_cmd "SELECT COUNT(*) FROM clinical_alerts WHERE encounter_id = '${encounter_id}' AND alert_type = 'SEPSIS_WARNING' AND status = 'OPEN'")"
    if [[ "${count}" == "1" ]]; then
      return 0
    fi
    sleep 1
    elapsed=$((elapsed + 1))
  done
  echo "SEPSIS_WARNING alert not found for encounter ${encounter_id} within ${SEPSIS_WAIT_SECS}s"
  return 1
}

wait_for_outbox_processed() {
  local outbox_id="$1"
  local elapsed=0
  while (( elapsed < RELAY_WAIT_SECS )); do
    local processed
    processed="$(psql_cmd "SELECT processed_at IS NOT NULL FROM outbox_events WHERE id = '${outbox_id}'")"
    if [[ "${processed}" == "t" ]]; then
      return 0
    fi
    sleep 1
    elapsed=$((elapsed + 1))
  done
  echo "Outbox row ${outbox_id} was not processed within ${RELAY_WAIT_SECS}s"
  return 1
}

latest_observation_outbox_id() {
  local encounter_id="$1"
  psql_cmd "SELECT id FROM outbox_events WHERE topic = 'observation.recorded' AND partition_key = '${encounter_id}' ORDER BY created_at DESC LIMIT 1"
}

# Observation ingest writes outbox first; relay publishes to Kafka; sepsis-engine consumes.
# Consumer lag alone is not enough — lag can be 0 before the relay publishes the new event.
wait_for_observation_pipeline() {
  local encounter_id="$1"
  local outbox_id
  outbox_id="$(latest_observation_outbox_id "${encounter_id}")"
  if [[ -z "${outbox_id}" ]]; then
    echo "No observation.recorded outbox row for encounter ${encounter_id}"
    return 1
  fi
  wait_for_outbox_processed "${outbox_id}"
  wait_for_consumer_lag_zero "${SEPSIS_CONSUMER_GROUP}"
}

redis_get() {
  local value
  value="$(redis_cmd GET "$1" | tr -d '\r')"
  if [[ "${value}" == "(nil)" ]]; then
    echo ""
  else
    echo "${value}"
  fi
}

redis_ttl() {
  redis_cmd TTL "$1" | tr -d '\r'
}

sirs_key() {
  local encounter_id="$1"
  local code="$2"
  echo "sirs:${encounter_id}:${code}"
}

TOTAL_STEPS=10

echo "Running sepsis / SIRS verification against ${BASE_URL}"
echo "Script run id: ${SCRIPT_RUN_ID}"

echo ""
echo "[0/${TOTAL_STEPS}] Preflight — API, Postgres, Redis, Kafka, and Elasticsearch 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

if ! redis_cmd PING >/dev/null 2>&1; then
  echo "Redis not reachable on port ${REDIS_PORT}."
  exit 1
fi

if ! wait_for_kafka; then
  exit 1
fi

es_health_status="$(curl -sS "${ES_URL}/_cluster/health" | jq -r '.status' || true)"
if [[ "${es_health_status}" != "green" && "${es_health_status}" != "yellow" ]]; then
  echo "Elasticsearch cluster health is '${es_health_status}' (expected green or yellow)."
  exit 1
fi
echo "OK: API, Postgres, Redis, Kafka, and Elasticsearch are up (cluster=${es_health_status})"

echo ""
echo "[1/${TOTAL_STEPS}] Creating patient, encounter, and baseline critical potassium alert"
patient_payload='{"firstName":"SIRS","lastName":"Verifier","dateOfBirth":"1975-04-12","gender":"M"}'
resp="$(request POST "${BASE_URL}/api/v1/patients" "${patient_payload}")"
assert_status "201" "${resp}"
patient_id="$(jq -r '.data.id' "${resp}")"

enc_payload='{"encounterType":"Inpatient","department":"ICU","attendingPhysician":"Dr. SIRS"}'
resp="$(request POST "${BASE_URL}/api/v1/patients/${patient_id}/encounters" "${enc_payload}")"
assert_status "201" "${resp}"
encounter_id="$(jq -r '.data.id' "${resp}")"
echo "OK: patient ${patient_id}, encounter ${encounter_id}"

critical_payload="$(jq -nc \
  --arg recordedAt "${RECORDED_AT}" \
  --arg key "sirs-potassium-${SCRIPT_RUN_ID}" \
  '{observations:[{observationCode:"POTASSIUM_MEQ_L",value:2.1,unit:"mEq/L",source:"LAB",recordedAt:$recordedAt,idempotencyKey:$key}]}')"
resp="$(request POST "${BASE_URL}/api/v1/encounters/${encounter_id}/observations" "${critical_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
echo "OK: baseline critical alert ingested (for openAlertCount=2 after sepsis)"

echo ""
echo "[2/${TOTAL_STEPS}] End-to-end SIRS — tachycardia sets Redis key, no SEPSIS_WARNING yet"
ingest_observation "${encounter_id}" "HEART_RATE" 95 "bpm" "DEVICE" "${RECORDED_AT}" \
  "sirs-hr1-${SCRIPT_RUN_ID}"

if ! wait_for_observation_pipeline "${encounter_id}"; then
  exit 1
fi

hr_key="$(sirs_key "${encounter_id}" "HEART_RATE")"
hr_value="$(redis_get "${hr_key}")"
if [[ "${hr_value}" != "1" ]]; then
  echo "Expected Redis ${hr_key} = 1, got '${hr_value}'"
  exit 1
fi

hr_ttl="$(redis_ttl "${hr_key}")"
if [[ "${hr_ttl}" -lt 1500 || "${hr_ttl}" -gt 1800 ]]; then
  echo "Expected HEART_RATE TTL between 1500 and 1800, got ${hr_ttl}"
  exit 1
fi

sepsis_count="$(psql_cmd "SELECT COUNT(*) FROM clinical_alerts WHERE encounter_id = '${encounter_id}' AND alert_type = 'SEPSIS_WARNING'")"
if [[ "${sepsis_count}" != "0" ]]; then
  echo "Expected no SEPSIS_WARNING after one criterion, got count=${sepsis_count}"
  exit 1
fi
echo "OK: HEART_RATE key set (TTL=${hr_ttl}s), no sepsis alert yet"

echo ""
echo "[3/${TOTAL_STEPS}] End-to-end SIRS — fever triggers SEPSIS_WARNING through Kafka pipeline"
recorded_fever="$(date -u +"%Y-%m-%dT%H:%M:%SZ")"
ingest_observation "${encounter_id}" "TEMP_C" 38.5 "°C" "DEVICE" "${recorded_fever}" \
  "sirs-temp-${SCRIPT_RUN_ID}"

if ! wait_for_observation_pipeline "${encounter_id}"; then
  exit 1
fi

if ! wait_for_sepsis_alert "${encounter_id}"; then
  exit 1
fi

alert_row="$(psql_cmd "SELECT alert_type || '|' || severity || '|' || status FROM clinical_alerts WHERE encounter_id = '${encounter_id}' AND alert_type = 'SEPSIS_WARNING' LIMIT 1")"
if [[ "${alert_row}" != "SEPSIS_WARNING|CRITICAL|OPEN" ]]; then
  echo "Unexpected SEPSIS_WARNING row: '${alert_row}'"
  exit 1
fi
echo "OK: SEPSIS_WARNING created (type=SEPSIS_WARNING, severity=CRITICAL, status=OPEN)"

echo ""
echo "[4/${TOTAL_STEPS}] Verifying alert.generated outbox row for sepsis alert"
outbox_row="$(psql_cmd "SELECT topic || '|' || partition_key FROM outbox_events WHERE topic = 'alert.generated' AND partition_key = '${encounter_id}' ORDER BY created_at DESC LIMIT 1")"
if [[ "${outbox_row}" != "alert.generated|${encounter_id}" ]]; then
  echo "Expected alert.generated outbox row for encounter ${encounter_id}, got '${outbox_row}'"
  exit 1
fi

sepsis_outbox_id="$(psql_cmd "SELECT id FROM outbox_events WHERE topic = 'alert.generated' AND partition_key = '${encounter_id}' ORDER BY created_at DESC LIMIT 1")"
wait_for_outbox_processed "${sepsis_outbox_id}"
echo "OK: alert.generated outbox row exists and was relayed"

echo ""
echo "[5/${TOTAL_STEPS}] Verifying SirsDetector uses MGET (one round-trip for four keys)"
monitor_out="$(mktemp)"
TMP_FILES+=("${monitor_out}")

redis_cmd MONITOR > "${monitor_out}" 2>&1 &
monitor_pid=$!
sleep 0.5

recorded_resp="$(date -u +"%Y-%m-%dT%H:%M:%SZ")"
ingest_observation "${encounter_id}" "RESP_RATE" 22 "breaths/min" "DEVICE" "${recorded_resp}" \
  "sirs-resp-monitor-${SCRIPT_RUN_ID}"

if ! wait_for_observation_pipeline "${encounter_id}"; then
  kill "${monitor_pid}" 2>/dev/null || true
  exit 1
fi

sleep 1
kill "${monitor_pid}" 2>/dev/null || true
wait "${monitor_pid}" 2>/dev/null || true

if grep -qi '"keys"' "${monitor_out}"; then
  echo "MONITOR output contains KEYS — SirsDetector must not scan the keyspace"
  exit 1
fi

mget_line="$(grep -i 'mget' "${monitor_out}" | grep -F "sirs:${encounter_id}" | head -n 1 || true)"
if [[ -z "${mget_line}" ]]; then
  echo "No MGET command found in Redis MONITOR output for encounter ${encounter_id}"
  echo "MONITOR tail:"
  tail -n 20 "${monitor_out}"
  exit 1
fi

for code in TEMP_C HEART_RATE RESP_RATE WBC_K_UL; do
  if ! grep -F "sirs:${encounter_id}:${code}" <<< "${mget_line}" >/dev/null; then
    echo "MGET line missing key sirs:${encounter_id}:${code}"
    echo "MGET line: ${mget_line}"
    exit 1
  fi
done

sirs_get_count="$(grep -cE '"get".*sirs:'"${encounter_id}" "${monitor_out}" || true)"
if [[ "${sirs_get_count}" -ge 4 ]]; then
  echo "Found ${sirs_get_count} individual GET commands on SIRS keys — expected MGET instead"
  exit 1
fi
echo "OK: single MGET with all four SIRS keys observed"

echo ""
echo "[6/${TOTAL_STEPS}] Verifying TTL sliding window on HEART_RATE key"
ttl_before_decay="$(redis_ttl "${hr_key}")"
if (( TTL_DECAY_WAIT_SECS > 0 )); then
  sleep "${TTL_DECAY_WAIT_SECS}"
  ttl_after_decay="$(redis_ttl "${hr_key}")"
  expected_max="$((ttl_before_decay - TTL_DECAY_WAIT_SECS + 2))"
  if [[ "${ttl_after_decay}" -gt "${expected_max}" ]]; then
    echo "Expected TTL to decay after ${TTL_DECAY_WAIT_SECS}s (${ttl_before_decay} -> <=${expected_max}), got ${ttl_after_decay}"
    exit 1
  fi
  echo "OK: TTL decayed (${ttl_before_decay}s -> ${ttl_after_decay}s over ${TTL_DECAY_WAIT_SECS}s wait)"
fi

recorded_hr2="$(date -u +"%Y-%m-%dT%H:%M:%SZ")"
ingest_observation "${encounter_id}" "HEART_RATE" 96 "bpm" "DEVICE" "${recorded_hr2}" \
  "sirs-hr2-${SCRIPT_RUN_ID}"
if ! wait_for_observation_pipeline "${encounter_id}"; then
  exit 1
fi

ttl_after_refresh="$(redis_ttl "${hr_key}")"
if [[ "${ttl_after_refresh}" -lt 1500 || "${ttl_after_refresh}" -gt 1800 ]]; then
  echo "Expected HEART_RATE TTL to reset near 1800 after re-qualifying observation, got ${ttl_after_refresh}"
  exit 1
fi
echo "OK: TTL reset after qualifying HEART_RATE observation (TTL=${ttl_after_refresh}s)"

echo ""
echo "[7/${TOTAL_STEPS}] Verifying consumer group independence (sepsis-engine and es-indexer)"
if ! wait_for_consumer_lag_zero "${SEPSIS_CONSUMER_GROUP}"; then
  exit 1
fi
if ! wait_for_consumer_lag_zero "${ES_CONSUMER_GROUP}"; then
  exit 1
fi
echo "OK: sepsis-engine and es-indexer both have LAG=0 on observation.recorded"

echo ""
echo "[8/${TOTAL_STEPS}] Verifying SEPSIS_WARNING in Elasticsearch and openAlertCount=2"
elapsed=0
while (( elapsed < INDEX_WAIT_SECS )); do
  es_hits="$(curl -sS "${ES_URL}/clinical_alerts/_search" \
    -H "Content-Type: application/json" \
    -d "{\"query\":{\"bool\":{\"must\":[{\"term\":{\"encounterId\":\"${encounter_id}\"}},{\"term\":{\"alertType\":\"SEPSIS_WARNING\"}}]}}}" \
    | jq -r '.hits.total.value')"
  if [[ "${es_hits}" == "1" ]]; then
    break
  fi
  sleep 2
  elapsed=$((elapsed + 2))
done

if [[ "${es_hits:-0}" != "1" ]]; then
  echo "Expected 1 SEPSIS_WARNING document in clinical_alerts for encounter ${encounter_id}, got ${es_hits:-0}"
  exit 1
fi

es_alert="$(curl -sS "${ES_URL}/clinical_alerts/_search" \
  -H "Content-Type: application/json" \
  -d "{\"query\":{\"bool\":{\"must\":[{\"term\":{\"encounterId\":\"${encounter_id}\"}},{\"term\":{\"alertType\":\"SEPSIS_WARNING\"}}]}}}" \
  | jq -r '.hits.hits[0]._source | [.severity, .status] | @tsv')"
if [[ "${es_alert}" != $'Critical\tOpen' ]]; then
  echo "Expected Elasticsearch alert severity=Critical status=Open, got '${es_alert}'"
  exit 1
fi

open_alert_count="$(curl -sS "${ES_URL}/patient_encounters/_source/${encounter_id}" | jq -r '.openAlertCount')"
if [[ "${open_alert_count}" != "2" ]]; then
  echo "Expected openAlertCount=2 (potassium + sepsis), got ${open_alert_count}"
  exit 1
fi
echo "OK: SEPSIS_WARNING indexed (Critical/Open) and openAlertCount=2"

echo ""
echo "[9/${TOTAL_STEPS}] Verifying criterion clears immediately but alert remains open"
ingest_observation "${encounter_id}" "TEMP_C" 38.5 "°C" "DEVICE" "$(date -u +"%Y-%m-%dT%H:%M:%SZ")" \
  "sirs-temp-qual-${SCRIPT_RUN_ID}"
if ! wait_for_observation_pipeline "${encounter_id}"; then
  exit 1
fi

temp_key="$(sirs_key "${encounter_id}" "TEMP_C")"
if [[ "$(redis_get "${temp_key}")" != "1" ]]; then
  echo "Expected qualifying temperature to set ${temp_key}"
  exit 1
fi

ingest_observation "${encounter_id}" "TEMP_C" 37.0 "°C" "DEVICE" "$(date -u +"%Y-%m-%dT%H:%M:%SZ")" \
  "sirs-temp-normal-${SCRIPT_RUN_ID}"
if ! wait_for_observation_pipeline "${encounter_id}"; then
  exit 1
fi

temp_value="$(redis_get "${temp_key}")"
if [[ -n "${temp_value}" ]]; then
  echo "Expected ${temp_key} to be cleared after normal temperature, got '${temp_value}'"
  exit 1
fi

alert_status="$(psql_cmd "SELECT status FROM clinical_alerts WHERE encounter_id = '${encounter_id}' AND alert_type = 'SEPSIS_WARNING' LIMIT 1")"
if [[ "${alert_status}" != "OPEN" ]]; then
  echo "Expected SEPSIS_WARNING to remain OPEN after criterion cleared, got '${alert_status}'"
  exit 1
fi
echo "OK: TEMP_C key deleted on normalisation; SEPSIS_WARNING remains OPEN"

echo ""
echo "[10/${TOTAL_STEPS}] Verifying non-SIRS observation is ignored by the sepsis engine"
ingest_observation "${encounter_id}" "POTASSIUM_MEQ_L" 3.2 "mEq/L" "LAB" "$(date -u +"%Y-%m-%dT%H:%M:%SZ")" \
  "sirs-non-sirs-${SCRIPT_RUN_ID}"
if ! wait_for_observation_pipeline "${encounter_id}"; then
  exit 1
fi

pot_key="$(sirs_key "${encounter_id}" "POTASSIUM_MEQ_L")"
if redis_cmd EXISTS "${pot_key}" | grep -q '^1'; then
  echo "Non-SIRS code must not create a Redis SIRS key (${pot_key})"
  exit 1
fi
echo "OK: non-SIRS observation did not create SIRS Redis keys"

echo ""
echo "All ${TOTAL_STEPS} sepsis / SIRS checks passed."
echo ""
echo "Prerequisites: docker compose up -d && dotnet run --project VigilCareClinicalAPI"
echo "Optional: set TTL_DECAY_WAIT_SECS=60 to match the full sliding-window decay check in the plan."