#!/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_LEDGER_TABLE="schema_migrations"
DEFAULT_DATABASE="${SC_MYSQL_DATABASE:-security_center}"
DEFAULT_ENV_FILE=""
DEFAULT_BASELINE_INIT_FILE=""

usage() {
  cat <<'EOF'
Enterprise DB migration runner

Applies SQL files from a migrations directory in lexicographic order and records
applied migrations in a schema_migrations ledger table.

Usage:
  ./scripts/enterprise/run-db-migrations.sh \
    --migrations-dir ./security-center-backend/migrations \
    --database security_center \
    --user sc_app \
    --password '...'

Options:
  --migrations-dir <path>   Directory containing *.sql migrations.
  --ledger-table <name>     Ledger table name. Default: schema_migrations
  --env-file <path>         Optional env file to read MYSQL_DATABASE, MYSQL_USER,
                            and MYSQL_PASSWORD from without shell sourcing.
  --baseline-init <path>    Optional mysql-init.sql path used to validate the
                            fresh-install schema_migrations baseline. Default:
                            auto-detects mysql-init.sql beside the script or
                            at the repo/rendered bundle root.
  --mysql-prefix <cmd>      Optional prefix before the mysql client, e.g.
                            "docker compose exec -T mysql". Default: mysql
  --host <host>             mysql connection host.
  --port <port>             mysql connection port.
  --socket <socket>         mysql connection socket.
  --protocol <value>        mysql protocol.
  --user <name>             mysql user.
  --password <secret>       mysql password.
  --database <name>         mysql database. Default: security_center
  --ssl-mode <mode>         mysql ssl mode.
  --defaults-extra-file <p>  mysql defaults file.
  --inventory              Print a read-only production schema inventory and exit.
  --adopt-existing         Adopt migration ledger rows for an existing database
                            only when live schema checks prove the migration is
                            already materialized. Use after backup + human check.
  --dry-run                 Print what would be applied without executing.
  --help                    Show this help.
EOF
}

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

info() {
  echo "INFO: $*"
}

warn() {
  echo "WARN: $*" >&2
}

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

resolve_path() {
  local path="$1"
  if [[ -z "${path}" ]]; then
    printf '%s' ""
    return 0
  fi
  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}"
}

require_command() {
  command -v "$1" >/dev/null 2>&1 || die "required command not found: $1"
}

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"
}

quote_shell() {
  printf '%q' "$1"
}

sql_escape_literal() {
  local value="${1:-}"
  value="${value//\'/\'\'}"
  printf '%s' "$value"
}

env_file_value() {
  local env_file="$1"
  local key="$2"
  [[ -f "${env_file}" ]] || return 0
  python3 - "${env_file}" "${key}" <<'PY'
from pathlib import Path
import sys

path = Path(sys.argv[1])
key = sys.argv[2]
for raw in path.read_text().splitlines():
    line = raw.strip()
    if not line or line.startswith("#") or "=" not in line:
        continue
    lhs, rhs = line.split("=", 1)
    if lhs.strip() == key:
        print(rhs.strip())
        break
PY
}

MYSQL_PREFIX="${SC_MIGRATION_MYSQL_PREFIX:-mysql}"
MYSQL_HOST="${SC_MIGRATION_MYSQL_HOST:-${MYSQL_HOST:-}}"
MYSQL_PORT="${SC_MIGRATION_MYSQL_PORT:-${MYSQL_PORT:-}}"
MYSQL_SOCKET="${SC_MIGRATION_MYSQL_SOCKET:-${MYSQL_SOCKET:-}}"
MYSQL_PROTOCOL="${SC_MIGRATION_MYSQL_PROTOCOL:-${MYSQL_PROTOCOL:-}}"
MYSQL_USER="${SC_MIGRATION_MYSQL_USER:-${MYSQL_USER:-}}"
MYSQL_PASSWORD="${SC_MIGRATION_MYSQL_PASSWORD:-${MYSQL_PASSWORD:-}}"
MYSQL_DATABASE="${SC_MIGRATION_MYSQL_DATABASE:-${MYSQL_DATABASE:-${DEFAULT_DATABASE}}}"
MYSQL_SSL_MODE="${SC_MIGRATION_MYSQL_SSL_MODE:-${MYSQL_SSL_MODE:-}}"
MYSQL_DEFAULTS_EXTRA_FILE="${SC_MIGRATION_MYSQL_DEFAULTS_EXTRA_FILE:-${MYSQL_DEFAULTS_EXTRA_FILE:-}}"
MIGRATIONS_DIR="${SC_MIGRATION_DIR:-${DEFAULT_MIGRATIONS_DIR}}"
LEDGER_TABLE="${SC_MIGRATION_LEDGER_TABLE:-${DEFAULT_LEDGER_TABLE}}"
ENV_FILE="${SC_MIGRATION_ENV_FILE:-${DEFAULT_ENV_FILE}}"
BASELINE_INIT_FILE="${SC_MIGRATION_BASELINE_INIT_FILE:-${DEFAULT_BASELINE_INIT_FILE}}"
MODE="${SC_MIGRATION_MODE:-apply}"
DRY_RUN="false"
mysql_query_args=("--batch" "--raw" "--skip-column-names" "--silent")

