loop-ops/db/migrate-one.sh

#!/usr/bin/env bash
# Migrate one installation DB: YC MDB → CNPG or Selectel Managed (no cutover).
# Usage:
#   migrate-one.sh <kube_context> <installation_id|uuid> [cnpg|selectel] [postgres://dsn]
#   migrate-one.sh <kube_context> <installation_id> --scale-0
#   TARGET=selectel migrate-one.sh <kube_context> <installation_id>
#   TRANSFER_MODE=file DUMP_PVC_SIZE=80Gi migrate-one.sh loop-ff <installation_id>
#
# Selectel: create user/password/database in UI with the same values as the MM pod secret.
# --scale-0 / MIGRATE_SCALE_0=1: scale MM to 0 before transfer, restore after (or keep 0 if Job still running).
set -euo pipefail

DIR="$(cd "$(dirname "${BASH_SOURCE[0]}")" && pwd)"
# shellcheck disable=SC1091
source "${DIR}/lib.sh"

[[ $# -ge 2 ]] || die "usage: $0 <kube_context> <installation_id|uuid> [cnpg|selectel|loop-cnpg-*] [dsn] [--scale-0]"
CTX="$1"
INSTALLATION_RAW="$2"
shift 2

SCALE_0="${MIGRATE_SCALE_0:-0}"
ARGS=()
for a in "$@"; do
  case "$a" in
    --scale-0|scale-0) SCALE_0=1 ;;
    *) ARGS+=("$a") ;;
  esac
done
set -- "${ARGS[@]+"${ARGS[@]}"}"

parse_target_dsn_args "$@"
OVERRIDE_DSN="${PARSE_DSN}"
if [[ -n "${PARSE_TARGET}" ]]; then
  TARGET="${PARSE_TARGET}"
fi
if [[ -n "${PARSE_CNPG_PROFILE:-}" ]]; then
  CNPG_CLUSTER_PROFILE="${PARSE_CNPG_PROFILE}"
fi
TARGET="${TARGET:-cnpg}"
TRANSFER_MODE="${TRANSFER_MODE:-stream}"
DUMP_VOLUME="${DUMP_VOLUME:-pvc}"
case "$TRANSFER_MODE" in
  stream|file) ;;
  *) die "TRANSFER_MODE must be stream|file (got: ${TRANSFER_MODE})" ;;
esac
case "$DUMP_VOLUME" in
  pvc|emptydir) ;;
  *) die "DUMP_VOLUME must be pvc|emptydir (got: ${DUMP_VOLUME})" ;;
esac
case "${DUMP_COMPRESS:-zstd}" in
  zstd|gzip|none) ;;
  *) die "DUMP_COMPRESS must be zstd|gzip|none (got: ${DUMP_COMPRESS})" ;;
esac

# ponytail: file-режим отключён — оборванный pg_dump мог считаться полным дампом (ревью 27.10.2026). Чинить test_archive/маркер, потом включать.
[[ "$TRANSFER_MODE" != "file" ]] || die "TRANSFER_MODE=file отключён до исправления, используй stream"

INSTALLATION="$(resolve_installation "$CTX" "$INSTALLATION_RAW")"

LOG_FILE="${DIR}/${LOG_DIR}/${INSTALLATION}.log"
exec > >(tee -a "$LOG_FILE") 2>&1

# MIGRATE_FORCE=1 (loopctl db move): статус в registry от прошлого переезда не повод пропускать копию на новый кластер
if [[ "${MIGRATE_FORCE:-0}" != "1" ]] && is_done "$INSTALLATION"; then
  cur="$(registry_get "$INSTALLATION" status 2>/dev/null || true)"
  log "SKIP ${INSTALLATION} (registry status=${cur:-done}; re-run: registry_lib.py set ${INSTALLATION} status=sized)"
  exit 0
fi

resolve_cnpg_cluster_profile "$INSTALLATION"
resolve_target "$TARGET"

