#!/usr/bin/env bash
#
# Повторяемая проверка, что DM-витрины и Superset metadata работают на данных
# стартовой истории генератора, а не на архивном сиде 2022 года.

set -euo pipefail

COMPOSE_BIN="${COMPOSE_BIN:-docker compose}"
CLICKHOUSE_SERVICE="${CLICKHOUSE_SERVICE:-clickhouse}"
CLICKHOUSE_USER="${CLICKHOUSE_USER:-default}"
CLICKHOUSE_PASSWORD="${CLICKHOUSE_PASSWORD:-123456}"
GEN_MODEL_T0="${GEN_MODEL_T0:-2026-01-01T00:00:00+00:00}"
GEN_MODEL_T_END="${GEN_MODEL_T_END:-2026-01-01T06:00:00+00:00}"
REQUIRE_SUPERSET="${REQUIRE_SUPERSET:-1}"
CHECK_LIVE_SEAM="${CHECK_LIVE_SEAM:-1}"
GEN_LIVE_CHECK_MINUTES="${GEN_LIVE_CHECK_MINUTES:-10}"
WAIT_LIVE_ROWS_SECONDS="${WAIT_LIVE_ROWS_SECONDS:-10}"

fail() {
  echo "Ошибка: $*" >&2
  exit 1
}

clickhouse_datetime_literal() {
  local value="$1"
  value="${value/T/ }"
  value="${value%Z}"
  if [[ "${value}" =~ ^(.*)[+-][0-9]{2}:[0-9]{2}$ ]]; then
    value="${BASH_REMATCH[1]}"
  fi
  echo "${value}"
}

ch_query() {
  ${COMPOSE_BIN} exec -T "${CLICKHOUSE_SERVICE}" clickhouse-client \
    --user="${CLICKHOUSE_USER}" \
    --password="${CLICKHOUSE_PASSWORD}" \
    --query "$1"
}

pg_query() {
  ${COMPOSE_BIN} exec -T postgres-metadata psql \
    -U airflow \
    -d superset \
    -At \
    -c "$1" | tr -d '\r'
}

CH_MODEL_T0="$(clickhouse_datetime_literal "${GEN_MODEL_T0}")"
CH_MODEL_T_END="$(clickhouse_datetime_literal "${GEN_MODEL_T_END}")"

[[ "${GEN_LIVE_CHECK_MINUTES}" =~ ^[0-9]+$ ]] \
  || fail "GEN_LIVE_CHECK_MINUTES должен быть целым числом минут"
[[ "${WAIT_LIVE_ROWS_SECONDS}" =~ ^[0-9]+$ ]] \
  || fail "WAIT_LIVE_ROWS_SECONDS должен быть целым числом секунд"
case "${CHECK_LIVE_SEAM}" in
  0|1|auto) ;;
  *) fail "CHECK_LIVE_SEAM должен быть 0, 1 или auto" ;;
esac

echo "=== Проверка startup-history manifest ==="

manifest_summary="$(${COMPOSE_BIN} run --rm --no-deps \
  generator python -m clickstream_generator.startup_history_artifact kafka-manifest-summary)"

IFS=$'\t' read -r manifest_events manifest_visits manifest_users manifest_min_ts manifest_max_ts manifest_model_t0 manifest_model_t_end manifest_launch_profile <<< "${manifest_summary}"

[[ "${manifest_events}" =~ ^[0-9]+$ ]] || fail "не удалось прочитать startup-history manifest из Kafka"
[[ -n "${manifest_model_t0}" ]] || fail "startup-history manifest не содержит model_t0"
[[ -n "${manifest_model_t_end}" ]] || fail "startup-history manifest не содержит model_t_end"

CH_MODEL_T0="$(clickhouse_datetime_literal "${manifest_model_t0}")"
CH_MODEL_T_END="$(clickhouse_datetime_literal "${manifest_model_t_end}")"

