#!/usr/bin/env bash
# =============================================================================
# eBdesk SiCABE — Verifikasi Phase 5 (Ingestion): Airflow
#
# Gate "kontrak ingestion": memastikan Airflow siap menjalankan pipeline —
# termasuk saat Airflow EXISTING (dikelola sistem lain).
#
# Yang diperiksa:
#   A. Kontrak service : webserver + scheduler + dag-processor, health API
#                        (metadatabase & scheduler), airflow-init pernah jalan
#   B. DAG             : tanpa import error, sicabe_batch_daily ada & aktif,
#                        4 task sesuai kontrak pipeline
#   C. Connections     : minio_s3 (endpoint/kredensial) DAN trino_default yang
#                        BENAR-BENAR DAPAT DIPAKAI (diuji konek nyata) — utang
#                        dari review Fase 3 (§7.6)
#   D. Sumber & aset   : sicabe_data.public.orders (sumber DAG), sling/env.yaml,
#                        DBT_TRINO_* untuk service Airflow
#   E. Temuan/antarmuka: berkas yatim connections.sh, docstring DAG basi, dan
#                        validasi GE yang menunjuk tabel sumber → WARN; consumer
#                        fase 6 → SKIP
#
# Catatan: setiap koneksi diuji dengan PARAMETER TERSIMPAN-nya (bukan asumsi),
# dan kata sandi tidak pernah dicetak / tidak lewat argumen proses (via stdin).
#
# Exit code: 0 = semua gate lolos (WARN/SKIP boleh), 1 = ada FAIL.
#
# Usage:
#   ./scripts/verify-phase5-ingestion.sh
#   AIRFLOW_HOST=airflow.corp AIRFLOW_PORT=8080 ./scripts/verify-phase5-ingestion.sh
# =============================================================================
set -uo pipefail

SCRIPT_DIR="$(cd "$(dirname "${BASH_SOURCE[0]}")/.." && pwd)"
# shellcheck source=../lib/common.sh
source "${SCRIPT_DIR}/lib/common.sh"
# shellcheck source=../config/environment.defaults.sh
source "${SCRIPT_DIR}/config/environment.defaults.sh"
sicabe_load_env "${SCRIPT_DIR}" "$@"

# Config RELEVAN: mode host (--env-file) = BUNDLE host (ter-render), bukan deploy/config/ sumber.
if [[ -n "${SICABE_BUNDLE_DIR:-}" ]]; then CONFIG_DIR="${SICABE_BUNDLE_DIR}/config"; else CONFIG_DIR="${SCRIPT_DIR}/config"; fi

AIRFLOW_HOST="${AIRFLOW_HOST:-localhost}"
AIRFLOW_PORT="${AIRFLOW_PORT:-12082}"
AIRFLOW_URL="http://$(sicabe_local_host AIRFLOW_WEBSERVER_HOST AIRFLOW_HOST):${AIRFLOW_PORT}"
BATCH_DAG="${AIRFLOW_BATCH_DAG:-sicabe_batch_daily}"
CDC_DAG="${AIRFLOW_CDC_DAG:-sicabe_cdc_realtime}"
S3_ENDPOINT="${S3_ENDPOINT:-http://minio:9000}"
BUCKET="${MINIO_BUCKET_ICEBERG_DATA:-iceberg-data}"
EXPECTED_TASKS="bronze_to_silver ingest_to_bronze silver_to_gold validate_gold"
WEB_CONTAINER="sicabe-airflow-webserver"
AIRFLOW_EXTERNAL=0
if [[ "${AIRFLOW_HOST}" != "localhost" && "${AIRFLOW_HOST}" != "127.0.0.1" ]]; then
  AIRFLOW_EXTERNAL=1
fi

PASS_N=0
FAIL_N=0
WARN_N=0
SKIP_N=0
pass() { echo -e "  ${PASS} $1"; PASS_N=$((PASS_N + 1)); }
FAIL_MSGS=()
fail() { echo -e "  ${FAIL} $1"; FAIL_N=$((FAIL_N + 1)); FAIL_MSGS+=("$1"); }
warn() { echo -e "  ${WARN} $1"; WARN_N=$((WARN_N + 1)); }
skip() { echo -e "  ${YELLOW}[SKIP]${NC} $1"; SKIP_N=$((SKIP_N + 1)); }

container_running() {
  docker ps --format "{{.Names}}" 2>/dev/null | grep -q "^$1$"
}

