#!/usr/bin/env bash
# =============================================================================
# eBdesk SiCABE — Verifikasi Phase 3 (Processing): Trino + OPA
#
# Gate "kontrak processing": memastikan query engine (Trino) DAN titik keputusan
# otorisasi (OPA) benar-benar siap dipakai fase berikutnya (StarRocks, Airflow/dbt,
# OpenMetadata) — termasuk saat memakai Trino/OPA EXISTING (bukan container deploy ini).
#
# Yang diperiksa:
#   A. Trino — kontrak service : container, node ACTIVE, HTTPS saja, katalog
#                                `iceberg` + `system`, berkas .properties kunci
#   B. OPA — kebijakan          : deny-by-default + uji keputusan LANGSUNG ke policy
#                                (tanpa Trino) untuk 4 peran & batas per-kolom
#   C. Penegakan Trino↔OPA      : bukti izin/penolakan NYATA saat query dijalankan
#                                (SKIP bila tabel probe belum di-seed — fase 5)
#   D. Authenticator            : password.db 4 service account, bcrypt, HTTP 200/401
#   E. Antarmuka konsumen       : dbt / StarRocks / Airflow / OpenMetadata (WARN/SKIP —
#                                artefaknya milik fase 4/5/6, lihat §2.10 dokumen fase)
#
# Catatan desain:
#   - OPA menyimpan data ACL di memori (bind-mount read-only) → tidak ada state disk.
#   - Gate ini TIDAK mengubah apa pun; murni verifikasi.
#
# Mode --deep: disiapkan untuk penambahan uji lintas-komponen di masa depan
#   (mis. query nyata dari dbt / Airflow). Saat ini tidak ada pemeriksaan tambahan.
#
# Exit code: 0 = semua gate lolos (WARN/SKIP boleh), 1 = ada FAIL.
#
# Usage:
#   ./scripts/verify-phase3-processing.sh
#   # menunjuk Trino/OPA existing:
#   TRINO_URL=https://trino.corp:8443 OPA_URL=http://opa.corp:8181 TRINO_TLS_VERIFY=1 \
#     ./scripts/verify-phase3-processing.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 yang RELEVAN: di mode host (--env-file) = BUNDLE host (yang di-mount container,
# sudah ter-render dgn endpoint host); bukan deploy/config/ sumber. Sejalan dgn
# generate_runtime_credentials() yang menulis ke generated/host-<x>/config saat --host.
if [[ -n "${SICABE_BUNDLE_DIR:-}" ]]; then
  CONFIG_DIR="${SICABE_BUNDLE_DIR}/config"
else
  CONFIG_DIR="${SCRIPT_DIR}/config"
fi

TRINO_URL="$(sicabe_local_url https TRINO_HOST "${TRINO_HTTPS_PORT:-12443}" TRINO_URL)"
OPA_URL="${OPA_URL:-http://localhost:${OPA_PORT:-12181}}"
OPA_POLICY_PATH="${OPA_POLICY_PATH:-/v1/data/trino/allow}"
LDAP_URL="${LDAP_URL:-ldap://openldap:389}"

# User manusia untuk uji ACL lewat jalur OIDC (Keycloak → grup LDAP → OPA).
# Jalur ini menguji rantai penuh TANPA memerlukan akun Trino ber-role terbatas.
OIDC_ANALYST_USER="${OIDC_ANALYST_USER:-analyst}"
OIDC_ANALYST_PASSWORD="${OIDC_ANALYST_PASSWORD:-analyst123}"