registry_set "$INSTALLATION" "status=migrating" "kube_context=${CTX}" || true
if [[ "${TARGET_KIND}" == "cnpg" && -n "${CNPG_CLUSTER_PROFILE:-}" ]]; then
  registry_set "$INSTALLATION" "target_cluster=${CNPG_CLUSTER_PROFILE}" || true
fi

log "=== migrate-one ${CTX} ${INSTALLATION} target=${TARGET_KIND} cnpg=${CNPG_CLUSTER_PROFILE:-${CNPG_CLUSTER:-?}} endpoint=${TARGET_HOST}:${TARGET_PORT} mode=${TRANSFER_MODE} scale-0=${SCALE_0} ==="
SCALE_0_ACTIVE=0
SCALE_0_PREV=""

DSN="${OVERRIDE_DSN}"
if [[ -z "$DSN" ]]; then
  DSN="$(get_secret_dsn "$CTX" "$INSTALLATION")"
fi
parse_dsn "$DSN"
[[ -n "$SRC_USER" ]] || die "empty user in DSN"
[[ -n "$SRC_DB" ]] || die "empty db in DSN"

log "Source host=${SRC_HOST}:${SRC_PORT} db=${SRC_DB} user=${SRC_USER}"
printf '%s\n' "$DSN" >"${DIR}/${LOG_DIR}/${INSTALLATION}.source.dsn"
chmod 600 "${DIR}/${LOG_DIR}/${INSTALLATION}.source.dsn"
save_installation_target "$INSTALLATION" "$TARGET_KIND"

log "Target ${TARGET_KIND} ${TARGET_HOST}:${TARGET_PORT} sslmode=${TARGET_SSLMODE}"

SUPERUSER_CLEARED=1  # selectel: never toggles superuser
if [[ "$TARGET_KIND" == "cnpg" ]]; then
  ensure_role_and_db "$SRC_USER" "$SRC_PASS" "$SRC_DB"
  set_role_superuser "$SRC_USER" ON
  SUPERUSER_CLEARED=0
else
  log "Selectel: using pre-created DB/user/password from UI (must match pod secret)"
fi

JOB_NAME="pgxfer-$(echo "$INSTALLATION" | tr '[:upper:]' '[:lower:]' | cut -c1-40)-$(date +%s | tail -c 5)"
SECRET_NAME="${JOB_NAME}-env"
PVC_NAME="pgxfer-dump-$(echo "$INSTALLATION" | tr '[:upper:]' '[:lower:]' | cut -c1-48)"
JOB_FILE="$(mktemp)"
KEEP_JOB=0

cleanup() {
  if [[ "$SUPERUSER_CLEARED" != "1" && "${TARGET_KIND:-}" == "cnpg" ]]; then
    set_role_superuser "$SRC_USER" OFF 2>/dev/null || true
  fi
  # Never delete a still-running Job (timeout, Ctrl-C, or bash crash after editing this file mid-wait)
  local active=0
  if [[ -n "${JOB_NAME:-}" ]]; then
    active="$(kubectl --context "$CTX" -n "$JOB_NS" get job "$JOB_NAME" -o jsonpath='{.status.active}' 2>/dev/null || echo 0)"
  fi
  if [[ "$KEEP_JOB" == "1" || "${active:-0}" -ge 1 ]]; then
    log "Keeping Job ${JOB_NAME} and secret ${SECRET_NAME} (active=${active:-0})"
    if [[ "${TRANSFER_MODE}" == "file" && "${DUMP_VOLUME}" == "pvc" ]]; then
      log "  dump PVC: ${PVC_NAME} (kept)"
    elif [[ "${TRANSFER_MODE}" == "file" ]]; then
      log "  dump volume: emptyDir (node disk, lost if pod is rescheduled)"
    fi
    log "  kubectl --context ${CTX} -n ${JOB_NS} logs -f job/${JOB_NAME}"
    if [[ "${MIGRATE_JOB_TIMEOUT}" == "0" ]]; then
      log "  kubectl --context ${CTX} -n ${JOB_NS} wait --for=condition=complete job/${JOB_NAME} --timeout=-1s"
    else
      log "  kubectl --context ${CTX} -n ${JOB_NS} wait --for=condition=complete job/${JOB_NAME} --timeout=${MIGRATE_JOB_TIMEOUT}s"
    fi
  else
    kubectl --context "$CTX" -n "$JOB_NS" delete job "$JOB_NAME" --ignore-not-found --wait=false >/dev/null 2>&1 || true
    kubectl --context "$CTX" -n "$JOB_NS" delete secret "$SECRET_NAME" --ignore-not-found --wait=false >/dev/null 2>&1 || true
    if [[ "${TRANSFER_MODE}" == "file" && "${DUMP_VOLUME}" == "pvc" && "${DUMP_KEEP_PVC}" != "1" ]]; then
      log "Deleting dump PVC ${PVC_NAME}"
      kubectl --context "$CTX" -n "$JOB_NS" delete pvc "$PVC_NAME" --ignore-not-found --wait=false >/dev/null 2>&1 || true
    elif [[ "${TRANSFER_MODE}" == "file" && "${DUMP_VOLUME}" == "pvc" ]]; then
      log "Keeping dump PVC ${PVC_NAME} (DUMP_KEEP_PVC=1)"
    fi
  fi
  # Restore MM replicas unless transfer Job is still running
  restore_mm_app_replicas "$CTX" "$INSTALLATION" || true
  rm -f "$JOB_FILE"
}
trap cleanup EXIT