af_cli() { docker exec "${WEB_CONTAINER}" airflow "$@" 2>/dev/null; }

# ─────────────────────────────────────────────────────────────────────────────
# A. Kontrak service
# ─────────────────────────────────────────────────────────────────────────────
check_service() {
  echo -e "${BOLD}A. Kontrak service Airflow${NC}"

  local c runtime=""
  for c in "${WEB_CONTAINER}" sicabe-airflow-scheduler sicabe-airflow-dag-processor; do
    if ! sicabe_host_expects_container "${c}"; then
      skip "Container ${c} di host lain (mode host) - dilewati"
    elif container_running "${c}"; then
      runtime+=" ${c}"
    fi
  done
  if [[ -n "${runtime}" ]]; then
    pass "Container berjalan:${runtime}"
  fi
  if [[ ${AIRFLOW_EXTERNAL} -eq 1 ]]; then
    skip "Airflow dikelola eksternal (${AIRFLOW_HOST}:${AIRFLOW_PORT}) — uji container dilewati"
  else
    for c in "${WEB_CONTAINER}" sicabe-airflow-scheduler sicabe-airflow-dag-processor; do
      if ! sicabe_host_expects_container "${c}"; then
        skip "Container ${c} di host lain (mode host) - dilewati"
      else
        container_running "${c}" || fail "Container ${c} tidak berjalan"
      fi
    done
    if docker ps -a --format "{{.Names}}" 2>/dev/null | grep -q "^sicabe-airflow-init$"; then
      pass "Container one-shot sicabe-airflow-init ada (db migrate + users create)"
    else
      warn "Container sicabe-airflow-init tidak ditemukan — metadata DB mungkin belum di-migrate"
    fi
  fi

  local health
  health="$(curl -sf --max-time 15 "${AIRFLOW_URL}/api/v2/monitor/health" 2>/dev/null)"
  if [[ -z "${health}" ]]; then
    fail "Health API tidak menjawab di ${AIRFLOW_URL}/api/v2/monitor/health"
    return 1
  fi
  local meta sched
  meta="$(echo "${health}" | python3 -c "import json,sys;print(json.load(sys.stdin).get('metadatabase',{}).get('status',''))" 2>/dev/null)"
  sched="$(echo "${health}" | python3 -c "import json,sys;print(json.load(sys.stdin).get('scheduler',{}).get('status',''))" 2>/dev/null)"
  if [[ "${meta}" == "healthy" ]]; then
    pass "metadatabase.status = healthy"
  else
    fail "metadatabase.status = '${meta:-tak terbaca}' (harus healthy)"
  fi
  if [[ "${sched}" == "healthy" ]]; then
    pass "scheduler.status = healthy"
  else
    warn "scheduler.status = '${sched:-tak terbaca}'"
  fi
}

# ─────────────────────────────────────────────────────────────────────────────
# B. DAG
# ─────────────────────────────────────────────────────────────────────────────
check_dags() {
  echo -e "${BOLD}B. DAG pipeline${NC}"

  if [[ ${AIRFLOW_EXTERNAL} -eq 1 ]]; then
    skip "Airflow eksternal — pemeriksaan DAG via CLI dilewati (gunakan UI/API di sana)"
    return 0
  fi

  local errs
  errs="$(af_cli dags list-import-errors)"
  if [[ "${errs}" == *"No data found"* || -z "${errs}" ]]; then
    pass "Tidak ada DAG import error"
  else
    fail "Ada DAG import error: $(echo "${errs}" | tr '\n' ' ' | cut -c1-120)"
  fi

  local list
  list="$(af_cli dags list)"
  if [[ "${list}" == *"${BATCH_DAG}"* ]]; then
    pass "DAG '${BATCH_DAG}' ada"
  else
    fail "DAG '${BATCH_DAG}' tidak ditemukan — sudah disalin? (bootstrap.sh --step 6)"
  fi
  if [[ "${list}" == *"${CDC_DAG}"* ]]; then
    pass "DAG '${CDC_DAG}' ada (opsional, default paused)"
  else
    warn "DAG '${CDC_DAG}' tidak ditemukan"
  fi
  # Baris DAG pada `dags list`: "<dag_id> | <fileloc> | <owner> | <is_paused> | ..."
  local row
  row="$(echo "${list}" | grep -m1 "^${BATCH_DAG} " || true)"
  if [[ "${row}" == *"| False"* ]]; then
    pass "DAG '${BATCH_DAG}' aktif (is_paused=False)"
  else
    # Gate ini SENGAJA read-only: ia melaporkan, tidak memperbaiki. Pengaktifan DAG
    # dilakukan di bootstrap.sh step 6 (skrip yang memang memiliki DAG-nya), supaya
    # hasil verifikasi tidak bergantung pada apakah gate pernah dijalankan.
    warn "DAG '${BATCH_DAG}' paused — pipeline harian tidak akan jalan. Aktifkan: docker exec sicabe-airflow-webserver airflow dags unpause ${BATCH_DAG}"
  fi

  local tasks
  tasks="$(af_cli tasks list "${BATCH_DAG}" | tr '\n' ' ' | tr -s ' ' | sed 's/ $//')"
  local t missing=""
  for t in ${EXPECTED_TASKS}; do
    [[ " ${tasks} " == *" ${t} "* ]] || missing+=" ${t}"
  done
  if [[ -z "${missing}" ]]; then
    pass "4 task sesuai kontrak: ${EXPECTED_TASKS}"
  else
    fail "Task hilang di '${BATCH_DAG}':${missing}"
  fi
}

