#!/usr/bin/env bash
#
# MySQL 跨机房实时同步脚本（交互菜单 + 主从/双主）
# 直接运行进入菜单: ./mysql_sync.sh
#

set -euo pipefail

SCRIPT_DIR="$(cd "$(dirname "${BASH_SOURCE[0]}")" && pwd)"
CONF_FILE="${SCRIPT_DIR}/mysql_sync.conf"
CONF_EXAMPLE="${SCRIPT_DIR}/mysql_sync.conf.example"

# ==================== 通用工具 ====================

log() {
  local msg="[$(date '+%Y-%m-%d %H:%M:%S')] $*"
  echo "$msg"
  if [[ -n "${LOG_FILE:-}" ]]; then
    mkdir -p "$(dirname "$LOG_FILE")" 2>/dev/null || true
    echo "$msg" >> "$LOG_FILE"
  fi
}

die() {
  log "ERROR: $*"
  exit 1
}

pause_enter() {
  read -r -p "按 Enter 返回菜单..." _ </dev/tty
}

clear_screen() {
  printf '\033[2J\033[H' 2>/dev/null || true
}

load_config_raw() {
  if [[ ! -f "$CONF_FILE" ]]; then
    return 1
  fi
  local tmp_conf
  tmp_conf="$(mktemp "${TMPDIR:-/tmp}/mysql_sync.conf.XXXXXX")"
  sed 's/\r$//' "$CONF_FILE" > "$tmp_conf"
  # shellcheck source=/dev/null
  source "$tmp_conf"
  rm -f "$tmp_conf"
  return 0
}

is_placeholder_password() {
  local val="$1"
  [[ -z "$val" ]] && return 0
  case "$val" in
    your_password_*|your_repl_password|change_me|password_here) return 0 ;;
  esac
  return 1
}

escape_config_value() {
  local s="$1"
  s="${s//\\/\\\\}"
  s="${s//\"/\\\"}"
  echo "$s"
}

sql_escape() {
  local s="$1"
  echo "${s//\'/\'\'}"
}

validate_config() {
  : "${SITE_A_HOST:?请先在菜单中配置主库/机房A}"
  : "${SITE_B_HOST:?请先在菜单中配置从库/机房B}"
  : "${REPL_USER:?请配置复制账号 REPL_USER}"
  : "${REPL_PASSWORD:?请配置复制账号密码 REPL_PASSWORD}"

  if [[ "${SITE_A_SERVER_ID}" == "${SITE_B_SERVER_ID}" ]]; then
    die "主库与从库 server-id 不能相同（当前均为 ${SITE_A_SERVER_ID}）"
  fi

  if is_placeholder_password "${SITE_A_ROOT_PASSWORD:-}"; then
    die "主库 root 密码未配置，请在菜单「配置向导」中填写"
  fi
  if is_placeholder_password "${SITE_B_ROOT_PASSWORD:-}"; then
    die "从库 root 密码未配置，请在菜单「配置向导」中填写"
  fi
  if is_placeholder_password "${REPL_PASSWORD:-}"; then
    die "复制账号密码未配置，请在菜单「配置向导」中填写"
  fi

  if [[ -n "${SYNC_TABLES:-}" ]]; then
    local spec
    for spec in $SYNC_TABLES; do
      [[ "$spec" =~ ^[A-Za-z0-9_]+\.[A-Za-z0-9_]+$ ]] \
        || die "SYNC_TABLES 格式错误: $spec（应为 db.table）"
    done
  fi
}

load_config() {
  load_config_raw || die "配置文件不存在，请先使用菜单「1) 配置向导」"
  validate_config
}

sync_mode_label() {
  case "${SYNC_MODE:-dual_master}" in
    master_slave) echo "主从同步（一主一从）" ;;
    dual_master)  echo "双主同步（跨机房）" ;;
    *)            echo "${SYNC_MODE:-未知}" ;;
  esac
}

is_dual_master() {
  [[ "${SYNC_MODE:-dual_master}" == "dual_master" ]]
}

is_master_slave() {
  [[ "${SYNC_MODE:-dual_master}" == "master_slave" ]]
}