if [[ "$TRANSFER_MODE" == "file" && "$DUMP_VOLUME" == "pvc" ]]; then
  log "Ensuring dump PVC ${PVC_NAME} (${DUMP_PVC_SIZE}, sc=${DUMP_STORAGE_CLASS}, compress=${DUMP_COMPRESS})"
  cat <<PVC | kubectl --context "$CTX" -n "$JOB_NS" apply -f -
apiVersion: v1
kind: PersistentVolumeClaim
metadata:
  name: ${PVC_NAME}
  namespace: ${JOB_NS}
  labels:
    app: pgxfer
    installation: ${INSTALLATION}
spec:
  accessModes:
    - ReadWriteOnce
  storageClassName: ${DUMP_STORAGE_CLASS}
  resources:
    requests:
      storage: ${DUMP_PVC_SIZE}
PVC
elif [[ "$TRANSFER_MODE" == "file" ]]; then
  log "Dump volume=emptyDir sizeLimit=${DUMP_EMPTYDIR_SIZE_LIMIT} compress=${DUMP_COMPRESS}"
fi

if [[ "$SCALE_0" == "1" ]]; then
  scale_mm_app_to_zero "$CTX" "$INSTALLATION"
fi

log "Creating transfer Job ${JOB_NAME} in ${CTX}/${JOB_NS}"

kubectl --context "$CTX" -n "$JOB_NS" create secret generic "$SECRET_NAME" \
  --from-literal=SRC_HOST="$SRC_HOST" \
  --from-literal=SRC_PORT="$SRC_PORT" \
  --from-literal=SRC_USER="$SRC_USER" \
  --from-literal=SRC_PASS="$SRC_PASS" \
  --from-literal=SRC_DB="$SRC_DB" \
  --from-literal=SRC_SSLMODE="${SRC_SSLMODE}" \
  --from-literal=DST_HOST="$TARGET_HOST" \
  --from-literal=DST_PORT="$TARGET_PORT" \
  --from-literal=DST_SSLMODE="$TARGET_SSLMODE" \
  --from-literal=TARGET_KIND="$TARGET_KIND" \
  --from-literal=TRANSFER_MODE="${TRANSFER_MODE}" \
  --from-literal=TRANSFER_MAX_ATTEMPTS="${TRANSFER_MAX_ATTEMPTS}" \
  --from-literal=TRANSFER_RETRY_SLEEP_SEC="${TRANSFER_RETRY_SLEEP_SEC}" \
  --from-literal=PG_KEEPALIVES_IDLE="${PG_KEEPALIVES_IDLE}" \
  --from-literal=PG_KEEPALIVES_INTERVAL="${PG_KEEPALIVES_INTERVAL}" \
  --from-literal=PG_KEEPALIVES_COUNT="${PG_KEEPALIVES_COUNT}" \
  --from-literal=DUMP_COMPRESS="${DUMP_COMPRESS}" \
  --from-literal=DUMP_COMPRESS_LEVEL="${DUMP_COMPRESS_LEVEL}" \
  --dry-run=client -o yaml | kubectl --context "$CTX" -n "$JOB_NS" apply -f -

