#!/usr/bin/env bash
#
# Проверяет завершённые стыки между порциями модельной истории.

set -euo pipefail

COMPOSE_BIN="${COMPOSE_BIN:-docker compose}"
CLICKHOUSE_SERVICE="${CLICKHOUSE_SERVICE:-clickhouse}"
CLICKHOUSE_USER="${CLICKHOUSE_USER:-default}"
CLICKHOUSE_PASSWORD="${CLICKHOUSE_PASSWORD:-123456}"

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

clickhouse_datetime_literal() {
  local value="$1"
  local normalized
  if ! normalized="$(date --utc --date="${value}" '+%Y-%m-%d %H:%M:%S.%6N')"; then
    fail "не удалось привести границу к UTC: ${value}"
  fi
  echo "${normalized}"
}

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

mapfile -t boundaries < <(
  ${COMPOSE_BIN} run --rm --no-deps generator \
    python -m clickstream_generator.startup_history_artifact kafka-boundaries
)

boundary_count="${#boundaries[@]}"
(( boundary_count >= 2 )) || fail "manifest boundaries должен содержать T0 и T_end"

manifest_topic_rows="$(${COMPOSE_BIN} run --rm --no-deps generator \
  python -m clickstream_generator.startup_history_artifact kafka-topic-rows)"
IFS=$'\t' read -r expected_browser_rows expected_location_rows \
  expected_device_rows expected_geo_rows <<< "${manifest_topic_rows}"
for rows in \
  "${expected_browser_rows}" \
  "${expected_location_rows}" \
  "${expected_device_rows}" \
  "${expected_geo_rows}"; do
  [[ "${rows}" =~ ^[0-9]+$ ]] || fail "manifest содержит неверные счётчики топиков"
done

echo "=== Проверка порций истории ==="
for ((index = 0; index < boundary_count - 1; index++)); do
  left="$(clickhouse_datetime_literal "${boundaries[index]}")"
  right="$(clickhouse_datetime_literal "${boundaries[index + 1]}")"
  rows="$(ch_query "
WITH
    toDateTime64('${left}', 6, 'UTC') AS segment_start,
    toDateTime64('${right}', 6, 'UTC') AS segment_end
SELECT count()
FROM dm.v_events_enriched
WHERE parseDateTime64BestEffortOrNull(toString(event_ts), 6, 'UTC') >= segment_start
  AND parseDateTime64BestEffortOrNull(toString(event_ts), 6, 'UTC') < segment_end
FORMAT TabSeparated")"
  [[ "${rows}" =~ ^[0-9]+$ ]] || fail "не удалось прочитать порцию ${boundaries[index]}"
  (( rows > 0 )) || fail "порция [${boundaries[index]}, ${boundaries[index + 1]}) пуста"
  echo "segment_${index}_rows=${rows}"
done

actual_topic_rows="$(ch_query "
SELECT
    (SELECT count() FROM stg.browser_raw) AS actual_topic_rows,
    (SELECT count() FROM stg.location_raw) AS actual_location_rows,
    (SELECT count() FROM stg.device_raw) AS actual_device_rows,
    (SELECT count() FROM stg.geo_raw) AS actual_geo_rows
FORMAT TabSeparated")"
IFS=$'\t' read -r actual_browser_rows actual_location_rows \
  actual_device_rows actual_geo_rows <<< "${actual_topic_rows}"
