#!/usr/bin/env bash
set -euo pipefail

# Uso:
#   ALLOW_DROP_DATABASE=1 ./mysqlsh_restore_hardened.sh <DUMP_DIR> <DATABASE> [EXCLUDE_TABLES] [SKIP_VIEWS_TRIGGERS]
# Ejemplos:
#   ALLOW_DROP_DATABASE=1 ./mysqlsh_restore_hardened.sh /backups/mbinv_20260602_010000 mbinv "" 1
#   INCLUDE_TABLES='"'"'mbinv.tabla1'"'"','"'"'origen.tabla1'"'"'' ALLOW_DROP_DATABASE=1 ./mysqlsh_restore_hardened.sh /backups/dump mbinv "" 1
#
# Carga híbrida con detección automática de tablas anchas:
#   - Primero se intenta `util.loadDump()` (MySQL Shell nativo) para TODO el
#     dump. Es la única vía que decodifica correctamente las columnas
#     BLOB/BINARY: MySQL Shell envuelve su contenido en base64 dentro de los
#     chunks .tsv para transportarlo de forma segura, y solo su propio
#     cargador sabe revertir eso. Un `LOAD DATA LOCAL INFILE` genérico lo
#     deja el base64 crudo metido en la columna (bug encontrado 2026-07-11:
#     fotos, fotos_detalle, parametros_imagenes*, touch_boton/menu,
#     huella_maestro, etc. quedaban con el binario corrupto tras cada
#     restore).
#   - En MySQL 8.0.46-cloud (versión forzada por OCI), `util.loadDump()`
#     truena con "Row size too large (> 8126)" en tablas con demasiadas
#     columnas VARCHAR anchas — varía por cliente (en una base pueden ser 3
#     tablas, en otra 7, con nombres distintos) y depende de si esa base ya
#     tiene aplicado el fix a nivel de proyecto (pasar esas columnas a
#     TEXT). En vez de mantener una lista fija a mano, el script detecta la
#     tabla exacta que falló en el mensaje de error, la excluye, y
#     reintenta `util.loadDump()` — repite hasta que cargue todo o hasta
#     MAX_AUTO_RETRIES intentos. Las tablas así descubiertas se cargan
#     aparte con el método "old-school" (LOAD DATA LOCAL INFILE).
#   - WIDE_TABLES (opcional, CSV) sigue existiendo como semilla: tablas que
#     ya se sabe que van a fallar, para no perder un intento de
#     util.loadDump() en descubrirlas de nuevo. No hace falta mantenerla
#     actualizada — es solo una optimización, no la fuente de verdad.
#
# Variables de entorno adicionales:
#   LOAD_THREADS=8            Hilos para util.loadDump() en la Fase 2 (1-64)
#   CREATE_INVISIBLE_PKS=1    Pasa createInvisiblePKs a util.loadDump() (0/1)
#   MAX_AUTO_RETRIES=25       Máximo de reintentos de util.loadDump() al ir
#                             descubriendo tablas anchas una por una
#   ENV_FILE=/home/macrobase/backup_mysql.env
#                             Credenciales de SendGrid para el correo de resumen
#                             (RECIPIENT_EMAIL, SENDGRID_API_KEY, FROM_EMAIL)

DUMP_DIR="${1:-}"
DATABASE="${2:-}"
EXCLUDE_TABLES="${3:-}"
SKIP_VIEWS_TRIGGERS="${4:-0}"  # 1 = omitir vistas/triggers/.post.sql en la fase old-school
INCLUDE_TABLES="${INCLUDE_TABLES:-}"
WIDE_TABLES="${WIDE_TABLES:-}"
LOAD_THREADS="${LOAD_THREADS:-8}"
CREATE_INVISIBLE_PKS="${CREATE_INVISIBLE_PKS:-1}"
MAX_AUTO_RETRIES="${MAX_AUTO_RETRIES:-25}"

if [[ ! "$MAX_AUTO_RETRIES" =~ ^[0-9]+$ ]] || [[ "$MAX_AUTO_RETRIES" -lt 1 ]]; then
  echo "Error: MAX_AUTO_RETRIES debe ser un entero mayor a 0"
  exit 1
fi

if [[ ! "$LOAD_THREADS" =~ ^[0-9]+$ ]] || [[ "$LOAD_THREADS" -lt 1 ]] || [[ "$LOAD_THREADS" -gt 64 ]]; then
  echo "Error: LOAD_THREADS debe ser un entero entre 1 y 64"
  exit 1
fi
if [[ "$CREATE_INVISIBLE_PKS" != "0" && "$CREATE_INVISIBLE_PKS" != "1" ]]; then
  echo "Error: CREATE_INVISIBLE_PKS debe ser 0 o 1"
  exit 1
fi

# Correo de resumen (mismo mecanismo que backup_produccion_optimizado_v12.sh:
# SendGrid vía curl). Si no hay credenciales configuradas, simplemente no se
# envía nada — el resto del script funciona igual.
ENV_FILE="${ENV_FILE:-/home/macrobase/backup_mysql.env}"
if [[ -f "$ENV_FILE" ]]; then
  set -a
  # shellcheck disable=SC1090
  . "$ENV_FILE"
  set +a
fi
RECIPIENT_EMAIL="${RECIPIENT_EMAIL:-}"
SENDGRID_API_KEY="${SENDGRID_API_KEY:-}"
FROM_EMAIL="${FROM_EMAIL:-}"

format_hms() {
  local total_seconds="$1"
  printf '%02d:%02d:%02d' $((total_seconds/3600)) $((total_seconds%3600/60)) $((total_seconds%60))
}

