#!/usr/bin/env bash
#
# Быстрый runtime gate для issue 17: проверяет, что daily-wave доходит до
# manifest/check, а live-продолжение реально записывает стык backfill/live.

set -euo pipefail

SCRIPT_DIR="$(cd "$(dirname "${BASH_SOURCE[0]}")" && pwd)"
REPO_ROOT="$(cd "${SCRIPT_DIR}/.." && pwd)"

COMPOSE_BIN="${COMPOSE_BIN:-docker compose}"
PROFILE="${PROFILE:-daily-wave}"
GEN_HISTORY_DURATION="${GEN_HISTORY_DURATION:-1h}"
WAIT_CLICKHOUSE_SECONDS="${WAIT_CLICKHOUSE_SECONDS:-60}"
WAIT_STG_SECONDS="${WAIT_STG_SECONDS:-5}"
LIVE_SECONDS="${LIVE_SECONDS:-25}"
GEN_LIVE_CHECK_MINUTES="${GEN_LIVE_CHECK_MINUTES:-10}"
WAIT_LIVE_ROWS_SECONDS="${WAIT_LIVE_ROWS_SECONDS:-2}"

cd "${REPO_ROOT}"

launch_env="$(
  PYTHONPATH="${REPO_ROOT}/generator/src" \
  uv run python -m clickstream_generator.launch \
    backfill \
    --profile "${PROFILE}" \
    --duration "${GEN_HISTORY_DURATION}"
)"
eval "${launch_env}"

wait_for_clickhouse() {
  local deadline
  deadline=$((SECONDS + WAIT_CLICKHOUSE_SECONDS))

  until ${COMPOSE_BIN} exec -T clickhouse clickhouse-client \
      --user=default \
      --password=123456 \
      --query "SELECT 1" >/dev/null 2>&1; do
    if (( SECONDS >= deadline )); then
      echo "Ошибка: ClickHouse не ответил за ${WAIT_CLICKHOUSE_SECONDS} сек." >&2
      exit 1
    fi
    sleep 2
  done
}

stg_total_rows() {
  ${COMPOSE_BIN} exec -T clickhouse clickhouse-client \
    --user=default \
    --password=123456 \
    --query "
SELECT
    (SELECT count() FROM stg.browser_raw)
  + (SELECT count() FROM stg.location_raw)
  + (SELECT count() FROM stg.device_raw)
  + (SELECT count() FROM stg.geo_raw)
FORMAT TabSeparated"
}

stg_crossing_visits() {
  ${COMPOSE_BIN} exec -T clickhouse clickhouse-client \
    --user=default \
    --password=123456 \
    --query "
WITH
    parseDateTime64BestEffort('${GEN_MODEL_T_END}', 6, 'UTC') AS t_end,
    t_end + INTERVAL ${GEN_LIVE_CHECK_MINUTES} MINUTE AS t_live_end,
    browser AS (
        SELECT
            JSONExtractString(raw, 'click_id') AS click_id,
            parseDateTime64BestEffortOrNull(
                JSONExtractString(raw, 'event_timestamp'), 6, 'UTC'
            ) AS event_ts
        FROM stg.browser_raw
        WHERE click_id != ''
          AND event_ts >= t_end - INTERVAL 30 MINUTE
          AND event_ts < t_live_end
    )
SELECT count()
FROM
(
    SELECT
        click_id,
        min(event_ts) AS first_ts,
        max(event_ts) AS last_ts
    FROM browser
    GROUP BY click_id
    HAVING first_ts < t_end AND last_ts >= t_end
)
FORMAT TabSeparated"
}

cleanup_live_generator() {
  ${COMPOSE_BIN} stop generator >/dev/null 2>&1 || true
}

stop_live_generator() {
  # docker compose stop возвращается только после фактической остановки контейнера.
  ${COMPOSE_BIN} stop generator
}
trap cleanup_live_generator EXIT

echo "=== Быстрая runtime-проверка startup-history/live seam ==="
echo "GEN_LAUNCH_PROFILE=${GEN_LAUNCH_PROFILE}"
echo "GEN_HISTORY_DURATION=${GEN_HISTORY_DURATION}"
echo "GEN_MODEL_T_END=${GEN_MODEL_T_END}"
echo ""

echo "Шаг 0: очистка volumes ClickHouse/Kafka/state"
${COMPOSE_BIN} --profile live-generator down -v --remove-orphans

echo "Шаг 1: запуск ClickHouse и Kafka"
${COMPOSE_BIN} up -d clickhouse kafka
wait_for_clickhouse

echo "Шаг 2: применение DDL"
bash "${SCRIPT_DIR}/apply_clickhouse_ddl.sh"