PULL_SECRETS='[]'
if kubectl --context "$CTX" -n "$JOB_NS" get secret docker.wilix.dev >/dev/null 2>&1; then
  PULL_SECRETS='[{"name":"docker.wilix.dev"}]'
fi

JOB_NAME="$JOB_NAME" JOB_NS="$JOB_NS" INSTALLATION="$INSTALLATION" \
  JOB_TTL_SECONDS="$JOB_TTL_SECONDS" PG_IMAGE="$PG_IMAGE" \
  SECRET_NAME="$SECRET_NAME" PULL_SECRETS="$PULL_SECRETS" \
  TARGET_KIND="$TARGET_KIND" TRANSFER_MODE="$TRANSFER_MODE" \
  PVC_NAME="$PVC_NAME" DUMP_VOLUME="$DUMP_VOLUME" \
  DUMP_EMPTYDIR_SIZE_LIMIT="${DUMP_EMPTYDIR_SIZE_LIMIT}" \
  XFER_CPU_REQUEST="${XFER_CPU_REQUEST}" XFER_CPU_LIMIT="${XFER_CPU_LIMIT}" \
  XFER_MEMORY_REQUEST="${XFER_MEMORY_REQUEST}" XFER_MEMORY_LIMIT="${XFER_MEMORY_LIMIT}" \
  python3 - "$JOB_FILE" <<'PY'
import json, os, sys
path = sys.argv[1]
kind = os.environ.get("TARGET_KIND", "cnpg")
mode = os.environ.get("TRANSFER_MODE", "stream")
pvc_name = os.environ.get("PVC_NAME", "")
dump_volume = os.environ.get("DUMP_VOLUME", "pvc")
emptydir_limit = os.environ.get("DUMP_EMPTYDIR_SIZE_LIMIT", "50Gi")
xfer_cpu_req = os.environ.get("XFER_CPU_REQUEST", "200m")
xfer_cpu_lim = os.environ.get("XFER_CPU_LIMIT", "2")
xfer_mem_req = os.environ.get("XFER_MEMORY_REQUEST", "512Mi")
xfer_mem_lim = os.environ.get("XFER_MEMORY_LIMIT", "2Gi")
if kind == "cnpg":
  pre_sql = r"""
DROP SCHEMA public CASCADE;
CREATE SCHEMA public AUTHORIZATION CURRENT_USER;
GRANT ALL ON SCHEMA public TO CURRENT_USER;
CREATE OR REPLACE FUNCTION public.user_search(uname text)
 RETURNS TABLE(usename name, passwd text)
 LANGUAGE sql SECURITY DEFINER
AS $f$SELECT usename, passwd FROM pg_catalog.pg_shadow WHERE usename=$1;$f$;
REVOKE ALL ON FUNCTION public.user_search(text) FROM PUBLIC;
GRANT USAGE ON SCHEMA public TO cnpg_pooler_pgbouncer;
GRANT EXECUTE ON FUNCTION public.user_search(text) TO cnpg_pooler_pgbouncer;
"""
else:
  pre_sql = r"""
DROP SCHEMA public CASCADE;
CREATE SCHEMA public AUTHORIZATION CURRENT_USER;
GRANT ALL ON SCHEMA public TO CURRENT_USER;
"""