send_email() {
  local subject="$1"
  local body="$2"

  if [[ -z "$SENDGRID_API_KEY" || -z "$RECIPIENT_EMAIL" || -z "$FROM_EMAIL" ]]; then
    echo "Aviso: no se envía correo porque falta SENDGRID_API_KEY, RECIPIENT_EMAIL o FROM_EMAIL"
    return 0
  fi

  if ! command -v python3 >/dev/null 2>&1 || ! command -v curl >/dev/null 2>&1; then
    echo "Aviso: no se envía correo porque falta python3 o curl"
    return 0
  fi

  python3 - "$subject" "$body" "$RECIPIENT_EMAIL" "$FROM_EMAIL" <<'PY' | \
  curl --silent --show-error --fail \
    --connect-timeout "${EMAIL_CONNECT_TIMEOUT:-10}" \
    --max-time "${EMAIL_MAX_TIME:-60}" \
    --request POST \
    --url https://api.sendgrid.com/v3/mail/send \
    --header "Authorization: Bearer ${SENDGRID_API_KEY}" \
    --header 'Content-Type: application/json' \
    --data-binary @- || echo "Aviso: fallo el envío de correo (curl/SendGrid)"
import html, json, sys
subject, body, to, from_email = sys.argv[1:5]
html_body = "<pre style=\"font-family: monospace, monospace;\">" + html.escape(body) + "</pre>"
print(json.dumps({
  "personalizations": [{"to": [{"email": to}]}],
  "from": {"email": from_email},
  "subject": subject,
  "content": [{"type": "text/html", "value": html_body}]
}))
PY
}

if [[ -z "$DUMP_DIR" || -z "$DATABASE" ]]; then
  echo "Uso: $0 <DUMP_DIR> <DATABASE> [EXCLUDE_TABLES] [SKIP_VIEWS_TRIGGERS]"
  exit 1
fi

if [[ ! -d "$DUMP_DIR" ]]; then
  echo "Error: no existe el directorio de dump: $DUMP_DIR"
  exit 1
fi

if [[ "$SKIP_VIEWS_TRIGGERS" != "0" && "$SKIP_VIEWS_TRIGGERS" != "1" ]]; then
  echo "Error: SKIP_VIEWS_TRIGGERS debe ser 0 o 1"
  exit 1
fi

ALLOW_DROP_DATABASE="${ALLOW_DROP_DATABASE:-0}"
RESTORE_RESUME="${RESTORE_RESUME:-0}"
START_FROM_TABLE="${START_FROM_TABLE:-}"
MYSQLSH_BIN="${MYSQLSH_BIN:-mysqlsh}"

# Conexion (se puede sobreescribir por env o MYSQL_CNF)
DB_USER="${DB_USER:-manuel}"
DB_PASSWORD="${DB_PASSWORD:-}"
DB_HOST="${DB_HOST:-}"
DB_PORT="${DB_PORT:-3306}"
MYSQL_CNF="${MYSQL_CNF:-}"

if [[ -n "$MYSQL_CNF" ]]; then
  if [[ ! -f "$MYSQL_CNF" ]]; then
    echo "Error: MYSQL_CNF no existe: $MYSQL_CNF"
    exit 1
  fi
  CNF_DB_USER="$(awk -F= '/^[[:space:]]*user[[:space:]]*=/{gsub(/^[[:space:]]+|[[:space:]]+$/, "", $2); print $2; exit}' "$MYSQL_CNF" || true)"
  CNF_DB_PASSWORD="$(awk -F= '/^[[:space:]]*password[[:space:]]*=/{gsub(/^[[:space:]]+|[[:space:]]+$/, "", $2); print $2; exit}' "$MYSQL_CNF" || true)"
  CNF_DB_HOST="$(awk -F= '/^[[:space:]]*host[[:space:]]*=/{gsub(/^[[:space:]]+|[[:space:]]+$/, "", $2); print $2; exit}' "$MYSQL_CNF" || true)"
  CNF_DB_PORT="$(awk -F= '/^[[:space:]]*port[[:space:]]*=/{gsub(/^[[:space:]]+|[[:space:]]+$/, "", $2); print $2; exit}' "$MYSQL_CNF" || true)"

  [[ -z "$DB_USER" && -n "$CNF_DB_USER" ]] && DB_USER="$CNF_DB_USER"
  [[ -z "$DB_PASSWORD" && -n "$CNF_DB_PASSWORD" ]] && DB_PASSWORD="$CNF_DB_PASSWORD"
  [[ -z "$DB_HOST" && -n "$CNF_DB_HOST" ]] && DB_HOST="$CNF_DB_HOST"
  [[ -z "$DB_PORT" && -n "$CNF_DB_PORT" ]] && DB_PORT="$CNF_DB_PORT"
fi

if [[ -z "$DB_USER" || -z "$DB_HOST" || -z "$DB_PORT" ]]; then
  echo "Error: faltan DB_USER/DB_HOST/DB_PORT. DB_HOST es obligatorio y explícito"
  echo "(sin valor por defecto a propósito: nunca debe adivinar un servidor si la"
  echo "variable de entorno no llegó — evita restaurar/borrar en el server equivocado)."
  exit 1
fi
if [[ -z "$DB_PASSWORD" ]]; then
  echo "Error: DB_PASSWORD vacío. Define DB_PASSWORD o MYSQL_CNF"
  exit 1
fi

if ! command -v mysql >/dev/null 2>&1; then
  echo "Error: no se encontró el cliente mysql"
  exit 1
fi
if ! command -v "$MYSQLSH_BIN" >/dev/null 2>&1; then
  echo "Error: no se encontró MYSQLSH_BIN=$MYSQLSH_BIN"
  exit 1
fi
if ! command -v find >/dev/null 2>&1; then
  echo "Error: no se encontró find"
  exit 1
fi
if ! command -v awk >/dev/null 2>&1; then
  echo "Error: no se encontró awk"
  exit 1
fi
if ! command -v sed >/dev/null 2>&1; then
  echo "Error: no se encontró sed"
  exit 1
fi
if ! command -v tr >/dev/null 2>&1; then
  echo "Error: no se encontró tr"
  exit 1
fi
if ! command -v mktemp >/dev/null 2>&1; then
  echo "Error: no se encontró mktemp"
  exit 1
fi

TARGET_DB="$DATABASE"
export TZ="America/Guatemala"
RESTORE_START_TS="$(date +%s)"

if [[ ! "$TARGET_DB" =~ ^[A-Za-z0-9_]+$ ]]; then
  echo "Error: nombre de base de datos inválido: $TARGET_DB"
  exit 1
fi

case "$TARGET_DB" in
  mysql|sys|information_schema|performance_schema)
    echo "Error: base protegida, restore cancelado: $TARGET_DB"
    exit 1
    ;;
esac