# TLS: dev memakai sertifikat self-signed → -k. Untuk Trino existing dengan
# sertifikat valid, set TRINO_TLS_VERIFY=1 agar verifikasi TLS tetap aktif.
TRINO_CURL_OPTS=()
if [[ "${TRINO_URL}" == https://* && "${TRINO_TLS_VERIFY:-0}" != "1" ]]; then
  TRINO_CURL_OPTS+=("-k")
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$"
}

# ─────────────────────────────────────────────────────────────────────────────
# Helper: Trino REST + OPA decision
# ─────────────────────────────────────────────────────────────────────────────
# _trino_stmt <sql> [arg curl auth...] → "<state>|<errorName>|<data-json>"
# Mengakumulasi `data` lintas-halaman: Trino mengirim hasil bertahap dan halaman
# terakhir (FINISHED) sering kosong.
_trino_stmt() {
  local sql="$1"; shift
  local resp next state tries=0 rows="" err=""
  resp=$(curl -s "${TRINO_CURL_OPTS[@]}" --max-time 25 -X POST "${TRINO_URL}/v1/statement" \
    "$@" -H "Content-Type: text/plain" -d "${sql}" 2>/dev/null || echo "")
  [[ -z "${resp}" ]] && { echo "NO_RESPONSE||"; return 0; }

  while [[ ${tries} -lt 20 ]]; do
    rows+=$(echo "${resp}" | python3 -c "
import sys,json
try: print(json.dumps(json.load(sys.stdin).get('data') or []))
except Exception: pass" 2>/dev/null)
    state=$(echo "${resp}" | python3 -c "
import sys,json
try: print(json.load(sys.stdin).get('stats',{}).get('state',''))
except Exception: print('')" 2>/dev/null)
    err=$(echo "${resp}" | python3 -c "
import sys,json
try:
    d=json.load(sys.stdin)
    print(d['error'].get('errorName','') if d.get('error') else '')
except Exception: print('')" 2>/dev/null)
    [[ "${state}" == "FINISHED" || "${state}" == "FAILED" ]] && break
    next=$(echo "${resp}" | python3 -c "
import sys,json
try: print(json.load(sys.stdin).get('nextUri',''))
except Exception: print('')" 2>/dev/null)
    [[ -z "${next}" ]] && break
    sleep 1
    resp=$(curl -s "${TRINO_CURL_OPTS[@]}" --max-time 25 "${next}" 2>/dev/null || echo "")
    tries=$((tries + 1))
  done
  echo "${state}|${err}|${rows}"
}

# trino_exec <user> <password> <sql> — HTTP Basic (service account)
trino_exec() { _trino_stmt "$3" -u "$1:$2"; }

# trino_exec_oidc <bearer-token> <sql> — jalur user manusia (OIDC Keycloak)
trino_exec_oidc() { _trino_stmt "$2" -H "Authorization: Bearer $1"; }

# oidc_token <username> <password> → access token (client 'trino', scope openid)
oidc_token() {
  curl -s --max-time 20 -X POST \
    "$(sicabe_local_url http KEYCLOAK_HOST "${KEYCLOAK_PORT:-12080}" KEYCLOAK_URL)/realms/${SICABE_REALM:-sicabe}/protocol/openid-connect/token" \
    -d grant_type=password -d client_id=trino -d "client_secret=${TRINO_CLIENT_SECRET}" \
    -d "username=$1" -d "password=$2" -d scope=openid 2>/dev/null |
    python3 -c "import sys,json;print(json.load(sys.stdin).get('access_token',''))" 2>/dev/null
}

# trino_http <user> <password> → kode HTTP untuk SELECT 1 (uji authenticator saja)
trino_http() {
  curl -s -o /dev/null -w "%{http_code}" "${TRINO_CURL_OPTS[@]}" --max-time 15 \
    -X POST "${TRINO_URL}/v1/statement" -u "$1:$2" \
    -H "Content-Type: text/plain" -d "SELECT 1" 2>/dev/null
}

# opa_decide <input-json> → "true" | "false" | "" (tak terjawab)
opa_decide() {
  curl -sf --max-time 10 -X POST "${OPA_URL}${OPA_POLICY_PATH}" \
    -H "Content-Type: application/json" -d "$1" 2>/dev/null |
    python3 -c "import sys,json
try:
    r=json.load(sys.stdin).get('result')
    print('true' if r is True else 'false' if r is False else '')
except Exception: print('')" 2>/dev/null || echo ""
}

# opa_input <op> <catalog> <schema> <table> <columns-json> <user> <groups-json>
opa_input() {
  printf '{"input":{"action":{"operation":"%s","resource":{"table":{"catalogName":"%s","schemaName":"%s","tableName":"%s","columns":%s}}},"context":{"identity":{"user":"%s","groups":%s}}}}' \
    "$1" "$2" "$3" "$4" "$5" "$6" "$7"
}

# ─────────────────────────────────────────────────────────────────────────────
# A. Trino — kontrak service
# ─────────────────────────────────────────────────────────────────────────────
check_trino_service() {
  echo -e "${BOLD}A. Trino — kontrak service${NC}"

  if ! sicabe_host_expects_container sicabe-trino-coordinator; then
    skip "Container sicabe-trino-coordinator di host lain (mode host) - dilewati"
  elif container_running sicabe-trino-coordinator; then
    pass "Container sicabe-trino-coordinator berjalan"
  elif [[ "${TRINO_URL}" == *localhost* ]]; then
    fail "Container sicabe-trino-coordinator tidak berjalan"
  else
    skip "Trino dikelola di luar compose ini (${TRINO_URL}) — lewati cek container"
  fi

  # Worker berdiri sendiri sebagai service (tier `engine`): tanpa worker,
  # coordinator tidak punya tempat eksekusi task.
  if ! sicabe_host_expects_container sicabe-trino-worker; then
    skip "Container sicabe-trino-worker di host lain (mode host) - dilewati"
  elif container_running sicabe-trino-worker; then
    pass "Container sicabe-trino-worker berjalan"
  elif [[ "${TRINO_URL}" == *localhost* ]]; then
    fail "Container sicabe-trino-worker tidak berjalan — task tidak akan dieksekusi"
  else
    skip "Trino worker dikelola eksternal — lewati cek container"
  fi

  # Semantik split: coordinator TIDAK boleh mengeksekusi task.
  if grep -q "^node-scheduler.include-coordinator=false" \
       "${CONFIG_DIR}/trino/config-coordinator.properties" 2>/dev/null; then
    pass "Coordinator tidak mengeksekusi task (node-scheduler.include-coordinator=false)"
  else
    fail "config-coordinator.properties: include-coordinator bukan false"
  fi

  # Bukti fungsional worker terpakai. Dua batasan yang harus dihormati SEKALIGUS:
  #  (a) tabel system.runtime.nodes TIDAK di-whitelist OPA (least privilege), jadi
  #      tidak bisa dipakai untuk melihat daftar node;
  #  (b) gate ini TIDAK boleh bergantung pada data pipeline.
  # Agregasi membutuhkan stage task, dan karena include-coordinator=false stage itu
  # hanya bisa dijalankan worker — jadi agregasi atas nilai literal sudah cukup.
  # Dulu kueri memakai iceberg.gold.orders_gold (dibuat oleh DAG, BUKAN oleh seed),
  # sehingga instalasi baru gagal dengan pesan menyesatkan "worker tidak mengeksekusi
  # task" padahal penyebabnya schema `gold` belum ada.
  local dist st err
  dist=$(trino_exec "svc_seed" "${TRINO_SVC_SEED_PASSWORD}" \
        "SELECT count(*) FROM (VALUES 1,2,3) t(x)")
  IFS='|' read -r st err _ <<< "${dist}"
  if [[ "${st}" == "FINISHED" ]]; then
    pass "Query agregasi selesai — worker mengeksekusi task"
  else
    fail "Query agregasi '${st}' — worker tidak mengeksekusi task${err:+ (error Trino: ${err})}"
  fi

  local info
  info=$(curl -s "${TRINO_CURL_OPTS[@]}" --max-time 15 "${TRINO_URL}/v1/info" 2>/dev/null || echo "")
  if [[ -z "${info}" ]]; then
    fail "Trino tidak menjawab di ${TRINO_URL}/v1/info"
    return 1
  fi
  if echo "${info}" | grep -qE '"state":\s*"ACTIVE"'; then
    pass "Node Trino ACTIVE"
  else
    fail "Node Trino tidak ACTIVE: $(echo "${info}" | head -c 120)"
  fi

  # Trino hanya melayani HTTPS: HTTP polos harus MATI (allow-insecure-over-http=false)
  local http_code
  http_code=$(curl -s -o /dev/null -w "%{http_code}" --max-time 5 \
    "http://localhost:${TRINO_PORT:-12081}/v1/info" 2>/dev/null)
  if [[ "${http_code}" == "000" ]]; then
    pass "HTTP polos (${TRINO_PORT:-12081}) tidak dilayani — sesuai allow-insecure-over-http=false"
  else
    warn "Port HTTP ${TRINO_PORT:-12081} menjawab (HTTP ${http_code}) — pastikan ini memang diinginkan"
  fi

  # Katalog via service account (HTTP Basic) — bukti authenticator + katalog hidup
  local res state data
  res=$(trino_exec "svc_seed" "${TRINO_SVC_SEED_PASSWORD:?}" "SHOW CATALOGS")
  state="${res%%|*}"; data="${res##*|}"
  if [[ "${state}" == "FINISHED" && "${data}" == *iceberg* ]]; then
    pass "Catalog 'iceberg' terdeteksi via service account svc_seed"
  else
    fail "Catalog 'iceberg' tidak terverifikasi (state=${state})"
  fi
  if [[ "${state}" == "FINISHED" && "${data}" == *system* ]]; then
    pass "Catalog 'system' terdeteksi"
  else
    fail "Catalog 'system' tidak terverifikasi (state=${state})"
  fi
}

# ─────────────────────────────────────────────────────────────────────────────
# B. OPA — titik keputusan & kebijakan (deny-by-default)
# ─────────────────────────────────────────────────────────────────────────────
opa_expect() { # opa_expect <deskripsi> <harapan true|false> <input-json>
  local got
  got=$(opa_decide "$3")
  if [[ "${got}" == "$2" ]]; then
    pass "$1 → allow=${got}"
  else
    fail "$1 → allow=${got:-tak terjawab} (diharapkan $2)"
  fi
}

check_opa_policy() {
  echo -e "${BOLD}B. OPA — kebijakan (deny-by-default)${NC}"

  if ! sicabe_host_expects_container sicabe-opa; then
    skip "Container sicabe-opa di host lain (mode host) - dilewati"
  elif container_running sicabe-opa; then
    pass "Container sicabe-opa berjalan"
  elif [[ "${OPA_URL}" == *localhost* ]]; then
    fail "Container sicabe-opa tidak berjalan"
  else
    skip "OPA dikelola di luar compose ini (${OPA_URL}) — lewati cek container"
  fi

  if [[ -z "$(opa_decide '{"input":{}}')" ]]; then
    fail "Policy OPA tidak menjawab di ${OPA_URL}${OPA_POLICY_PATH}"
    return 1
  fi
  opa_expect "Deny-by-default (input kosong)" "false" '{"input":{}}'

  local eng='["data-engineer"]' ana='["data-analyst"]'
  opa_expect "data-admin (superuser) DROPTABLE gold" \
    "true" "$(opa_input 'DROPTABLE' iceberg gold orders_gold '[]' admin '["data-admin"]')"
  opa_expect "data-engineer SELECT bronze.orders" \
    "true" "$(opa_input 'SELECTFROMCOLUMNS' iceberg bronze orders '["id"]' u "$eng")"
  opa_expect "data-engineer DROPTABLE gold.orders_gold" \
    "true" "$(opa_input 'DROPTABLE' iceberg gold orders_gold '[]' u "$eng")"
  opa_expect "data-engineer CREATE SCHEMA" \
    "true" "$(printf '{"input":{"action":{"operation":"CREATESCHEMA","resource":{}},"context":{"identity":{"user":"u","groups":%s}}}}' "$eng")"
  opa_expect "data-analyst SELECT gold.orders_gold" \
    "true" "$(opa_input 'SELECTFROMCOLUMNS' iceberg gold orders_gold '["id"]' u "$ana")"
  opa_expect "data-analyst SELECT bronze.orders (kolom tak diberi)" \
    "false" "$(opa_input 'SELECTFROMCOLUMNS' iceberg bronze orders '["id"]' u "$ana")"
  opa_expect "data-analyst SELECT silver.orders_silver (kolom diizinkan)" \
    "true" "$(opa_input 'SELECTFROMCOLUMNS' iceberg silver orders_silver '["id","total_value"]' u "$ana")"
  opa_expect "data-analyst SELECT silver.orders_silver (kolom terlarang)" \
    "false" "$(opa_input 'SELECTFROMCOLUMNS' iceberg silver orders_silver '["id","order_date"]' u "$ana")"
  opa_expect "data-analyst SELECT agregat tanpa kolom (count(*))" \
    "true" "$(opa_input 'SELECTFROMCOLUMNS' iceberg silver orders_silver '[]' u "$ana")"
  opa_expect "data-analyst DROP TABLE (tanpa hak ddl)" \
    "false" "$(opa_input 'DROPTABLE' iceberg gold orders_gold '[]' u "$ana")"
  opa_expect "data-viewer SELECT gold.orders_gold (tanpa grant)" \
    "false" "$(opa_input 'SELECTFROMCOLUMNS' iceberg gold orders_gold '["id"]' u '["data-viewer"]')"
  opa_expect "Service account svc_om via data.users (tanpa grup LDAP)" \
    "true" "$(opa_input 'SELECTFROMCOLUMNS' iceberg gold orders_gold '["id"]' svc_om '[]')"
  opa_expect "User tak dikenal" \
    "false" "$(opa_input 'SELECTFROMCOLUMNS' iceberg gold orders_gold '["id"]' ghost '[]')"
  opa_expect "Metadata information_schema" \
    "true" "$(opa_input 'SELECTFROMCOLUMNS' system information_schema tables '["table_name"]' ghost '[]')"

  # FilterColumns — operasi yang dipakai Trino untuk memfilter daftar kolom di
  # information_schema / SHOW COLUMNS. Bila policy tidak memuatnya (deny-by-default),
  # SEMUA kolom tersembunyi walau SELECT-nya diizinkan → katalog (mis. ingestion
  # OpenMetadata) melihat tabel TANPA kolom. Batas per-kolom harus tetap dihormati.
  opa_expect "FilterColumns: data-engineer gold.orders_gold" \
    "true" "$(opa_input 'FilterColumns' iceberg gold orders_gold '["komoditas"]' u "$eng")"
  opa_expect "FilterColumns: data-analyst silver kolom diizinkan (id)" \
    "true" "$(opa_input 'FilterColumns' iceberg silver orders_silver '["id"]' u "$ana")"
  opa_expect "FilterColumns: data-analyst silver kolom terlarang (order_date)" \
    "false" "$(opa_input 'FilterColumns' iceberg silver orders_silver '["order_date"]' u "$ana")"

  # Cakupan data ACL (statis — artefak Phase 3)
  local data_file="${CONFIG_DIR}/opa/data.json"
  local missing="" r
  for r in data-admin data-engineer data-analyst data-viewer; do
    grep -q "\"${r}\"" "${data_file}" 2>/dev/null || missing+=" ${r}"
  done
  if [[ -z "${missing}" ]]; then
    pass "data.json memuat 4 peran (data-admin/engineer/analyst/viewer)"
  else
    fail "data.json kehilangan peran:${missing}"
  fi
  local svc missing_svc=""
  for svc in svc_seed svc_dbt svc_airflow svc_om; do
    grep -q "\"${svc}\"" "${data_file}" 2>/dev/null || missing_svc+=" ${svc}"
  done
  if [[ -z "${missing_svc}" ]]; then
    pass "data.json memetakan 4 service account ke peran"
  else
    warn "data.json belum memetakan:${missing_svc} (hanya perlu bila service account tidak ada di LDAP)"
  fi
}

# ─────────────────────────────────────────────────────────────────────────────
# C. Penegakan Trino ↔ OPA (butuh tabel probe → fase 5)
# ─────────────────────────────────────────────────────────────────────────────
_trino_assert() { # <desc> <auth-label> <state-harapan> <substring-error> <hasil>
  local desc="$1" auth="$2" want="$3" suberr="$4" res="$5"
  local state err
  state="${res%%|*}"; err=$(echo "${res}" | cut -d'|' -f2)
  if [[ "${state}" == "${want}" ]] && { [[ -z "${suberr}" ]] || [[ "${err}" == *"${suberr}"* ]]; }; then
    pass "${desc} (state=${state}${err:+, ${err}})"
  else
    fail "${desc} → state=${state}${err:+, ${err}} (diharapkan ${want}${suberr:+, ${suberr}}) [${auth}]"
  fi
}

trino_expect() { # <desc> <user> <pass> <sql> <state> [substring-error]
  _trino_assert "$1" "basic:$2" "$5" "${6:-}" "$(trino_exec "$2" "$3" "$4")"
}

trino_expect_oidc() { # <desc> <token> <sql> <state> [substring-error]
  _trino_assert "$1" "oidc" "$4" "${5:-}" "$(trino_exec_oidc "$2" "$3")"
}

check_enforcement() {
  echo -e "${BOLD}C. Penegakan Trino ↔ OPA (query nyata)${NC}"

  if [[ -z "${TRINO_SVC_OM_PASSWORD:-}" ]]; then
    skip "TRINO_SVC_OM_PASSWORD tidak di-set — lewati uji penegakan"
    return 0
  fi

  local tables
  tables=$(trino_exec "svc_seed" "${TRINO_SVC_SEED_PASSWORD:?}" "SHOW TABLES FROM iceberg.bronze" | tail -1)
  if [[ "${tables}" != *orders* ]]; then
    skip "Tabel probe iceberg.bronze.orders belum ada (di-seed fase 5) — lewati uji penegakan"
    return 0
  fi

  # Positif: pemilik data boleh baca
  trino_expect "svc_seed boleh baca bronze.orders" svc_seed "${TRINO_SVC_SEED_PASSWORD:?}" \
    "SELECT count(*) FROM iceberg.bronze.orders" FINISHED

  # Metadata kolom HARUS terlihat: regresi yang pernah terjadi adalah policy tanpa
  # rule FilterColumns → information_schema/SHOW COLUMNS kosong walau SELECT boleh.
  local meta_res ncols
  meta_res=$(trino_exec "svc_seed" "${TRINO_SVC_SEED_PASSWORD:?}" \
    "SELECT count(*) FROM iceberg.information_schema.columns WHERE table_schema='bronze' AND table_name='orders'")
  ncols=$(echo "${meta_res##*|}" | grep -oE '[0-9]+' | head -1)
  if [[ "${ncols}" =~ ^[0-9]+$ ]] && [[ ${ncols} -gt 0 ]]; then
    pass "Metadata kolom terlihat via Trino (${ncols} kolom di bronze.orders)"
  else
    fail "Metadata kolom TIDAK terlihat via Trino (information_schema → '${ncols:-kosong}')"
    warn "  cek rule FilterColumns di config/opa/policies/trino.rego (lihat catatan §5.6)"
  fi
  trino_expect "svc_dbt boleh baca bronze.orders" svc_dbt "${TRINO_SVC_DBT_PASSWORD:?}" \
    "SELECT count(*) FROM iceberg.bronze.orders" FINISHED
  trino_expect "svc_airflow boleh baca bronze.orders" svc_airflow "${TRINO_SVC_AIRFLOW_PASSWORD:?}" \
    "SELECT count(*) FROM iceberg.bronze.orders" FINISHED

  # svc_om = identitas AGENT katalog (role data-catalog: baca-saja seluruh tabel).
  # Positif: boleh baca semua, tetapi TIDAK boleh menulis (least privilege).
  trino_expect "svc_om (agent katalog) boleh baca bronze.orders" svc_om "${TRINO_SVC_OM_PASSWORD:-}" \
    "SELECT count(*) FROM iceberg.bronze.orders" FINISHED
  trino_expect "svc_om DITOLAK menulis (INSERT) di bronze.orders" svc_om "${TRINO_SVC_OM_PASSWORD:-}" \
    "INSERT INTO iceberg.bronze.orders (id) VALUES (999)" FAILED PERMISSION_DENIED

  # ── Negatif end-to-end lewat jalur USER MANUSIA (OIDC) ────────────────────
  # Menguji OIDC Keycloak + group provider LDAP + OPA sekaligus; tidak perlu akun
  # Trino ber-role terbatas hanya untuk pengujian.
  local oidc_tok
  oidc_tok="$(oidc_token "${OIDC_ANALYST_USER}" "${OIDC_ANALYST_PASSWORD}")"
  if [[ -z "${oidc_tok}" ]]; then
    skip "Tidak dapat token OIDC untuk '${OIDC_ANALYST_USER}' — uji ACL jalur OIDC dilewati"
  else
    pass "Token OIDC '${OIDC_ANALYST_USER}' diperoleh (jalur OIDC aktif)"
    trino_expect_oidc "'${OIDC_ANALYST_USER}' (data-analyst) DITOLAK baca bronze.orders" "${oidc_tok}" \
      "SELECT count(*) FROM iceberg.bronze.orders" FAILED PERMISSION_DENIED
    trino_expect_oidc "'${OIDC_ANALYST_USER}' boleh baca kolom yang diizinkan di silver" "${oidc_tok}" \
      "SELECT id, total_value FROM iceberg.silver.orders_silver" FINISHED
    trino_expect_oidc "'${OIDC_ANALYST_USER}' DITOLAK SELECT * di silver (batas kolom)" "${oidc_tok}" \
      "SELECT * FROM iceberg.silver.orders_silver" FAILED PERMISSION_DENIED
    trino_expect_oidc "'${OIDC_ANALYST_USER}' boleh baca gold.orders_gold" "${oidc_tok}" \
      "SELECT count(*) FROM iceberg.gold.orders_gold" FINISHED
    trino_expect_oidc "'${OIDC_ANALYST_USER}' DITOLAK DROP TABLE gold" "${oidc_tok}" \
      "DROP TABLE iceberg.gold.orders_gold" FAILED PERMISSION_DENIED
  fi
}

# ─────────────────────────────────────────────────────────────────────────────
# D. Authenticator service account (password.db)
# ─────────────────────────────────────────────────────────────────────────────
check_authenticator() {
  echo -e "${BOLD}D. Authenticator service account (password.db)${NC}"

  local pw_file="${CONFIG_DIR}/trino/password.db"
  if [[ -d "${pw_file}" ]]; then
    fail "config/trino/password.db berupa DIREKTORI (Docker membuatnya saat bind-mount gagal) — jalankan bootstrap.sh step 1"
  elif [[ ! -f "${pw_file}" ]]; then
    fail "config/trino/password.db tidak ada — di-generate bootstrap.sh (butuh htpasswd)"
  else
    pass "config/trino/password.db ada (berkas biasa)"
    local svc missing=""
    for svc in svc_seed svc_dbt svc_airflow svc_om; do
      grep -q "^${svc}:" "${pw_file}" || missing+=" ${svc}"
    done
    if [[ -z "${missing}" ]]; then
      pass "4 service account terdaftar: svc_seed, svc_dbt, svc_airflow, svc_om"
    else
      fail "password.db kehilangan:${missing}"
    fi
    if grep -qE '^\w+:\$2[aby]\$' "${pw_file}"; then
      pass "Hash bcrypt (\$2a/\$2b/\$2y) terdeteksi"
    else
      warn "Format hash tidak dikenali sebagai bcrypt"
    fi
  fi

  # Uji login nyata (hanya kode HTTP — kata sandi tidak pernah dicetak)
  local pair u p
  for pair in "svc_seed:${TRINO_SVC_SEED_PASSWORD:-}" "svc_dbt:${TRINO_SVC_DBT_PASSWORD:-}" \
              "svc_airflow:${TRINO_SVC_AIRFLOW_PASSWORD:-}" "svc_om:${TRINO_SVC_OM_PASSWORD:-}"; do
    u="${pair%%:*}"; p="${pair#*:}"
    [[ -z "${p}" ]] && continue
    if [[ "$(trino_http "${u}" "${p}")" == "200" ]]; then
      pass "Login HTTP Basic ${u} → 200"
    else
      fail "Login HTTP Basic ${u} gagal (bukan 200)"
    fi
  done

  if [[ "$(trino_http "svc_seed" "salah-sekali")" == "401" ]]; then
    pass "Kata sandi salah → 401"
  else
    fail "Kata sandi salah tidak ditolak dengan 401"
  fi
  if [[ "$(trino_http "tidak-ada-user" "apa-saja")" == "401" ]]; then
    pass "User tak dikenal → 401"
  else
    fail "User tak dikenal tidak ditolak dengan 401"
  fi
}

# ─────────────────────────────────────────────────────────────────────────────
# E. Antarmuka konsumen — milik fase 4/5/6 (WARN/SKIP)
# ─────────────────────────────────────────────────────────────────────────────
check_consumer_interfaces() {
  echo -e "${BOLD}E. Antarmuka konsumen — dbt / StarRocks / Airflow / OpenMetadata (informatif)${NC}"

  local prof="${CONFIG_DIR}/airflow/dbt/profiles.yml"
  if [[ -f "${prof}" ]]; then
    local port scheme
    port=$(sed -n 's/^ *port: *//p' "${prof}" | head -1)
    scheme=$(sed -n 's/^ *http_scheme: *//p' "${prof}" | head -1)
    if [[ "${port}" == "8443" && "${scheme}" == "https" ]]; then
      pass "dbt profiles.yml → https://…:8443 (sesuai HTTPS-only Trino)"
    else
      warn "dbt profiles.yml port=${port:-?} scheme=${scheme:-?} — pastikan cocok dengan endpoint Trino"
    fi
  else
    skip "config/airflow/dbt/profiles.yml belum ada"
  fi

  local props="${CONFIG_DIR}/trino/config-coordinator.properties"
  if grep -q "internal-communication.shared-secret=trino-internal-secret-dev-only" "${props}" 2>/dev/null; then
    warn "internal-communication.shared-secret masih nilai DEV — ganti untuk instalasi nyata"
  else
    pass "internal-communication.shared-secret tidak memakai nilai dev default"
  fi
  if grep -qE '^http-server.https.keystore.path=' "${props}" 2>/dev/null; then
    warn "keystore.p12 ikut ter-commit (rahasia DEV 'trino_ssl_dev') — regenerate untuk instalasi nyata"
  fi

  if ! sicabe_host_expects_container sicabe-starrocks-fe; then
    skip "Container sicabe-starrocks-fe di host lain (mode host) - dilewati"
  elif container_running sicabe-starrocks-fe; then
    pass "StarRocks berjalan — external catalog Trino diuji di gate Phase 4"
  else
    skip "StarRocks belum berjalan (fase 4) — lewati pemeriksaan antarmukanya"
  fi
  if ! sicabe_host_expects_container sicabe-airflow-webserver; then
    skip "Container sicabe-airflow-webserver di host lain (mode host) - dilewati"
  elif container_running sicabe-airflow-webserver; then
    pass "Airflow berjalan — koneksi Trino/dbt diuji di gate Phase 5"
  else
    skip "Airflow belum berjalan (fase 5) — lewati pemeriksaan antarmukanya"
  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 Trino diuji di gate Phase 6"
  else
    skip "OpenMetadata belum berjalan (fase 6) — lewati pemeriksaan antarmukanya"
  fi
}

# ─────────────────────────────────────────────────────────────────────────────
# Main
# ─────────────────────────────────────────────────────────────────────────────
main() {
  print_section "Verifikasi Phase 3 — Kontrak Processing (Trino + OPA)"
  echo -e "  Trino : ${CYAN}${TRINO_URL}${NC} (HTTPS) · OPA: ${CYAN}${OPA_URL}${NC}"
  echo -e "  Authz : ${CYAN}${OPA_URL}${OPA_POLICY_PATH}${NC} (deny-by-default, ACL di memori)"
  echo ""

  check_trino_service
  check_opa_policy
  check_enforcement
  check_authenticator
  check_consumer_interfaces

  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 processing Phase 3 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 "$@"