common = r"""set -uo pipefail
echo "dump ${SRC_HOST}:${SRC_PORT}/${SRC_DB} -> ${DST_HOST}:${DST_PORT}/${SRC_DB} (target=${TARGET_KIND} mode=${TRANSFER_MODE:-stream})"
export PGCONNECT_TIMEOUT=60
export PGOPTIONS="-c statement_timeout=0 -c idle_in_transaction_session_timeout=0"

MAX_ATTEMPTS="${TRANSFER_MAX_ATTEMPTS:-30}"
SLEEP_SEC="${TRANSFER_RETRY_SLEEP_SEC:-60}"
KA_IDLE="${PG_KEEPALIVES_IDLE:-30}"
KA_INT="${PG_KEEPALIVES_INTERVAL:-10}"
KA_CNT="${PG_KEEPALIVES_COUNT:-9}"

src_conn() {
  printf 'host=%s port=%s dbname=%s user=%s sslmode=%s keepalives=1 keepalives_idle=%s keepalives_interval=%s keepalives_count=%s connect_timeout=60' \
    "${SRC_HOST}" "${SRC_PORT}" "${SRC_DB}" "${SRC_USER}" "${SRC_SSLMODE}" "${KA_IDLE}" "${KA_INT}" "${KA_CNT}"
}
dst_conn() {
  printf 'host=%s port=%s dbname=%s user=%s sslmode=%s keepalives=1 keepalives_idle=%s keepalives_interval=%s keepalives_count=%s connect_timeout=60' \
    "${DST_HOST}" "${DST_PORT}" "${SRC_DB}" "${SRC_USER}" "${DST_SSLMODE}" "${KA_IDLE}" "${KA_INT}" "${KA_CNT}"
}

reset_dst() {
  PGPASSWORD="${SRC_PASS}" psql "$(dst_conn)" -v ON_ERROR_STOP=1 <<'SQL'
""" + pre_sql + r"""
SQL
}

rewrite_ext() {
  sed -E '/^CREATE EXTENSION/s/pgvector/vector/g'
}
"""