DB_EXISTE="$(MYSQL_PWD="$DB_PASSWORD" mysql -h "$DB_HOST" -u "$DB_USER" -P "$DB_PORT" -Nse "SELECT SCHEMA_NAME FROM INFORMATION_SCHEMA.SCHEMATA WHERE SCHEMA_NAME='${TARGET_DB}' LIMIT 1;" 2>/dev/null || true)"
if [[ -n "$DB_EXISTE" && "$ALLOW_DROP_DATABASE" != "1" && "$RESTORE_RESUME" != "1" && -z "$START_FROM_TABLE" ]]; then
  echo "Error: la base '$TARGET_DB' ya existe en ${DB_HOST}:${DB_PORT}."
  echo "Para continuar con reemplazo seguro, ejecuta con ALLOW_DROP_DATABASE=1"
  exit 1
fi

normalize_list() {
  local raw="$1"
  printf '%s' "$raw" | tr -d " '\""
}

in_csv_list() {
  local needle="$1"
  local csv="$2"
  local item

  [[ -z "$csv" ]] && return 1

  IFS=',' read -r -a _items <<< "$csv"
  for item in "${_items[@]}"; do
    [[ -z "$item" ]] && continue
    if [[ "$item" == "$needle" ]]; then
      return 0
    fi
    if [[ "$item" == *.* ]]; then
      if [[ "${item##*.}" == "$needle" ]]; then
        return 0
      fi
    fi
  done

  return 1
}

INCLUDE_TABLES_NORM="$(normalize_list "$INCLUDE_TABLES")"
EXCLUDE_TABLES_NORM="$(normalize_list "$EXCLUDE_TABLES")"
START_FROM_TABLE_NORM="$(normalize_list "$START_FROM_TABLE")"
WIDE_TABLES_NORM="$(normalize_list "$WIDE_TABLES")"

checkpoint_key="${DUMP_DIR}|${TARGET_DB}|${INCLUDE_TABLES_NORM}|${EXCLUDE_TABLES_NORM}|${SKIP_VIEWS_TRIGGERS}"
checkpoint_key_safe="$(printf '%s' "$checkpoint_key" | tr -c 'A-Za-z0-9' '_' | cut -c1-160)"
CHECKPOINT_FILE="${RESTORE_CHECKPOINT_FILE:-/tmp/restore_checkpoint_${checkpoint_key_safe}.txt}"

if [[ "$RESTORE_RESUME" != "0" && "$RESTORE_RESUME" != "1" ]]; then
  echo "Error: RESTORE_RESUME debe ser 0 o 1"
  exit 1
fi

if [[ -n "$START_FROM_TABLE_NORM" ]] && [[ ! "$START_FROM_TABLE_NORM" =~ ^[A-Za-z0-9_]+$ ]]; then
  echo "Error: START_FROM_TABLE inválida: $START_FROM_TABLE_NORM"
  exit 1
fi

is_wide_table() {
  local table_name="$1"
  [[ -z "$WIDE_TABLES_NORM" ]] && return 1
  in_csv_list "$table_name" "$WIDE_TABLES_NORM"
}

should_skip_table() {
  local table_name="$1"

  if [[ -n "$INCLUDE_TABLES_NORM" ]]; then
    if ! in_csv_list "$table_name" "$INCLUDE_TABLES_NORM"; then
      return 0
    fi
  fi

  if [[ -n "$EXCLUDE_TABLES_NORM" ]]; then
    if in_csv_list "$table_name" "$EXCLUDE_TABLES_NORM"; then
      return 0
    fi
  fi

  return 1
}

list_data_files() {
  find "$DUMP_DIR" -type f \( -name "*@*@*.tsv.zst" -o -name "*@*@*.tsv" -o -name "*@*@*.tsv.gz" \) 2>/dev/null | sort -V
}

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