expected_profile="${PROFILE:-${GEN_LAUNCH_PROFILE:-}}"
if [[ -n "${expected_profile}" && -n "${manifest_launch_profile}" && "${expected_profile}" != "${manifest_launch_profile}" ]]; then
  fail "проверка запущена для профиля ${expected_profile}, но manifest от ${manifest_launch_profile}"
fi

echo "manifest_events=${manifest_events}"
echo "manifest_visits=${manifest_visits}"
echo "manifest_users=${manifest_users}"
echo "manifest_model_t0=${manifest_model_t0}"
echo "manifest_model_t_end=${manifest_model_t_end}"
echo "manifest_launch_profile=${manifest_launch_profile}"

echo "=== Проверка ClickHouse: данные генерации в DM ==="

stats_query="
WITH
    toDateTime64('${CH_MODEL_T0}', 6) AS t0,
    toDateTime64('${CH_MODEL_T_END}', 6) AS t_end
SELECT
    count() AS events,
    uniqExact(click_id) AS visits,
    uniqExact(user_domain_id) AS users,
    toString(min(event_ts)) AS min_event_ts,
    toString(max(event_ts)) AS max_event_ts,
    users < visits AND visits < events AS pyramid_ok,
    min(event_ts) >= t0 AND max(event_ts) < t_end AS half_open_ok,
    countIf(toYear(event_ts) = 2022) = 0 AS no_2022_rows,
    hex(sipHash128(groupArray(tuple(event_id, click_id, user_domain_id, event_ts, page_url_path)))) AS digest
FROM
(
    SELECT
        event_id,
        click_id,
        user_domain_id,
        event_ts,
        page_url_path
    FROM dm.v_events_enriched
    WHERE event_ts >= t0 AND event_ts < t_end
    ORDER BY event_id
)
FORMAT TabSeparated"

stats="$(ch_query "${stats_query}")"
IFS=$'\t' read -r events visits users min_event_ts max_event_ts pyramid_ok half_open_ok no_2022_rows digest <<< "${stats}"

[[ "${events}" =~ ^[0-9]+$ ]] || fail "не удалось прочитать число событий из dm.v_events_enriched"
(( events > 0 )) || fail "dm.v_events_enriched пустая в модельном диапазоне"
[[ "${pyramid_ok}" == "1" ]] || fail "нарушена пирамида users < visits < events"
[[ "${half_open_ok}" == "1" ]] || fail "данные вышли за диапазон [GEN_MODEL_T0, GEN_MODEL_T_END)"
[[ "${no_2022_rows}" == "1" ]] || fail "найдены строки 2022 года, похожие на архивный сид"

echo "events=${events}"
echo "visits=${visits}"
echo "users=${users}"
echo "min_event_ts=${min_event_ts}"
echo "max_event_ts=${max_event_ts}"
echo "digest=${digest}"

echo ""
echo "=== Проверка ClickHouse: возвраты пользователей ==="

returns_query="
WITH
    toDateTime64('${CH_MODEL_T0}', 6) AS t0,
    toDateTime64('${CH_MODEL_T_END}', 6) AS t_end,
    users AS (
        SELECT user_domain_id, uniqExact(click_id) AS visits
        FROM dm.v_events_enriched
        WHERE event_ts >= t0 AND event_ts < t_end
          AND user_domain_id IS NOT NULL
        GROUP BY user_domain_id
    )
SELECT
    count() AS users,
    countIf(visits > 1) AS returning_users,
    returning_users / users AS returning_share,
    max(visits) AS max_visits_per_user
FROM users
FORMAT TabSeparated"

returns="$(ch_query "${returns_query}")"
IFS=$'\t' read -r total_users returning_users returning_share max_visits_per_user <<< "${returns}"

[[ "${total_users}" =~ ^[0-9]+$ ]] || fail "не удалось прочитать число пользователей"
(( total_users > 0 )) || fail "нет пользователей для проверки возвратов"
(( returning_users > 0 )) || fail "нет пользователей с повторными визитами"