# ─────────────────────────────────────────────────────────────────────────────
# C. Connections — inti utang review Fase 3
# ─────────────────────────────────────────────────────────────────────────────
check_connections() {
  echo -e "${BOLD}C. Connections (dipakai DAG user/operator)${NC}"

  if [[ ${AIRFLOW_EXTERNAL} -eq 1 ]]; then
    skip "Airflow eksternal — pemeriksaan connection via CLI dilewati"
    return 0
  fi

  local json
  json="$(af_cli connections export -)"
  if [[ -z "${json}" ]]; then
    fail "Tidak bisa membaca connections (airflow connections export)"
    return 1
  fi

  # ── minio_s3 ──────────────────────────────────────────────────────────────
  local m_ep m_login expected_login
  m_ep="$(echo "${json}" | python3 -c "
import json,sys
try:
    c=json.load(sys.stdin)['minio_s3']; print(json.loads(c.get('extra') or '{}').get('endpoint_url',''))
except Exception: print('')" 2>/dev/null)"
  m_login="$(echo "${json}" | python3 -c "
import json,sys
try: print(json.load(sys.stdin)['minio_s3'].get('login') or '')
except Exception: print('')" 2>/dev/null)"
  if [[ "${json}" != *'"minio_s3"'* ]]; then
    fail "Connection 'minio_s3' tidak ada (bootstrap.sh --step 6)"
  elif [[ "${m_ep}" == "${S3_ENDPOINT}" ]]; then
    pass "minio_s3: endpoint_url = ${m_ep}"
  else
    fail "minio_s3: endpoint_url '${m_ep:-kosong}' ≠ S3_ENDPOINT '${S3_ENDPOINT}'"
  fi
  # Kredensial harus mengikuti S3_ACCESS_KEY (fallback saat-pakai), bukan MINIO_ROOT_USER
  # — kalau tidak, salah saat memakai S3 existing (§2.9).
  expected_login="${S3_ACCESS_KEY:-${MINIO_ROOT_USER:-minioadmin}}"
  if [[ -z "${m_login}" ]]; then
    fail "minio_s3: login/kredensial kosong"
  elif [[ "${m_login}" == "${expected_login}" ]]; then
    pass "minio_s3: kredensial sesuai S3 aktif (login '${m_login}', password disamarkan)"
  else
    fail "minio_s3: login '${m_login}' ≠ kredensial S3 aktif '${expected_login}' — salah saat S3 existing"
    warn "  perbaiki: bootstrap.sh --step 6 (memakai \${S3_ACCESS_KEY} dengan fallback \${MINIO_ROOT_USER})"
  fi

  # ── trino_default — SENGAJA tidak disediakan (keputusan review Fase 5) ────
  # Tidak ada DAG yang memakainya (task memakai dbt HTTPS 8443 + Sling), dan versi
  # lamanya menunjuk http://trino:8080 tanpa kredensial → selalu 403.
  if [[ "${json}" == *'"trino_default"'* ]]; then
    local t_parsed
    t_parsed="$(echo "${json}" | python3 -c "
import json,sys
try:
    c=json.load(sys.stdin)['trino_default']
    print('{0}:{1}'.format(c.get('host') or '', c.get('port') or ''))
except Exception: print('')" 2>/dev/null)"
    fail "Connection 'trino_default' masih ada (${t_parsed}) — sudah diputuskan TIDAK disediakan"
    warn "  hapus: docker exec sicabe-airflow-webserver airflow connections delete trino_default"
  else
    pass "tidak ada connection 'trino_default' (sesuai keputusan — DAG memakai dbt + Sling)"
  fi
}