SOURCE_SCHEMA=""
if [[ -f "$DUMP_DIR/@.json" ]]; then
  SOURCE_SCHEMA="$(sed -nE 's/.*"schemas"[[:space:]]*:[[:space:]]*\[[[:space:]]*"([^"]+)".*/\1/p' "$DUMP_DIR/@.json" | sed -n '1p' || true)"
fi

if [[ -z "$SOURCE_SCHEMA" ]]; then
  first_data_file="$(find "$DUMP_DIR" -type f \( -name '*@*@*.tsv.zst' -o -name '*@*@*.tsv' -o -name '*@*@*.tsv.gz' \) | sort -V | sed -n '1p' || true)"
  if [[ -n "$first_data_file" ]]; then
    SOURCE_SCHEMA="$(basename "$first_data_file" | awk -F'@' '{print $1}')"
  fi
fi

echo "================================================================="
echo " RESTORE HÍBRIDO (util.loadDump + old-school para tablas anchas)"
echo "================================================================="
echo "Destino         : ${DB_USER}@${DB_HOST}:${DB_PORT}"
echo "Dump origen     : $DUMP_DIR"
echo "Base destino    : $TARGET_DB"
echo "Esquema origen  : ${SOURCE_SCHEMA:-desconocido}"
echo "Tablas anchas (semilla): ${WIDE_TABLES_NORM:-ninguna, se detectan solas}"
if [[ -n "$INCLUDE_TABLES_NORM" ]]; then
  echo "Include tables  : $INCLUDE_TABLES_NORM"
fi
if [[ -n "$EXCLUDE_TABLES_NORM" ]]; then
  echo "Exclude tables  : $EXCLUDE_TABLES_NORM"
fi
if [[ -n "$START_FROM_TABLE_NORM" ]]; then
  echo "Start from table: $START_FROM_TABLE_NORM"
fi
if [[ "$RESTORE_RESUME" == "1" ]]; then
  echo "Resume          : habilitado"
  echo "Checkpoint file : $CHECKPOINT_FILE"
fi
echo "-----------------------------------------------------------------"

if [[ "$RESTORE_RESUME" == "1" ]]; then
  if [[ ! -f "$CHECKPOINT_FILE" ]]; then
    : > "$CHECKPOINT_FILE"
  fi
  echo "-> Resume mode: conservando base existente y checkpoint actual"
  MYSQL_PWD="$DB_PASSWORD" mysql -h "$DB_HOST" -u "$DB_USER" -P "$DB_PORT" \
    -e "CREATE DATABASE IF NOT EXISTS \`$TARGET_DB\` CHARACTER SET utf8mb4 COLLATE utf8mb4_unicode_ci;"
else
  : > "$CHECKPOINT_FILE"
  echo "-> Inicializando base de datos limpia..."
  if [[ "$ALLOW_DROP_DATABASE" == "1" ]]; then
    MYSQL_PWD="$DB_PASSWORD" mysql -h "$DB_HOST" -u "$DB_USER" -P "$DB_PORT" \
      -e "DROP DATABASE IF EXISTS \`$TARGET_DB\`; CREATE DATABASE \`$TARGET_DB\` CHARACTER SET utf8mb4 COLLATE utf8mb4_unicode_ci;"
  else
    MYSQL_PWD="$DB_PASSWORD" mysql -h "$DB_HOST" -u "$DB_USER" -P "$DB_PORT" \
      -e "CREATE DATABASE IF NOT EXISTS \`$TARGET_DB\` CHARACTER SET utf8mb4 COLLATE utf8mb4_unicode_ci;"
  fi
  echo "   [OK] Base de datos vacía creada."
fi

# FASE 2: util.loadDump con detección automática de tablas anchas.
# Se intenta cargar todo; si falla por "Row size too large" en una tabla
# específica, se excluye esa tabla y se reintenta, hasta que cargue todo o
# se agoten los reintentos. Corre ANTES que la Fase 1 porque hasta que esto
# termina no se sabe con certeza qué tablas son "anchas" en esta base.
echo "-> Fase 2: Cargando con util.loadDump (detecta tablas anchas automáticamente)..."

PHASE2_STATUS="OK"
PHASE2_ELAPSED=0
DISCOVERED_WIDE_TABLES_NORM="$WIDE_TABLES_NORM"
DISCOVERED_DEPENDENT_VIEWS_NORM=""

if [[ "$RESTORE_RESUME" == "1" ]] && grep -Fxq "__PHASE2_LOADDUMP_DONE__" "$CHECKPOINT_FILE" 2>/dev/null; then
  echo "   [SKIP] Fase 2 omitida por checkpoint (ya se completó en una corrida previa)"
  PHASE2_STATUS="OMITIDA (checkpoint)"
  PREV_WIDE="$(grep '^__WIDE_TABLES_DISCOVERED__:' "$CHECKPOINT_FILE" 2>/dev/null | tail -1 | cut -d: -f2-)"
  [[ -n "$PREV_WIDE" ]] && DISCOVERED_WIDE_TABLES_NORM="$PREV_WIDE"
  PREV_VIEWS="$(grep '^__DEPENDENT_VIEWS_DISCOVERED__:' "$CHECKPOINT_FILE" 2>/dev/null | tail -1 | cut -d: -f2-)"
  [[ -n "$PREV_VIEWS" ]] && DISCOVERED_DEPENDENT_VIEWS_NORM="$PREV_VIEWS"
else
  qualify_table() {
    local item="$1"
    if [[ "$item" == *.* ]]; then
      printf '%s' "$item"
    else
      printf '%s.%s' "${SOURCE_SCHEMA:-$TARGET_DB}" "$item"
    fi
  }

  build_loaddump_js() {
    local out_file="$1"
    local loaddump_exclude_js="["
    local first=1

    if [[ -n "$EXCLUDE_TABLES_NORM" ]]; then
      IFS=',' read -r -a _ex_items <<< "$EXCLUDE_TABLES_NORM"
      for item in "${_ex_items[@]}"; do
        [[ -z "$item" ]] && continue
        [[ "$first" -eq 0 ]] && loaddump_exclude_js+=","
        loaddump_exclude_js+="$(js_quote "$(qualify_table "$item")")"
        first=0
      done
    fi
    if [[ -n "$DISCOVERED_WIDE_TABLES_NORM" ]]; then
      IFS=',' read -r -a _wide_items <<< "$DISCOVERED_WIDE_TABLES_NORM"
      for item in "${_wide_items[@]}"; do
        [[ -z "$item" ]] && continue
        [[ "$first" -eq 0 ]] && loaddump_exclude_js+=","
        loaddump_exclude_js+="$(js_quote "$(qualify_table "$item")")"
        first=0
      done
    fi
    if [[ -n "$DISCOVERED_DEPENDENT_VIEWS_NORM" ]]; then
      IFS=',' read -r -a _view_items <<< "$DISCOVERED_DEPENDENT_VIEWS_NORM"
      for item in "${_view_items[@]}"; do
        [[ -z "$item" ]] && continue
        [[ "$first" -eq 0 ]] && loaddump_exclude_js+=","
        loaddump_exclude_js+="$(js_quote "$(qualify_table "$item")")"
        first=0
      done
    fi
    loaddump_exclude_js+="]"

    local loaddump_include_js="null"
    if [[ -n "$INCLUDE_TABLES_NORM" ]]; then
      loaddump_include_js="["
      first=1
      IFS=',' read -r -a _inc_items <<< "$INCLUDE_TABLES_NORM"
      for item in "${_inc_items[@]}"; do
        [[ -z "$item" ]] && continue
        [[ "$first" -eq 0 ]] && loaddump_include_js+=","
        loaddump_include_js+="$(js_quote "$(qualify_table "$item")")"
        first=0
      done
      loaddump_include_js+="]"
    fi

    {
      printf 'var options = {\n'
      printf '  schema: %s,\n' "$(js_quote "$TARGET_DB")"
      printf '  ignoreVersion: true,\n'
      printf '  resetProgress: %s,\n' "$([[ "$RESTORE_RESUME" == "1" ]] && echo false || echo true)"
      printf '  showProgress: true,\n'
      printf '  threads: %s,\n' "$LOAD_THREADS"
      printf '  createInvisiblePKs: %s,\n' "$([[ "$CREATE_INVISIBLE_PKS" == "1" ]] && echo true || echo false)"
      printf '  excludeTables: %s\n' "$loaddump_exclude_js"
      printf '};\n'
      if [[ "$loaddump_include_js" != "null" ]]; then
        printf 'options.includeTables = %s;\n' "$loaddump_include_js"
      fi
      if [[ -n "$SOURCE_SCHEMA" && "$SOURCE_SCHEMA" != "$TARGET_DB" ]]; then
        printf 'options.includeSchemas = [%s];\n' "$(js_quote "$SOURCE_SCHEMA")"
      fi
      printf 'util.loadDump(%s, options);\n' "$(js_quote "$DUMP_DIR")"
    } > "$out_file"
  }

  fail_phase2() {
    local motivo="$1"
    RESTORE_END_TS="$(date +%s)"
    EMAIL_BODY="Restore de '$TARGET_DB' DETENIDO en Fase 2 (util.loadDump)."$'\n'
    EMAIL_BODY+="Dump origen: $DUMP_DIR"$'\n'
    EMAIL_BODY+="Motivo: $motivo"$'\n'
    EMAIL_BODY+="Tiempo hasta el fallo: $(format_hms "$((RESTORE_END_TS - RESTORE_START_TS))")"$'\n\n'
    EMAIL_BODY+="Revisar el log de la corrida en el servidor para el detalle del error de MySQL Shell."
    send_email "Restore FALLIDO: $TARGET_DB" "$EMAIL_BODY"
    exit 2
  }

  PHASE2_T0="$(date +%s)"
  attempt=0
  while :; do
    attempt=$((attempt + 1))
    if [[ "$attempt" -gt "$MAX_AUTO_RETRIES" ]]; then
      PHASE2_STATUS="FALLÓ"
      echo "   [ERROR] Se alcanzó MAX_AUTO_RETRIES ($MAX_AUTO_RETRIES) detectando tablas anchas."
      fail_phase2 "se alcanzó el máximo de $MAX_AUTO_RETRIES reintentos automáticos"
    fi

    if [[ "$attempt" -gt 1 && "$RESTORE_RESUME" != "1" ]]; then
      # Un intento previo puede haber cargado varias tablas antes de abortar
      # por la que falló (los workers de loadDump corren en paralelo). Sin
      # recrear la base, el reintento choca con "objeto ya existe" en esas
      # tablas que sí habían cargado bien.
      echo "   -> Reintentando: recreando base destino limpia antes del siguiente intento..."
      MYSQL_PWD="$DB_PASSWORD" mysql -h "$DB_HOST" -u "$DB_USER" -P "$DB_PORT" \
        -e "DROP DATABASE IF EXISTS \`$TARGET_DB\`; CREATE DATABASE \`$TARGET_DB\` CHARACTER SET utf8mb4 COLLATE utf8mb4_unicode_ci;"
    fi

    LOADDUMP_JS_FILE="$(mktemp /tmp/restore_loaddump_XXXXXX.js)"
    LOADDUMP_LOG="$(mktemp /tmp/restore_loaddump_log_XXXXXX.txt)"
    build_loaddump_js "$LOADDUMP_JS_FILE"

    if "$MYSQLSH_BIN" --host="$DB_HOST" --port="$DB_PORT" --user="$DB_USER" --password="$DB_PASSWORD" \
      --verbose=0 --js < "$LOADDUMP_JS_FILE" > "$LOADDUMP_LOG" 2>&1; then
      cat "$LOADDUMP_LOG"
      rm -f "$LOADDUMP_JS_FILE" "$LOADDUMP_LOG"
      break
    fi

    cat "$LOADDUMP_LOG"

    newly_found="$(grep -oE 'Error processing table `[^`]+`\.`[^`]+`: MySQL Error 1118' "$LOADDUMP_LOG" \
      | sed -E 's/.*`([^`]+)`: MySQL Error 1118/\1/' | sort -u || true)"
    newly_found_views="$(grep -oE 'Error executing DDL script for view `[^`]+`\.`[^`]+`: MySQL Error 1146' "$LOADDUMP_LOG" \
      | sed -E 's/.*for view `[^`]+`\.`([^`]+)`: MySQL Error 1146/\1/' | sort -u || true)"
    rm -f "$LOADDUMP_JS_FILE" "$LOADDUMP_LOG"

    if [[ -z "$newly_found" && -z "$newly_found_views" ]]; then
      PHASE2_STATUS="FALLÓ"
      echo "   [ERROR] util.loadDump falló por un motivo no reconocido (ni 'Row size too large' ni vista dependiente de tabla ancha) — no se puede auto-recuperar."
      fail_phase2 "error de util.loadDump no reconocido (ver log arriba)"
    fi

    added_any=0
    while IFS= read -r t; do
      [[ -z "$t" ]] && continue
      if ! in_csv_list "$t" "$DISCOVERED_WIDE_TABLES_NORM"; then
        echo "   -> Tabla ancha detectada automáticamente: $t (intento $attempt, reintentando sin ella)"
        DISCOVERED_WIDE_TABLES_NORM="${DISCOVERED_WIDE_TABLES_NORM:+$DISCOVERED_WIDE_TABLES_NORM,}$t"
        added_any=1
      fi
    done <<< "$newly_found"
    while IFS= read -r v; do
      [[ -z "$v" ]] && continue
      if ! in_csv_list "$v" "$DISCOVERED_DEPENDENT_VIEWS_NORM"; then
        echo "   -> Vista dependiente de tabla ancha detectada: $v (intento $attempt, se crea después de las tablas anchas)"
        DISCOVERED_DEPENDENT_VIEWS_NORM="${DISCOVERED_DEPENDENT_VIEWS_NORM:+$DISCOVERED_DEPENDENT_VIEWS_NORM,}$v"
        added_any=1
      fi
    done <<< "$newly_found_views"

    if [[ "$added_any" -eq 0 ]]; then
      PHASE2_STATUS="FALLÓ"
      echo "   [ERROR] util.loadDump sigue fallando con los mismos objetos ya excluidos — no se puede auto-recuperar."
      fail_phase2 "el mismo error persiste en objetos ya excluidos (posible bug o error transitorio)"
    fi
  done
  PHASE2_ELAPSED="$(( $(date +%s) - PHASE2_T0 ))"
  echo "   [OK] Fase 2 (util.loadDump) completada en $(format_hms "$PHASE2_ELAPSED") tras $attempt intento(s)"
  if [[ "$RESTORE_RESUME" == "1" ]]; then
    printf '%s\n' "__PHASE2_LOADDUMP_DONE__" >> "$CHECKPOINT_FILE"
    printf '__WIDE_TABLES_DISCOVERED__:%s\n' "$DISCOVERED_WIDE_TABLES_NORM" >> "$CHECKPOINT_FILE"
    printf '__DEPENDENT_VIEWS_DISCOVERED__:%s\n' "$DISCOVERED_DEPENDENT_VIEWS_NORM" >> "$CHECKPOINT_FILE"
  fi
fi

WIDE_TABLES_NORM="$DISCOVERED_WIDE_TABLES_NORM"
echo "   Tablas anchas descubiertas: ${WIDE_TABLES_NORM:-ninguna}"

# FASE 1: Estructuras de las tablas anchas descubiertas en la Fase 2 (old-school)
echo "-> Fase 1: Creando estructuras de tablas anchas (old-school)..."

if [[ "$RESTORE_RESUME" == "1" ]] && grep -Fxq "__PHASE1_WIDE_TABLES_DONE__" "$CHECKPOINT_FILE" 2>/dev/null; then
  echo "   [SKIP] Fase 1 omitida por checkpoint (ya se completó en una corrida previa)"
  sql_count=0
elif [[ -z "$WIDE_TABLES_NORM" ]]; then
  echo "   [SKIP] Ninguna tabla ancha, nada que crear aquí"
  sql_count=0
else

sql_count=0
while IFS= read -r sql_file; do
  [[ -e "$sql_file" ]] || continue
  filename="$(basename "$sql_file")"

  if [[ "$SKIP_VIEWS_TRIGGERS" == "1" ]]; then
    if [[ "$filename" == *".post.sql"* || "$filename" == *"trigger"* || "$filename" == *"view"* || "$filename" == *"@.sql"* ]]; then
      continue
    fi
  fi

  table_name=""
  if [[ "$filename" == *"@"* ]]; then
    table_name="$(echo "$filename" | awk -F'@' '{print $2}' | sed -E 's/\..*$//')"
  fi

  # Old-school solo se encarga de las tablas anchas descubiertas; el resto ya lo cargó util.loadDump en la Fase 2.
  if [[ -z "$table_name" ]] || ! is_wide_table "$table_name"; then
    continue
  fi

  if [[ -n "$table_name" ]] && should_skip_table "$table_name"; then
    continue
  fi

  if [[ -n "$SOURCE_SCHEMA" && "$SOURCE_SCHEMA" != "$TARGET_DB" ]]; then
    sed_schema=(
      -e "s/\`$SOURCE_SCHEMA\`\./\`$TARGET_DB\`./g"
      -e "s/USE \`$SOURCE_SCHEMA\`/USE \`$TARGET_DB\`/g"
      -e "s/CREATE DATABASE IF NOT EXISTS \`$SOURCE_SCHEMA\`/CREATE DATABASE IF NOT EXISTS \`$TARGET_DB\`/g"
    )
  else
    sed_schema=()
  fi

  (
    echo "SET FOREIGN_KEY_CHECKS=0;"
    echo "SET UNIQUE_CHECKS=0;"
    tr -d '\r' < "$sql_file" | sed -E \
      -e 's/\/\*![0-9]{5} DEFINER=`[^`]+`@`[^`]+` \*\//\/* stripped DEFINER *\//g' \
      -e 's/DEFINER=`[^`]+`@`[^`]+`[ ]*//g' \
      -e 's/SQL SECURITY DEFINER/SQL SECURITY INVOKER/g' \
      "${sed_schema[@]+"${sed_schema[@]}"}"
    echo ";SET FOREIGN_KEY_CHECKS=1;"
    echo "SET UNIQUE_CHECKS=1;"
  ) | MYSQL_PWD="$DB_PASSWORD" mysql -h "$DB_HOST" -u "$DB_USER" -P "$DB_PORT" --default-character-set=utf8mb4 "$TARGET_DB" 2>/dev/null || true

  sql_count=$((sql_count + 1))
done < <(find "$DUMP_DIR" -type f -name "*.sql" 2>/dev/null | sort)
fi

echo "   [OK] Estructuras de tablas anchas procesadas: $sql_count"

if [[ "$RESTORE_RESUME" == "1" ]]; then
  printf '%s\n' "__PHASE1_WIDE_TABLES_DONE__" >> "$CHECKPOINT_FILE"
fi

# FASE 1.5: Vistas que dependen de tablas anchas (se crean ahora que esas
# tablas ya existen; en la Fase 2 fallaban porque loadDump intenta crear las
# vistas antes de que la Fase 1 haya corrido). Respeta SKIP_VIEWS_TRIGGERS
# igual que la Fase 1 old-school, ya que conceptualmente es parte de ella.
if [[ "$SKIP_VIEWS_TRIGGERS" != "1" && -n "$DISCOVERED_DEPENDENT_VIEWS_NORM" ]]; then
  echo "-> Fase 1.5: Creando vistas dependientes de tablas anchas..."
  view_count=0
  while IFS= read -r sql_file; do
    [[ -e "$sql_file" ]] || continue
    filename="$(basename "$sql_file")"

    view_name=""
    if [[ "$filename" == *"@"* ]]; then
      view_name="$(echo "$filename" | awk -F'@' '{print $2}' | sed -E 's/\..*$//')"
    fi

    if [[ -z "$view_name" ]] || ! in_csv_list "$view_name" "$DISCOVERED_DEPENDENT_VIEWS_NORM"; then
      continue
    fi

    if [[ -n "$SOURCE_SCHEMA" && "$SOURCE_SCHEMA" != "$TARGET_DB" ]]; then
      sed_schema=(
        -e "s/\`$SOURCE_SCHEMA\`\./\`$TARGET_DB\`./g"
        -e "s/USE \`$SOURCE_SCHEMA\`/USE \`$TARGET_DB\`/g"
      )
    else
      sed_schema=()
    fi

    tr -d '\r' < "$sql_file" | sed -E \
      -e 's/\/\*![0-9]{5} DEFINER=`[^`]+`@`[^`]+` \*\//\/* stripped DEFINER *\//g' \
      -e 's/DEFINER=`[^`]+`@`[^`]+`[ ]*//g' \
      -e 's/SQL SECURITY DEFINER/SQL SECURITY INVOKER/g' \
      "${sed_schema[@]+"${sed_schema[@]}"}" \
      | MYSQL_PWD="$DB_PASSWORD" mysql -h "$DB_HOST" -u "$DB_USER" -P "$DB_PORT" --default-character-set=utf8mb4 "$TARGET_DB" 2>&1 \
      | sed "s/^/   [vista $view_name] /" || true

    view_count=$((view_count + 1))
  done < <(find "$DUMP_DIR" -type f -name "*.sql" 2>/dev/null | sort)
  echo "   [OK] Vistas dependientes procesadas: $view_count"
fi

# FASE 3: Datos de las tablas anchas (old-school LOAD DATA LOCAL INFILE)
echo "-> Fase 3: Cargando datos de tablas anchas (old-school)..."

if [[ -z "$WIDE_TABLES_NORM" ]]; then
  echo "   [SKIP] WIDE_TABLES vacío, nada que cargar aquí"
  loaded_chunks=0
  total_chunks=0
  failed_chunks=0
  skipped_view_chunks=0
  skipped_missing_chunks=0
  skipped_checkpoint_chunks=0
  skipped_start_from_chunks=0
  declare -a failed_items=()
  tmp_wide_status="$(mktemp /tmp/restore_wide_status_XXXXXX.tsv)"
  trap 'rm -f "$tmp_wide_status"' EXIT
else

if ! command -v zstd >/dev/null 2>&1; then
  echo "Error: zstd no está instalado y es requerido para archivos .zst"
  exit 1
fi

tmp_infile="$(mktemp /tmp/restore_load_data_XXXXXX.tsv)"
tmp_tiempos_tabla="$(mktemp /tmp/restore_table_times_XXXXXX.tsv)"
tmp_wide_status="$(mktemp /tmp/restore_wide_status_XXXXXX.tsv)"
cleanup() {
  rm -f "$tmp_infile"
  rm -f "$tmp_tiempos_tabla"
  rm -f "$tmp_wide_status"
}
trap cleanup EXIT

loaded_chunks=0
total_chunks=0
failed_chunks=0
skipped_view_chunks=0
skipped_missing_chunks=0
skipped_checkpoint_chunks=0
skipped_start_from_chunks=0
declare -a failed_items=()

if [[ -n "$START_FROM_TABLE_NORM" ]]; then
  start_table_exists=0
  while IFS= read -r data_file; do
    filename_check="$(basename "$data_file")"
    table_check="$(echo "$filename_check" | awk -F'@' '{print $2}' | sed -E 's/@@.*$//')"
    if [[ -z "$table_check" ]]; then
      continue
    fi
    if ! is_wide_table "$table_check"; then
      continue
    fi
    if should_skip_table "$table_check"; then
      continue
    fi
    if [[ "$table_check" == "$START_FROM_TABLE_NORM" ]]; then
      start_table_exists=1
      break
    fi
  done < <(list_data_files)

  if [[ "$start_table_exists" -ne 1 ]]; then
    echo "Aviso: START_FROM_TABLE '$START_FROM_TABLE_NORM' no es una tabla ancha; no aplica a la Fase 3."
  fi
fi

while IFS= read -r data_file; do
  filename="$(basename "$data_file")"
  table_name="$(echo "$filename" | awk -F'@' '{print $2}' | sed -E 's/@@.*$//')"

  if [[ -z "$table_name" ]]; then
    continue
  fi

  # Old-school (Fase 3) solo carga datos de las tablas anchas; el resto ya lo cargó util.loadDump en la Fase 2.
  if ! is_wide_table "$table_name"; then
    continue
  fi

  if should_skip_table "$table_name"; then
    continue
  fi

  if [[ -n "$START_FROM_TABLE_NORM" ]] && [[ "$table_name" < "$START_FROM_TABLE_NORM" ]]; then
    echo "   -> Saltando (start_from_table): $table_name :: $filename"
    skipped_start_from_chunks=$((skipped_start_from_chunks + 1))
    continue
  fi

  total_chunks=$((total_chunks + 1))

  table_type="$(MYSQL_PWD="$DB_PASSWORD" mysql -Nse "SELECT TABLE_TYPE FROM information_schema.TABLES WHERE TABLE_SCHEMA='${TARGET_DB}' AND TABLE_NAME='${table_name}' LIMIT 1;" -h "$DB_HOST" -u "$DB_USER" -P "$DB_PORT" </dev/null 2>/dev/null || true)"
  if [[ "$table_type" == "VIEW" ]]; then
    echo "   -> Omitiendo vista: $table_name :: $filename"
    skipped_view_chunks=$((skipped_view_chunks + 1))
    failed_chunks=$((failed_chunks + 1))
    failed_items+=("$filename || vista omitida")
    printf '%s\tFALLÓ\t0\tvista omitida\n' "$table_name" >> "$tmp_wide_status"
    continue
  fi
  if [[ "$table_type" != "BASE TABLE" ]]; then
    echo "   -> Omitiendo objeto no cargable (tabla no existe): $table_name :: $filename"
    skipped_missing_chunks=$((skipped_missing_chunks + 1))
    failed_chunks=$((failed_chunks + 1))
    failed_items+=("$filename || tabla base no existe en destino")
    printf '%s\tFALLÓ\t0\ttabla no se pudo crear (probable fix VARCHAR->TEXT pendiente en esta base)\n' "$table_name" >> "$tmp_wide_status"
    continue
  fi

  if [[ "$RESTORE_RESUME" == "1" ]] && grep -Fxq "$filename" "$CHECKPOINT_FILE"; then
    echo "   -> Saltando (checkpoint): $table_name :: $filename"
    skipped_checkpoint_chunks=$((skipped_checkpoint_chunks + 1))
    continue
  fi

  echo "   -> Inyectando: $table_name :: $filename"

  rm -f "$tmp_infile"
  decode_ok=1

  case "$data_file" in
    *.zst)
      if ! zstd -d -c "$data_file" > "$tmp_infile"; then
        decode_ok=0
      fi
      ;;
    *.gz)
      if ! gzip -dc "$data_file" > "$tmp_infile"; then
        decode_ok=0
      fi
      ;;
    *)
      if ! cp "$data_file" "$tmp_infile"; then
        decode_ok=0
      fi
      ;;
  esac

  if [[ "$decode_ok" -ne 1 ]]; then
    echo "   [ERROR] No se pudo preparar chunk: $filename"
    failed_chunks=$((failed_chunks + 1))
    failed_items+=("$filename || error al descomprimir/copiar chunk")
    printf '%s\tFALLÓ\t0\terror al descomprimir/copiar chunk\n' "$table_name" >> "$tmp_wide_status"
    rm -f "$tmp_infile"
    continue
  fi

  t0="$(date +%s)"

  if MYSQL_PWD="$DB_PASSWORD" mysql --local-infile=1 -h "$DB_HOST" -u "$DB_USER" -P "$DB_PORT" "$TARGET_DB" </dev/null \
    -e "SET foreign_key_checks=0; SET unique_checks=0; LOAD DATA LOCAL INFILE '$tmp_infile' INTO TABLE \`$table_name\` FIELDS TERMINATED BY '\\t' LINES TERMINATED BY '\\n'; SET unique_checks=1; SET foreign_key_checks=1;"; then
    t1="$(date +%s)"
    elapsed="$((t1 - t0))"
    printf '%s\t%s\n' "$table_name" "$elapsed" >> "$tmp_tiempos_tabla"
    printf '%s\tOK\t%s\t\n' "$table_name" "$elapsed" >> "$tmp_wide_status"

    rm -f "$tmp_infile"
    loaded_chunks=$((loaded_chunks + 1))
    if [[ "$RESTORE_RESUME" == "1" ]]; then
      printf '%s\n' "$filename" >> "$CHECKPOINT_FILE"
    fi
  else
    echo "   [ERROR] Falló la carga de: $table_name :: $filename"
    failed_chunks=$((failed_chunks + 1))
    failed_items+=("$filename || error al cargar en tabla $table_name")
    printf '%s\tFALLÓ\t0\terror al ejecutar LOAD DATA\n' "$table_name" >> "$tmp_wide_status"
    rm -f "$tmp_infile"
    continue
  fi