save_config_file() {
  cat > "$CONF_FILE" <<EOF
# MySQL 同步配置（由菜单向导生成）
# 生成时间: $(date '+%Y-%m-%d %H:%M:%S')

SYNC_MODE="${SYNC_MODE:-dual_master}"

# ========== 主库 / 机房 A ==========
SITE_A_NAME="$(escape_config_value "${SITE_A_NAME:-主库}")"
SITE_A_HOST="$(escape_config_value "${SITE_A_HOST}")"
SITE_A_PORT="${SITE_A_PORT:-3306}"
SITE_A_ROOT_USER="$(escape_config_value "${SITE_A_ROOT_USER:-root}")"
SITE_A_ROOT_PASSWORD="$(escape_config_value "${SITE_A_ROOT_PASSWORD}")"
SITE_A_SERVER_ID="${SITE_A_SERVER_ID:-101}"
SITE_A_AUTO_INCREMENT_OFFSET="${SITE_A_AUTO_INCREMENT_OFFSET:-1}"
SITE_A_REPL_ALLOW_HOST="$(escape_config_value "${SITE_A_REPL_ALLOW_HOST:-%}")"

# ========== 从库 / 机房 B ==========
SITE_B_NAME="$(escape_config_value "${SITE_B_NAME:-从库}")"
SITE_B_HOST="$(escape_config_value "${SITE_B_HOST}")"
SITE_B_PORT="${SITE_B_PORT:-3306}"
SITE_B_ROOT_USER="$(escape_config_value "${SITE_B_ROOT_USER:-root}")"
SITE_B_ROOT_PASSWORD="$(escape_config_value "${SITE_B_ROOT_PASSWORD}")"
SITE_B_SERVER_ID="${SITE_B_SERVER_ID:-102}"
SITE_B_AUTO_INCREMENT_OFFSET="${SITE_B_AUTO_INCREMENT_OFFSET:-2}"
SITE_B_REPL_ALLOW_HOST="$(escape_config_value "${SITE_B_REPL_ALLOW_HOST:-%}")"

# ========== 复制账号 ==========
REPL_USER="$(escape_config_value "${REPL_USER:-repl}")"
REPL_PASSWORD="$(escape_config_value "${REPL_PASSWORD}")"

# ========== 同步范围 ==========
SYNC_DATABASES="$(escape_config_value "${SYNC_DATABASES:-}")"
# 指定表同步，格式: db.table（空格分隔），留空=同步整库
SYNC_TABLES="$(escape_config_value "${SYNC_TABLES:-}")"

# ========== 初始全量 ==========
INITIAL_SYNC_DIRECTION="${INITIAL_SYNC_DIRECTION:-A_TO_B}"
DUMP_DIR="$(escape_config_value "${DUMP_DIR:-/tmp/mysql_sync_dump}")"

# ========== 跨机房参数 ==========
MYSQL_CONNECT_TIMEOUT="${MYSQL_CONNECT_TIMEOUT:-30}"
REPL_LAG_WARN_SECONDS="${REPL_LAG_WARN_SECONDS:-300}"
BINLOG_RETENTION_DAYS="${BINLOG_RETENTION_DAYS:-7}"
ENABLE_SEMI_SYNC="${ENABLE_SEMI_SYNC:-0}"
SEMI_SYNC_TIMEOUT_MS="${SEMI_SYNC_TIMEOUT_MS:-30000}"
AUTO_INCREMENT_INCREMENT="${AUTO_INCREMENT_INCREMENT:-2}"
SLAVE_SKIP_ERRORS="${SLAVE_SKIP_ERRORS:-0}"
SLAVE_READ_ONLY="${SLAVE_READ_ONLY:-1}"
STRICT_MYCNF_CHECK="${STRICT_MYCNF_CHECK:-0}"
LOG_FILE="$(escape_config_value "${LOG_FILE:-/var/log/mysql_sync.log}")"
EOF
  log "配置已保存: $CONF_FILE"
}

# ==================== 交互输入 ====================

prompt_value() {
  local var_name="$1"
  local label="$2"
  local default="${3:-}"
  local current="${!var_name:-$default}"
  local input=""

  read -r -p "${label} [${current}]: " input </dev/tty
  if [[ -n "$input" ]]; then
    printf -v "$var_name" '%s' "$input"
  else
    printf -v "$var_name" '%s' "$current"
  fi
}

prompt_password() {
  local var_name="$1"
  local label="$2"
  local input=""

  read -r -s -p "${label}（直接回车保持不变）: " input </dev/tty
  echo
  if [[ -n "$input" ]]; then
    printf -v "$var_name" '%s' "$input"
  fi
}

prompt_yes_no() {
  local var_name="$1"
  local label="$2"
  local default="${3:-n}"
  local input=""
  local hint="y/n"

  [[ "$default" == "y" ]] && hint="Y/n" || hint="y/N"
  read -r -p "${label} [${hint}]: " input </dev/tty
  input="${input:-$default}"
  case "$input" in
    y|Y|yes|YES) printf -v "$var_name" '%s' "1" ;;
    *)           printf -v "$var_name" '%s' "0" ;;
  esac
}

