#!/usr/bin/env bash

set -euo pipefail

SCRIPT_DIR="$(cd "$(dirname "${BASH_SOURCE[0]}")" && pwd)"
ROOT_DIR="$(cd "${SCRIPT_DIR}/../.." && pwd)"
DEFAULT_MIGRATIONS_DIR="${ROOT_DIR}/security-center-backend/migrations"
DEFAULT_DATABASE="${SC_MYSQL_DATABASE:-security_center}"

usage() {
  cat <<'EOF'
Read-only production schema compatibility preflight.

Checks whether an existing MySQL database is compatible with recent CAASM and
vulnerability changes before deploying the current application code.

Usage:
  ./scripts/enterprise/preflight-schema-compatibility.sh \
    --host 127.0.0.1 \
    --port 3306 \
    --user root \
    --password '...' \
    --database security_center \
    --migrations-dir ./security-center-backend/migrations

Environment variables are also supported:
  SC_SCHEMA_CHECK_MYSQL_HOST
  SC_SCHEMA_CHECK_MYSQL_PORT
  SC_SCHEMA_CHECK_MYSQL_USER
  SC_SCHEMA_CHECK_MYSQL_PASSWORD
  SC_SCHEMA_CHECK_MYSQL_DATABASE
  SC_SCHEMA_CHECK_MYSQL_SSL_MODE
  SC_SCHEMA_CHECK_MIGRATIONS_DIR

Fallback environment variables:
  MYSQL_HOST, MYSQL_PORT, MYSQL_USER, MYSQL_PASSWORD, MYSQL_PWD, MYSQL_DATABASE,
  MYSQL_SSL_MODE, SC_MYSQL_DATABASE

The script does not modify data or schema. It exits non-zero when required
tables, columns, indexes, or data invariants are missing.
EOF
}

die() {
  echo "ERROR: $*" >&2
  exit 1
}