echo "Шаг 3: сборка образа генератора"
${COMPOSE_BIN} build generator

echo "Шаг 4: короткий backfill стартовой истории"
${COMPOSE_BIN} run --rm --no-deps \
  -e GEN_RUN_MODE=backfill \
  -e GEN_STATE_RESET=true \
  -e GEN_LAUNCH_PROFILE="${GEN_LAUNCH_PROFILE}" \
  -e GEN_SEED="${GEN_SEED}" \
  -e GEN_MODEL_T0="${GEN_MODEL_T0}" \
  -e GEN_MODEL_T_END="${GEN_MODEL_T_END}" \
  -e GEN_MODEL_TIMEZONE="${GEN_MODEL_TIMEZONE}" \
  -e GEN_MODEL_TIME_SPEED="${GEN_MODEL_TIME_SPEED}" \
  -e GEN_TICK_SECONDS="${GEN_TICK_SECONDS}" \
  -e GEN_LAMBDA_BASE_PER_MIN="${GEN_LAMBDA_BASE_PER_MIN}" \
  -e GEN_JITTER_PCT="${GEN_JITTER_PCT}" \
  -e GEN_MIN_EVENTS_PER_TICK="${GEN_MIN_EVENTS_PER_TICK}" \
  -e GEN_MAX_EVENTS_PER_TICK="${GEN_MAX_EVENTS_PER_TICK}" \
  generator

echo "Шаг 5: первый batch STG -> ODS -> DDS -> DM"
sleep "${WAIT_STG_SECONDS}"
bash "${SCRIPT_DIR}/run_batch.sh"
stg_rows_before_live="$(stg_total_rows)"

echo "Шаг 6: live-продолжение, ждём переходящий визит в STG до ${LIVE_SECONDS} сек."
GEN_RUN_MODE=live \
GEN_STATE_RESET=false \
GEN_LAUNCH_PROFILE="${GEN_LAUNCH_PROFILE}" \
GEN_SEED="${GEN_SEED}" \
GEN_MODEL_T0="${GEN_MODEL_T0}" \
GEN_MODEL_T_END="${GEN_MODEL_T_END}" \
GEN_MODEL_TIMEZONE="${GEN_MODEL_TIMEZONE}" \
GEN_MODEL_TIME_SPEED="${GEN_MODEL_TIME_SPEED}" \
GEN_TICK_SECONDS="${GEN_TICK_SECONDS}" \
GEN_LAMBDA_BASE_PER_MIN="${GEN_LAMBDA_BASE_PER_MIN}" \
GEN_JITTER_PCT="${GEN_JITTER_PCT}" \
GEN_MIN_EVENTS_PER_TICK="${GEN_MIN_EVENTS_PER_TICK}" \
GEN_MAX_EVENTS_PER_TICK="${GEN_MAX_EVENTS_PER_TICK}" \
${COMPOSE_BIN} up -d --build generator
deadline=$((SECONDS + LIVE_SECONDS))
while true; do
  crossing_visits="$(stg_crossing_visits)"
  [[ "${crossing_visits}" =~ ^[0-9]+$ ]] \
    || { echo "Ошибка: не удалось прочитать переходящие визиты из STG." >&2; exit 1; }
  if (( crossing_visits > 0 )); then
    stg_rows_after_live="$(stg_total_rows)"
    echo "live_stg_rows_before=${stg_rows_before_live}"
    echo "live_stg_rows_after=${stg_rows_after_live}"
    echo "live_stg_crossing_visits=${crossing_visits}"
    break
  fi
  if (( SECONDS >= deadline )); then
    echo "Ошибка: live-продолжение не создало переходящий визит в STG за ${LIVE_SECONDS} сек." >&2
    exit 1
  fi
  sleep 1
done
stop_live_generator

echo "Шаг 7: второй batch после live"
sleep "${WAIT_STG_SECONDS}"
bash "${SCRIPT_DIR}/run_batch.sh"

echo "Шаг 8: проверка manifest/profile/live seam без Superset"
PROFILE="${PROFILE}" \
GEN_LAUNCH_PROFILE="${GEN_LAUNCH_PROFILE}" \
GEN_LIVE_CHECK_MINUTES="${GEN_LIVE_CHECK_MINUTES}" \
WAIT_LIVE_ROWS_SECONDS="${WAIT_LIVE_ROWS_SECONDS}" \
CHECK_LIVE_SEAM=1 \
REQUIRE_SUPERSET=0 \
COMPOSE_BIN="${COMPOSE_BIN}" \
bash "${SCRIPT_DIR}/check_generated_analytics.sh"

echo ""
echo "Готово: runtime-проверка startup-history/live seam прошла."