select_option() {
  local prompt="$1"
  shift
  local options=("$@")
  local i choice

  echo "$prompt"
  for i in "${!options[@]}"; do
    echo "  $((i + 1))) ${options[$i]}"
  done

  while true; do
    read -r -p "请输入序号: " choice </dev/tty
    if [[ "$choice" =~ ^[0-9]+$ ]] && (( choice >= 1 && choice <= ${#options[@]} )); then
      REPLY="$((choice - 1))"
      return 0
    fi
    echo "无效选项，请重新输入"
  done
}

# ==================== 配置向导 ====================

wizard_load_existing() {
  SYNC_MODE="dual_master"
  SITE_A_NAME="主库"; SITE_A_HOST="127.0.0.1"; SITE_A_PORT="3306"
  SITE_A_ROOT_USER="root"; SITE_A_ROOT_PASSWORD=""
  SITE_A_SERVER_ID="101"; SITE_A_AUTO_INCREMENT_OFFSET="1"; SITE_A_REPL_ALLOW_HOST="%"
  SITE_B_NAME="从库"; SITE_B_HOST="127.0.0.1"; SITE_B_PORT="3306"
  SITE_B_ROOT_USER="root"; SITE_B_ROOT_PASSWORD=""
  SITE_B_SERVER_ID="102"; SITE_B_AUTO_INCREMENT_OFFSET="2"; SITE_B_REPL_ALLOW_HOST="%"
  REPL_USER="repl"; REPL_PASSWORD=""
  SYNC_DATABASES=""
  SYNC_TABLES=""
  INITIAL_SYNC_DIRECTION="A_TO_B"
  DUMP_DIR="/tmp/mysql_sync_dump"
  MYSQL_CONNECT_TIMEOUT="30"
  REPL_LAG_WARN_SECONDS="300"
  ENABLE_SEMI_SYNC="0"
  SLAVE_READ_ONLY="1"
  STRICT_MYCNF_CHECK="0"
  LOG_FILE="/var/log/mysql_sync.log"

  load_config_raw || true
}

wizard_configure_master() {
  clear_screen
  echo "========== 主库配置 =========="
  prompt_value SITE_A_NAME "机房/主库名称" "${SITE_A_NAME:-主库}"
  prompt_value SITE_A_HOST "主库 IP 地址" "${SITE_A_HOST:-}"
  prompt_value SITE_A_PORT "主库端口" "${SITE_A_PORT:-3306}"
  prompt_value SITE_A_ROOT_USER "主库 root 用户名" "${SITE_A_ROOT_USER:-root}"
  prompt_password SITE_A_ROOT_PASSWORD "主库 root 密码"
  prompt_value SITE_A_SERVER_ID "主库 server-id（唯一）" "${SITE_A_SERVER_ID:-101}"
  prompt_value SITE_A_REPL_ALLOW_HOST "允许复制的客户端 IP/网段" "${SITE_A_REPL_ALLOW_HOST:-%}"
}

wizard_configure_slave() {
  clear_screen
  echo "========== 从库配置 =========="
  prompt_value SITE_B_NAME "机房/从库名称" "${SITE_B_NAME:-从库}"
  prompt_value SITE_B_HOST "从库 IP 地址" "${SITE_B_HOST:-}"
  prompt_value SITE_B_PORT "从库端口" "${SITE_B_PORT:-3306}"
  prompt_value SITE_B_ROOT_USER "从库 root 用户名" "${SITE_B_ROOT_USER:-root}"
  prompt_password SITE_B_ROOT_PASSWORD "从库 root 密码"
  prompt_value SITE_B_SERVER_ID "从库 server-id（唯一，勿与主库相同）" "${SITE_B_SERVER_ID:-102}"
  prompt_value SITE_B_REPL_ALLOW_HOST "允许复制的客户端 IP/网段" "${SITE_B_REPL_ALLOW_HOST:-%}"
}

wizard_configure_common() {
  clear_screen
  echo "========== 复制与同步选项 =========="
  prompt_value REPL_USER "复制账号用户名" "${REPL_USER:-repl}"
  prompt_password REPL_PASSWORD "复制账号密码"
  prompt_value SYNC_DATABASES "同步的数据库（空格分隔，留空=全部）" "${SYNC_DATABASES:-}"
  prompt_value SYNC_TABLES "指定同步的表 db.table（空格分隔，留空=整库）" "${SYNC_TABLES:-}"
  prompt_value DUMP_DIR "全量备份目录" "${DUMP_DIR:-/tmp/mysql_sync_dump}"
  prompt_value REPL_LAG_WARN_SECONDS "延迟告警阈值(秒)" "${REPL_LAG_WARN_SECONDS:-300}"
  prompt_yes_no ENABLE_SEMI_SYNC "是否启用半同步复制" "n"

  if is_dual_master; then
    prompt_value SITE_A_AUTO_INCREMENT_OFFSET "机房A 自增 offset" "${SITE_A_AUTO_INCREMENT_OFFSET:-1}"
    prompt_value SITE_B_AUTO_INCREMENT_OFFSET "机房B 自增 offset" "${SITE_B_AUTO_INCREMENT_OFFSET:-2}"
    echo
    echo "初始全量数据以哪边为准:"
    select_option "请选择" "以主库/机房A为准 (A→B)" "以从库/机房B为准 (B→A)" "跳过全量"
    case "$REPLY" in
      0) INITIAL_SYNC_DIRECTION="A_TO_B" ;;
      1) INITIAL_SYNC_DIRECTION="B_TO_A" ;;
      2) INITIAL_SYNC_DIRECTION="skip" ;;
    esac
  else
    INITIAL_SYNC_DIRECTION="A_TO_B"
    prompt_yes_no SLAVE_READ_ONLY "从库是否只读" "y"
  fi
}

menu_config_wizard() {
  wizard_load_existing

  clear_screen
  echo "========== 同步模式选择 =========="
  select_option "请选择同步模式" "主从同步（一主一从，主写从读）" "双主同步（跨机房，双向可写）"
  case "$REPLY" in
    0) SYNC_MODE="master_slave"; SITE_A_NAME="${SITE_A_NAME:-主库}"; SITE_B_NAME="${SITE_B_NAME:-从库}" ;;
    1) SYNC_MODE="dual_master" ;;
  esac

  wizard_configure_master
  wizard_configure_slave
  wizard_configure_common

  clear_screen
  echo "========== 配置确认 =========="
  echo "同步模式: $(sync_mode_label)"
  echo "主库: ${SITE_A_NAME} ${SITE_A_HOST}:${SITE_A_PORT} (server-id=${SITE_A_SERVER_ID})"
  echo "从库: ${SITE_B_NAME} ${SITE_B_HOST}:${SITE_B_PORT} (server-id=${SITE_B_SERVER_ID})"
  echo "复制账号: ${REPL_USER}"
  echo "同步库: ${SYNC_DATABASES:-<全部用户库>}"
  echo "同步表: ${SYNC_TABLES:-<整库>}"
  echo "全量方向: ${INITIAL_SYNC_DIRECTION}"
  echo
  select_option "确认保存?" "保存并返回菜单" "返回继续修改" "取消"
  case "$REPLY" in
    0)
      save_config_file
      log "配置保存成功，可选择「3) 初始化部署」"
      ;;
    1)
      menu_config_wizard
      ;;
    *) log "已取消保存" ;;
  esac
}