resolve_path() {
  local path="$1"
  if [[ "${path}" = /* ]]; then
    printf '%s' "${path}"
    return 0
  fi
  if [[ -e "${path}" ]]; then
    printf '%s' "${path}"
    return 0
  fi
  if [[ -e "${ROOT_DIR}/${path}" ]]; then
    printf '%s' "${ROOT_DIR}/${path}"
    return 0
  fi
  printf '%s' "${path}"
}

sha256_file() {
  local path="$1"
  if command -v sha256sum >/dev/null 2>&1; then
    sha256sum "${path}" | awk '{print $1}'
    return 0
  fi
  if command -v shasum >/dev/null 2>&1; then
    shasum -a 256 "${path}" | awk '{print $1}'
    return 0
  fi
  die "sha256sum or shasum is required"
}

trim() {
  local value="${1:-}"
  value="${value#"${value%%[![:space:]]*}"}"
  value="${value%"${value##*[![:space:]]}"}"
  printf '%s' "${value}"
}

MYSQL_HOST="${SC_SCHEMA_CHECK_MYSQL_HOST:-${MYSQL_HOST:-127.0.0.1}}"
MYSQL_PORT="${SC_SCHEMA_CHECK_MYSQL_PORT:-${MYSQL_PORT:-3306}}"
MYSQL_USER="${SC_SCHEMA_CHECK_MYSQL_USER:-${MYSQL_USER:-root}}"
MYSQL_PASSWORD="${SC_SCHEMA_CHECK_MYSQL_PASSWORD:-${MYSQL_PASSWORD:-${MYSQL_PWD:-}}}"
MYSQL_DATABASE="${SC_SCHEMA_CHECK_MYSQL_DATABASE:-${MYSQL_DATABASE:-${DEFAULT_DATABASE}}}"
MYSQL_SSL_MODE="${SC_SCHEMA_CHECK_MYSQL_SSL_MODE:-${MYSQL_SSL_MODE:-}}"
MIGRATIONS_DIR="${SC_SCHEMA_CHECK_MIGRATIONS_DIR:-${DEFAULT_MIGRATIONS_DIR}}"

while [[ $# -gt 0 ]]; do
  case "$1" in
    --host)
      MYSQL_HOST="${2:-}"
      shift 2
      ;;
    --port)
      MYSQL_PORT="${2:-}"
      shift 2
      ;;
    --user)
      MYSQL_USER="${2:-}"
      shift 2
      ;;
    --password)
      MYSQL_PASSWORD="${2:-}"
      shift 2
      ;;
    --database)
      MYSQL_DATABASE="${2:-}"
      shift 2
      ;;
    --ssl-mode)
      MYSQL_SSL_MODE="${2:-}"
      shift 2
      ;;
    --migrations-dir)
      MIGRATIONS_DIR="${2:-}"
      shift 2
      ;;
    --help|-h)
      usage
      exit 0
      ;;
    *)
      die "unknown argument: $1"
      ;;
  esac
done

[[ -n "${MYSQL_HOST}" ]] || die "--host is required"
[[ -n "${MYSQL_PORT}" ]] || die "--port is required"
[[ -n "${MYSQL_USER}" ]] || die "--user is required"
[[ -n "${MYSQL_DATABASE}" ]] || die "--database is required"
[[ "${MYSQL_DATABASE}" =~ ^[A-Za-z0-9_]+$ ]] || die "database name may only contain letters, numbers, and underscores"
MIGRATIONS_DIR="$(resolve_path "${MIGRATIONS_DIR}")"
[[ -d "${MIGRATIONS_DIR}" ]] || die "migrations directory not found: ${MIGRATIONS_DIR}"
command -v mysql >/dev/null 2>&1 || die "mysql client is required"

mysql_args=(
  "--batch"
  "--raw"
  "--skip-column-names"
  "--silent"
  "--connect-timeout=10"
  "--host=${MYSQL_HOST}"
  "--port=${MYSQL_PORT}"
  "--user=${MYSQL_USER}"
)

if [[ -n "${MYSQL_SSL_MODE}" ]]; then
  mysql_args+=("--ssl-mode=${MYSQL_SSL_MODE}")
fi

mysql_base() {
  local sql="$1"
  MYSQL_PWD="${MYSQL_PASSWORD}" mysql "${mysql_args[@]}" -e "${sql}"
}

mysql_db() {
  local sql="$1"
  MYSQL_PWD="${MYSQL_PASSWORD}" mysql "${mysql_args[@]}" "--database=${MYSQL_DATABASE}" -e "${sql}"
}

scalar_base() {
  local sql="$1"
  trim "$(mysql_base "${sql}" | head -n 1)"
}

scalar_db() {
  local sql="$1"
  trim "$(mysql_db "${sql}" | head -n 1)"
}

checks=0
failures=0
warnings=0

pass_check() {
  checks=$((checks + 1))
  echo "[PASS] $*"
}

fail_check() {
  checks=$((checks + 1))
  failures=$((failures + 1))
  echo "[FAIL] $*" >&2
}

warn_check() {
  checks=$((checks + 1))
  warnings=$((warnings + 1))
  echo "[WARN] $*" >&2
}

info() {
  echo "[INFO] $*"
}

has_table() {
  local table="$1"
  [[ "$(scalar_base "SELECT COUNT(*) FROM information_schema.TABLES WHERE TABLE_SCHEMA = '${MYSQL_DATABASE}' AND TABLE_NAME = '${table}';")" == "1" ]]
}

has_column() {
  local table="$1"
  local column="$2"
  [[ "$(scalar_base "SELECT COUNT(*) FROM information_schema.COLUMNS WHERE TABLE_SCHEMA = '${MYSQL_DATABASE}' AND TABLE_NAME = '${table}' AND COLUMN_NAME = '${column}';")" == "1" ]]
}

has_index() {
  local table="$1"
  local index="$2"
  [[ "$(scalar_base "SELECT COUNT(*) FROM information_schema.STATISTICS WHERE TABLE_SCHEMA = '${MYSQL_DATABASE}' AND TABLE_NAME = '${table}' AND INDEX_NAME = '${index}';")" != "0" ]]
}

check_table() {
  local table="$1"
  if has_table "${table}"; then
    pass_check "table exists: ${table}"
  else
    fail_check "missing table: ${table}"
  fi
}

check_column() {
  local table="$1"
  local column="$2"
  if has_column "${table}" "${column}"; then
    pass_check "column exists: ${table}.${column}"
  else
    fail_check "missing column: ${table}.${column}"
  fi
}

check_index() {
  local table="$1"
  local index="$2"
  if has_index "${table}" "${index}"; then
    pass_check "index exists: ${table}.${index}"
  else
    fail_check "missing index: ${table}.${index}"
  fi
}

check_zero_count() {
  local label="$1"
  local sql="$2"
  local count
  count="$(scalar_db "${sql}")"
  if [[ "${count}" == "0" ]]; then
    pass_check "${label}: 0"
  else
    fail_check "${label}: ${count}"
  fi
}

echo "== Schema compatibility preflight =="
info "target=${MYSQL_USER}@${MYSQL_HOST}:${MYSQL_PORT}/${MYSQL_DATABASE}"
echo

if [[ "$(scalar_base "SELECT COUNT(*) FROM information_schema.SCHEMATA WHERE SCHEMA_NAME = '${MYSQL_DATABASE}';")" != "1" ]]; then
  fail_check "database does not exist: ${MYSQL_DATABASE}"
  echo
  echo "Preflight failed: database is missing."
  exit 1
fi
pass_check "database exists: ${MYSQL_DATABASE}"

if [[ "$(scalar_base "SELECT 1;")" != "1" ]]; then
  die "mysql connectivity check failed"
fi
pass_check "mysql connectivity"

echo
echo "== Required tables =="
required_tables=(
  "schema_migrations"
  "cloud_accounts"
  "assets"
  "asset_relationships"
  "ip_assets"
  "domain_assets"
  "ip_port_index"
  "dns_records"
  "sync_jobs"
  "sync_job_events"
  "caasm_api_route_configs"
  "caasm_api_route_assets"
  "vulnerabilities"
  "vulnerability_public_id_sequences"
)

for table in "${required_tables[@]}"; do
  check_table "${table}"
done

echo
echo "== Required columns =="
required_columns=(
  "schema_migrations:filename"
  "schema_migrations:checksum"
  "schema_migrations:applied_at"
  "cloud_accounts:deleted_at"
  "cloud_accounts:provider"
  "cloud_accounts:external_id"
  "assets:deleted_at"
  "assets:cloud_account_id"
  "assets:native_id"
  "assets:name"
  "assets:service"
  "assets:region"
  "assets:provider"
  "asset_relationships:deleted_at"
  "ip_assets:deleted_at"
  "ip_assets:asset_id"
  "ip_assets:ip_address"
  "ip_assets:hostname"
  "ip_assets:is_public"
  "ip_assets:geo_location"
  "domain_assets:deleted_at"
  "domain_assets:asset_id"
  "domain_assets:domain_name"
  "domain_assets:domain_class"
  "domain_assets:zone_id"
  "domain_assets:zone_type"
  "domain_assets:status"
  "ip_port_index:deleted_at"
  "ip_port_index:asset_id"
  "ip_port_index:ip_address"
  "ip_port_index:port"
  "ip_port_index:protocol"
  "ip_port_index:observation_source"
  "ip_port_index:exposure"
  "dns_records:deleted_at"
  "dns_records:domain_asset_id"
  "dns_records:name"
  "dns_records:type"
  "dns_records:name_key"
  "dns_records:type_key"
  "dns_records:set_identifier_key"
  "dns_records:region_key"
  "dns_records:failover_key"
  "sync_jobs:deleted_at"
  "sync_jobs:cloud_account_id"
  "sync_jobs:status"
  "sync_job_events:deleted_at"
  "caasm_api_route_configs:deleted_at"
  "caasm_api_route_configs:is_active"
  "caasm_api_route_assets:deleted_at"
  "caasm_api_route_assets:config_id"
  "caasm_api_route_assets:route_key"
  "vulnerabilities:deleted_at"
  "vulnerabilities:public_id"
  "vulnerabilities:authorized_viewer_employee_ids"
)

for item in "${required_columns[@]}"; do
  IFS=":" read -r table column <<<"${item}"
  check_column "${table}" "${column}"
done

echo
echo "== Required indexes =="
required_indexes=(
  "schema_migrations:PRIMARY"
  "cloud_accounts:idx_cloud_accounts_deleted_at"
  "assets:idx_assets_deleted_at"
  "assets:idx_assets_cloud_account_native"
  "asset_relationships:idx_asset_relationships_deleted_at"
  "ip_assets:idx_ip_assets_asset_ip"
  "ip_assets:idx_ip_assets_ip"
  "domain_assets:idx_domain_assets_asset_domain"
  "domain_assets:idx_domain_assets_domain_name"
  "ip_port_index:idx_ip_port_asset_endpoint"
  "ip_port_index:idx_exposure_port"
  "dns_records:idx_dns_records_identity"
  "dns_records:idx_dns_records_domain_updated_at"
  "sync_jobs:idx_sync_jobs_cloud_account_status_created"
  "sync_jobs:idx_sync_jobs_status_started_created"
  "sync_job_events:idx_sync_job_events_deleted_at"
  "caasm_api_route_configs:idx_api_route_configs_active_deleted"
  "caasm_api_route_assets:uk_api_route_config_route_key"
  "caasm_api_route_assets:idx_api_route_assets_config_deleted"
  "vulnerabilities:idx_vulnerabilities_public_id"
  "vulnerabilities:idx_vulnerabilities_deleted_at"
)

for item in "${required_indexes[@]}"; do
  IFS=":" read -r table index <<<"${item}"
  check_index "${table}" "${index}"
done

echo
echo "== Required migration ledger rows =="
critical_migrations=(
  "202604100003_add_vulnerability_authorized_viewers.sql"
  "202604100004_add_vulnerability_public_id.sql"
  "202604100005_add_caasm_query_indexes.sql"
  "202604100008_add_remaining_soft_delete_columns.sql"
  "202604110001_add_asset_identity_unique_key.sql"
  "202604110002_add_dns_record_identity_key_columns.sql"
  "202605120001_add_caasm_api_route_assets.sql"
)

if has_table "schema_migrations" && has_column "schema_migrations" "filename"; then
  for migration in "${critical_migrations[@]}"; do
    if [[ "$(scalar_db "SELECT COUNT(*) FROM schema_migrations WHERE filename = '${migration}';")" == "1" ]]; then
      pass_check "migration ledger row exists: ${migration}"
      if has_column "schema_migrations" "checksum"; then
        migration_path="${MIGRATIONS_DIR}/${migration}"
        if [[ ! -f "${migration_path}" ]]; then
          fail_check "migration file missing locally: ${migration_path}"
          continue
        fi
        ledger_checksum="$(scalar_db "SELECT checksum FROM schema_migrations WHERE filename = '${migration}' LIMIT 1;")"
        file_checksum="$(sha256_file "${migration_path}")"
        if [[ "${ledger_checksum}" == "${file_checksum}" ]]; then
          pass_check "migration checksum matches current file: ${migration}"
        else
          fail_check "migration checksum mismatch: ${migration}"
        fi
      else
        fail_check "cannot verify migration checksum because schema_migrations.checksum is missing: ${migration}"
      fi
    else
      fail_check "missing migration ledger row: ${migration}"
    fi
  done
else
  for migration in "${critical_migrations[@]}"; do
    fail_check "cannot verify migration ledger row because schema_migrations is missing: ${migration}"
  done
fi

echo
echo "== Data compatibility checks =="
if has_table "assets" && has_column "assets" "cloud_account_id" && has_column "assets" "native_id"; then
  check_zero_count "duplicate CAASM asset identities" "SELECT COUNT(*) FROM (SELECT cloud_account_id, native_id FROM assets WHERE native_id IS NOT NULL AND TRIM(native_id) <> '' GROUP BY cloud_account_id, native_id HAVING COUNT(*) > 1) duplicate_asset_identities;"
else
  warn_check "skipped duplicate CAASM asset identity check because required assets columns are missing"
fi

if has_table "ip_assets" && has_column "ip_assets" "asset_id" && has_column "ip_assets" "ip_address"; then
  check_zero_count "duplicate CAASM IP asset identities" "SELECT COUNT(*) FROM (SELECT asset_id, ip_address FROM ip_assets WHERE ip_address IS NOT NULL AND TRIM(ip_address) <> '' GROUP BY asset_id, ip_address HAVING COUNT(*) > 1) duplicate_ip_asset_identities;"
else
  warn_check "skipped duplicate CAASM IP identity check because required ip_assets columns are missing"
fi

if has_table "domain_assets" && has_column "domain_assets" "asset_id" && has_column "domain_assets" "domain_name" && has_column "domain_assets" "domain_class"; then
  check_zero_count "duplicate CAASM domain asset identities" "SELECT COUNT(*) FROM (SELECT asset_id, domain_name, domain_class FROM domain_assets WHERE domain_name IS NOT NULL AND TRIM(domain_name) <> '' GROUP BY asset_id, domain_name, domain_class HAVING COUNT(*) > 1) duplicate_domain_asset_identities;"
else
  warn_check "skipped duplicate CAASM domain identity check because required domain_assets columns are missing"
fi

if has_table "ip_port_index" && has_column "ip_port_index" "asset_id" && has_column "ip_port_index" "ip_address" && has_column "ip_port_index" "port" && has_column "ip_port_index" "protocol" && has_column "ip_port_index" "observation_source"; then
  check_zero_count "duplicate CAASM port endpoint identities" "SELECT COUNT(*) FROM (SELECT asset_id, ip_address, port, protocol, observation_source FROM ip_port_index WHERE ip_address IS NOT NULL AND TRIM(ip_address) <> '' GROUP BY asset_id, ip_address, port, protocol, observation_source HAVING COUNT(*) > 1) duplicate_port_identities;"
else
  warn_check "skipped duplicate CAASM port endpoint check because required ip_port_index columns are missing"
fi

if has_table "dns_records" && has_column "dns_records" "domain_asset_id" && has_column "dns_records" "name_key" && has_column "dns_records" "type_key" && has_column "dns_records" "set_identifier_key" && has_column "dns_records" "region_key" && has_column "dns_records" "failover_key"; then
  check_zero_count "duplicate CAASM DNS identities" "SELECT COUNT(*) FROM (SELECT domain_asset_id, name_key, type_key, set_identifier_key, region_key, failover_key FROM dns_records GROUP BY domain_asset_id, name_key, type_key, set_identifier_key, region_key, failover_key HAVING COUNT(*) > 1) duplicate_dns_identities;"
else
  warn_check "skipped duplicate CAASM DNS identity check because required dns_records columns are missing"
fi

if has_table "caasm_api_route_assets" && has_column "caasm_api_route_assets" "config_id" && has_column "caasm_api_route_assets" "route_key"; then
  check_zero_count "duplicate CAASM API route identities" "SELECT COUNT(*) FROM (SELECT config_id, route_key FROM caasm_api_route_assets GROUP BY config_id, route_key HAVING COUNT(*) > 1) duplicate_api_route_identities;"
else
  warn_check "skipped duplicate CAASM API route identity check because required caasm_api_route_assets columns are missing"
fi

if has_table "vulnerabilities" && has_column "vulnerabilities" "public_id"; then
  check_zero_count "vulnerabilities with empty public_id" "SELECT COUNT(*) FROM vulnerabilities WHERE public_id IS NULL OR TRIM(public_id) = '';"
  check_zero_count "duplicate vulnerability public_id values" "SELECT COUNT(*) FROM (SELECT public_id FROM vulnerabilities WHERE public_id IS NOT NULL AND TRIM(public_id) <> '' GROUP BY public_id HAVING COUNT(*) > 1) duplicate_public_ids;"
else
  warn_check "skipped vulnerability public_id data check because required vulnerabilities columns are missing"
fi

echo
echo "== Informational row counts =="
for table in "${required_tables[@]}"; do
  if has_table "${table}"; then
    info "${table}: $(scalar_db "SELECT COUNT(*) FROM ${table};") rows"
  fi
done

if has_table "domain_assets" && has_column "domain_assets" "zone_type" && has_column "domain_assets" "deleted_at"; then
  info "active non-public domain_assets: $(scalar_db "SELECT COUNT(*) FROM domain_assets WHERE deleted_at IS NULL AND LOWER(COALESCE(zone_type, '')) NOT IN ('public', 'public-and-private', '');") rows"
fi

echo
echo "== Result =="
info "checks=${checks}, failures=${failures}, warnings=${warnings}"

if [[ "${failures}" -gt 0 ]]; then
  echo "Preflight failed. Do not deploy until the missing schema or incompatible data is resolved." >&2
  exit 1
fi

echo "Preflight passed. The checked schema and data invariants are compatible with the current code."