# ─────────────────────────────────────────────────────────────────────────────
# D. Sumber data & aset pipeline
# ─────────────────────────────────────────────────────────────────────────────
check_source_and_assets() {
  echo -e "${BOLD}D. Sumber data & aset pipeline${NC}"

  # Sumber DAG: sicabe_data.public.orders (di-seed bootstrap step 6)
  if ! sicabe_host_expects_container sicabe-postgres; then
    skip "Container sicabe-postgres di host lain (mode host) - dilewati"
  elif container_running sicabe-postgres; then
    local n
    n="$(docker exec -e PGPASSWORD="${POSTGRES_PASSWORD:-}" sicabe-postgres \
      psql -U "${POSTGRES_USER:-sicabe_admin}" -d sicabe_data -tAc \
      "SELECT count(*) FROM public.orders" 2>/dev/null || echo "")"
    if [[ "${n}" =~ ^[0-9]+$ ]] && [[ ${n} -ge 1 ]]; then
      pass "Sumber sicabe_data.public.orders → ${n} baris"
    elif [[ -z "${n}" ]]; then
      fail "sicabe_data.public.orders tidak ada / tidak terbaca (jalankan scripts/seed-source-postgres.sh)"
    else
      fail "sicabe_data.public.orders kosong (0 baris) — task ingest_to_bronze tidak akan menghasilkan data"
    fi
  else
    skip "Container sicabe-postgres tidak berjalan — lewati cek sumber contoh"
  fi

  # Aset kredensial runtime Sling
  local sling="${CONFIG_DIR}/airflow/sling/env.yaml"
  if [[ -f "${sling}" ]]; then
    pass "config/airflow/sling/env.yaml ada"
    local key missing=""
    for key in POSTGRES_DB ICEBERG rest_uri s3_endpoint; do
      grep -q "${key}" "${sling}" || missing+=" ${key}"
    done
    if [[ -z "${missing}" ]]; then
      pass "  memuat koneksi POSTGRES_DB + ICEBERG (rest_uri & s3_endpoint)"
    else
      fail "  env.yaml kehilangan:${missing} — generate ulang (bootstrap.sh / generate_runtime_credentials)"
    fi
  else
    fail "config/airflow/sling/env.yaml tidak ada — task ingest_to_bronze akan gagal"
  fi

  # dbt → Trino: env DBT_TRINO_* harus SAMPAI ke container Airflow yang mengeksekusi
  # task. Dicek dari ENV CONTAINER — bukan hasil render `docker compose config`.
  # Alasan: (a) env container adalah yang benar-benar dipakai task dbt; (b) cek lama
  # pernah salah lapor (19 PASS/1 FAIL, lalu 20/0 pada 3 pengulangan) sementara
  # rendernya sendiri selalu memuat key itu → mekanisme render tidak dapat dipercaya
  # sebagai bukti. `docker compose ... 2>/dev/null` juga menelan pesan error compose.
  if [[ ${AIRFLOW_EXTERNAL} -eq 1 ]]; then
    skip "Airflow eksternal (${AIRFLOW_HOST}:${AIRFLOW_PORT}) — cek env DBT_TRINO_* dilewati"
  elif ! container_running sicabe-airflow-scheduler; then
    fail "Container sicabe-airflow-scheduler tidak berjalan — env DBT_TRINO_* tidak dapat diverifikasi"
  else
    local dbt_user dbt_pw
    dbt_user="$(docker exec sicabe-airflow-scheduler printenv DBT_TRINO_USER 2>/dev/null || true)"
    dbt_pw="$(docker exec sicabe-airflow-scheduler printenv DBT_TRINO_PASSWORD 2>/dev/null || true)"
    if [[ -n "${dbt_user}" ]]; then
      pass "Aset dbt: DBT_TRINO_USER='${dbt_user}' ada di env container Airflow"
    else
      fail "DBT_TRINO_USER tidak ada di env container Airflow — task dbt tidak bisa autentikasi"
    fi
    if [[ -n "${dbt_pw}" ]]; then
      pass "Aset dbt: DBT_TRINO_PASSWORD ada di env container Airflow (nilai disamarkan)"
    else
      fail "DBT_TRINO_PASSWORD tidak ada di env container Airflow"
    fi
  fi
}