menu_show_config() {
  clear_screen
  echo "========== 当前配置 =========="
  if ! load_config_raw; then
    echo "尚未配置，请先使用「1) 配置向导」"
    return
  fi
  echo "同步模式: $(sync_mode_label)"
  echo
  echo "[主库 / 机房 A] ${SITE_A_NAME}"
  echo "  地址: ${SITE_A_HOST}:${SITE_A_PORT}"
  echo "  用户: ${SITE_A_ROOT_USER}"
  echo "  server-id: ${SITE_A_SERVER_ID}"
  echo
  echo "[从库 / 机房 B] ${SITE_B_NAME}"
  echo "  地址: ${SITE_B_HOST}:${SITE_B_PORT}"
  echo "  用户: ${SITE_B_ROOT_USER}"
  echo "  server-id: ${SITE_B_SERVER_ID}"
  echo
  echo "复制账号: ${REPL_USER}"
  echo "同步库: ${SYNC_DATABASES:-<全部>}"
  echo "同步表: ${SYNC_TABLES:-<整库>}"
  echo "配置文件: ${CONF_FILE}"
}

# ==================== MySQL 操作 ====================

site_var() {
  local site="$1"
  local field="$2"
  local name="SITE_${site}_${field}"
  echo "${!name}"
}

mysql_site() {
  local site="$1"
  shift
  mysql --batch \
    -h"$(site_var "$site" HOST)" \
    -P"$(site_var "$site" PORT)" \
    -u"$(site_var "$site" ROOT_USER)" \
    --password="$(site_var "$site" ROOT_PASSWORD)" \
    --connect-timeout="${MYSQL_CONNECT_TIMEOUT:-30}" \
    -N -s "$@"
}

mysql_site_raw() {
  local site="$1"
  shift
  mysql --batch \
    -h"$(site_var "$site" HOST)" \
    -P"$(site_var "$site" PORT)" \
    -u"$(site_var "$site" ROOT_USER)" \
    --password="$(site_var "$site" ROOT_PASSWORD)" \
    --connect-timeout="${MYSQL_CONNECT_TIMEOUT:-30}" \
    "$@"
}

site_label() {
  local site="$1"
  echo "$(site_var "$site" NAME) ($(site_var "$site" HOST):$(site_var "$site" PORT))"
}

peer_site() {
  [[ "$1" == "A" ]] && echo "B" || echo "A"
}

check_deps() {
  command -v mysql >/dev/null 2>&1 || die "未找到 mysql 客户端"
  command -v mysqldump >/dev/null 2>&1 || die "未找到 mysqldump"
}

check_connectivity() {
  log "检测主库 $(site_label A)..."
  mysql_site A -e "SELECT 1" >/dev/null || die "无法连接主库"
  log "检测从库 $(site_label B)..."
  mysql_site B -e "SELECT 1" >/dev/null || die "无法连接从库"
  log "连接检测通过"
}

# ==================== 版本检测 ====================

declare -A SITE_MYSQL_VERSION=()

version_ge() {
  printf '%s\n%s\n' "$2" "$1" | sort -V -C 2>/dev/null
}

mysql_global_var() {
  local site="$1"
  local var="$2"
  mysql_site "$site" -e "SELECT @@GLOBAL.${var}" 2>/dev/null | tr -d '\r'
}

cache_mysql_versions() {
  local site ver
  for site in A B; do
    ver="$(mysql_site "$site" -e "SELECT VERSION()" | tr -d '\r')"
    SITE_MYSQL_VERSION["$site"]="$ver"
  done
}

detect_and_show_versions() {
  cache_mysql_versions
  log "========== MySQL 版本 =========="
  log "主库 $(site_label A): ${SITE_MYSQL_VERSION[A]}"
  log "从库 $(site_label B): ${SITE_MYSQL_VERSION[B]}"

  local major_a major_b
  major_a="$(echo "${SITE_MYSQL_VERSION[A]}" | grep -oE '^[0-9]+\.[0-9]+' | head -1)"
  major_b="$(echo "${SITE_MYSQL_VERSION[B]}" | grep -oE '^[0-9]+\.[0-9]+' | head -1)"

  if [[ "$major_a" != "$major_b" ]]; then
    log "WARN: 主从 MySQL 大版本不一致 (${major_a} vs ${major_b})，建议保持一致"
    return 1
  fi

  if ! version_ge "${SITE_MYSQL_VERSION[A]}" "5.7.0"; then
    log "WARN: MySQL 版本过低，建议 5.7+ 且启用 GTID"
    return 1
  fi

  log "版本检测通过"
  return 0
}

# ==================== my.cnf / 参数预检 ====================

check_var_eq() {
  local label="$1"
  local actual="$2"
  local expected="$3"
  if [[ "$actual" != "$expected" ]]; then
    log "  [FAIL] ${label}: 当前=${actual}，需要=${expected}"
    return 1
  fi
  log "  [ OK ] ${label}=${actual}"
  return 0
}

check_var_on() {
  local label="$1"
  local actual="$2"
  if [[ "$actual" == "ON" || "$actual" == "1" ]]; then
    log "  [ OK ] ${label}=${actual}"
    return 0
  fi
  log "  [FAIL] ${label}: 当前=${actual}，需要=ON 或 1"
  return 1
}