done < <(list_data_files)

fi

echo "   [OK] Chunks de tablas anchas cargados: $loaded_chunks"
echo "   [OK] Chunks no cargados: $failed_chunks"
echo "   [OK] Vistas omitidas: $skipped_view_chunks"
if [[ "$skipped_missing_chunks" -gt 0 ]]; then
  echo "   [OK] Objetos no encontrados en destino: $skipped_missing_chunks"
fi
if [[ "$skipped_checkpoint_chunks" -gt 0 ]]; then
  echo "   [OK] Chunks saltados por checkpoint: $skipped_checkpoint_chunks"
fi
if [[ "$skipped_start_from_chunks" -gt 0 ]]; then
  echo "   [OK] Chunks saltados por start_from_table: $skipped_start_from_chunks"
fi

if [[ "${#failed_items[@]}" -gt 0 ]]; then
  echo "-----------------------------------------------------------------"
  echo " DETALLE DE CHUNKS NO CARGADOS (tablas anchas)"
  echo "-----------------------------------------------------------------"
  for item in "${failed_items[@]}"; do
    echo " - $item"
  done
fi

WIDE_TABLE_REPORT=""
if [[ -s "$tmp_wide_status" ]]; then
  WIDE_TABLE_REPORT="$(
    {
      printf '%-30s | %-8s | %10s | %s\n' "Tabla" "Estado" "Tiempo" "Detalle"
      printf -- '-------------------------------+----------+------------+----------------------------------------\n'
      sort -k1,1 "$tmp_wide_status" | while IFS=$'\t' read -r tabla estado segundos detalle; do
        printf '%-30s | %-8s | %10s | %s\n' "$tabla" "$estado" "$(format_hms "${segundos:-0}")" "$detalle"
      done
    }
  )"