echo "users=${total_users}"
echo "returning_users=${returning_users}"
echo "returning_share=${returning_share}"
echo "max_visits_per_user=${max_visits_per_user}"

echo ""
echo "=== Проверка ClickHouse: форма длины визита ==="

visit_shape_query="
WITH
    toDateTime64('${CH_MODEL_T0}', 6) AS t0,
    toDateTime64('${CH_MODEL_T_END}', 6) AS t_end,
    30 AS max_session_events,
    sessions AS (
        SELECT
            click_id,
            count() AS events_count,
            dateDiff('second', min(event_ts), max(event_ts)) AS duration_sec
        FROM dm.v_events_enriched
        WHERE event_ts >= t0 AND event_ts < t_end
        GROUP BY click_id
    )
SELECT
    count() AS visits,
    countIf(events_count <= 2) / visits AS short_visit_share,
    quantileExact(0.5)(events_count) AS median_events_per_visit,
    avg(events_count) AS avg_events_per_visit,
    countIf(events_count = max_session_events) / visits AS capped_visit_share,
    quantileExact(0.5)(duration_sec) AS median_duration_sec,
    quantileExact(0.95)(duration_sec) AS p95_duration_sec,
    max(events_count) AS max_events_per_visit
FROM sessions
FORMAT TabSeparated"

visit_shape="$(ch_query "${visit_shape_query}")"
IFS=$'\t' read -r shape_visits short_visit_share median_events_per_visit avg_events_per_visit capped_visit_share median_duration_sec p95_duration_sec max_events_per_visit <<< "${visit_shape}"

[[ "${shape_visits}" =~ ^[0-9]+$ ]] || fail "не удалось прочитать число визитов"
(( shape_visits > 0 )) || fail "нет визитов для проверки формы"
(( max_events_per_visit <= 30 )) || fail "длина визита превысила GEN_MAX_SESSION_EVENTS"

echo "visits=${shape_visits}"
echo "short_visit_share=${short_visit_share}"
echo "median_events_per_visit=${median_events_per_visit}"
echo "avg_events_per_visit=${avg_events_per_visit}"
echo "capped_visit_share=${capped_visit_share}"
echo "median_duration_sec=${median_duration_sec}"
echo "p95_duration_sec=${p95_duration_sec}"
echo "max_events_per_visit=${max_events_per_visit}"

echo ""
echo "=== Проверка ClickHouse: ordered funnel ==="

ordered_funnel_query="
WITH
    toDateTime64('${CH_MODEL_T0}', 6) AS t0,
    toDateTime64('${CH_MODEL_T_END}', 6) AS t_end,
    sessions AS (
        SELECT
            click_id,
            minIf(event_ts, page_url_path = '/home') AS home_ts,
            minIf(event_ts, page_url_path IN ('/product_a', '/product_b')) AS product_ts,
            minIf(event_ts, page_url_path = '/cart') AS cart_ts,
            minIf(event_ts, page_url_path = '/payment') AS payment_ts,
            minIf(event_ts, page_url_path = '/confirmation') AS confirmation_ts
        FROM dm.v_events_enriched
        WHERE event_ts >= t0 AND event_ts < t_end
        GROUP BY click_id
    )
SELECT
    countIf(home_ts IS NOT NULL) AS home,
    countIf(home_ts IS NOT NULL AND product_ts > home_ts) AS products,
    countIf(home_ts IS NOT NULL AND product_ts > home_ts AND cart_ts > product_ts) AS cart,
    countIf(home_ts IS NOT NULL AND product_ts > home_ts AND cart_ts > product_ts AND payment_ts > cart_ts) AS payment,
    countIf(home_ts IS NOT NULL AND product_ts > home_ts AND cart_ts > product_ts AND payment_ts > cart_ts AND confirmation_ts > payment_ts) AS confirmation,
    products <= home AND cart <= products AND payment <= cart AND confirmation <= payment AS monotonic_ok,
    confirmation / home AS confirmation_share