check_node_mycnf() {
  local site="$1"
  local role="$2"
  local issues=0
  local gtid enforce log_bin fmt sid ro log_updates

  log "---------- $(site_label "$site") [${role}] ----------"

  gtid="$(mysql_global_var "$site" gtid_mode)"
  enforce="$(mysql_global_var "$site" enforce_gtid_consistency)"
  log_bin="$(mysql_global_var "$site" log_bin)"
  fmt="$(mysql_global_var "$site" binlog_format)"
  sid="$(mysql_global_var "$site" server_id)"
  ro="$(mysql_global_var "$site" read_only)"
  log_updates="$(mysql_global_var "$site" log_slave_updates)"

  check_var_on "gtid_mode" "$gtid" || issues=$((issues + 1))
  check_var_on "enforce_gtid_consistency" "$enforce" || issues=$((issues + 1))
  check_var_on "log_bin" "$log_bin" || issues=$((issues + 1))
  check_var_eq "binlog_format" "$fmt" "ROW" || issues=$((issues + 1))

  local expect_sid
  expect_sid="$(site_var "$site" SERVER_ID)"
  check_var_eq "server_id" "$sid" "$expect_sid" || issues=$((issues + 1))

  if [[ "$role" == "slave" && "${SLAVE_READ_ONLY:-1}" == "1" ]]; then
    check_var_on "read_only" "$ro" || issues=$((issues + 1))
  fi

  if [[ "$role" == "dual" ]]; then
    check_var_on "log_slave_updates" "$log_updates" || issues=$((issues + 1))
  fi

  if [[ -n "${SYNC_TABLES:-}" ]] && ! version_ge "${SITE_MYSQL_VERSION[$site]:-8.0.0}" "8.0.0"; then
    log "  [WARN] 指定表同步需 MySQL 8.0+ 或手动在 my.cnf 配置 replicate-do-table"
    issues=$((issues + 1))
  fi

  return "$issues"
}

preflight_check() {
  local strict="${STRICT_MYCNF_CHECK:-0}"
  local ver_ok mycnf_ok total=0

  log "========== 部署前环境预检 =========="
  detect_and_show_versions && ver_ok=0 || ver_ok=1
  [[ "$ver_ok" -eq 0 ]] || total=$((total + 1))

  log "========== my.cnf / 运行时参数检查 =========="
  local issues_a issues_b
  if is_master_slave; then
    check_node_mycnf A "master"; issues_a=$?
    check_node_mycnf B "slave"; issues_b=$?
  else
    check_node_mycnf A "dual"; issues_a=$?
    check_node_mycnf B "dual"; issues_b=$?
  fi
  mycnf_ok=$((issues_a + issues_b))

  if [[ "$mycnf_ok" -gt 0 ]]; then
    log "WARN: 发现 ${mycnf_ok} 项参数不符合要求"
    log "      可先执行菜单「9) 输出 my.cnf 参考」，配置后重启 MySQL"
    total=$((total + mycnf_ok))
  else
    log "参数检查全部通过"
  fi

  if [[ -n "${SYNC_TABLES:-}" ]]; then
    log "表级同步: ${SYNC_TABLES}"
    log "      增量复制将仅同步以上表（需 MySQL 8.0+ CHANGE REPLICATION FILTER）"
  fi

  if [[ "$total" -gt 0 && "$strict" == "1" ]]; then
    die "预检未通过（STRICT_MYCNF_CHECK=1），请修正后重试"
  fi

  if [[ "$total" -gt 0 ]]; then
    log "WARN: 预检有问题但将继续（可在配置中设 STRICT_MYCNF_CHECK=1 强制拦截）"
    return 1
  fi

  log "========== 预检通过 =========="
  return 0
}

menu_preflight() {
  load_config
  check_deps
  check_connectivity
  preflight_check || true
}

repl_status_field() {
  local site="$1"
  local field="$2"
  local sql="SHOW SLAVE STATUS"
  if version_ge "${SITE_MYSQL_VERSION[$site]:-5.7.0}" "8.0.22"; then
    sql="SHOW REPLICA STATUS"
  fi
  mysql_site_raw "$site" -e "${sql}\G" \
    | grep -F "${field}:" | head -1 | awk -F': ' '{print $2}' | tr -d '\r'
}

stop_replica_sql() {
  local site="$1"
  if version_ge "${SITE_MYSQL_VERSION[$site]:-5.7.0}" "8.0.22"; then
    mysql_site_raw "$site" -e "STOP REPLICA;" 2>/dev/null \
      || mysql_site_raw "$site" -e "STOP SLAVE;"
  else
    mysql_site_raw "$site" -e "STOP SLAVE;"
  fi
}

start_replica_sql() {
  local site="$1"
  if version_ge "${SITE_MYSQL_VERSION[$site]:-5.7.0}" "8.0.22"; then
    mysql_site_raw "$site" -e "START REPLICA;" 2>/dev/null \
      || mysql_site_raw "$site" -e "START SLAVE;"
  else
    mysql_site_raw "$site" -e "START SLAVE;"
  fi
}

reset_replica_sql() {
  local site="$1"
  if version_ge "${SITE_MYSQL_VERSION[$site]:-5.7.0}" "8.0.22"; then
    mysql_site_raw "$site" -e "RESET REPLICA ALL;" 2>/dev/null \
      || mysql_site_raw "$site" -e "RESET SLAVE ALL;"
  else
    mysql_site_raw "$site" -e "RESET SLAVE ALL;"
  fi
}

apply_replication_filters() {
  local site="$1"
  [[ -n "${SYNC_TABLES:-}" ]] || return 0

  if ! version_ge "${SITE_MYSQL_VERSION[$site]:-5.7.0}" "8.0.0"; then
    log "WARN: $(site_label "$site") MySQL 5.7 请在 my.cnf 添加 replicate-do-table 后重启"
    return 0
  fi

  local table_list="" spec
  for spec in $SYNC_TABLES; do
    [[ -n "$table_list" ]] && table_list+=","
    table_list+="${spec}"
  done

  log "在 $(site_label "$site") 设置表级复制过滤: ${table_list}"
  mysql_site_raw "$site" -e "CHANGE REPLICATION FILTER REPLICATE_DO_TABLE = (${table_list});"
}