build_mysql_command() {
  local shell_cmd="${MYSQL_PREFIX:-mysql}"

  local part=""
  for part in "${mysql_query_args[@]}"; do
    shell_cmd+=" $(quote_shell "${part}")"
  done

  if [[ -n "${MYSQL_DEFAULTS_EXTRA_FILE}" ]]; then
    shell_cmd+=" $(quote_shell "--defaults-extra-file=${MYSQL_DEFAULTS_EXTRA_FILE}")"
  fi
  if [[ -n "${MYSQL_HOST}" ]]; then
    shell_cmd+=" $(quote_shell "--host=${MYSQL_HOST}")"
  fi
  if [[ -n "${MYSQL_PORT}" ]]; then
    shell_cmd+=" $(quote_shell "--port=${MYSQL_PORT}")"
  fi
  if [[ -n "${MYSQL_SOCKET}" ]]; then
    shell_cmd+=" $(quote_shell "--socket=${MYSQL_SOCKET}")"
  fi
  if [[ -n "${MYSQL_PROTOCOL}" ]]; then
    shell_cmd+=" $(quote_shell "--protocol=${MYSQL_PROTOCOL}")"
  fi
  if [[ -n "${MYSQL_USER}" ]]; then
    shell_cmd+=" $(quote_shell "--user=${MYSQL_USER}")"
  fi
  if [[ -n "${MYSQL_PASSWORD}" ]]; then
    shell_cmd+=" $(quote_shell "--password=${MYSQL_PASSWORD}")"
  fi
  if [[ -n "${MYSQL_SSL_MODE}" ]]; then
    shell_cmd+=" $(quote_shell "--ssl-mode=${MYSQL_SSL_MODE}")"
  fi
  if [[ -n "${MYSQL_DATABASE}" ]]; then
    shell_cmd+=" $(quote_shell "--database=${MYSQL_DATABASE}")"
  fi

  printf '%s' "${shell_cmd}"
}

run_mysql_sql() {
  local sql_input="$1"
  local command
  command="$(build_mysql_command)"
  bash -lc "${command}" <<<"${sql_input}"
}

mysql_scalar() {
  local sql_input="$1"
  trim "$(run_mysql_sql "${sql_input}" | head -n 1)"
}

run_mysql_file() {
  local file="$1"
  local command
  command="$(build_mysql_command)"
  bash -lc "${command}" < "${file}"
}

