#!/usr/bin/env bash

set -euo pipefail

SCRIPT_DIR="$(cd -- "$(dirname -- "${BASH_SOURCE[0]}")" && pwd)"
# shellcheck source=tools/migration/import_into/lib.sh
source "${SCRIPT_DIR}/lib.sh"

usage() {
  cat <<'USAGE'
Usage:
  OLD_DB_HOST="old-host" OLD_DB_PORT="3306" OLD_DB_USER="user" OLD_DB_PASSWORD="pass" OLD_DB_NAME="old_db" \
  NEW_DB_HOST="new-host" NEW_DB_PORT="3306" NEW_DB_USER="user" NEW_DB_PASSWORD="pass" NEW_DB_NAME="new_db" \
  CLUSTER_IDS="1001,1002" \
  SNAPSHOT_DIR="./snapshot" \
  PROBE_IMPORT_ID="12345" \
  ./tools/migration/import_into/verify_parity.sh

Required environment variables:
  OLD_DB_HOST, OLD_DB_PORT, OLD_DB_USER, OLD_DB_PASSWORD, OLD_DB_NAME
  NEW_DB_HOST, NEW_DB_PORT, NEW_DB_USER, NEW_DB_PASSWORD, NEW_DB_NAME
                    MySQL connection fields for old/new metadb
  SNAPSHOT_DIR     Directory to store parity output artifacts

Optional:
  CLUSTER_IDS      Comma-separated cluster IDs. If unset, verify whole tables
  PROBE_IMPORT_ID  Probe import task ID for targeted parity check
USAGE
}

if [[ "${1:-}" == "-h" || "${1:-}" == "--help" ]]; then
  usage
  exit 0
fi

required_envs=(
  OLD_DB_HOST OLD_DB_PORT OLD_DB_USER OLD_DB_PASSWORD OLD_DB_NAME
  NEW_DB_HOST NEW_DB_PORT NEW_DB_USER NEW_DB_PASSWORD NEW_DB_NAME
  SNAPSHOT_DIR
)

for name in "${required_envs[@]}"; do
  if [[ -z "${!name:-}" ]]; then
    usage
    die "missing required environment variable: ${name}"
  fi
done

cluster_filter="$(build_cluster_filter "${CLUSTER_IDS:-}")"
scope_desc="all rows"
if [[ -n "${CLUSTER_IDS:-}" ]]; then
  scope_desc="cluster ids: ${CLUSTER_IDS//[[:space:]]/}"
fi

probe_import_id=""
if [[ -n "${PROBE_IMPORT_ID:-}" ]]; then
  probe_import_id="$(sanitize_import_id "${PROBE_IMPORT_ID}")"
fi

timestamp="$(date -u +%Y%m%dT%H%M%SZ)"
run_dir="$(prepare_output_dir "${SNAPSHOT_DIR}" "verify_${timestamp}")"
echo "scope: ${scope_desc}"

compare_template_query() {
  local name="$1"
  local template_file="$2"

  local rendered_sql="${run_dir}/${name}.sql"
  local old_file="${run_dir}/${name}.old.txt"
  local new_file="${run_dir}/${name}.new.txt"
  local diff_file="${run_dir}/${name}.diff.txt"

  render_sql_template "${template_file}" "${rendered_sql}" "${cluster_filter}" "${probe_import_id:-0}"

  run_mysql_old --batch --raw --skip-column-names < "${rendered_sql}" > "${old_file}"
  run_mysql_new --batch --raw --skip-column-names < "${rendered_sql}" > "${new_file}"

  if ! diff -u "${old_file}" "${new_file}" > "${diff_file}"; then
    echo "parity check failed: ${name}" >&2
    echo "diff saved to: ${diff_file}" >&2
    exit 1
  fi
  rm -f "${diff_file}"
  echo "parity check passed: ${name}"
}

verify_templates=(
  "import_into_tasks_count"
  "import_into_precheck_tasks_count"
  "import_into_tasks_status_distribution"
  "import_into_tasks_inflight"
)

for check_name in "${verify_templates[@]}"; do
  compare_template_query \
    "${check_name}" \
    "${SCRIPT_DIR}/sql/verify/${check_name}.sql.tmpl"
done

if [[ -n "${probe_import_id}" ]]; then
  compare_template_query \
    "probe_import_task" \
    "${SCRIPT_DIR}/sql/verify/probe_import_task.sql.tmpl"

  probe_old_file="${run_dir}/probe_import_task.old.txt"
  probe_new_file="${run_dir}/probe_import_task.new.txt"

  if [[ ! -s "${probe_old_file}" ]]; then
    die "probe parity check failed: probe task not found in old metadb"
  fi
  if [[ ! -s "${probe_new_file}" ]]; then
    die "probe parity check failed: probe task not found in new metadb"
  fi
fi

echo "All parity checks passed."
echo "  run dir: ${run_dir}"