enable_semi_sync() {
  local site="$1"
  mysql_site_raw "$site" -e "INSTALL PLUGIN rpl_semi_sync_master SONAME 'semisync_master.so';" 2>/dev/null || true
  mysql_site_raw "$site" -e "INSTALL PLUGIN rpl_semi_sync_slave SONAME 'semisync_slave.so';" 2>/dev/null || true
  mysql_site_raw "$site" -e "SET GLOBAL rpl_semi_sync_master_enabled = 1;"
  mysql_site_raw "$site" -e "SET GLOBAL rpl_semi_sync_slave_enabled = 1;"
  mysql_site_raw "$site" -e "SET GLOBAL rpl_semi_sync_master_timeout = ${SEMI_SYNC_TIMEOUT_MS:-30000};"
}

setup_master_node() {
  local allow_host="${SITE_A_REPL_ALLOW_HOST:-%}"
  local repl_pass
  repl_pass="$(sql_escape "$REPL_PASSWORD")"
  log "配置主库 $(site_label A)..."

  mysql_site_raw A <<EOF
SET GLOBAL gtid_mode = ON;
SET GLOBAL enforce_gtid_consistency = ON;
SET GLOBAL read_only = 0;
SET GLOBAL super_read_only = 0;
CREATE USER IF NOT EXISTS '${REPL_USER}'@'${allow_host}' IDENTIFIED BY '${repl_pass}';
GRANT REPLICATION SLAVE, REPLICATION CLIENT ON *.* TO '${REPL_USER}'@'${allow_host}';
FLUSH PRIVILEGES;
EOF

  if [[ "${ENABLE_SEMI_SYNC:-0}" == "1" ]]; then
    enable_semi_sync A
  fi
}

setup_slave_node() {
  local read_only="${SLAVE_READ_ONLY:-1}"
  log "配置从库 $(site_label B)..."

  mysql_site_raw B <<EOF
SET GLOBAL gtid_mode = ON;
SET GLOBAL enforce_gtid_consistency = ON;
SET GLOBAL read_only = ${read_only};
SET GLOBAL super_read_only = ${read_only};
EOF

  if [[ "${ENABLE_SEMI_SYNC:-0}" == "1" ]]; then
    enable_semi_sync B
  fi
}

setup_dual_node() {
  local site="$1"
  local allow_host offset increment repl_pass
  allow_host="$(site_var "$site" REPL_ALLOW_HOST)"
  offset="$(site_var "$site" AUTO_INCREMENT_OFFSET)"
  increment="${AUTO_INCREMENT_INCREMENT:-2}"
  repl_pass="$(sql_escape "$REPL_PASSWORD")"

  log "配置节点 $(site_label "$site")..."

  mysql_site_raw "$site" <<EOF
SET GLOBAL gtid_mode = ON;
SET GLOBAL enforce_gtid_consistency = ON;
SET GLOBAL auto_increment_increment = ${increment};
SET GLOBAL auto_increment_offset = ${offset};
SET GLOBAL read_only = 0;
SET GLOBAL super_read_only = 0;
CREATE USER IF NOT EXISTS '${REPL_USER}'@'${allow_host}' IDENTIFIED BY '${repl_pass}';
GRANT REPLICATION SLAVE, REPLICATION CLIENT ON *.* TO '${REPL_USER}'@'${allow_host}';
FLUSH PRIVILEGES;
EOF

  if [[ "${ENABLE_SEMI_SYNC:-0}" == "1" ]]; then
    enable_semi_sync "$site"
  fi
}

configure_replication_on() {
  local site="$1"
  local peer repl_pass
  peer="$(peer_site "$site")"
  repl_pass="$(sql_escape "$REPL_PASSWORD")"
  log "在 $(site_label "$site") 上指向 $(site_label "$peer") ..."

  stop_replica_sql "$site"
  reset_replica_sql "$site"
  apply_replication_filters "$site"

  mysql_site_raw "$site" <<EOF
CHANGE MASTER TO
  MASTER_HOST='$(site_var "$peer" HOST)',
  MASTER_PORT=$(site_var "$peer" PORT),
  MASTER_USER='${REPL_USER}',
  MASTER_PASSWORD='${repl_pass}',
  MASTER_AUTO_POSITION=1,
  MASTER_CONNECT_RETRY=10;
EOF
  start_replica_sql "$site"
}

start_replication() {
  if is_master_slave; then
    log "启动主从复制（主→从）..."
    configure_replication_on B
  else
    log "启动双主复制（双向）..."
    configure_replication_on A
    configure_replication_on B
  fi
  sleep 2
  show_status
}

stop_replication() {
  if is_master_slave; then
    stop_replica_sql B && log "已停止从库复制"
  else
    stop_replica_sql A && log "已停止 $(site_label A) 复制"
    stop_replica_sql B && log "已停止 $(site_label B) 复制"
  fi
}

list_sync_databases() {
  local site="${1:-A}"
  if [[ -n "${SYNC_TABLES:-}" ]]; then
    local spec db
    for spec in $SYNC_TABLES; do
      db="${spec%%.*}"
      echo "$db"
    done | sort -u
    return
  fi
  if [[ -n "${SYNC_DATABASES:-}" ]]; then
    echo "$SYNC_DATABASES"
    return
  fi
  mysql_site "$site" -e "SHOW DATABASES" \
    | grep -Ev '^(Database|information_schema|performance_schema|mysql|sys)$' || true
}