FROM sessions
FORMAT TabSeparated"

ordered_funnel="$(ch_query "${ordered_funnel_query}")"
IFS=$'\t' read -r ordered_home ordered_products ordered_cart ordered_payment ordered_confirmation ordered_monotonic_ok ordered_confirmation_share <<< "${ordered_funnel}"

[[ "${ordered_home}" =~ ^[0-9]+$ ]] || fail "не удалось прочитать ordered funnel"
(( ordered_home > 0 )) || fail "ordered funnel: нет /home"
[[ "${ordered_monotonic_ok}" == "1" ]] || fail "ordered funnel не монотонен"

echo "home=${ordered_home}"
echo "products=${ordered_products}"
echo "cart=${ordered_cart}"
echo "payment=${ordered_payment}"
echo "confirmation=${ordered_confirmation}"
echo "monotonic_ok=${ordered_monotonic_ok}"
echo "confirmation_share=${ordered_confirmation_share}"

echo ""
echo "=== Проверка ClickHouse: contains funnel ==="

contains_funnel_query="
WITH
    toDateTime64('${CH_MODEL_T0}', 6) AS t0,
    toDateTime64('${CH_MODEL_T_END}', 6) AS t_end,
    sessions AS (
        SELECT
            click_id,
            countIf(page_url_path = '/home') > 0 AS has_home,
            countIf(page_url_path IN ('/product_a', '/product_b')) > 0 AS has_product,
            countIf(page_url_path = '/cart') > 0 AS has_cart,
            countIf(page_url_path = '/payment') > 0 AS has_payment,
            countIf(page_url_path = '/confirmation') > 0 AS has_confirmation
        FROM dm.v_events_enriched
        WHERE event_ts >= t0 AND event_ts < t_end
        GROUP BY click_id
    )
SELECT
    countIf(has_home) AS home,
    countIf(has_home AND has_product) AS products,
    countIf(has_home AND has_product AND has_cart) AS cart,
    countIf(has_home AND has_product AND has_cart AND has_payment) AS payment,
    countIf(has_home AND has_product AND has_cart AND has_payment AND has_confirmation) AS confirmation,
    products <= home AND cart <= products AND payment <= cart AND confirmation <= payment AS monotonic_ok,
    confirmation / home AS confirmation_share
FROM sessions
FORMAT TabSeparated"

contains_funnel="$(ch_query "${contains_funnel_query}")"
IFS=$'\t' read -r contains_home contains_products contains_cart contains_payment contains_confirmation contains_monotonic_ok contains_confirmation_share <<< "${contains_funnel}"

[[ "${contains_home}" =~ ^[0-9]+$ ]] || fail "не удалось прочитать contains funnel"
(( contains_home > 0 )) || fail "contains funnel: нет /home"
[[ "${contains_monotonic_ok}" == "1" ]] || fail "contains funnel не монотонен"

echo "home=${contains_home}"
echo "products=${contains_products}"
echo "cart=${contains_cart}"
echo "payment=${contains_payment}"
echo "confirmation=${contains_confirmation}"
echo "monotonic_ok=${contains_monotonic_ok}"
echo "confirmation_share=${contains_confirmation_share}"

should_check_live_seam="${CHECK_LIVE_SEAM}"
if [[ "${should_check_live_seam}" != "0" ]]; then
  live_rows_query="
WITH
    toDateTime64('${CH_MODEL_T_END}', 6) AS t_end,
    t_end + INTERVAL ${GEN_LIVE_CHECK_MINUTES} MINUTE AS t_live_end