# ─────────────────────────────────────────────────────────────────────────────
# E. Temuan statis & antarmuka konsumen (WARN/SKIP)
# ─────────────────────────────────────────────────────────────────────────────
check_findings_and_consumers() {
  echo -e "${BOLD}E. Temuan statis & antarmuka konsumen (informatif)${NC}"

  local orphan="${CONFIG_DIR}/airflow/connections.sh"
  if [[ -f "${orphan}" ]]; then
    warn "config/airflow/connections.sh ADA tetapi tidak dirujuk siapa pun dan isinya menyimpang"
    warn "  (menyebut trino http:8080, postgres_default & openmetadata_default yang tidak pernah dibuat)"
  else
    pass "Tidak ada berkas connection yatim"
  fi

  local dag_file="${CONFIG_DIR}/airflow/dags/example_sicabe_pipeline.py"
  if [[ -f "${dag_file}" ]]; then
    if grep -q "postgres_default" "${dag_file}" 2>/dev/null; then
      warn "Docstring DAG menyebut connection 'postgres_default' yang tidak dibuat bootstrap (DAG memakai env var)"
    fi
    if grep -q "trino-proxy" "${dag_file}" 2>/dev/null; then
      warn "Komentar DAG masih menyebut 'trino-proxy' (service yang tidak ada; dbt konek langsung HTTPS 8443)"
    fi
    if grep -q "_postgres_conn_string" "${dag_file}" 2>/dev/null; then
      warn "validate_gold/validate_bronze memvalidasi tabel di PostgreSQL (sicabe_data.orders = SUMBER),"
      warn "  bukan tabel gold/bronze di Iceberg — hasil 'LULUS' tidak membuktikan output pipeline"
    fi
  fi

  if ! sicabe_host_expects_container sicabe-openmetadata; then
    skip "Container sicabe-openmetadata di host lain (mode host) - dilewati"
  elif container_running sicabe-openmetadata; then
    pass "OpenMetadata berjalan — konektor Airflow diuji di gate Phase 6"
  else
    skip "OpenMetadata belum berjalan (fase 6) — lewati pemeriksaan antarmukanya"
  fi
  if ! sicabe_host_expects_container sicabe-grafana; then
    skip "Container sicabe-grafana di host lain (mode host) - dilewati"
  elif container_running sicabe-grafana; then
    pass "Grafana berjalan — dashboard Airflow diuji di gate Phase 6"
  else
    skip "Grafana belum berjalan (fase 6) — lewati pemeriksaan antarmukanya"
  fi
}

# ─────────────────────────────────────────────────────────────────────────────
# Main
# ─────────────────────────────────────────────────────────────────────────────
main() {
  print_section "Verifikasi Phase 5 — Kontrak Ingestion (Airflow)"
  echo -e "  Airflow: ${CYAN}${AIRFLOW_URL}${NC} · DAG utama: ${CYAN}${BATCH_DAG}${NC}"
  echo -e "  Sumber : sicabe_data.public.orders (PostgreSQL) → Sling → dbt/Trino"
  echo ""

  check_service
  check_dags
  check_connections
  check_source_and_assets
  check_findings_and_consumers

  echo ""
  echo -e "${BOLD}Ringkasan:${NC} ${GREEN}${PASS_N} PASS${NC} · ${RED}${FAIL_N} FAIL${NC} · ${YELLOW}${WARN_N} WARN${NC} · ${YELLOW}${SKIP_N} SKIP${NC}"
  if [[ ${FAIL_N} -gt 0 ]]; then
    echo ""
    echo -e "${BOLD}Yang gagal (${FAIL_N}):${NC}"
    local _i=1 _m
    for _m in "${FAIL_MSGS[@]}"; do
      echo -e "  ${RED}${_i}.${NC} ${_m}"
      _i=$((_i + 1))
    done
    echo -e "${INFO} Perbaiki item di atas, lalu jalankan ulang: ./scripts/${0##*/}"
  fi
  if [[ ${FAIL_N} -eq 0 ]]; then
    echo -e "${OK} ${GREEN}Kontrak ingestion Phase 5 terpenuhi — aman lanjut ke fase berikutnya.${NC}"
    exit 0
  fi
  echo -e "${ERR} Ada ${FAIL_N} gate yang gagal — perbaiki sebelum melanjutkan."
  exit 1
}

main "$@"