dump_table_specs() {
  local site="$1"
  local db tbl spec
  if [[ -n "${SYNC_TABLES:-}" ]]; then
    for spec in $SYNC_TABLES; do
      echo "$spec"
    done
    return
  fi
  while read -r db; do
    [[ -n "$db" ]] || continue
    echo "${db}.*"
  done < <(list_sync_databases "$site")
}

initial_dump() {
  local direction="${INITIAL_SYNC_DIRECTION:-A_TO_B}"
  if [[ "$direction" == "skip" ]]; then
    log "跳过全量同步"
    return 0
  fi

  local src dst
  case "$direction" in
    A_TO_B) src="A"; dst="B" ;;
    B_TO_A) src="B"; dst="A" ;;
    *) die "INITIAL_SYNC_DIRECTION 无效: $direction" ;;
  esac

  log "全量同步: $(site_label "$src") → $(site_label "$dst")"
  mkdir -p "$DUMP_DIR"

  local dump_opts=(
    -h"$(site_var "$src" HOST)"
    -P"$(site_var "$src" PORT)"
    -u"$(site_var "$src" ROOT_USER)"
    --password="$(site_var "$src" ROOT_PASSWORD)"
    --single-transaction --master-data=2 --set-gtid-purged=ON
    --routines --triggers --events --hex-blob
  )

  local spec db tbl outfile imported_dbs=""
  while read -r spec; do
    [[ -n "$spec" ]] || continue
    if [[ "$spec" == *.* && "$spec" != *".*" ]]; then
      db="${spec%%.*}"
      tbl="${spec#*.}"
      outfile="${DUMP_DIR}/${db}.${tbl}.sql"
      log "导出表: ${db}.${tbl}"
      mysqldump "${dump_opts[@]}" "$db" "$tbl" > "$outfile"
      imported_dbs+=" ${db}"
    else
      db="${spec%%.*}"
      outfile="${DUMP_DIR}/${db}.sql"
      log "导出库: ${db}"
      mysqldump "${dump_opts[@]}" "$db" > "$outfile"
      imported_dbs+=" ${db}"
    fi
  done < <(dump_table_specs "$src")

  local seen="" db
  for db in $imported_dbs; do
    [[ " ${seen} " == *" ${db} "* ]] && continue
    seen+=" ${db}"
    if [[ -n "${SYNC_TABLES:-}" ]]; then
      mysql_site_raw "$dst" -e "CREATE DATABASE IF NOT EXISTS \`${db}\`;"
      while read -r spec; do
        [[ -n "$spec" ]] || continue
        [[ "${spec%%.*}" == "$db" ]] || continue
        tbl="${spec#*.}"
        [[ -f "${DUMP_DIR}/${db}.${tbl}.sql" ]] || continue
        log "导入表: ${db}.${tbl} → $(site_label "$dst")"
        mysql_site_raw "$dst" < "${DUMP_DIR}/${db}.${tbl}.sql"
      done < <(dump_table_specs "$src")
    else
      [[ -f "${DUMP_DIR}/${db}.sql" ]] || continue
      log "导入库: ${db} → $(site_label "$dst")"
      mysql_site_raw "$dst" < "${DUMP_DIR}/${db}.sql"
    fi
  done

  log "全量同步完成: $DUMP_DIR"
}

show_site_status() {
  local site="$1"
  local sql="SHOW SLAVE STATUS"
  if version_ge "${SITE_MYSQL_VERSION[$site]:-5.7.0}" "8.0.22"; then
    sql="SHOW REPLICA STATUS"
  fi
  log "---- $(site_label "$site") ----"
  mysql_site_raw "$site" -e "${sql}\G" \
    | grep -E 'Master_Host|Master_Port|Slave_IO_Running|Slave_SQL_Running|Replica_IO_Running|Replica_SQL_Running|Seconds_Behind_Master|Last_IO_Error|Last_SQL_Error' \
    || log "  无复制信息"
}

show_status() {
  log "========== $(sync_mode_label) 状态 =========="
  if is_master_slave; then
    show_site_status B
  else
    show_site_status A
    show_site_status B
  fi
}

check_site_health() {
  local site="$1"
  local io sql lag warn
  warn="${REPL_LAG_WARN_SECONDS:-300}"
  io="$(repl_status_field "$site" "Slave_IO_Running")"
  [[ -z "$io" || "$io" == "NULL" ]] && io="$(repl_status_field "$site" "Replica_IO_Running")"
  sql="$(repl_status_field "$site" "Slave_SQL_Running")"
  [[ -z "$sql" || "$sql" == "NULL" ]] && sql="$(repl_status_field "$site" "Replica_SQL_Running")"
  lag="$(repl_status_field "$site" "Seconds_Behind_Master")"
  log "$(site_label "$site"): IO=${io:-N/A} SQL=${sql:-N/A} 延迟=${lag:-N/A}s"
  [[ "$io" == "Yes" && "$sql" == "Yes" ]] || return 1
  [[ -n "$lag" && "$lag" != "NULL" && "$lag" =~ ^[0-9]+$ && "$lag" -gt "$warn" ]] && return 1
  return 0
}

check_health() {
  load_config
  check_deps
  check_connectivity
  cache_mysql_versions
  local ok=0
  if is_master_slave; then
    check_site_health B || ok=1
  else
    check_site_health A || ok=1
    check_site_health B || ok=1
  fi
  if [[ "$ok" -eq 0 ]]; then
    log "复制健康"
    return 0
  fi
  log "ERROR: 复制异常，请选菜单「4) 查看复制状态」"
  return 1
}