if mode == "file":
  script = common + r"""
DUMP_DIR="${DUMP_DIR:-/dump}"
COMPRESS="${DUMP_COMPRESS:-zstd}"
LEVEL="${DUMP_COMPRESS_LEVEL:-3}"
case "${COMPRESS}" in
  zstd) DUMP_FILE="${DUMP_DIR}/dump.sql.zst" ;;
  gzip) DUMP_FILE="${DUMP_DIR}/dump.sql.gz" ;;
  none) DUMP_FILE="${DUMP_DIR}/dump.sql" ;;
  *) echo "bad DUMP_COMPRESS=${COMPRESS}"; exit 2 ;;
esac
MARKER="${DUMP_DIR}/dump.complete"
PARTIAL="${DUMP_FILE}.partial"
mkdir -p "${DUMP_DIR}"

compress_cmd() {
  case "${COMPRESS}" in
    zstd) zstd -T0 -"${LEVEL}" -c ;;
    gzip) gzip -"${LEVEL}" -c ;;
    none) cat ;;
  esac
}
decompress_to_stdout() {
  case "${COMPRESS}" in
    zstd) zstd -dc "${DUMP_FILE}" ;;
    gzip) gzip -dc "${DUMP_FILE}" ;;
    none) cat "${DUMP_FILE}" ;;
  esac
}
test_archive() {
  local f="$1"
  case "${COMPRESS}" in
    zstd) zstd -t "$f" ;;
    gzip) gzip -t "$f" ;;
    none) [[ -s "$f" ]] ;;
  esac
}
dump_file_ready() {
  [[ -f "${MARKER}" && -s "${DUMP_FILE}" ]] && test_archive "${DUMP_FILE}"
}

# Dump from MDB once: write compressed file on PVC and stream to dst.
# If psql dies mid-way, drain the pipe so pg_dump+compress can finish the file.
dump_and_stream() {
  echo "=== dump+stream -> ${DUMP_FILE} ==="
  reset_dst
  rm -f "${MARKER}" "${PARTIAL}"
  local dst_cs pipe_rc=0 rc=0
  dst_cs="$(dst_conn)"
  set +e
  # tee writes compressed file; if psql dies, cat drains so pg_dump can finish the archive
  PGPASSWORD="${SRC_PASS}" pg_dump "$(src_conn)" \
    --no-owner --no-acl --no-comments \
  | rewrite_ext \
  | tee >(compress_cmd > "${PARTIAL}") \
  | {
      PGPASSWORD="${SRC_PASS}" psql "${dst_cs}" -v ON_ERROR_STOP=1
      rc=$?
      cat >/dev/null
      exit "${rc}"
    }
  pipe_rc=$?
  set -e

  if [[ ! -s "${PARTIAL}" ]]; then
    echo "ERROR: dump partial empty (pipeline rc=${pipe_rc})"
    return 1
  fi
  if ! test_archive "${PARTIAL}"; then
    echo "ERROR: dump archive failed integrity check"
    rm -f "${PARTIAL}"
    return 1
  fi
  mv -f "${PARTIAL}" "${DUMP_FILE}"
  sync
  touch "${MARKER}"
  echo "dump file ready: $(du -h "${DUMP_FILE}" | awk '{print $1}') (${DUMP_FILE}) pipeline_rc=${pipe_rc}"
  return "${pipe_rc}"
}

restore_from_file() {
  echo "=== restore from file ${DUMP_FILE} ($(du -h "${DUMP_FILE}" | awk '{print $1}')) ==="
  reset_dst
  decompress_to_stdout | PGPASSWORD="${SRC_PASS}" psql "$(dst_conn)" -v ON_ERROR_STOP=1
}

attempt=1
while [[ "${attempt}" -le "${MAX_ATTEMPTS}" ]]; do
  echo "=== transfer attempt ${attempt}/${MAX_ATTEMPTS} at $(date -u +%Y-%m-%dT%H:%M:%SZ) ==="
  ok=0
  if dump_file_ready; then
    if restore_from_file; then ok=1; fi
  else
    if dump_and_stream; then
      ok=1
    elif dump_file_ready; then
      echo "stream failed but dump file is complete — will retry restore from PVC"
    fi
  fi
  if [[ "${ok}" -eq 1 ]]; then
    echo "transfer OK (attempt ${attempt})"
    exit 0
  fi
  echo "transfer FAILED attempt ${attempt}/${MAX_ATTEMPTS}; sleeping ${SLEEP_SEC}s before retry..."
  sleep "${SLEEP_SEC}"
  attempt=$((attempt + 1))
done
echo "transfer gave up after ${MAX_ATTEMPTS} attempts"
exit 1
"""
else:
  script = common + r"""
do_transfer() {
  set -euo pipefail
  reset_dst
  PGPASSWORD="${SRC_PASS}" pg_dump "$(src_conn)" \
    --no-owner --no-acl --no-comments \
  | rewrite_ext \
  | PGPASSWORD="${SRC_PASS}" psql "$(dst_conn)" -v ON_ERROR_STOP=1
}

attempt=1
while [[ "${attempt}" -le "${MAX_ATTEMPTS}" ]]; do
  echo "=== transfer attempt ${attempt}/${MAX_ATTEMPTS} at $(date -u +%Y-%m-%dT%H:%M:%SZ) ==="
  if do_transfer; then
    echo "transfer OK (attempt ${attempt})"
    exit 0
  fi
  rc=$?
  echo "transfer FAILED attempt ${attempt}/${MAX_ATTEMPTS} rc=${rc}; sleeping ${SLEEP_SEC}s before retry..."
  sleep "${SLEEP_SEC}"
  attempt=$((attempt + 1))
done
echo "transfer gave up after ${MAX_ATTEMPTS} attempts"
exit 1
"""