SELECT count()
FROM dds.event
WHERE event_ts >= t_end AND event_ts < t_live_end
FORMAT TabSeparated"
  live_rows=0
  deadline=$((SECONDS + WAIT_LIVE_ROWS_SECONDS))
  while true; do
    live_rows="$(ch_query "${live_rows_query}")"
    [[ "${live_rows}" =~ ^[0-9]+$ ]] || fail "не удалось прочитать live-строки DDS после GEN_MODEL_T_END"
    if (( live_rows > 0 || SECONDS >= deadline )); then
      break
    fi
    sleep 1
  done
  if [[ "${should_check_live_seam}" == "1" ]] && (( live_rows == 0 )); then
    fail "live-продолжение не записало строки в первые ${GEN_LIVE_CHECK_MINUTES} минут после manifest model_t_end"
  fi
  if [[ "${CHECK_LIVE_SEAM}" == "auto" ]]; then
    if (( live_rows > 0 )); then
      should_check_live_seam="1"
    else
      should_check_live_seam="0"
    fi
  fi
fi

if [[ "${should_check_live_seam}" == "1" ]]; then
  echo ""
  echo "=== Проверка ClickHouse: стык backfill/live в DDS ==="

  seam_duplicates_query="
WITH
    toDateTime64('${CH_MODEL_T0}', 6) AS t0,
    toDateTime64('${CH_MODEL_T_END}', 6) AS t_end,
    t_end + INTERVAL ${GEN_LIVE_CHECK_MINUTES} MINUTE AS t_live_end
SELECT
    count() AS events,
    uniqExact(event_id) AS unique_events,
    events - unique_events AS duplicate_events
FROM dds.event
WHERE event_ts >= t0 AND event_ts < t_live_end
FORMAT TabSeparated"

  seam_duplicates="$(ch_query "${seam_duplicates_query}")"
  IFS=$'\t' read -r seam_events seam_unique_events seam_duplicate_events <<< "${seam_duplicates}"

  [[ "${seam_events}" =~ ^[0-9]+$ ]] || fail "не удалось прочитать события DDS на стыке"
  (( seam_events > 0 )) || fail "нет событий DDS в диапазоне проверки стыка"
  (( seam_duplicate_events == 0 )) || fail "на стыке найдены дубли event_id: ${seam_duplicate_events}"

  echo "events=${seam_events}"
  echo "unique_events=${seam_unique_events}"
  echo "duplicate_events=${seam_duplicate_events}"

  seam_context_query="
WITH
    toDateTime64('${CH_MODEL_T_END}', 6) AS t_end,
    t_end + INTERVAL ${GEN_LIVE_CHECK_MINUTES} MINUTE AS t_live_end,
    crossing AS (
        SELECT
            e.click_id,
            min(e.event_ts) AS first_ts,
            max(e.event_ts) AS last_ts,
            groupUniqArray(coalesce(toString(e.browser_name), '__NULL__')) AS browser_names,
            groupUniqArray(coalesce(toString(e.browser_language), '__NULL__')) AS browser_languages,
            groupUniqArray(coalesce(toString(e.browser_user_agent), '__NULL__')) AS browser_user_agents,
            groupUniqArray(coalesce(toString(e.referer_url), '__NULL__')) AS referer_urls,
            groupUniqArray(coalesce(toString(e.referer_medium), '__NULL__')) AS referer_mediums,
            groupUniqArray(coalesce(toString(e.utm_medium), '__NULL__')) AS utm_mediums,
            groupUniqArray(coalesce(toString(e.utm_source), '__NULL__')) AS utm_sources,
            groupUniqArray(coalesce(toString(e.utm_content), '__NULL__')) AS utm_contents,
            groupUniqArray(coalesce(toString(e.utm_campaign), '__NULL__')) AS utm_campaigns,
            groupUniqArray(coalesce(toString(c.device_type), '__NULL__')) AS device_types,
            groupUniqArray(coalesce(toString(c.os_name), '__NULL__')) AS os_names,
            groupUniqArray(coalesce(toString(c.geo_country), '__NULL__')) AS geo_countries
        FROM dds.event AS e
        LEFT JOIN dds.click AS c ON c.click_id = e.click_id
        WHERE e.event_ts >= t_end - INTERVAL 30 MINUTE
          AND e.event_ts < t_live_end
        GROUP BY e.click_id
        HAVING first_ts < t_end AND last_ts >= t_end
    )