print_cnf() {
  load_config
  local days="${BINLOG_RETENTION_DAYS:-7}"
  local seconds=$((days * 86400))
  local inc="${AUTO_INCREMENT_INCREMENT:-2}"
  local table_filter=""
  if [[ -n "${SYNC_TABLES:-}" ]]; then
    table_filter="# 表级同步（MySQL 5.7 需使用 replicate-do-table）"
    local spec
    for spec in $SYNC_TABLES; do
      table_filter+="
replicate-do-table = ${spec}"
    done
  fi
  cat <<EOF
# 主库 $(site_var A NAME) - ${SITE_A_HOST}
[mysqld]
server-id = $(site_var A SERVER_ID)
log-bin = mysql-bin
relay-log = relay-bin
binlog_format = ROW
gtid_mode = ON
enforce_gtid_consistency = ON
binlog_expire_logs_seconds = ${seconds}
sync_binlog = 1
slave_net_timeout = 120
max_allowed_packet = 64M
log_slave_updates = 1
${table_filter}
$(is_dual_master && echo "auto_increment_increment = ${inc}
auto_increment_offset = $(site_var A AUTO_INCREMENT_OFFSET)")

# 从库 $(site_var B NAME) - ${SITE_B_HOST}
[mysqld]
server-id = $(site_var B SERVER_ID)
log-bin = mysql-bin
relay-log = relay-bin
binlog_format = ROW
gtid_mode = ON
enforce_gtid_consistency = ON
binlog_expire_logs_seconds = ${seconds}
log_slave_updates = 1
${table_filter}
$(is_master_slave && echo "read_only = 1
super_read_only = 1")
$(is_dual_master && echo "auto_increment_increment = ${inc}
auto_increment_offset = $(site_var B AUTO_INCREMENT_OFFSET)")
EOF
}

cmd_init() {
  load_config
  check_deps
  check_connectivity
  cache_mysql_versions
  preflight_check || true
  log "========== 开始部署: $(sync_mode_label) =========="
  if is_master_slave; then
    setup_master_node
    setup_slave_node
  else
    setup_dual_node A
    setup_dual_node B
  fi
  initial_dump
  start_replication
  log "========== 部署完成 =========="
}

cmd_resync() {
  load_config
  check_deps
  check_connectivity
  cache_mysql_versions
  stop_replication
  initial_dump
  start_replication
  log "重新同步完成"
}

# ==================== 主菜单 ====================

show_main_menu() {
  clear_screen
  echo "=============================================="
  echo "       MySQL 跨机房实时同步管理菜单"
  echo "=============================================="
  if load_config_raw 2>/dev/null; then
    echo "当前模式: $(sync_mode_label)"
    echo "主库: ${SITE_A_HOST}:${SITE_A_PORT}  |  从库: ${SITE_B_HOST}:${SITE_B_PORT}"
  else
    echo "当前模式: 未配置"
  fi
  echo "----------------------------------------------"
  echo "  1) 配置向导（选择主从/双主 + 填写主副库）"
  echo "  2) 查看当前配置"
  echo "  3) 初始化部署"
  echo "  4) 查看复制状态"
  echo "  5) 启动复制"
  echo "  6) 停止复制"
  echo "  7) 重新全量同步"
  echo "  8) 健康检查"
  echo "  9) 输出 my.cnf 参考"
  echo " 10) 环境预检（版本 + my.cnf 参数）"
  echo "  0) 退出"
  echo "=============================================="
}

menu_main() {
  [[ -t 0 && -t 1 ]] || die "交互菜单需要终端，请直接运行: ./mysql_sync.sh"
  set +e
  while true; do
    show_main_menu
    local choice=""
    read -r -p "请选择 [0-10]: " choice </dev/tty
    case "$choice" in
      1) menu_config_wizard; pause_enter ;;
      2) menu_show_config; pause_enter ;;
      3) cmd_init; pause_enter ;;
      4) load_config; check_deps; cache_mysql_versions; show_status; pause_enter ;;
      5) load_config; check_deps; cache_mysql_versions; start_replication; pause_enter ;;
      6) load_config; check_deps; cache_mysql_versions; stop_replication; pause_enter ;;
      7) cmd_resync; pause_enter ;;
      8) load_config; check_deps; cache_mysql_versions; check_health; pause_enter ;;
      9) print_cnf; pause_enter ;;
     10) menu_preflight; pause_enter ;;
      0) echo "再见"; exit 0 ;;
      *) echo "无效选项"; sleep 1 ;;
    esac
  done
}

usage() {
  cat <<EOF
用法:
  $0              进入交互菜单（推荐）
  $0 menu         进入交互菜单
  $0 init|status|start|stop|resync|check|print-cnf|preflight

配置文件: ${CONF_FILE}
EOF
}

main() {
  local cmd="${1:-menu}"
  case "$cmd" in
    menu|"")   menu_main ;;
    init)      cmd_init ;;
    status)    load_config; check_deps; cache_mysql_versions; show_status ;;
    start)     load_config; check_deps; cache_mysql_versions; start_replication ;;
    stop)      load_config; check_deps; cache_mysql_versions; stop_replication ;;
    resync)    cmd_resync ;;
    check)     check_health || exit 1 ;;
    preflight) menu_preflight ;;
    print-cnf) print_cnf ;;
    config)    menu_config_wizard ;;
    -h|--help) usage ;;
    *)         usage; exit 1 ;;
  esac
}

main "$@"