fi

RESTORE_END_TS="$(date +%s)"
RESTORE_TOTAL_SEC="$((RESTORE_END_TS - RESTORE_START_TS))"
RESTORE_TOTAL_HMS="$(format_hms "$RESTORE_TOTAL_SEC")"

echo "-----------------------------------------------------------------"
if [[ -n "$WIDE_TABLE_REPORT" ]]; then
  echo " RESUMEN POR TABLA ANCHA"
  echo "-----------------------------------------------------------------"
  echo "$WIDE_TABLE_REPORT"
  echo "-----------------------------------------------------------------"
fi
echo " Fase 2 (util.loadDump, tablas normales): $PHASE2_STATUS en $(format_hms "$PHASE2_ELAPSED")"
echo " TIEMPO TOTAL RESTORE: $RESTORE_TOTAL_HMS (${RESTORE_TOTAL_SEC}s)"
echo " RESUMEN RESTORE (tablas anchas old-school): total=${total_chunks}, cargados=${loaded_chunks}, no_cargados=${failed_chunks}"

EMAIL_BODY="Restore de '$TARGET_DB'"$'\n'
EMAIL_BODY+="Dump origen: $DUMP_DIR"$'\n\n'
EMAIL_BODY+="Fase 2 (util.loadDump, tablas normales): $PHASE2_STATUS en $(format_hms "$PHASE2_ELAPSED")"$'\n\n'
if [[ -n "$WIDE_TABLE_REPORT" ]]; then
  EMAIL_BODY+="Tablas anchas (old-school):"$'\n'
  EMAIL_BODY+="$WIDE_TABLE_REPORT"$'\n\n'