SELECT
    count() AS crossing_visits,
    countIf(
        length(browser_names) = 1
        AND length(browser_languages) = 1
        AND length(browser_user_agents) = 1
        AND length(referer_urls) = 1
        AND length(referer_mediums) = 1
        AND length(utm_mediums) = 1
        AND length(utm_sources) = 1
        AND length(utm_contents) = 1
        AND length(utm_campaigns) = 1
    ) AS per_event_homogeneous_visits,
    countIf(
        length(device_types) = 1
        AND length(os_names) = 1
        AND length(geo_countries) = 1
    ) AS click_context_homogeneous_visits
FROM crossing
FORMAT TabSeparated"

  seam_context="$(ch_query "${seam_context_query}")"
  IFS=$'\t' read -r crossing_visits per_event_homogeneous_visits click_context_homogeneous_visits <<< "${seam_context}"

  [[ "${crossing_visits}" =~ ^[0-9]+$ ]] || fail "не удалось прочитать переходящие визиты"
  (( crossing_visits > 0 )) || fail "нет визитов, переходящих через GEN_MODEL_T_END"
  (( per_event_homogeneous_visits == crossing_visits )) \
    || fail "per-event фактура меняется на стыке: ${per_event_homogeneous_visits}/${crossing_visits}"
  (( click_context_homogeneous_visits == crossing_visits )) \
    || fail "device/os/geo меняются на стыке: ${click_context_homogeneous_visits}/${crossing_visits}"

  echo "crossing_visits=${crossing_visits}"
  echo "per_event_homogeneous_visits=${per_event_homogeneous_visits}"
  echo "click_context_homogeneous_visits=${click_context_homogeneous_visits}"

  ods_context_query="
WITH
    toDateTime64('${CH_MODEL_T_END}', 6) AS t_end,
    t_end + INTERVAL ${GEN_LIVE_CHECK_MINUTES} MINUTE AS t_live_end,
    crossing AS (
        SELECT
            click_id
        FROM dds.event
        WHERE event_ts >= t_end - INTERVAL 30 MINUTE
          AND event_ts < t_live_end
        GROUP BY click_id
        HAVING min(event_ts) < t_end AND max(event_ts) >= t_end
    ),
    device_conflicts AS (
        SELECT
            d.click_id
        FROM ods.device_by_click AS d
        INNER JOIN crossing AS c ON c.click_id = d.click_id
        WHERE d.click_id IS NOT NULL
        GROUP BY d.click_id
        HAVING
            uniqExact(coalesce(toString(device_type), '__NULL__')) > 1
            OR uniqExact(coalesce(toString(device_is_mobile), '__NULL__')) > 1
            OR uniqExact(coalesce(toString(os_name), '__NULL__')) > 1
            OR uniqExact(coalesce(toString(os), '__NULL__')) > 1
            OR uniqExact(coalesce(toString(os_timezone), '__NULL__')) > 1
            OR uniqExact(coalesce(toString(user_domain_id), '__NULL__')) > 1
    ),
    geo_conflicts AS (
        SELECT
            g.click_id
        FROM ods.geo_by_click AS g
        INNER JOIN crossing AS c ON c.click_id = g.click_id
        WHERE g.click_id IS NOT NULL
        GROUP BY g.click_id
        HAVING
            uniqExact(coalesce(toString(geo_country), '__NULL__')) > 1
            OR uniqExact(coalesce(toString(geo_timezone), '__NULL__')) > 1
            OR uniqExact(coalesce(toString(geo_region_name), '__NULL__')) > 1
            OR uniqExact(coalesce(toString(geo_latitude), '__NULL__')) > 1
            OR uniqExact(coalesce(toString(geo_longitude), '__NULL__')) > 1
            OR uniqExact(coalesce(toString(ip_address), '__NULL__')) > 1
    )
