#!/bin/bash
# Live acceptance for issue #5: incremental manifest counters.
# Focus: next-day runs must be O(new day) — flat wall time across days,
# no full Kafka reread; chain-check green on the grown world.
# UI actions emulated via airflow CLI in the scheduler container
# (session-auth-only REST on this stand; trigger with no conf == empty form).
set -u
cd /home/dementev/sources/clickstream-ch-kafka-superset-demo || exit 1

AF() { docker compose exec -T airflow-scheduler airflow "$@"; }
CH_QUERY() { docker compose exec -T clickhouse clickhouse-client --user=default --password=123456 --query "$1"; }

step() { echo; echo "=== [$(date +%H:%M:%S)] $*"; }
fail() { echo "!!! ACCEPTANCE FAILED: $*"; exit 1; }

latest_run_state() { # dag_id -> "run_id state" of newest run
  AF dags list-runs -d "$1" -o json 2>/dev/null | python3 -c '
import sys, json
rs = json.load(sys.stdin)
rs.sort(key=lambda r: r["execution_date"], reverse=True)
print(rs[0]["run_id"], rs[0]["state"]) if rs else print("none", "none")'
}

runs_count() {
  AF dags list-runs -d "$1" -o json 2>/dev/null | python3 -c 'import sys,json;print(len(json.load(sys.stdin)))'
}

wait_dag_done() { # dag_id timeout_sec
  local dag="$1" t="$2" line="" state=""
  local deadline=$(( $(date +%s) + t ))
  while [ "$(date +%s)" -lt "$deadline" ]; do
    line=$(latest_run_state "$dag"); state="${line##* }"
    case "$state" in
      success) echo "$dag ${line% *}: success"; return 0 ;;
      failed)  echo "$dag ${line% *}: FAILED"; return 1 ;;
      *) sleep 20 ;;
    esac
  done
  echo "$dag: TIMEOUT (last: $line)"; return 1
}

run_durations() { # dag_id -> per-run "run_id duration_s" sorted by exec date
  AF dags list-runs -d "$1" -o json 2>/dev/null | python3 -c '
import sys, json, datetime as dt
def p(s): return dt.datetime.fromisoformat(s.replace("Z", "+00:00"))
rs = json.load(sys.stdin)
rs.sort(key=lambda r: r["execution_date"])
for r in rs:
    if r.get("start_date") and r.get("end_date"):
        d = (p(r["end_date"]) - p(r["start_date"])).total_seconds()
        print(f"{r[\"run_id\"]} {r[\"state\"]} {d:.0f}s")
    else:
        print(f"{r[\"run_id\"]} {r[\"state\"]} -")'
}

world_stats() {
  CH_QUERY "SELECT uniqExact(event_date) AS days, count() AS events, uniqExact(user_domain_id) AS users FROM dm.v_events_enriched FORMAT TSVWithNames"
}

# --- 0. Pre-flight ---
step "Pre-flight"
[ -S /var/run/docker.sock ] || fail "docker socket absent"
docker info >/dev/null 2>&1 || fail "docker daemon not answering"
echo "docker OK"

# --- 1. Clean stand + up ---
step "make clean (wipe volumes)"
make clean || fail "make clean"
step "make up"
make up || fail "make up"

step "wait for Airflow (scheduler parsed DAGs, webserver answers)"
ok=""
for i in $(seq 1 60); do
  if AF dags list -o plain 2>/dev/null | grep -q "world_init"; then ok=1; break; fi
  sleep 10
done
[ -n "$ok" ] || fail "world_init not parsed after 10 min"
ok=""
for i in $(seq 1 18); do
  if curl -sf -o /dev/null http://localhost:8080/login/; then ok=1; break; fi
  sleep 10
done
[ -n "$ok" ] || fail "webserver UI not answering after 3 min"
echo "Airflow up"

# --- 2. Ladder: ddl_init, etl, world_init (empty form => import) ---
step "ddl_init: unpause + trigger"
AF dags unpause ddl_init >/dev/null || fail "unpause ddl_init"
AF dags trigger ddl_init >/dev/null || fail "trigger ddl_init"
wait_dag_done ddl_init 600 || fail "ddl_init run"

step "unpause etl_pipeline; world_init trigger (empty form)"
AF dags unpause etl_pipeline >/dev/null || fail "unpause etl_pipeline"
AF dags unpause world_init >/dev/null || fail "unpause world_init"
t0=$(date +%s)
AF dags trigger world_init >/dev/null || fail "trigger world_init"
wait_dag_done world_init 2400 || fail "world_init run (import)"
echo "IMPORT_WALL_TIME=$(( $(date +%s) - t0 ))s"

step "world after import"
world_stats || fail "dm query after import"

# --- 3. Three next-day runs, each timed (core of #5) ---
# Unpause => catchup=False gives run 1 immediately; runs 2-3 by manual
# trigger (same empty-form semantics). max_active_runs=1 serializes any
# stray :30 boundary run; durations are read from run metadata per run.
step "world_next_day: unpause (run 1)"
AF dags unpause world_next_day >/dev/null || fail "unpause world_next_day"
sleep 60
[ "$(runs_count world_next_day)" -ge 1 ] || fail "no immediate run after unpause"
wait_dag_done world_next_day 3600 || fail "next-day run 1"

for i in 2 3; do
  step "world_next_day: manual trigger (run $i)"
  AF dags trigger world_next_day >/dev/null || fail "trigger run $i"
  sleep 20
  wait_dag_done world_next_day 3600 || fail "next-day run $i"
done

step "re-pause world_next_day"
AF dags pause world_next_day >/dev/null || fail "re-pause"

step "next-day run durations (must be roughly flat: O(new day))"
run_durations world_next_day

step "world after 3 next days"
world_stats || fail "dm query after next days"

# --- 4. Chain check on the grown world ---
step "make generated-history-chain-check"
make generated-history-chain-check || fail "chain-check"

echo
echo "=== ACCEPTANCE PASSED [$(date +%H:%M:%S)] ==="