fi
EMAIL_BODY+="TIEMPO TOTAL: $RESTORE_TOTAL_HMS"$'\n'

if [[ "$failed_chunks" -gt 0 ]]; then
  echo " PROCESO FINALIZADO CON ERRORES PARCIALES: base '$TARGET_DB' restaurada parcialmente"
  echo " CHECKPOINT: $CHECKPOINT_FILE"
  echo "================================================================="
  EMAIL_BODY+=$'\n'"ATENCIÓN: $failed_chunks tabla(s) ancha(s) no se cargaron. Revisar arriba"
  EMAIL_BODY+=" — lo más probable es que a esa base todavía le falte el fix VARCHAR->TEXT"
  EMAIL_BODY+=" para poder crearse en MySQL 8.0.46-cloud."$'\n'
  EMAIL_BODY+="Checkpoint: $CHECKPOINT_FILE"
  send_email "Restore con errores parciales: $TARGET_DB" "$EMAIL_BODY"
  exit 2
fi

echo " PROCESO FINALIZADO: base '$TARGET_DB' restaurada"
echo " CHECKPOINT: $CHECKPOINT_FILE"
echo "================================================================="
EMAIL_BODY+=$'\n'"Restore completado sin errores."
send_email "Restore OK: $TARGET_DB" "$EMAIL_BODY"
exit 0