SELECT
    (SELECT count() FROM crossing) AS crossing_visits,
    (
      SELECT count()
      FROM ods.device_by_click AS d
      INNER JOIN crossing AS c ON c.click_id = d.click_id
      WHERE d.click_id IS NOT NULL
    ) AS ods_device_rows,
    (
      SELECT count()
      FROM ods.geo_by_click AS g
      INNER JOIN crossing AS c ON c.click_id = g.click_id
      WHERE g.click_id IS NOT NULL
    ) AS ods_geo_rows,
    (SELECT count() FROM device_conflicts) AS ods_device_conflicts,
    (SELECT count() FROM geo_conflicts) AS ods_geo_conflicts
FORMAT TabSeparated"

  ods_context="$(ch_query "${ods_context_query}")"
  IFS=$'\t' read -r ods_crossing_visits ods_device_rows ods_geo_rows ods_device_conflicts ods_geo_conflicts <<< "${ods_context}"

  [[ "${ods_crossing_visits}" =~ ^[0-9]+$ ]] || fail "не удалось прочитать ODS-проверку стыка"
  (( ods_crossing_visits == crossing_visits )) \
    || fail "ODS и DDS нашли разное число переходящих визитов: ${ods_crossing_visits}/${crossing_visits}"
  (( ods_device_rows > 0 )) || fail "ODS device пустой на стыке"
  (( ods_geo_rows > 0 )) || fail "ODS geo пустой на стыке"
  (( ods_device_conflicts == 0 )) || fail "ODS device конфликтует на стыке: ${ods_device_conflicts}"
  (( ods_geo_conflicts == 0 )) || fail "ODS geo конфликтует на стыке: ${ods_geo_conflicts}"

  echo "ods_device_rows=${ods_device_rows}"
  echo "ods_geo_rows=${ods_geo_rows}"
  echo "ods_device_conflicts=${ods_device_conflicts}"
  echo "ods_geo_conflicts=${ods_geo_conflicts}"
else
  echo ""
  echo "Проверка стыка backfill/live пропущена: CHECK_LIVE_SEAM=${CHECK_LIVE_SEAM}, live_rows=${live_rows:-0}"
fi

echo ""
echo "=== Проверка ClickHouse: основные DM-витрины не пустые ==="

views_query="
SELECT source, rows
FROM
(
    SELECT 'dm.v_events_enriched' AS source, count() AS rows FROM dm.v_events_enriched
    UNION ALL SELECT 'dm.v_daily_traffic', count() FROM dm.v_daily_traffic
    UNION ALL SELECT 'dm.v_top_pages_daily', count() FROM dm.v_top_pages_daily
    UNION ALL SELECT 'dm.v_utm_effectiveness', count() FROM dm.v_utm_effectiveness
    UNION ALL SELECT 'dm.v_session_overview', count() FROM dm.v_session_overview
    UNION ALL SELECT 'dm.dq_summary', count() FROM dm.dq_summary
)
ORDER BY source
FORMAT TabSeparated"

while IFS=$'\t' read -r source rows; do
  [[ -n "${source}" ]] || continue
  [[ "${rows}" =~ ^[0-9]+$ ]] || fail "не удалось прочитать число строк для ${source}"
  (( rows > 0 )) || fail "${source} пустая"
  echo "${source}=${rows}"
done < <(ch_query "${views_query}")

if [[ "${REQUIRE_SUPERSET}" != "1" ]]; then
  echo ""
  echo "Проверка Superset пропущена: REQUIRE_SUPERSET=${REQUIRE_SUPERSET}"
  exit 0
fi

echo ""
echo "=== Проверка Superset: datasets, charts и dashboard созданы ==="

if ! ${COMPOSE_BIN} ps --services --filter "status=running" | grep -qx "superset"; then
  fail "сервис superset не запущен"