container = {
  "name": "transfer",
  "image": os.environ["PG_IMAGE"],
  "imagePullPolicy": "IfNotPresent",
  "envFrom": [{"secretRef": {"name": os.environ["SECRET_NAME"]}}],
  "command": ["bash", "-c"],
  "args": [script],
  "resources": {
    "requests": {"cpu": xfer_cpu_req, "memory": xfer_mem_req},
    "limits": {"cpu": xfer_cpu_lim, "memory": xfer_mem_lim},
  },
}
pod_spec = {
  "restartPolicy": "Never",
  "imagePullSecrets": json.loads(os.environ["PULL_SECRETS"]),
  "containers": [container],
}
if mode == "file":
  container["volumeMounts"] = [{"name": "dump", "mountPath": "/dump"}]
  # postgres image has zstd/gzip as root; keep root for compress tools + volume perms
  pod_spec["securityContext"] = {"runAsUser": 0, "fsGroup": 0}
  if dump_volume == "emptydir":
    pod_spec["volumes"] = [{
      "name": "dump",
      "emptyDir": {"sizeLimit": emptydir_limit},
    }]
  else:
    pod_spec["volumes"] = [{
      "name": "dump",
      "persistentVolumeClaim": {"claimName": pvc_name},
    }]

job = {
  "apiVersion": "batch/v1",
  "kind": "Job",
  "metadata": {
    "name": os.environ["JOB_NAME"],
    "namespace": os.environ["JOB_NS"],
    "labels": {
      "app": "pgxfer",
      "installation": os.environ["INSTALLATION"],
      "migrate-target": kind,
      "transfer-mode": mode,
    },
  },
  "spec": {
    "ttlSecondsAfterFinished": int(os.environ.get("JOB_TTL_SECONDS", "86400")),
    "backoffLimit": 2,
    "template": {
      "metadata": {
        "labels": {
          "app": "pgxfer",
          "installation": os.environ["INSTALLATION"],
          "migrate-target": kind,
          "transfer-mode": mode,
        },
      },
      "spec": pod_spec,
    },
  },
}
with open(path, "w") as f:
    json.dump(job, f)
PY

kubectl --context "$CTX" -n "$JOB_NS" apply -f "$JOB_FILE"

if [[ "${MIGRATE_JOB_TIMEOUT}" == "0" ]]; then
  log "Waiting for Job ${JOB_NAME} (no timeout) ..."
else
  log "Waiting for Job ${JOB_NAME} (timeout=${MIGRATE_JOB_TIMEOUT}s) ..."
fi
wait_rc=0
wait_job "$CTX" "$JOB_NS" "$JOB_NAME" "${MIGRATE_JOB_TIMEOUT}" || wait_rc=$?

if [[ "$wait_rc" -eq 0 ]]; then
  kubectl --context "$CTX" -n "$JOB_NS" logs "job/${JOB_NAME}" --tail=80 || true
  if [[ "$TARGET_KIND" == "cnpg" ]]; then
    log "Reinstalling CNPG pooler auth hooks on ${SRC_DB}"
    install_pooler_auth "$SRC_DB"
    set_role_superuser "$SRC_USER" OFF
    SUPERUSER_CLEARED=1
  fi
  mark_done "$INSTALLATION"
  log "SUCCESS ${INSTALLATION} → ${TARGET_KIND} (mode=${TRANSFER_MODE})"
  exit 0
fi

if [[ "$wait_rc" -eq 2 && "${MIGRATE_KEEP_JOB_ON_TIMEOUT}" == "1" ]]; then
  KEEP_JOB=1
  kubectl --context "$CTX" -n "$JOB_NS" logs "job/${JOB_NAME}" --tail=40 || true
  log "Transfer still running after ${MIGRATE_JOB_TIMEOUT}s — Job left alive."
  if [[ "${MIGRATE_JOB_TIMEOUT}" == "0" ]]; then
    log "Resume wait: kubectl --context ${CTX} -n ${JOB_NS} wait --for=condition=complete job/${JOB_NAME} --timeout=-1s"
  else
    log "Resume wait: kubectl --context ${CTX} -n ${JOB_NS} wait --for=condition=complete job/${JOB_NAME} --timeout=${MIGRATE_JOB_TIMEOUT}s"
  fi
  exit 2
fi

kubectl --context "$CTX" -n "$JOB_NS" logs "job/${JOB_NAME}" --tail=120 || true
mark_failed "$INSTALLATION"
die "Job ${JOB_NAME} failed for ${INSTALLATION}"