topic_names=(browser_events location_events device_events geo_events)
expected_rows=(
  "${expected_browser_rows}"
  "${expected_location_rows}"
  "${expected_device_rows}"
  "${expected_geo_rows}"
)
actual_rows=(
  "${actual_browser_rows}"
  "${actual_location_rows}"
  "${actual_device_rows}"
  "${actual_geo_rows}"
)
for ((index = 0; index < ${#topic_names[@]}; index++)); do
  [[ "${actual_rows[index]}" =~ ^[0-9]+$ ]] \
    || fail "не удалось прочитать число строк ${topic_names[index]} в STG"
  if (( actual_rows[index] > expected_rows[index] )); then
    fail "найден хвост ${topic_names[index]} после manifest: STG=${actual_rows[index]}, manifest=${expected_rows[index]}"
  fi
  if (( actual_rows[index] < expected_rows[index] )); then
    fail "в STG не хватает строк ${topic_names[index]}: STG=${actual_rows[index]}, manifest=${expected_rows[index]}"
  fi
done

last_boundary="$(clickhouse_datetime_literal "${boundaries[boundary_count - 1]}")"
data_tail_rows="$(ch_query "
SELECT count() AS data_tail_rows
FROM stg.browser_raw
WHERE parseDateTime64BestEffortOrNull(
          JSONExtractString(raw, 'event_timestamp'), 6, 'UTC'
      ) >= toDateTime64('${last_boundary}', 6, 'UTC')
FORMAT TabSeparated")"
[[ "${data_tail_rows}" =~ ^[0-9]+$ ]] \
  || fail "не удалось проверить хвост после ${boundaries[boundary_count - 1]}"
(( data_tail_rows == 0 )) \
  || fail "найден хвост данных после последней границы manifest: ${data_tail_rows}"

if (( boundary_count == 2 )); then
  echo "Завершённых внутренних стыков пока нет."
  exit 0
fi

echo "=== Проверка внутренних стыков boundaries ==="
for ((index = 1; index < boundary_count - 1; index++)); do
  boundary="$(clickhouse_datetime_literal "${boundaries[index]}")"

  pairing="$(ch_query "
WITH
    toDateTime64('${boundary}', 6, 'UTC') AS boundary,
    browser_rows AS (
        SELECT
            toUUIDOrNull(JSONExtractString(raw, 'event_id')) AS event_id,
            toUUIDOrNull(JSONExtractString(raw, 'click_id')) AS click_id,
            parseDateTime64BestEffortOrNull(
                JSONExtractString(raw, 'event_timestamp'), 6, 'UTC'
            ) AS event_ts
        FROM stg.browser_raw
        WHERE event_id IS NOT NULL AND click_id IS NOT NULL AND event_ts IS NOT NULL
        GROUP BY event_id, click_id, event_ts
    ),
    crossing AS (
        SELECT click_id
        FROM browser_rows
        GROUP BY click_id
        HAVING min(event_ts) < boundary AND max(event_ts) >= boundary
    ),
    browser_counts AS (
        SELECT click_id, count() AS browser_rows
        FROM browser_rows
        WHERE click_id IN (SELECT click_id FROM crossing)
        GROUP BY click_id
    ),
    location_ids AS (
        SELECT toUUIDOrNull(JSONExtractString(raw, 'event_id')) AS event_id
        FROM stg.location_raw
        WHERE event_id IS NOT NULL
        GROUP BY event_id
    ),
    device_counts AS (
        SELECT
            toUUIDOrNull(JSONExtractString(raw, 'click_id')) AS click_id,
            count() AS device_rows
        FROM stg.device_raw
        WHERE click_id IN (SELECT click_id FROM crossing)
        GROUP BY click_id
    ),
    geo_counts AS (
        SELECT
            toUUIDOrNull(JSONExtractString(raw, 'click_id')) AS click_id,
            count() AS geo_rows
        FROM stg.geo_raw
        WHERE click_id IN (SELECT click_id FROM crossing)
        GROUP BY click_id
    )
SELECT
    (SELECT count() FROM crossing) AS crossing_visits,
    (
        SELECT count()
        FROM browser_rows AS b
        LEFT JOIN location_ids AS l ON l.event_id = b.event_id
        WHERE b.click_id IN (SELECT click_id FROM crossing)
          AND l.event_id IS NULL
    ) AS unpaired_location_rows,
    (
        SELECT ifNull(sum(greatest(b.browser_rows - ifNull(d.device_rows, 0), 0)), 0)
        FROM browser_counts AS b
        LEFT JOIN device_counts AS d ON d.click_id = b.click_id
    ) AS unpaired_device_rows,
    (
        SELECT ifNull(sum(greatest(b.browser_rows - ifNull(g.geo_rows, 0), 0)), 0)
        FROM browser_counts AS b
        LEFT JOIN geo_counts AS g ON g.click_id = b.click_id
    ) AS unpaired_geo_rows
SETTINGS join_use_nulls = 1
FORMAT TabSeparated")"

  IFS=$'\t' read -r crossing_visits unpaired_location_rows unpaired_device_rows unpaired_geo_rows <<< "${pairing}"
  [[ "${crossing_visits}" =~ ^[0-9]+$ ]] \
    && [[ "${unpaired_location_rows}" =~ ^[0-9]+$ ]] \
    && [[ "${unpaired_device_rows}" =~ ^[0-9]+$ ]] \
    && [[ "${unpaired_geo_rows}" =~ ^[0-9]+$ ]] \
    || fail "не удалось прочитать пары на границе ${boundaries[index]}"
  if (( unpaired_location_rows > 0 || unpaired_device_rows > 0 || unpaired_geo_rows > 0 )); then
    fail "непарные строки на границе ${boundaries[index]}: location=${unpaired_location_rows}, device=${unpaired_device_rows}, geo=${unpaired_geo_rows}"
  fi

  context="$(ch_query "
WITH
    toDateTime64('${boundary}', 6, 'UTC') AS boundary,
    crossing AS (
        SELECT click_id
        FROM
        (
            SELECT
                click_id,
                parseDateTime64BestEffortOrNull(
                    toString(event_ts), 6, 'UTC'
                ) AS model_event_ts
            FROM dds.event
            WHERE event_ts IS NOT NULL
        )
        GROUP BY click_id
        HAVING min(model_event_ts) < boundary AND max(model_event_ts) >= boundary
    ),
    facts AS (
        SELECT
            e.click_id,
            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
        FROM dds.event AS e
        INNER JOIN crossing AS c ON c.click_id = e.click_id
        GROUP BY e.click_id
    )
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
FROM facts
FORMAT TabSeparated")"

  IFS=$'\t' read -r context_crossing_visits per_event_homogeneous_visits <<< "${context}"
  [[ "${context_crossing_visits}" =~ ^[0-9]+$ ]] \
    && [[ "${per_event_homogeneous_visits}" =~ ^[0-9]+$ ]] \
    || fail "не удалось прочитать фактуру на границе ${boundaries[index]}"
  (( context_crossing_visits == crossing_visits )) \
    || fail "STG и DDS нашли разное число переходящих визитов на ${boundaries[index]}"
  if (( crossing_visits == 0 )); then
    echo "boundary ${index}/${boundary_count}: crossing_visits=0 — однородность неприменима"
  else
    (( per_event_homogeneous_visits == crossing_visits )) \
      || fail "browser/referer/utm меняются на границе ${boundaries[index]}: ${per_event_homogeneous_visits}/${crossing_visits}"
    echo "boundary ${index}/${boundary_count}: crossing_visits=${crossing_visits} — однородность подтверждена"
  fi

  echo "boundary_${index}=${boundaries[index]}"
  echo "boundary_${index}_crossing_visits=${crossing_visits}"
  echo "boundary_${index}_unpaired_location_rows=${unpaired_location_rows}"
  echo "boundary_${index}_unpaired_device_rows=${unpaired_device_rows}"
  echo "boundary_${index}_unpaired_geo_rows=${unpaired_geo_rows}"
done