fi

${COMPOSE_BIN} exec -T superset curl -fsS http://localhost:8088/health >/dev/null \
  || fail "Superset health endpoint не отвечает"

dataset_count="$(pg_query "
SELECT count(*)
FROM tables
WHERE schema = 'dm'
  AND table_name IN (
    'v_events_enriched',
    'v_daily_traffic',
    'v_utm_effectiveness',
    'v_top_pages_daily',
    'v_session_overview',
    'dq_summary'
  );")"

dashboard_count="$(pg_query "
SELECT count(*)
FROM dashboards
WHERE slug = 'ecommerce-analytics';")"

chart_count="$(pg_query "
SELECT count(*)
FROM dashboard_slices ds
JOIN dashboards d ON d.id = ds.dashboard_id
WHERE d.slug = 'ecommerce-analytics';")"

[[ "${dataset_count}" == "6" ]] || fail "ожидалось 6 Superset datasets, найдено ${dataset_count}"
[[ "${dashboard_count}" == "1" ]] || fail "dashboard ecommerce-analytics не найден"
(( chart_count > 0 )) || fail "dashboard ecommerce-analytics не связан с chart"

echo "superset_datasets=${dataset_count}"
echo "superset_dashboards=${dashboard_count}"
echo "superset_dashboard_charts=${chart_count}"
echo "dashboard_url=http://localhost:8088/superset/dashboard/ecommerce-analytics/"

echo ""
echo "=== Проверка Superset: dashboard открывается и читает ClickHouse ==="

superset_probe="$(
  ${COMPOSE_BIN} exec -T superset bash -s <<'PY'
set -euo pipefail
python - <<'PYTHON'
import json
import urllib.request

from sqlalchemy import text

from superset.app import create_app

app = create_app()
with app.app_context():
    from superset.extensions import db
    from superset.models.core import Database
    from superset.models.dashboard import Dashboard

    dashboard = (
        db.session.query(Dashboard)
        .filter_by(slug="ecommerce-analytics")
        .one()
    )

    database = (
        db.session.query(Database)
        .filter_by(database_name="clickhouse_dwh")
        .one()
    )
    with database.get_sqla_engine() as engine:
      with engine.connect() as connection:
        events = connection.execute(
            text("SELECT count() FROM dm.v_events_enriched")
        ).scalar()

login_payload = json.dumps(
    {
        "username": "admin",
        "password": "admin",
        "provider": "db",
        "refresh": True,
    }
).encode("utf-8")

login_request = urllib.request.Request(
    "http://localhost:8088/api/v1/security/login",
    data=login_payload,
    headers={"Content-Type": "application/json"},
    method="POST",
)
with urllib.request.urlopen(login_request, timeout=10) as response:
    login_status = response.status
    token = json.loads(response.read().decode("utf-8"))["access_token"]

dashboard_request = urllib.request.Request(
    f"http://localhost:8088/api/v1/dashboard/{dashboard.id}",
    headers={"Authorization": f"Bearer {token}"},
    method="GET",
)
with urllib.request.urlopen(dashboard_request, timeout=10) as response:
    dashboard_status = response.status
    dashboard_payload = json.loads(response.read().decode("utf-8"))

dashboard_title = dashboard_payload["result"]["dashboard_title"]

print(f"login_api_status={login_status}")
print(f"dashboard_api_status={dashboard_status}")
print(f"dashboard_title={dashboard_title}")
print(f"superset_clickhouse_events={events}")

if login_status >= 400:
    raise SystemExit("Superset login API failed")
if dashboard_status >= 400:
    raise SystemExit("Superset dashboard API did not open")
if "E-commerce Analytics" not in dashboard_title:
    raise SystemExit("Unexpected dashboard title")
if not isinstance(events, int) or events <= 0:
    raise SystemExit("Superset ClickHouse query returned no data")
PYTHON
PY
)"

echo "${superset_probe}"