ensure_ledger_table() {
  local create_sql
  create_sql=$(cat <<EOF
CREATE TABLE IF NOT EXISTS \`${LEDGER_TABLE}\` (
  filename VARCHAR(255) NOT NULL PRIMARY KEY,
  checksum CHAR(64) NOT NULL,
  applied_at TIMESTAMP NOT NULL DEFAULT CURRENT_TIMESTAMP
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_unicode_ci;
EOF
)
  run_mysql_sql "${create_sql}" >/dev/null
}

get_ledger_checksum() {
  local filename="$1"
  local escaped_filename
  escaped_filename="$(sql_escape_literal "${filename}")"
  local sql
  sql=$(cat <<EOF
SELECT checksum
FROM \`${LEDGER_TABLE}\`
WHERE filename = '${escaped_filename}'
LIMIT 1;
EOF
)
  local output
  output="$(run_mysql_sql "${sql}" || true)"
  output="$(trim "${output}")"
  if [[ -n "${output}" ]]; then
    printf '%s' "${output}"
  fi
}

record_ledger_entry() {
  local filename="$1"
  local checksum="$2"
  local escaped_filename
  local escaped_checksum
  escaped_filename="$(sql_escape_literal "${filename}")"
  escaped_checksum="$(sql_escape_literal "${checksum}")"
  local sql
  sql=$(cat <<EOF
INSERT INTO \`${LEDGER_TABLE}\` (filename, checksum, applied_at)
VALUES ('${escaped_filename}', '${escaped_checksum}', CURRENT_TIMESTAMP)
ON DUPLICATE KEY UPDATE
  checksum = VALUES(checksum),
  applied_at = VALUES(applied_at);
EOF
)
  run_mysql_sql "${sql}" >/dev/null
}

ledger_table_exists() {
  local sql
  sql=$(cat <<EOF
SELECT COUNT(*)
FROM information_schema.TABLES
WHERE TABLE_SCHEMA = DATABASE()
  AND TABLE_NAME = '${LEDGER_TABLE}';
EOF
)
  [[ "$(mysql_scalar "${sql}")" == "1" ]]
}

database_user_table_count() {
  local escaped_ledger
  escaped_ledger="$(sql_escape_literal "${LEDGER_TABLE}")"
  local sql
  sql=$(cat <<EOF
SELECT COUNT(*)
FROM information_schema.TABLES
WHERE TABLE_SCHEMA = DATABASE()
  AND TABLE_TYPE = 'BASE TABLE'
  AND TABLE_NAME <> '${escaped_ledger}';
EOF
)
  mysql_scalar "${sql}"
}

require_safe_ledger_state_before_apply() {
  if ledger_table_exists; then
    return 0
  fi

  local table_count
  table_count="$(database_user_table_count)"
  if [[ "${table_count}" != "0" ]]; then
    die "existing database has ${table_count} tables but no ${LEDGER_TABLE} ledger. Refusing to replay historical migrations. Run --inventory, take a backup, follow docs/76-2026-05-13-enterprise-db-ledger-adoption-runbook.md, then run --adopt-existing after human approval."
  fi
}

list_migration_files() {
  local dir="$1"
  find "${dir}" -maxdepth 1 -type f -name '*.sql' ! -name '._*' | LC_ALL=C sort
}

schema_check_sql_for_migration() {
  case "$1" in
    202409200001_add_user_employee_link.sql)
      cat <<'EOF'
SELECT
  IF(EXISTS(SELECT 1 FROM information_schema.COLUMNS WHERE TABLE_SCHEMA=DATABASE() AND TABLE_NAME='users' AND COLUMN_NAME='employee_id'),0,1) +
  IF(EXISTS(SELECT 1 FROM information_schema.STATISTICS WHERE TABLE_SCHEMA=DATABASE() AND TABLE_NAME='users' AND INDEX_NAME='idx_employee_id'),0,1);
EOF
      ;;
    202409200002_refactor_asset_owner_fields.sql)
      cat <<'EOF'
SELECT
  IF(EXISTS(SELECT 1 FROM information_schema.COLUMNS WHERE TABLE_SCHEMA=DATABASE() AND TABLE_NAME='assets' AND COLUMN_NAME='owner_employee_id'),0,1) +
  IF(EXISTS(SELECT 1 FROM information_schema.COLUMNS WHERE TABLE_SCHEMA=DATABASE() AND TABLE_NAME='assets' AND COLUMN_NAME='department_id'),0,1);
EOF
      ;;
    202409200003_add_domain_asset_owner_fields.sql)
      cat <<'EOF'
SELECT
  IF(EXISTS(SELECT 1 FROM information_schema.COLUMNS WHERE TABLE_SCHEMA=DATABASE() AND TABLE_NAME='domain_assets' AND COLUMN_NAME='owner_employee_id'),0,1) +
  IF(EXISTS(SELECT 1 FROM information_schema.COLUMNS WHERE TABLE_SCHEMA=DATABASE() AND TABLE_NAME='domain_assets' AND COLUMN_NAME='department_id'),0,1);
EOF
      ;;
    202409200004_add_vulnerability_employee_fields.sql)
      cat <<'EOF'
SELECT
  IF(EXISTS(SELECT 1 FROM information_schema.COLUMNS WHERE TABLE_SCHEMA=DATABASE() AND TABLE_NAME='vulnerabilities' AND COLUMN_NAME='discoverer_employee_id'),0,1) +
  IF(EXISTS(SELECT 1 FROM information_schema.COLUMNS WHERE TABLE_SCHEMA=DATABASE() AND TABLE_NAME='vulnerabilities' AND COLUMN_NAME='fixer_employee_id'),0,1) +
  IF(EXISTS(SELECT 1 FROM information_schema.COLUMNS WHERE TABLE_SCHEMA=DATABASE() AND TABLE_NAME='vulnerabilities' AND COLUMN_NAME='submitter_user_id'),0,1) +
  IF(EXISTS(SELECT 1 FROM information_schema.COLUMNS WHERE TABLE_SCHEMA=DATABASE() AND TABLE_NAME='vulnerabilities' AND COLUMN_NAME='owner_employee_id'),0,1) +
  IF(EXISTS(SELECT 1 FROM information_schema.COLUMNS WHERE TABLE_SCHEMA=DATABASE() AND TABLE_NAME='vulnerabilities' AND COLUMN_NAME='department_id'),0,1);
EOF
      ;;
    202409200005_add_user_link_status_fields.sql)
      cat <<'EOF'
SELECT
  IF(EXISTS(SELECT 1 FROM information_schema.COLUMNS WHERE TABLE_SCHEMA=DATABASE() AND TABLE_NAME='users' AND COLUMN_NAME='first_login_at'),0,1) +
  IF(EXISTS(SELECT 1 FROM information_schema.COLUMNS WHERE TABLE_SCHEMA=DATABASE() AND TABLE_NAME='users' AND COLUMN_NAME='employee_link_status'),0,1);
EOF
      ;;
    202409200006_add_domain_name_server_alerts.sql)
      cat <<'EOF'
SELECT
  IF(EXISTS(SELECT 1 FROM information_schema.COLUMNS WHERE TABLE_SCHEMA=DATABASE() AND TABLE_NAME='domain_assets' AND COLUMN_NAME='name_servers_last_change_at'),0,1) +
  IF(EXISTS(SELECT 1 FROM information_schema.COLUMNS WHERE TABLE_SCHEMA=DATABASE() AND TABLE_NAME='domain_assets' AND COLUMN_NAME='name_servers_last_change_from'),0,1) +
  IF(EXISTS(SELECT 1 FROM information_schema.COLUMNS WHERE TABLE_SCHEMA=DATABASE() AND TABLE_NAME='domain_assets' AND COLUMN_NAME='name_servers_last_change_to'),0,1);
EOF
      ;;
    202409200007_normalize_dns_record_owner_employee_id.sql|202604010001_normalize_dns_record_owner_employee_id.sql)
      cat <<'EOF'
SELECT IF(EXISTS(
  SELECT 1 FROM information_schema.COLUMNS
  WHERE TABLE_SCHEMA=DATABASE()
    AND TABLE_NAME='dns_records'
    AND COLUMN_NAME='owner_employee_id'
    AND DATA_TYPE IN ('bigint')
),0,1);
EOF
      ;;
    202604010002_add_alert_recipient_targets.sql)
      cat <<'EOF'
SELECT
  IF(EXISTS(SELECT 1 FROM information_schema.COLUMNS WHERE TABLE_SCHEMA=DATABASE() AND TABLE_NAME='lark_alert_rules' AND COLUMN_NAME='recipient_target_keys'),0,1) +
  IF(EXISTS(SELECT 1 FROM information_schema.COLUMNS WHERE TABLE_SCHEMA=DATABASE() AND TABLE_NAME='lark_alert_deliveries' AND COLUMN_NAME='recipient_label'),0,1);
EOF
      ;;
    202604100003_add_vulnerability_authorized_viewers.sql)
      cat <<'EOF'
SELECT IF(EXISTS(SELECT 1 FROM information_schema.COLUMNS WHERE TABLE_SCHEMA=DATABASE() AND TABLE_NAME='vulnerabilities' AND COLUMN_NAME='authorized_viewer_employee_ids'),0,1);
EOF
      ;;
    202604100004_add_vulnerability_public_id.sql)
      cat <<'EOF'
SELECT
  IF(EXISTS(SELECT 1 FROM information_schema.COLUMNS WHERE TABLE_SCHEMA=DATABASE() AND TABLE_NAME='vulnerabilities' AND COLUMN_NAME='public_id'),0,1) +
  IF(EXISTS(SELECT 1 FROM information_schema.TABLES WHERE TABLE_SCHEMA=DATABASE() AND TABLE_NAME='vulnerability_public_id_sequences'),0,1) +
  IF(EXISTS(SELECT 1 FROM information_schema.STATISTICS WHERE TABLE_SCHEMA=DATABASE() AND TABLE_NAME='vulnerabilities' AND INDEX_NAME='idx_vulnerabilities_public_id'),0,1);
EOF
      ;;
    202604100005_add_caasm_query_indexes.sql)
      cat <<'EOF'
SELECT
  IF(EXISTS(SELECT 1 FROM information_schema.STATISTICS WHERE TABLE_SCHEMA=DATABASE() AND TABLE_NAME='cloud_accounts' AND INDEX_NAME='idx_cloud_accounts_status_last_sync'),0,1) +
  IF(EXISTS(SELECT 1 FROM information_schema.STATISTICS WHERE TABLE_SCHEMA=DATABASE() AND TABLE_NAME='dns_records' AND INDEX_NAME='idx_dns_records_updated_at'),0,1) +
  IF(EXISTS(SELECT 1 FROM information_schema.STATISTICS WHERE TABLE_SCHEMA=DATABASE() AND TABLE_NAME='sync_jobs' AND INDEX_NAME='idx_sync_jobs_cloud_account_status_created'),0,1);
EOF
      ;;
    202604100006_add_auth_soft_delete_columns.sql)
      cat <<'EOF'
SELECT
  IF(EXISTS(SELECT 1 FROM information_schema.COLUMNS WHERE TABLE_SCHEMA=DATABASE() AND TABLE_NAME='roles' AND COLUMN_NAME='deleted_at'),0,1) +
  IF(EXISTS(SELECT 1 FROM information_schema.COLUMNS WHERE TABLE_SCHEMA=DATABASE() AND TABLE_NAME='users' AND COLUMN_NAME='deleted_at'),0,1) +
  IF(EXISTS(SELECT 1 FROM information_schema.COLUMNS WHERE TABLE_SCHEMA=DATABASE() AND TABLE_NAME='permissions' AND COLUMN_NAME='deleted_at'),0,1);
EOF
      ;;
    202604100007_add_vulnerability_attachment_soft_delete.sql)
      cat <<'EOF'
SELECT IF(EXISTS(SELECT 1 FROM information_schema.COLUMNS WHERE TABLE_SCHEMA=DATABASE() AND TABLE_NAME='vulnerability_attachments' AND COLUMN_NAME='deleted_at'),0,1);
EOF
      ;;
    202604100008_add_remaining_soft_delete_columns.sql)
      cat <<'EOF'
SELECT
  IF(EXISTS(SELECT 1 FROM information_schema.COLUMNS WHERE TABLE_SCHEMA=DATABASE() AND TABLE_NAME='cloud_accounts' AND COLUMN_NAME='deleted_at'),0,1) +
  IF(EXISTS(SELECT 1 FROM information_schema.COLUMNS WHERE TABLE_SCHEMA=DATABASE() AND TABLE_NAME='assets' AND COLUMN_NAME='deleted_at'),0,1) +
  IF(EXISTS(SELECT 1 FROM information_schema.COLUMNS WHERE TABLE_SCHEMA=DATABASE() AND TABLE_NAME='asset_history' AND COLUMN_NAME='deleted_at'),0,1) +
  IF(EXISTS(SELECT 1 FROM information_schema.COLUMNS WHERE TABLE_SCHEMA=DATABASE() AND TABLE_NAME='asset_relationships' AND COLUMN_NAME='deleted_at'),0,1) +
  IF(EXISTS(SELECT 1 FROM information_schema.COLUMNS WHERE TABLE_SCHEMA=DATABASE() AND TABLE_NAME='sync_job_events' AND COLUMN_NAME='deleted_at'),0,1) +
  IF(EXISTS(SELECT 1 FROM information_schema.COLUMNS WHERE TABLE_SCHEMA=DATABASE() AND TABLE_NAME='vulnerabilities' AND COLUMN_NAME='deleted_at'),0,1);
EOF
      ;;
    202604110001_add_asset_identity_unique_key.sql)
      cat <<'EOF'
SELECT IF(EXISTS(SELECT 1 FROM information_schema.STATISTICS WHERE TABLE_SCHEMA=DATABASE() AND TABLE_NAME='assets' AND INDEX_NAME='idx_assets_cloud_account_native'),0,1);
EOF
      ;;
    202604110002_add_dns_record_identity_key_columns.sql)
      cat <<'EOF'
SELECT
  IF(EXISTS(SELECT 1 FROM information_schema.COLUMNS WHERE TABLE_SCHEMA=DATABASE() AND TABLE_NAME='dns_records' AND COLUMN_NAME='name_key'),0,1) +
  IF(EXISTS(SELECT 1 FROM information_schema.COLUMNS WHERE TABLE_SCHEMA=DATABASE() AND TABLE_NAME='dns_records' AND COLUMN_NAME='type_key'),0,1) +
  IF(EXISTS(SELECT 1 FROM information_schema.COLUMNS WHERE TABLE_SCHEMA=DATABASE() AND TABLE_NAME='dns_records' AND COLUMN_NAME='set_identifier_key'),0,1) +
  IF(EXISTS(SELECT 1 FROM information_schema.COLUMNS WHERE TABLE_SCHEMA=DATABASE() AND TABLE_NAME='dns_records' AND COLUMN_NAME='region_key'),0,1) +
  IF(EXISTS(SELECT 1 FROM information_schema.COLUMNS WHERE TABLE_SCHEMA=DATABASE() AND TABLE_NAME='dns_records' AND COLUMN_NAME='failover_key'),0,1) +
  IF(EXISTS(SELECT 1 FROM information_schema.STATISTICS WHERE TABLE_SCHEMA=DATABASE() AND TABLE_NAME='dns_records' AND INDEX_NAME='idx_dns_records_identity'),0,1);
EOF
      ;;
    202605120001_add_caasm_api_route_assets.sql)
      cat <<'EOF'
SELECT
  IF(EXISTS(SELECT 1 FROM information_schema.TABLES WHERE TABLE_SCHEMA=DATABASE() AND TABLE_NAME='caasm_api_route_configs'),0,1) +
  IF(EXISTS(SELECT 1 FROM information_schema.TABLES WHERE TABLE_SCHEMA=DATABASE() AND TABLE_NAME='caasm_api_route_assets'),0,1);
EOF
      ;;
    *)
      return 1
      ;;
  esac
}

migration_missing_schema_count() {
  local filename="$1"
  case "${filename}" in
    202604110001_add_asset_identity_unique_key.sql)
      migration_asset_identity_missing_count
      return 0
      ;;
    202604110002_add_dns_record_identity_key_columns.sql)
      migration_dns_identity_missing_count
      return 0
      ;;
  esac

  local check_sql
  if ! check_sql="$(schema_check_sql_for_migration "${filename}")"; then
    printf '%s' "unknown"
    return 0
  fi
  mysql_scalar "${check_sql}"
}

migration_asset_identity_missing_count() {
  mysql_scalar <<'EOF'
SELECT IF((
  SELECT COUNT(*)
  FROM information_schema.STATISTICS
  WHERE TABLE_SCHEMA = DATABASE()
    AND TABLE_NAME = 'assets'
    AND INDEX_NAME = 'idx_assets_cloud_account_native'
    AND NON_UNIQUE = 0
    AND (
      (SEQ_IN_INDEX = 1 AND COLUMN_NAME = 'cloud_account_id') OR
      (SEQ_IN_INDEX = 2 AND COLUMN_NAME = 'native_id')
    )
) = 2, 0, 1);
EOF
}

migration_dns_identity_missing_count() {
  local column_missing_count
  column_missing_count="$(mysql_scalar <<'EOF'
SELECT
  IF(EXISTS(SELECT 1 FROM information_schema.COLUMNS WHERE TABLE_SCHEMA=DATABASE() AND TABLE_NAME='dns_records' AND COLUMN_NAME='name_key'),0,1) +
  IF(EXISTS(SELECT 1 FROM information_schema.COLUMNS WHERE TABLE_SCHEMA=DATABASE() AND TABLE_NAME='dns_records' AND COLUMN_NAME='type_key'),0,1) +
  IF(EXISTS(SELECT 1 FROM information_schema.COLUMNS WHERE TABLE_SCHEMA=DATABASE() AND TABLE_NAME='dns_records' AND COLUMN_NAME='set_identifier_key'),0,1) +
  IF(EXISTS(SELECT 1 FROM information_schema.COLUMNS WHERE TABLE_SCHEMA=DATABASE() AND TABLE_NAME='dns_records' AND COLUMN_NAME='region_key'),0,1) +
  IF(EXISTS(SELECT 1 FROM information_schema.COLUMNS WHERE TABLE_SCHEMA=DATABASE() AND TABLE_NAME='dns_records' AND COLUMN_NAME='failover_key'),0,1);
EOF
)"
  if [[ "${column_missing_count}" != "0" ]]; then
    printf '%s' "${column_missing_count}"
    return 0
  fi

  local index_missing_count
  index_missing_count="$(mysql_scalar <<'EOF'
SELECT IF((
  SELECT COUNT(*)
  FROM information_schema.STATISTICS
  WHERE TABLE_SCHEMA = DATABASE()
    AND TABLE_NAME = 'dns_records'
    AND INDEX_NAME = 'idx_dns_records_identity'
    AND NON_UNIQUE = 0
    AND (
      (SEQ_IN_INDEX = 1 AND COLUMN_NAME = 'domain_asset_id') OR
      (SEQ_IN_INDEX = 2 AND COLUMN_NAME = 'name_key') OR
      (SEQ_IN_INDEX = 3 AND COLUMN_NAME = 'type_key') OR
      (SEQ_IN_INDEX = 4 AND COLUMN_NAME = 'set_identifier_key') OR
      (SEQ_IN_INDEX = 5 AND COLUMN_NAME = 'region_key') OR
      (SEQ_IN_INDEX = 6 AND COLUMN_NAME = 'failover_key')
    )
) = 6, 0, 1);
EOF
)"
  if [[ "${index_missing_count}" != "0" ]]; then
    printf '%s' "${index_missing_count}"
    return 0
  fi

  mysql_scalar <<'EOF'
SELECT COUNT(*)
FROM dns_records
WHERE COALESCE(name_key, '') <> LOWER(TRIM(TRAILING '.' FROM TRIM(COALESCE(name, ''))))
   OR COALESCE(type_key, '') <> UPPER(TRIM(COALESCE(type, '')))
   OR COALESCE(set_identifier_key, '') <> CASE
     WHEN set_identifier IS NULL OR TRIM(set_identifier) = '' OR LOWER(TRIM(set_identifier)) = '<nil>' THEN ''
     ELSE TRIM(set_identifier)
   END
   OR COALESCE(region_key, '') <> CASE
     WHEN region IS NULL OR TRIM(region) = '' OR LOWER(TRIM(region)) = '<nil>' THEN ''
     ELSE LOWER(TRIM(region))
   END
   OR COALESCE(failover_key, '') <> CASE
     WHEN failover IS NULL OR TRIM(failover) = '' OR LOWER(TRIM(failover)) = '<nil>' THEN ''
     ELSE UPPER(TRIM(failover))
   END;
EOF
}

print_inventory_section() {
  local title="$1"
  local sql
  sql="$(cat)"
  info "${title}"
  run_mysql_sql "${sql}" || true
}

run_schema_inventory() {
  info "schema inventory for database: ${MYSQL_DATABASE}"
  if ledger_table_exists; then
    local ledger_count
    ledger_count="$(mysql_scalar "SELECT COUNT(*) FROM \`${LEDGER_TABLE}\`;")"
    info "ledger table: present (${ledger_count} rows)"
  else
    info "ledger table: missing"
  fi
  info "application table count: $(database_user_table_count)"

  print_inventory_section "asset identity duplicates (cloud_account_id, native_id)" <<'EOF'
SELECT COALESCE(CAST(cloud_account_id AS CHAR), '<null>') AS cloud_account_id,
       COALESCE(native_id, '<null>') AS native_id,
       COUNT(*) AS duplicate_count,
       GROUP_CONCAT(id ORDER BY id) AS sample_ids
FROM assets
WHERE native_id IS NOT NULL AND TRIM(native_id) <> ''
GROUP BY cloud_account_id, native_id
HAVING COUNT(*) > 1
ORDER BY duplicate_count DESC
LIMIT 50;
EOF

  print_inventory_section "dns identity duplicates after normalization" <<'EOF'
SELECT domain_asset_id,
       LOWER(TRIM(TRAILING '.' FROM TRIM(name))) AS name_key,
       UPPER(TRIM(type)) AS type_key,
       CASE
         WHEN set_identifier IS NULL OR TRIM(set_identifier) = '' OR LOWER(TRIM(set_identifier)) = '<nil>' THEN ''
         ELSE TRIM(set_identifier)
       END AS set_identifier_key,
       CASE
         WHEN region IS NULL OR TRIM(region) = '' OR LOWER(TRIM(region)) = '<nil>' THEN ''
         ELSE LOWER(TRIM(region))
       END AS region_key,
       CASE
         WHEN failover IS NULL OR TRIM(failover) = '' OR LOWER(TRIM(failover)) = '<nil>' THEN ''
         ELSE UPPER(TRIM(failover))
       END AS failover_key,
       COUNT(*) AS duplicate_count,
       GROUP_CONCAT(id ORDER BY id) AS sample_ids
FROM dns_records
WHERE domain_asset_id IS NOT NULL
GROUP BY domain_asset_id, name_key, type_key, set_identifier_key, region_key, failover_key
HAVING COUNT(*) > 1
ORDER BY duplicate_count DESC
LIMIT 50;
EOF

  print_inventory_section "vulnerability public_id coverage" <<'EOF'
SELECT
  COUNT(*) AS total_vulnerabilities,
  SUM(CASE WHEN public_id IS NULL OR public_id = '' THEN 1 ELSE 0 END) AS missing_public_id,
  COUNT(DISTINCT public_id) AS distinct_public_id
FROM vulnerabilities;
EOF

  local file_path filename checksum current_checksum missing_count
  info "migration adoption candidates"
  while IFS= read -r file_path; do
    [[ -n "${file_path}" ]] || continue
    filename="$(basename "${file_path}")"
    checksum="$(sha256_file "${file_path}")"
    current_checksum=""
    if ledger_table_exists; then
      current_checksum="$(get_ledger_checksum "${filename}" || true)"
    fi
    missing_count="$(migration_missing_schema_count "${filename}")"
    if [[ -n "${current_checksum}" ]]; then
      printf 'applied\t%s\t%s\n' "${filename}" "${current_checksum}"
    elif [[ "${missing_count}" == "0" ]]; then
      printf 'adoptable\t%s\t%s\n' "${filename}" "${checksum}"
    else
      printf 'pending_or_unsafe\t%s\tmissing_checks=%s\n' "${filename}" "${missing_count}"
    fi
  done < <(list_migration_files "${MIGRATIONS_DIR}")
}

adopt_existing_ledger() {
  validate_baseline_ledger
  local file_path filename checksum current_checksum missing_count adopted_count=0 skipped_count=0

  if [[ "${DRY_RUN}" != "true" ]]; then
    ensure_ledger_table
  fi

  while IFS= read -r file_path; do
    [[ -n "${file_path}" ]] || continue
    filename="$(basename "${file_path}")"
    checksum="$(sha256_file "${file_path}")"
    current_checksum=""
    if ledger_table_exists; then
      current_checksum="$(get_ledger_checksum "${filename}" || true)"
    fi
    if [[ -n "${current_checksum}" ]]; then
      if [[ "${current_checksum}" != "${checksum}" ]]; then
        die "checksum drift for ${filename}: ledger=${current_checksum} file=${checksum}"
      fi
      skipped_count=$((skipped_count + 1))
      info "already adopted: ${filename}"
      continue
    fi

    missing_count="$(migration_missing_schema_count "${filename}")"
    if [[ "${missing_count}" != "0" ]]; then
      die "cannot adopt ${filename}: live schema check reports missing_checks=${missing_count}. Apply or fix the migration instead of adopting it."
    fi

    if [[ "${DRY_RUN}" == "true" ]]; then
      info "would adopt ledger row: ${filename} (${checksum})"
    else
      record_ledger_entry "${filename}" "${checksum}"
      info "adopted ledger row: ${filename}"
    fi
    adopted_count=$((adopted_count + 1))
  done < <(list_migration_files "${MIGRATIONS_DIR}")

  info "ledger adoption complete: adopted=${adopted_count} skipped=${skipped_count}"
}

detect_baseline_init_file() {
  local candidate=""
  local -a candidates=(
    "${BASELINE_INIT_FILE}"
    "${ROOT_DIR}/mysql-init.sql"
    "${SCRIPT_DIR}/mysql-init.sql"
    "${ROOT_DIR}/scripts/enterprise/mysql-init.sql"
  )
  for candidate in "${candidates[@]}"; do
    [[ -n "${candidate}" ]] || continue
    if [[ -f "${candidate}" ]]; then
      printf '%s' "${candidate}"
      return 0
    fi
  done
  printf '%s' ""
}

extract_baseline_rows() {
  local baseline_file="$1"
  python3 - "${baseline_file}" <<'PY'
from pathlib import Path
import re
import sys

path = Path(sys.argv[1])
text = path.read_text()
match = re.search(
    r"INSERT\s+INTO\s+schema_migrations\s*\(filename,\s*checksum,\s*applied_at\)\s*VALUES\s*(.*?)(?:\nON\s+DUPLICATE\s+KEY\s+UPDATE|\Z)",
    text,
    re.IGNORECASE | re.DOTALL,
)
if not match:
    raise SystemExit(0)

rows = re.findall(
    r"\(\s*'([^']+)'\s*,\s*'([0-9a-fA-F]{64})'\s*,\s*CURRENT_TIMESTAMP\s*\)",
    match.group(1),
    re.IGNORECASE,
)
for filename, checksum in rows:
    print(f"{filename}\t{checksum.lower()}")
PY
}

validate_baseline_ledger() {
  local baseline_file
  baseline_file="$(detect_baseline_init_file)"
  if [[ -z "${baseline_file}" ]]; then
    info "baseline ledger validation: no mysql-init.sql file found; skipping"
    return 0
  fi

  local -a baseline_rows=()
  local row=""
  while IFS= read -r row; do
    [[ -n "${row}" ]] || continue
    baseline_rows+=("${row}")
  done < <(extract_baseline_rows "${baseline_file}")

  if ((${#baseline_rows[@]} == 0)); then
    info "baseline ledger validation: no schema_migrations bootstrap rows found in $(basename "${baseline_file}")"
    return 0
  fi

  local -a migrations=()
  local migration=""
  while IFS= read -r migration; do
    [[ -n "${migration}" ]] || continue
    migrations+=("$(basename "${migration}")")
  done < <(list_migration_files "${MIGRATIONS_DIR}")

  if ((${#baseline_rows[@]} > ${#migrations[@]})); then
    die "baseline ledger has more entries than available migrations: baseline=${#baseline_rows[@]} migrations=${#migrations[@]}"
  fi

  local index=0
  local baseline_filename=""
  local baseline_checksum=""
  local migration_filename=""
  local migration_checksum=""
  for row in "${baseline_rows[@]}"; do
    baseline_filename="${row%%$'\t'*}"
    baseline_checksum="${row#*$'\t'}"
    migration_filename="${migrations[${index}]}"
    if [[ "${baseline_filename}" != "${migration_filename}" ]]; then
      die "baseline ledger order mismatch at position $((index + 1)): mysql-init.sql has ${baseline_filename} but migrations dir has ${migration_filename}"
    fi

    migration_checksum="$(sha256_file "${MIGRATIONS_DIR}/${migration_filename}")"
    if [[ "${baseline_checksum}" != "${migration_checksum}" ]]; then
      die "baseline checksum drift for ${baseline_filename}: init=${baseline_checksum} file=${migration_checksum}"
    fi
    index=$((index + 1))
  done

  info "baseline ledger validation passed: ${#baseline_rows[@]} bootstrap rows match the leading migration files in ${MIGRATIONS_DIR}"
}

parse_args() {
  while [[ $# -gt 0 ]]; do
    case "$1" in
      --migrations-dir)
        MIGRATIONS_DIR="${2:-}"
        shift 2
        ;;
      --ledger-table)
        LEDGER_TABLE="${2:-}"
        shift 2
        ;;
      --env-file)
        ENV_FILE="${2:-}"
        shift 2
        ;;
      --baseline-init)
        BASELINE_INIT_FILE="${2:-}"
        shift 2
        ;;
      --mysql-prefix)
        MYSQL_PREFIX="${2:-}"
        shift 2
        ;;
      --host)
        MYSQL_HOST="${2:-}"
        shift 2
        ;;
      --port)
        MYSQL_PORT="${2:-}"
        shift 2
        ;;
      --socket)
        MYSQL_SOCKET="${2:-}"
        shift 2
        ;;
      --protocol)
        MYSQL_PROTOCOL="${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
        ;;
      --defaults-extra-file)
        MYSQL_DEFAULTS_EXTRA_FILE="${2:-}"
        shift 2
        ;;
      --inventory)
        MODE="inventory"
        shift
        ;;
      --adopt-existing)
        MODE="adopt-existing"
        shift
        ;;
      --dry-run)
        DRY_RUN="true"
        shift
        ;;
      --help|-h)
        usage
        exit 0
        ;;
      *)
        die "unknown argument: $1"
        ;;
    esac
  done
}

main() {
  parse_args "$@"

  require_command find
  require_command sort

  MIGRATIONS_DIR="$(resolve_path "${MIGRATIONS_DIR}")"
  [[ -d "${MIGRATIONS_DIR}" ]] || die "migrations directory not found: ${MIGRATIONS_DIR}"
  if [[ -n "${BASELINE_INIT_FILE}" ]]; then
    BASELINE_INIT_FILE="$(resolve_path "${BASELINE_INIT_FILE}")"
  fi

  if [[ -n "${ENV_FILE}" ]]; then
    ENV_FILE="$(resolve_path "${ENV_FILE}")"
    [[ -f "${ENV_FILE}" ]] || die "env file not found: ${ENV_FILE}"
    if [[ -z "${MYSQL_DATABASE}" || "${MYSQL_DATABASE}" == "${DEFAULT_DATABASE}" ]]; then
      MYSQL_DATABASE="$(env_file_value "${ENV_FILE}" "SC_MYSQL_DATABASE")"
    fi
    if [[ -z "${MYSQL_USER}" ]]; then
      MYSQL_USER="$(env_file_value "${ENV_FILE}" "SC_MYSQL_USER")"
    fi
    if [[ -z "${MYSQL_PASSWORD}" ]]; then
      MYSQL_PASSWORD="$(env_file_value "${ENV_FILE}" "SC_MYSQL_APP_PASSWORD")"
    fi
  fi

  [[ -n "${MYSQL_DATABASE}" ]] || die "mysql database is required"
  [[ -n "${MYSQL_USER}" ]] || die "mysql user is required"

  local migrations=()
  local migration=""
  while IFS= read -r migration; do
    [[ -n "${migration}" ]] || continue
    migrations+=("${migration}")
  done < <(list_migration_files "${MIGRATIONS_DIR}")

  if [[ "${#migrations[@]}" -eq 0 ]]; then
    info "no SQL migrations found in ${MIGRATIONS_DIR}"
    exit 0
  fi

  case "${MODE}" in
    inventory)
      run_schema_inventory
      exit 0
      ;;
    adopt-existing)
      adopt_existing_ledger
      exit 0
      ;;
    apply)
      ;;
    *)
      die "unknown migration mode: ${MODE}"
      ;;
  esac

  validate_baseline_ledger
  if [[ "${DRY_RUN}" != "true" ]]; then
    require_safe_ledger_state_before_apply
  fi

  info "migration source directory: ${MIGRATIONS_DIR}"
  info "migration ledger table: ${LEDGER_TABLE}"
  info "mysql database: ${MYSQL_DATABASE}"
  if [[ "${DRY_RUN}" == "true" ]]; then
    info "dry-run mode enabled"
  fi

  if [[ "${DRY_RUN}" != "true" ]]; then
    ensure_ledger_table
  fi

  local applied_count=0
  local skipped_count=0
  local file_path=""
  local filename=""
  local checksum=""
  local current_checksum=""

  for file_path in "${migrations[@]}"; do
    filename="$(basename "${file_path}")"
    checksum="$(sha256_file "${file_path}")"

    current_checksum=""
    if [[ "${DRY_RUN}" != "true" ]]; then
      current_checksum="$(get_ledger_checksum "${filename}" || true)"
    fi

    if [[ -n "${current_checksum}" ]]; then
      if [[ "${current_checksum}" != "${checksum}" ]]; then
        die "checksum drift for ${filename}: ledger=${current_checksum} file=${checksum}"
      fi
      skipped_count=$((skipped_count + 1))
      info "skipped already applied migration: ${filename}"
      continue
    fi

    if [[ "${DRY_RUN}" == "true" ]]; then
      info "would apply migration: ${filename} (${checksum})"
      continue
    fi

    info "applying migration: ${filename}"
    run_mysql_file "${file_path}"
    record_ledger_entry "${filename}" "${checksum}"
    applied_count=$((applied_count + 1))
  done

  info "migration run complete: applied=${applied_count} skipped=${skipped_count} total=${#migrations[@]}"
}

main "$@"