init

db/bind.sh

@@ -0,0 +1,30 @@
+#!/bin/bash
+# После cutover базы: прописать новую базу в визоре, иначе следующий sync вернёт CR на старую (так было с korica-team).
+#   loopctl db bind <куб> <loop id>
+# Берёт MM_SQLSETTINGS_DATASOURCE(REPLICAS) из CR и кладёт в env визора через API,
+# провайдер БД инстанса = ON_PREM-провайдер визора с тем же host:port.
+set -euo pipefail
+. "$(dirname "$0")/../lib/common.sh"
+CTX=${1:?куб, например loop-3}; ID=${2:?loop id}
+
+CR=$(kubectl --context "$CTX" -n loop get mattermost "$ID" -o json)
+SRCENV=$(jq -c '[.spec.mattermostEnv[]|select(.name|test("^MM_SQLSETTINGS_DATASOURCE(REPLICAS)?$"))|{key:.name,value:.value}]' <<<"$CR")
+DBHOST=$(jq -r '.[]|select(.key=="MM_SQLSETTINGS_DATASOURCE").value' <<<"$SRCENV" | sed -E 's#.*@([^/]+)/.*#\1#')
+[ -n "$DBHOST" ] || { echo "в CR нет MM_SQLSETTINGS_DATASOURCE"; exit 1; }
+
+PROVIDER=$(echo "select id||'|'||name from database_provider where vendor='ON_PREM' and host_name='${DBHOST%:*}' and host_port::text='${DBHOST##*:}';" | vsql)
+[ "$(grep -c . <<<"$PROVIDER")" = 1 ] || { echo "в визоре не один ON_PREM-провайдер для $DBHOST: '$PROVIDER'"; exit 1; }
+PNAME=${PROVIDER#*|}; PROVIDER=${PROVIDER%%|*}
+FOUND=$(printf "select i.id||'|'||d.id from loop_instance i join database_instance d on d.id=i.database_instance_id where i.id like '%%%s' and i.deleted_at is null;\n" "${ID#loop}" | vsql)
+[ "$(grep -c . <<<"$FOUND")" = 1 ] || { echo "в визоре не один инстанс: $FOUND"; exit 1; }
+VID=${FOUND%%|*}; DBID=${FOUND#*|}
+
+confirm "В визоре $ID: провайдер БД -> $PNAME ($DBHOST), env DATASOURCE* из CR на $CTX." || exit 1
+ensure_token
+echo "update database_instance set vendor='ON_PREM', database_provider_id='$PROVIDER' where id='$DBID';" | vsql
+vapi "$VIZOR_API/$VID" | jq -c --argjson src "$SRCENV" \
+  '{environmentProperties: ([.environmentProperties[]|select(.key|test("^MM_SQLSETTINGS_DATASOURCE(REPLICAS)?$")|not)|{key,value}] + $src)}' \
+  | vapi -X PATCH "$VIZOR_API/$VID" -d @- -o /dev/null
+# sync diff (эндпоинт есть не на всех версиях визора): env DATASOURCE* в визоре и в CR должны совпасть
+D=$(vapi "$VIZOR_API/$VID/synchronization/diff" 2>/dev/null | jq -r '[..|objects|select((.path? // "")|startswith("env.MM_SQLSETTINGS_DATASOURCE"))]|length' 2>/dev/null) || D="diff недоступен"
+echo "-- визор: провайдер $PNAME, env обновлены. Расхождений DATASOURCE в sync diff: ${D:-diff недоступен} (должно быть 0)"

db/config.env

@@ -0,0 +1,90 @@
+# Default migration target: cnpg | selectel
+# Override per run: TARGET=selectel ./migrate-one.sh …
+# Or per list line: yc-loop-1  loopxxx  selectel
+TARGET="${TARGET:-cnpg}"
+
+# CNPG target (pooler NodePort; use a node that runs a pooler pod)
+CNPG_HOST="${CNPG_HOST:-95.143.178.247}"
+CNPG_PORT="${CNPG_PORT:-31574}"
+CNPG_SSLMODE="${CNPG_SSLMODE:-disable}"
+# 1 = always use CNPG_HOST (public VIP); 0 = try pooler pod hostIP first
+CNPG_PREFER_CONFIG_HOST="${CNPG_PREFER_CONFIG_HOST:-0}"
+CNPG_POOLER_RW="${CNPG_POOLER_RW:-pooler-cnpg-wilix-rw}"
+
+# Admin role for CREATE ROLE / CREATE DATABASE (kubectl --context local)
+CNPG_KUBE_CONTEXT="${CNPG_KUBE_CONTEXT:-local}"
+CNPG_NAMESPACE="${CNPG_NAMESPACE:-cnpg-loop-wilix}"
+CNPG_CLUSTER="${CNPG_CLUSTER:-loop-cnpg-wilix-1}"
+CNPG_ADMIN_SECRET="${CNPG_ADMIN_SECRET:-cnpg-vizor-admin-role-secret}"
+
+# Selectel Managed PostgreSQL (create user/pass/db in UI = same as MM pod secret)
+SELECTEL_HOST="${SELECTEL_HOST:-}"
+SELECTEL_PORT="${SELECTEL_PORT:-5432}"
+SELECTEL_SSLMODE="${SELECTEL_SSLMODE:-require}"
+# Source (YC MDB) SSL for pg_dump inside transfer Job
+SRC_SSLMODE="${SRC_SSLMODE:-disable}"
+
+# Source Mattermost installs
+SOURCE_NS="${SOURCE_NS:-loop}"
+JOB_NS="${JOB_NS:-loop}"
+# Pin minor to avoid pg_dump/psql \\restrict mismatches across floating :15 tags
+PG_IMAGE="${PG_IMAGE:-postgres:15.8}"
+JOB_TTL_SECONDS="${JOB_TTL_SECONDS:-86400}"
+# pgxfer Job resources (cluster often tight on memory)
+XFER_CPU_REQUEST="${XFER_CPU_REQUEST:-200m}"
+XFER_CPU_LIMIT="${XFER_CPU_LIMIT:-2}"
+XFER_MEMORY_REQUEST="${XFER_MEMORY_REQUEST:-512Mi}"
+XFER_MEMORY_LIMIT="${XFER_MEMORY_LIMIT:-2Gi}"
+
+# migrate-list parallelism
+PARALLELISM="${PARALLELISM:-1}"
+
+# migrate-one local wait for Job (seconds). 0 = wait forever (manual monitoring).
+MIGRATE_JOB_TIMEOUT="${MIGRATE_JOB_TIMEOUT:-86400}"
+# On wait timeout while Job still Running: keep Job (do not delete / mark failed)
+MIGRATE_KEEP_JOB_ON_TIMEOUT="${MIGRATE_KEEP_JOB_ON_TIMEOUT:-1}"
+
+# Inside transfer Job: retry full dump|restore after short network blips (still restarts copy).
+TRANSFER_MAX_ATTEMPTS="${TRANSFER_MAX_ATTEMPTS:-30}"
+TRANSFER_RETRY_SLEEP_SEC="${TRANSFER_RETRY_SLEEP_SEC:-60}"
+# libpq TCP keepalives (seconds) — reduces silent idle drops on long COPY/INDEX
+PG_KEEPALIVES_IDLE="${PG_KEEPALIVES_IDLE:-30}"
+PG_KEEPALIVES_INTERVAL="${PG_KEEPALIVES_INTERVAL:-10}"
+PG_KEEPALIVES_COUNT="${PG_KEEPALIVES_COUNT:-9}"
+
+# Transfer mode:
+#   stream — pg_dump | psql (default; retry = full re-dump from MDB)
+#   file   — pg_dump | tee >(zstd>PVC) | psql; on blip re-restore from file (no MDB re-dump)
+TRANSFER_MODE="${TRANSFER_MODE:-stream}"
+# file-mode volume: pvc | emptydir (emptydir = node disk; survives blips inside same pod only)
+DUMP_VOLUME="${DUMP_VOLUME:-pvc}"
+DUMP_PVC_SIZE="${DUMP_PVC_SIZE:-80Gi}"
+DUMP_STORAGE_CLASS="${DUMP_STORAGE_CLASS:-yc-network-ssd}"
+DUMP_EMPTYDIR_SIZE_LIMIT="${DUMP_EMPTYDIR_SIZE_LIMIT:-50Gi}"
+# zstd | gzip | none  (postgres:15.8 image already has zstd + gzip)
+DUMP_COMPRESS="${DUMP_COMPRESS:-zstd}"
+DUMP_COMPRESS_LEVEL="${DUMP_COMPRESS_LEVEL:-3}"
+# Keep PVC + dump after success (1) so you can re-restore manually
+DUMP_KEEP_PVC="${DUMP_KEEP_PVC:-1}"
+
+# cutover-one: MM needs long plugin sync after DB switch
+CUTOVER_LIVENESS_INITIAL_DELAY="${CUTOVER_LIVENESS_INITIAL_DELAY:-600}"
+CUTOVER_LIVENESS_FAILURE_THRESHOLD="${CUTOVER_LIVENESS_FAILURE_THRESHOLD:-60}"
+CUTOVER_READINESS_INITIAL_DELAY="${CUTOVER_READINESS_INITIAL_DELAY:-120}"
+CUTOVER_READINESS_FAILURE_THRESHOLD="${CUTOVER_READINESS_FAILURE_THRESHOLD:-60}"
+CUTOVER_MEMORY_LIMIT="${CUTOVER_MEMORY_LIMIT:-4Gi}"
+CUTOVER_MEMORY_REQUEST="${CUTOVER_MEMORY_REQUEST:-512Mi}"
+CUTOVER_CPU_REQUEST="${CUTOVER_CPU_REQUEST:-200m}"
+# wait loop: attempts * sleep (default ~30 min)
+CUTOVER_WAIT_ATTEMPTS="${CUTOVER_WAIT_ATTEMPTS:-180}"
+CUTOVER_WAIT_SLEEP_SEC="${CUTOVER_WAIT_SLEEP_SEC:-10}"
+# kill idle CNPG backends left by crashed MM/session-pooler clients
+CUTOVER_STALE_IDLE_MINUTES="${CUTOVER_STALE_IDLE_MINUTES:-5}"
+
+# Files relative to this directory
+TRANSFER_LIST="${TRANSFER_LIST:-transfer-list.txt}"
+DONE_FILE="${DONE_FILE:-done.txt}"
+FAILED_FILE="${FAILED_FILE:-failed.txt}"
+# Unified migration table (source of truth for status / sizes / batches)
+REGISTRY_FILE="${REGISTRY_FILE:-registry.tsv}"
+LOG_DIR="${LOG_DIR:-logs}"

db/cutover-list.sh

@@ -0,0 +1,183 @@
+#!/usr/bin/env bash
+# Run cutover-one for each line in a list file OR a registry batch.
+#
+# Usage:
+#   ./cutover-list.sh --batch wave1-lm
+#   ./cutover-list.sh batches/foo.txt
+#   PARALLELISM=2 ./cutover-list.sh --batch wave4-lp-3
+#   ./cutover-list.sh --batch wave1-lm --force   # also re-cutover status=done|cutover
+#   ./cutover-list.sh --batch wave4-lp-3 --mem-req 256Mi   # MM memory request after cutover
+#
+# Without --force: ONLY registry status=verified (batch export + per-row check).
+# With --force: also queue status=done|cutover (re-cutover).
+# --mem-req: optional k8s memory request for MM (256Mi, 250M, 1Gi); default from config.env.
+# Prefer PARALLELISM=1 for production cutovers unless you know you want overlap.
+set -euo pipefail
+
+DIR="$(cd "$(dirname "${BASH_SOURCE[0]}")" && pwd)"
+# shellcheck disable=SC1091
+source "${DIR}/lib.sh"
+
+LIST=""
+BATCH=""
+FORCE=0
+MEM_REQ=""
+# locked unless --force (then export may include done/cutover)
+STATUS="verified"
+
+while [[ $# -gt 0 ]]; do
+  case "$1" in
+    --batch) BATCH="${2:-}"; shift 2 ;;
+    --status)
+      # only meaningful with --force; without force we always use verified
+      STATUS="${2:-}"
+      shift 2
+      ;;
+    --mem-req|--memory-request)
+      MEM_REQ="$(normalize_mem_req "${2:-}")"
+      shift 2
+      ;;
+    --force) FORCE=1; shift ;;
+    -h|--help)
+      sed -n '2,16p' "$0"
+      exit 0
+      ;;
+    *)
+      LIST="$1"
+      shift
+      ;;
+  esac
+done
+
+if [[ -n "$MEM_REQ" ]]; then
+  export CUTOVER_MEMORY_REQUEST="$MEM_REQ"
+fi
+
+if [[ "$FORCE" != "1" ]]; then
+  STATUS="verified"
+else
+  # re-cutover already finished installs too
+  if [[ "$STATUS" == "verified" ]]; then
+    STATUS="verified,done,cutover"
+  fi
+fi
+
+registry_status() {
+  local installation="$1" st
+  st="$(registry_get "$installation" status 2>/dev/null || true)"
+  printf '%s' "$st" | tr -d '[:space:]'
+}
+
+# Return 0 if this install may be cut over under current FORCE/STATUS rules.
+cutover_allowed() {
+  local installation="$1" st allowed
+  st="$(registry_status "$installation")"
+  if [[ "$FORCE" == "1" ]]; then
+    case "$st" in
+      verified|done|cutover) return 0 ;;
+      *)
+        # honor custom --status list when --force
+        IFS=',' read -r -a allowed <<<"$STATUS"
+        for a in "${allowed[@]}"; do
+          [[ "$st" == "$a" ]] && return 0
+        done
+        return 1
+        ;;
+    esac
+  fi
+  [[ "$st" == "verified" ]]
+}
+
+WORK="$(mktemp)"
+NORM="$(mktemp)"
+trap 'rm -f "$WORK" "$NORM"' EXIT
+
+if [[ -n "$BATCH" ]]; then
+  # --include-done: verified/done/cutover are in DONE_LIKE and would be skipped otherwise
+  "${DIR}/registry-export-batch.sh" "$BATCH" --status "$STATUS" --include-done -o "$WORK"
+  log "=== cutover-list batch=${BATCH} status=${STATUS} parallelism=${PARALLELISM} force=${FORCE} mem-req=${CUTOVER_MEMORY_REQUEST:-default} ==="
+elif [[ -n "$LIST" ]]; then
+  [[ -f "$LIST" ]] || die "list file not found: $LIST"
+  while IFS= read -r line || [[ -n "$line" ]]; do
+    [[ -z "$line" || "$line" =~ ^[[:space:]]*# ]] && continue
+    echo "$line" >>"${WORK}.raw"
+  done <"$LIST"
+  mv "${WORK}.raw" "$WORK"
+  log "=== cutover-list ${LIST} parallelism=${PARALLELISM} force=${FORCE} mem-req=${CUTOVER_MEMORY_REQUEST:-default} (filter status=${STATUS}) ==="
+else
+  die "usage: $0 --batch <id> | <list_file> [--force] [--mem-req 256Mi]"
+fi
+
+skipped=0
+queued_preview=0
+while IFS= read -r line || [[ -n "$line" ]]; do
+  [[ -z "$line" || "$line" =~ ^[[:space:]]*# ]] && continue
+  # shellcheck disable=SC2086
+  set -- $line
+  ctx="${1:-}"; installation="${2:-}"
+  [[ -n "$ctx" && -n "$installation" ]] || continue
+  shift 2 2>/dev/null || true
+  if ! cutover_allowed "$installation"; then
+    st="$(registry_status "$installation")"
+    if [[ "$FORCE" == "1" ]]; then
+      log "SKIP ${installation} (status=${st:-empty}; --force allows verified|done|cutover)"
+    else
+      log "SKIP ${installation} (status=${st:-empty}; need verified, or pass --force)"
+    fi
+    skipped=$((skipped + 1))
+    continue
+  fi
+  printf '%s\t%s' "$ctx" "$installation" >>"$NORM"
+  for a in "$@"; do
+    printf '\t%s' "$a" >>"$NORM"
+  done
+  printf '\n' >>"$NORM"
+  queued_preview=$((queued_preview + 1))
+done <"$WORK"
+
+total="$(wc -l <"$NORM" | tr -d ' ')"
+log "Queued ${total} installations for cutover (skipped=${skipped})"
+[[ "$total" -gt 0 ]] || { log "nothing to cutover"; exit 0; }
+
+run_one() {
+  local ctx="$1" installation="$2" st
+  shift 2
+  # re-check: parallel workers / stale export
+  if ! cutover_allowed "$installation"; then
+    st="$(registry_status "$installation")"
+    log "SKIP ${installation} (status=${st:-empty} at run time)"
+    return 0
+  fi
+  local extra=()
+  if [[ -n "${MEM_REQ:-}" ]]; then
+    extra+=(--mem-req "$MEM_REQ")
+  fi
+  "${DIR}/cutover-one.sh" "$ctx" "$installation" "${extra[@]+"${extra[@]}"}" "$@" \
+    && LOOPCTL_YES=1 "${DIR}/bind.sh" "$ctx" "$installation" \
+    || log "FAILED cutover/bind $installation (continuing)"
+}
+
+if [[ "${PARALLELISM}" -le 1 ]]; then
+  while IFS=$'\t' read -r ctx installation a3 a4 || [[ -n "${ctx:-}" ]]; do
+    [[ -z "${ctx:-}" ]] && continue
+    args=()
+    [[ -n "${a3:-}" ]] && args+=("$a3")
+    [[ -n "${a4:-}" ]] && args+=("$a4")
+    run_one "$ctx" "$installation" "${args[@]+"${args[@]}"}"
+  done <"$NORM"
+else
+  while IFS=$'\t' read -r ctx installation a3 a4 || [[ -n "${ctx:-}" ]]; do
+    [[ -z "${ctx:-}" ]] && continue
+    while [[ "$(jobs -pr | wc -l | tr -d ' ')" -ge "${PARALLELISM}" ]]; do
+      sleep 5
+    done
+    args=()
+    [[ -n "${a3:-}" ]] && args+=("$a3")
+    [[ -n "${a4:-}" ]] && args+=("$a4")
+    run_one "$ctx" "$installation" "${args[@]+"${args[@]}"}" &
+  done <"$NORM"
+  wait || true
+fi
+
+log "=== cutover-list finished ==="
+python3 "${DIR}/registry_lib.py" --registry "${REGISTRY_PATH}" summary || true

db/cutover-one.sh

@@ -0,0 +1,261 @@
+#!/usr/bin/env bash
+# Point MM installation at target DB (CNPG pooler or Selectel Managed) and wait until healthy.
+# Applies canary learnings: long probes, more memory, stale session cleanup (CNPG), orphan RS.
+# Usage: cutover-one.sh <kube_context> <installation_id|uuid> [cnpg|selectel] [--mem-req 256Mi]
+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-*] [--mem-req 256Mi]"
+CTX="$1"
+INSTALLATION_RAW="$2"
+shift 2
+
+ARGS=()
+while [[ $# -gt 0 ]]; do
+  case "$1" in
+    --mem-req|--memory-request)
+      CUTOVER_MEMORY_REQUEST="$(normalize_mem_req "${2:-}")"
+      export CUTOVER_MEMORY_REQUEST
+      shift 2
+      ;;
+    --mem-req=*|--memory-request=*)
+      CUTOVER_MEMORY_REQUEST="$(normalize_mem_req "${1#*=}")"
+      export CUTOVER_MEMORY_REQUEST
+      shift
+      ;;
+    *)
+      ARGS+=("$1")
+      shift
+      ;;
+  esac
+done
+set -- "${ARGS[@]+"${ARGS[@]}"}"
+
+parse_target_dsn_args "$@"
+if [[ -n "${PARSE_TARGET}" ]]; then
+  TARGET="${PARSE_TARGET}"
+fi
+if [[ -n "${PARSE_CNPG_PROFILE:-}" ]]; then
+  CNPG_CLUSTER_PROFILE="${PARSE_CNPG_PROFILE}"
+fi
+
+INSTALLATION="$(resolve_installation "$CTX" "$INSTALLATION_RAW")"
+if [[ -z "${PARSE_TARGET:-}" ]]; then
+  TARGET="$(load_installation_target "$INSTALLATION")"
+fi
+
+LOG_FILE="${DIR}/${LOG_DIR}/${INSTALLATION}.cutover.log"
+exec > >(tee -a "$LOG_FILE") 2>&1
+
+resolve_cnpg_cluster_profile "$INSTALLATION"
+resolve_target "${TARGET:-cnpg}"
+
+log "=== cutover-one ${CTX} ${INSTALLATION} target=${TARGET_KIND} cnpg=${CNPG_CLUSTER_PROFILE:-${CNPG_CLUSTER:-?}} endpoint=${TARGET_HOST}:${TARGET_PORT} mem-req=${CUTOVER_MEMORY_REQUEST} ==="
+
+DSN="$(get_secret_dsn "$CTX" "$INSTALLATION")"
+parse_dsn "$DSN"
+
+NEW_DSN="$(build_target_dsn "$SRC_USER" "$SRC_PASS" "$SRC_DB")"
+NEW_DSN_RO="$(build_target_dsn_ro "$SRC_USER" "$SRC_PASS" "$SRC_DB")"
+EXPECTED_HP="${TARGET_HOST}:${TARGET_PORT}"
+EXPECTED_HP_RO="${TARGET_HOST_RO:-${TARGET_HOST}}:${TARGET_PORT_RO:-${TARGET_PORT}}"
+
+if [[ "$SRC_HOST" == "$TARGET_HOST" && "${SRC_PORT}" == "${TARGET_PORT}" ]]; then
+  log "Secret already points at ${TARGET_KIND} ${EXPECTED_HP} — will still retune probes/resources and bounce pods"
+fi
+
+# Drop leftover session-pooler backends before MM reconnects (CNPG only)
+if [[ "$TARGET_KIND" == "cnpg" ]]; then
+  terminate_stale_cnpg_sessions "$SRC_DB" "$SRC_USER" "${CUTOVER_STALE_IDLE_MINUTES}"
+fi
+
+log "Updating secret ${INSTALLATION}-db-credentials → ${EXPECTED_HP}/${SRC_DB} (ro ${EXPECTED_HP_RO})"
+kubectl --context "$CTX" -n "$SOURCE_NS" create secret generic "${INSTALLATION}-db-credentials" \
+  --from-literal=DB_CONNECTION_STRING="$NEW_DSN" \
+  --dry-run=client -o yaml | kubectl --context "$CTX" -n "$SOURCE_NS" apply -f -
+
+# Patch MM CR: DSN env + probes + memory (single replace)
+TMP_MM="$(mktemp)"
+kubectl --context "$CTX" -n "$SOURCE_NS" get mm "$INSTALLATION" -o json >"$TMP_MM"
+
+python3 - "$TMP_MM" "$NEW_DSN" "$NEW_DSN_RO" \
+  "${CUTOVER_LIVENESS_INITIAL_DELAY}" "${CUTOVER_LIVENESS_FAILURE_THRESHOLD}" \
+  "${CUTOVER_READINESS_INITIAL_DELAY}" "${CUTOVER_READINESS_FAILURE_THRESHOLD}" \
+  "${CUTOVER_MEMORY_LIMIT}" "${CUTOVER_MEMORY_REQUEST}" "${CUTOVER_CPU_REQUEST}" <<'PY'
+import json, os, sys
+
+path, new_dsn, new_dsn_ro = sys.argv[1], sys.argv[2], sys.argv[3]
+live_delay, live_fail = int(sys.argv[4]), int(sys.argv[5])
+ready_delay, ready_fail = int(sys.argv[6]), int(sys.argv[7])
+mem_limit, mem_req, cpu_req = sys.argv[8], sys.argv[9], sys.argv[10]
+
+with open(path) as f:
+    mm = json.load(f)
+spec = mm.setdefault("spec", {})
+
+# Probes/ресурсы только поднимаем до минимума: свои (больше) у крупных инстансов не трогаем, визор о них не знает
+def at_least(d, key, val):
+    if int(d.get(key) or 0) < val:
+        d[key] = val
+
+probes = spec.setdefault("probes", {})
+lp = probes.setdefault("livenessProbe", {})
+rp = probes.setdefault("readinessProbe", {})
+at_least(lp, "initialDelaySeconds", live_delay); at_least(lp, "failureThreshold", live_fail)
+at_least(rp, "initialDelaySeconds", ready_delay); at_least(rp, "failureThreshold", ready_fail)
+
+sched = spec.setdefault("scheduling", {})
+res = sched.setdefault("resources", {})
+res.setdefault("limits", {}).setdefault("memory", mem_limit)
+requests = res.setdefault("requests", {})
+requests.setdefault("memory", mem_req)
+requests.setdefault("cpu", cpu_req)
+
+# Force inline DSN via mattermostEnv only (spec.env is not a valid MM CR field)
+def upsert_env(envs, name, value):
+    for e in envs:
+        if e.get("name") == name:
+            e["value"] = value
+            e.pop("valueFrom", None)
+            return envs
+    envs.append({"name": name, "value": value})
+    return envs
+
+envs = list(spec.get("mattermostEnv") or [])
+envs = upsert_env(envs, "MM_SQLSETTINGS_DATASOURCE", new_dsn)
+envs = upsert_env(envs, "MM_SQLSETTINGS_DATASOURCEREPLICAS", new_dsn_ro)
+spec["mattermostEnv"] = envs
+# Drop invalid field if present from older script runs / edits
+spec.pop("env", None)
+
+# CUTOVER_REPLICAS (loopctl db move --scale-0) — вернуть число реплик до scale-0.
+# Иначе replicas не трогаем: приостановленный (0) инстанс остаётся на паузе, ему меняем только DSN.
+want = os.environ.get("CUTOVER_REPLICAS")
+if want:
+    spec["replicas"] = int(want)
+print("replicas=%s" % spec.get("replicas"))
+
+with open(path, "w") as f:
+    json.dump(mm, f)
+print("patched probes+resources+datasource")
+PY
+
+log "Replacing MM CR (DSN + probes + resources)"
+kubectl --context "$CTX" -n "$SOURCE_NS" replace -f "$TMP_MM"
+rm -f "$TMP_MM"
+
+# Also setenv on Deployment immediately (operator reconcile can lag)
+if kubectl --context "$CTX" -n "$SOURCE_NS" get deploy "$INSTALLATION" >/dev/null 2>&1; then
+  log "Forcing deploy/${INSTALLATION} MM_SQLSETTINGS_* → ${TARGET_KIND} rw=${EXPECTED_HP} ro=${EXPECTED_HP_RO}"
+  ENV_FILE="$(mktemp)"
+  printf 'MM_SQLSETTINGS_DATASOURCE=%s\nMM_SQLSETTINGS_DATASOURCEREPLICAS=%s\n' "$NEW_DSN" "$NEW_DSN_RO" >"$ENV_FILE"
+  kubectl --context "$CTX" -n "$SOURCE_NS" set env "deploy/${INSTALLATION}" --env - <"$ENV_FILE" >/dev/null
+  rm -f "$ENV_FILE"
+  # ensure deploy is not left at 0 while MM CR says 1
+  dep_rep="$(kubectl --context "$CTX" -n "$SOURCE_NS" get deploy "$INSTALLATION" \
+    -o jsonpath='{.spec.replicas}' 2>/dev/null || echo 0)"
+  if [[ "${dep_rep:-0}" -lt 1 && "${CUTOVER_REPLICAS:-0}" -ge 1 ]]; then
+    log "Scaling deploy/${INSTALLATION} replicas 0→1"
+    kubectl --context "$CTX" -n "$SOURCE_NS" scale deploy "$INSTALLATION" --replicas=1 >/dev/null
+  fi
+fi
+
+scale_orphan_replicasets "$CTX" "$INSTALLATION"
+
+log "Deleting MM pods to pick up new DSN"
+kubectl --context "$CTX" -n "$SOURCE_NS" delete pod \
+  -l "installation.mattermost.com/installation=${INSTALLATION}" --wait=false || true
+
+# Wait until stable + app ready + DSN on CNPG (+ HTTPS if endpoint known)
+stale_swept=0
+for i in $(seq 1 "${CUTOVER_WAIT_ATTEMPTS}"); do
+  state="$(kubectl --context "$CTX" -n "$SOURCE_NS" get mm "$INSTALLATION" \
+    -o jsonpath='{.status.state}' 2>/dev/null || echo unknown)"
+  endpoint="$(kubectl --context "$CTX" -n "$SOURCE_NS" get mm "$INSTALLATION" \
+    -o jsonpath='{.status.endpoint}' 2>/dev/null || true)"
+  ready="false"
+  if mm_app_ready "$CTX" "$INSTALLATION"; then
+    ready="true"
+  fi
+  dsn_hp="$(mm_pod_dsn_hostport "$CTX" "$INSTALLATION" 2>/dev/null || echo no-pod)"
+  ping_ok="no"
+  if mm_https_ping_ok "$endpoint"; then
+    ping_ok="yes"
+  fi
+
+  restarts="$(kubectl --context "$CTX" -n "$SOURCE_NS" get pods \
+    -l "app=mattermost,installation.mattermost.com/installation=${INSTALLATION}" \
+    -o jsonpath='{.items[0].status.containerStatuses[0].restartCount}' 2>/dev/null || echo 0)"
+  mm_rep="$(kubectl --context "$CTX" -n "$SOURCE_NS" get mm "$INSTALLATION" \
+    -o jsonpath='{.spec.replicas}' 2>/dev/null || echo '?')"
+
+  log "wait ${i}/${CUTOVER_WAIT_ATTEMPTS}: state=${state} ready=${ready} dsn=${dsn_hp} expected=${EXPECTED_HP} replicas=${mm_rep} https=${ping_ok} restarts=${restarts}"
+
+  # Приостановленный инстанс не будим: DSN уже в секрете и CR, подов ждать нечего
+  if [[ "${mm_rep:-0}" == "0" ]]; then
+    log "SUCCESS cutover ${INSTALLATION}: на паузе (replicas=0), DSN переключён, поды не поднимаю"
+    mark_cutover_done "$INSTALLATION"
+    exit 0
+  fi
+
+  # Periodically collapse duplicate ReplicaSets
+  if (( i % 3 == 0 )); then
+    scale_orphan_replicasets "$CTX" "$INSTALLATION"
+  fi
+
+  # If pods keep crashing, clear stale DB sessions once
+  if [[ "$TARGET_KIND" == "cnpg" && "${restarts:-0}" -ge 2 && "$stale_swept" -eq 0 ]]; then
+    log "Restarts=${restarts}: sweeping stale CNPG sessions again"
+    terminate_stale_cnpg_sessions "$SRC_DB" "$SRC_USER" "${CUTOVER_STALE_IDLE_MINUTES}"
+    stale_swept=1
+  fi
+
+  # Wrong DSN → force again
+  if [[ "$dsn_hp" != "no-pod" && "$dsn_hp" != "unknown" && "$dsn_hp" != "${EXPECTED_HP}" && "$dsn_hp" != ":" ]]; then
+    log "Pod DSN is ${dsn_hp}, expected ${EXPECTED_HP} — re-applying deploy env"
+    ENV_FILE="$(mktemp)"
+    printf 'MM_SQLSETTINGS_DATASOURCE=%s\nMM_SQLSETTINGS_DATASOURCEREPLICAS=%s\n' "$NEW_DSN" "$NEW_DSN_RO" >"$ENV_FILE"
+    kubectl --context "$CTX" -n "$SOURCE_NS" set env "deploy/${INSTALLATION}" --env - <"$ENV_FILE" >/dev/null 2>&1 || true
+    rm -f "$ENV_FILE"
+    kubectl --context "$CTX" -n "$SOURCE_NS" delete pod \
+      -l "app=mattermost,installation.mattermost.com/installation=${INSTALLATION}" --wait=false >/dev/null 2>&1 || true
+  fi
+
+  if [[ "$ready" == "true" && "$dsn_hp" == "${EXPECTED_HP}" ]]; then
+    if [[ "$state" == "stable" ]] || [[ "$ping_ok" == "yes" ]]; then
+      log "SUCCESS cutover ${INSTALLATION} (state=${state} endpoint=${endpoint} dsn=${dsn_hp})"
+      mark_cutover_done "$INSTALLATION"
+      exit 0
+    fi
+    # ready+correct DSN but state still reconciling — accept after HTTPS OK or enough ready cycles
+    if [[ "$i" -ge 12 && "$ready" == "true" ]]; then
+      # in-pod ping as fallback when ingress endpoint empty
+      pod="$(kubectl --context "$CTX" -n "$SOURCE_NS" get pods \
+        -l "app=mattermost,installation.mattermost.com/installation=${INSTALLATION}" \
+        -o jsonpath='{.items[0].metadata.name}' 2>/dev/null || true)"
+      if [[ -n "$pod" ]] && kubectl --context "$CTX" -n "$SOURCE_NS" exec "$pod" -- \
+          curl -sS -m 5 http://127.0.0.1:8065/api/v4/system/ping 2>/dev/null | grep -q OK; then
+        log "SUCCESS cutover ${INSTALLATION} (app ready+ping; state=${state})"
+        mark_cutover_done "$INSTALLATION"
+        exit 0
+      fi
+    fi
+  fi
+
+  sleep "${CUTOVER_WAIT_SLEEP_SEC}"
+done
+
+log "Diagnostics:"
+kubectl --context "$CTX" -n "$SOURCE_NS" get pods -l "installation.mattermost.com/installation=${INSTALLATION}" || true
+kubectl --context "$CTX" -n "$SOURCE_NS" get mm "$INSTALLATION" -o jsonpath='state={.status.state} endpoint={.status.endpoint}{"\n"}' || true
+pod="$(kubectl --context "$CTX" -n "$SOURCE_NS" get pods \
+  -l "app=mattermost,installation.mattermost.com/installation=${INSTALLATION}" \
+  -o jsonpath='{.items[0].metadata.name}' 2>/dev/null || true)"
+if [[ -n "$pod" ]]; then
+  kubectl --context "$CTX" -n "$SOURCE_NS" logs "$pod" --tail=40 || true
+fi
+
+die "cutover did not become stable in time for ${INSTALLATION}"

db/healthcheck-mm-one.sh

@@ -0,0 +1,223 @@
+#!/usr/bin/env bash
+# Post-cutover health check for one Mattermost installation.
+# Verifies kube readiness + HTTP API (/api/v4/system/ping, config/client).
+# Writes registry notes: live | err:<reason>
+#
+# Usage:
+#   ./healthcheck-mm-one.sh <kube_context> <installation_id>
+#   ./healthcheck-mm-one.sh <ctx> <id> --require-dsn-host <host>
+#   ./healthcheck-mm-one.sh <ctx> <id> --skip-https
+#   ./healthcheck-mm-one.sh <ctx> <id> --no-registry   # do not update notes
+#
+# Exit 0 if Ready + API answers OK (HTTPS preferred, in-pod fallback). Exit 1 otherwise.
+set -euo pipefail
+
+DIR="$(cd "$(dirname "${BASH_SOURCE[0]}")" && pwd)"
+# shellcheck source=lib.sh
+source "${DIR}/lib.sh"
+
+CTX=""
+INSTALLATION=""
+REQUIRE_DSN_HOST=""
+SKIP_HTTPS=0
+NO_REGISTRY=0
+
+while [[ $# -gt 0 ]]; do
+  case "$1" in
+    --require-dsn-host) REQUIRE_DSN_HOST="${2:-}"; shift 2 ;;
+    --skip-https) SKIP_HTTPS=1; shift ;;
+    --no-registry) NO_REGISTRY=1; shift ;;
+    -h|--help)
+      sed -n '2,14p' "$0"
+      exit 0
+      ;;
+    *)
+      if [[ -z "$CTX" ]]; then CTX="$1"
+      elif [[ -z "$INSTALLATION" ]]; then INSTALLATION="$1"
+      else die "unknown arg: $1"
+      fi
+      shift
+      ;;
+  esac
+done
+
+[[ -n "$CTX" && -n "$INSTALLATION" ]] || die "usage: $0 <kube_context> <installation_id> [--require-dsn-host HOST]"
+
+SOURCE_NS="${SOURCE_NS:-mm-cloud}"
+fail=0
+api_ok=0
+ERR_REASONS=()
+API_VIA=""
+MM_VERSION=""
+
+note_sanitize() {
+  # single-line TSV-safe note
+  printf '%s' "$1" | tr '\t\n\r' '   ' | sed 's/  */ /g' | cut -c1-200
+}
+
+hc_finish() {
+  local note="$1" rc="$2"
+  note="$(note_sanitize "$note")"
+  if [[ "$NO_REGISTRY" != "1" ]]; then
+    if registry_set "$INSTALLATION" "notes=${note}"; then
+      log "REGISTRY ${INSTALLATION} notes=${note}"
+    else
+      log "WARN ${INSTALLATION}: failed to write notes=${note}"
+    fi
+  fi
+  exit "$rc"
+}
+
+push_err() { ERR_REASONS+=("$1"); fail=1; }
+
+err_note() {
+  local joined
+  if [[ ${#ERR_REASONS[@]} -eq 0 ]]; then
+    printf 'err:unknown'
+    return
+  fi
+  joined="$(IFS=','; echo "${ERR_REASONS[*]}")"
+  printf 'err:%s' "$joined"
+}
+
+mm_json="$(kubectl --context "$CTX" -n "$SOURCE_NS" get mm "$INSTALLATION" -o json 2>/dev/null || true)"
+if [[ -z "$mm_json" ]]; then
+  log "FAIL ${INSTALLATION}: Mattermost CR not found in ${CTX}/${SOURCE_NS}"
+  hc_finish "err:no_cr" 1
+fi
+
+MM_STATE="$(python3 -c 'import json,sys; print((json.loads(sys.argv[1]).get("status") or {}).get("state") or "")' "$mm_json")"
+MM_ENDPOINT="$(python3 -c 'import json,sys; print((json.loads(sys.argv[1]).get("status") or {}).get("endpoint") or "")' "$mm_json")"
+MM_REPLICAS="$(python3 -c 'import json,sys; r=(json.loads(sys.argv[1]).get("spec") or {}).get("replicas"); print("" if r is None else r)' "$mm_json")"
+MM_MEM_REQ="$(python3 -c 'import json,sys
+mm=json.loads(sys.argv[1]); spec=mm.get("spec") or {}
+res=(spec.get("scheduling") or {}).get("resources") or spec.get("resources") or {}
+print((res.get("requests") or {}).get("memory") or "")' "$mm_json")"
+
+log "=== healthcheck ${CTX} ${INSTALLATION} state=${MM_STATE:-?} endpoint=${MM_ENDPOINT:-?} replicas=${MM_REPLICAS:-?} mem=${MM_MEM_REQ:-?} ==="
+
+pod_line="$(kubectl --context "$CTX" -n "$SOURCE_NS" get pods \
+  -l "app=mattermost,installation.mattermost.com/installation=${INSTALLATION}" \
+  -o jsonpath='{range .items[*]}{.metadata.name}{"\t"}{.status.phase}{"\t"}{.status.containerStatuses[0].ready}{"\t"}{.status.podIP}{"\n"}{end}' 2>/dev/null || true)"
+pod_name=""; pod_phase=""; pod_ready="false"; pod_ip=""
+if [[ -n "$pod_line" ]]; then
+  while IFS=$'\t' read -r n ph rd ip; do
+    [[ -z "${n:-}" ]] && continue
+    pod_name="$n"; pod_phase="$ph"; pod_ready="${rd:-false}"; pod_ip="${ip:-}"
+    [[ "$pod_ready" == "true" ]] && break
+  done <<<"$pod_line"
+fi
+
+if [[ -z "$pod_name" ]]; then
+  log "FAIL ${INSTALLATION}: no mattermost pod"
+  push_err "no_pod"
+else
+  log "POD ${pod_name} phase=${pod_phase} ready=${pod_ready} ip=${pod_ip:-?}"
+  if [[ "$pod_ready" != "true" ]]; then
+    log "WARN ${INSTALLATION}: container not Ready"
+    push_err "not_ready"
+  fi
+fi
+
+dsn_hp="$(mm_pod_dsn_hostport "$CTX" "$INSTALLATION" 2>/dev/null || true)"
+if [[ -n "$dsn_hp" ]]; then
+  log "DSN hostport=${dsn_hp}"
+  if [[ -n "$REQUIRE_DSN_HOST" ]]; then
+    dsn_host="${dsn_hp%%:*}"
+    if [[ "$dsn_host" != "$REQUIRE_DSN_HOST" ]]; then
+      log "FAIL ${INSTALLATION}: DSN host ${dsn_host} != required ${REQUIRE_DSN_HOST}"
+      push_err "dsn_host=${dsn_host}"
+    else
+      log "OK DSN host matches ${REQUIRE_DSN_HOST}"
+    fi
+  fi
+else
+  log "WARN ${INSTALLATION}: could not read DATASOURCE hostport"
+fi
+
+mm_https_api_check() {
+  local endpoint="$1"
+  local url body code version
+  [[ -n "$endpoint" ]] || return 1
+  url="https://${endpoint}/api/v4/system/ping"
+  body="$(curl -sS -m 10 -w '\n%{http_code}' "$url" 2>/dev/null || true)"
+  code="$(printf '%s' "$body" | tail -n1)"
+  body="$(printf '%s' "$body" | sed '$d')"
+  log "HTTPS GET ${url} → HTTP ${code:-?} body=${body:-<empty>}"
+  if [[ "$code" != "200" ]]; then
+    return 1
+  fi
+  if [[ "$body" != *'"status":"OK"'* && "$body" != *'"status": "OK"'* && "$body" != *OK* ]]; then
+    return 1
+  fi
+  url="https://${endpoint}/api/v4/config/client?format=old"
+  body="$(curl -sS -m 10 -w '\n%{http_code}' "$url" 2>/dev/null || true)"
+  code="$(printf '%s' "$body" | tail -n1)"
+  body="$(printf '%s' "$body" | sed '$d')"
+  version="$(printf '%s' "$body" | python3 -c 'import sys,json
+try:
+  d=json.load(sys.stdin)
+  print(d.get("Version") or d.get("version") or "")
+except Exception:
+  print("")' 2>/dev/null || true)"
+  MM_VERSION="$version"
+  log "HTTPS GET config/client → HTTP ${code:-?} Version=${version:-?}"
+  [[ "$code" == "200" && -n "$version" ]] || log "WARN ${INSTALLATION}: config/client incomplete (ping OK is enough)"
+  return 0
+}
+
+mm_inpod_ping() {
+  local pod="$1"
+  [[ -n "$pod" ]] || return 1
+  local body
+  body="$(kubectl --context "$CTX" -n "$SOURCE_NS" exec "$pod" -- \
+    curl -sS -m 5 http://127.0.0.1:8065/api/v4/system/ping 2>/dev/null || true)"
+  log "IN-POD ping → ${body:-<empty>}"
+  [[ "$body" == *OK* ]]
+}
+
+if [[ "$SKIP_HTTPS" != "1" ]]; then
+  if mm_https_api_check "${MM_ENDPOINT:-}"; then
+    api_ok=1
+    API_VIA="https"
+    log "OK ${INSTALLATION}: HTTPS API healthy"
+  else
+    log "WARN ${INSTALLATION}: HTTPS API failed or no endpoint; trying in-pod"
+    if [[ -n "$pod_name" ]] && mm_inpod_ping "$pod_name"; then
+      api_ok=1
+      API_VIA="inpod"
+      log "OK ${INSTALLATION}: in-pod API healthy (HTTPS unavailable)"
+    else
+      log "FAIL ${INSTALLATION}: API not responding"
+      push_err "api"
+    fi
+  fi
+else
+  if [[ -n "$pod_name" ]] && mm_inpod_ping "$pod_name"; then
+    api_ok=1
+    API_VIA="inpod"
+    log "OK ${INSTALLATION}: in-pod API healthy (--skip-https)"
+  else
+    log "FAIL ${INSTALLATION}: in-pod API failed"
+    push_err "api"
+  fi
+fi
+
+case "${MM_STATE:-}" in
+  stable|Stable|"") ;;
+  *)
+    log "WARN ${INSTALLATION}: MM state=${MM_STATE} (expected stable)"
+    push_err "state=${MM_STATE}"
+    ;;
+esac
+
+if [[ "$api_ok" -eq 1 && "$fail" -eq 0 ]]; then
+  log "SUCCESS healthcheck ${INSTALLATION}"
+  hc_finish "live" 0
+fi
+if [[ "$api_ok" -eq 1 ]]; then
+  log "PARTIAL ${INSTALLATION}: API OK but other checks failed"
+  hc_finish "$(err_note),api_ok=${API_VIA}" 1
+fi
+log "FAILED healthcheck ${INSTALLATION}"
+hc_finish "$(err_note)" 1

db/lib.sh

@@ -0,0 +1,800 @@
+#!/usr/bin/env bash
+# Shared helpers for MDB → CNPG migration scripts.
+set -euo pipefail
+
+MIGRATE_DIR="$(cd "$(dirname "${BASH_SOURCE[0]}")" && pwd)"
+# shellcheck disable=SC1091
+source "${MIGRATE_DIR}/config.env"
+
+mkdir -p "${MIGRATE_DIR}/${LOG_DIR}"
+DONE_PATH="${MIGRATE_DIR}/${DONE_FILE}"
+FAILED_PATH="${MIGRATE_DIR}/${FAILED_FILE}"
+LIST_PATH="${MIGRATE_DIR}/${TRANSFER_LIST}"
+REGISTRY_PATH="${MIGRATE_DIR}/${REGISTRY_FILE}"
+touch "${DONE_PATH}" "${FAILED_PATH}"
+if [[ ! -f "${REGISTRY_PATH}" ]]; then
+  printf 'loop_id\tdomain\tkube_context\tyc_cluster\ttarget_cluster\tsize_bytes\tsize_pretty\tbatch_id\tstatus\tnotes\n' \
+    >"${REGISTRY_PATH}"
+fi
+
+log() { printf '[%s] %s\n' "$(date -u +%Y-%m-%dT%H:%M:%SZ)" "$*"; }
+die() { printf '[%s] ERROR: %s\n' "$(date -u +%Y-%m-%dT%H:%M:%SZ)" "$*" >&2; exit 1; }
+
+# Normalize memory quantity for k8s (250M|256Mi|1G|512 → 250Mi|256Mi|1Gi|512Mi).
+normalize_mem_req() {
+  local v="${1:-}"
+  [[ -n "$v" ]] || { printf '%s\n' ""; return 0; }
+  if [[ "$v" =~ ^[0-9]+$ ]]; then
+    printf '%sMi\n' "$v"
+    return 0
+  fi
+  if [[ "$v" =~ ^([0-9]+)([KkMmGg])$ ]]; then
+    printf '%s%si\n' "${BASH_REMATCH[1]}" "${BASH_REMATCH[2]}"
+    return 0
+  fi
+  if [[ "$v" =~ ^[0-9]+([Ee][0-9]+)?$ || "$v" =~ ^[0-9]+([KkMmGg]i)$ ]]; then
+    printf '%s\n' "$v"
+    return 0
+  fi
+  die "invalid memory quantity '${v}' (examples: 256Mi, 250M, 1Gi)"
+}
+
+registry_set() {
+  # registry_set <loop_id> key=value [key=value ...]
+  # done.txt/failed.txt mirrors are updated inside registry_lib.py (flock-safe, shrink-guarded).
+  local installation="$1"
+  shift
+  [[ -n "$installation" ]] || return 0
+  local attempt=1 err="${MIGRATE_DIR}/${LOG_DIR}/.registry-set.err"
+  while ! python3 "${MIGRATE_DIR}/registry_lib.py" --registry "${REGISTRY_PATH}" set "$installation" "$@" >/dev/null 2>>"$err"; do
+    if [[ "$attempt" -ge 8 ]]; then
+      log "WARN registry_set failed for ${installation} after ${attempt} attempts: $(tail -3 "$err" 2>/dev/null | tr '\n' ' ')"
+      return 1
+    fi
+    sleep "0.$((RANDOM % 5 + 1))"
+    attempt=$((attempt + 1))
+  done
+  return 0
+}
+
+registry_get() {
+  # registry_get <loop_id> [field]
+  python3 "${MIGRATE_DIR}/registry_lib.py" --registry "${REGISTRY_PATH}" get "$@"
+}
+
+sql_quote() {
+  # Escape single quotes for SQL string literals
+  printf "%s" "${1//\'/\'\'}"
+}
+
+url_encode() {
+  python3 -c 'import sys,urllib.parse; print(urllib.parse.quote(sys.argv[1], safe=""))' "$1"
+}
+
+parse_dsn() {
+  # $1 → exports SRC_USER SRC_PASS SRC_HOST SRC_PORT SRC_DB SRC_QUERY
+  local dsn="${1:-}"
+  [[ -n "$dsn" ]] || die "parse_dsn: empty DSN"
+  local tmp
+  tmp="$(mktemp)"
+  python3 - "$dsn" "$tmp" <<'PY'
+import sys, urllib.parse, shlex
+u = urllib.parse.urlparse(sys.argv[1])
+db = (u.path or "/").lstrip("/")
+with open(sys.argv[2], "w") as f:
+    f.write("export SRC_USER=%s\n" % shlex.quote(u.username or ""))
+    f.write("export SRC_PASS=%s\n" % shlex.quote(u.password or ""))
+    f.write("export SRC_HOST=%s\n" % shlex.quote(u.hostname or ""))
+    f.write("export SRC_PORT=%s\n" % shlex.quote(str(u.port or 5432)))
+    f.write("export SRC_DB=%s\n" % shlex.quote(db))
+    f.write("export SRC_QUERY=%s\n" % shlex.quote(u.query or ""))
+PY
+  # shellcheck disable=SC1090
+  source "$tmp"
+  rm -f "$tmp"
+}
+
+get_secret_dsn() {
+  local ctx="$1" installation="$2"
+  kubectl --context "$ctx" -n "$SOURCE_NS" get secret "${installation}-db-credentials" \
+    -o jsonpath='{.data.DB_CONNECTION_STRING}' | base64 --decode
+}
+
+# Resolve cloud UUID / partial id → Mattermost installation name (e.g. loop036d0e599ad3).
+# Accepts: loop… id, raw UUID, or unique substring of the installation name.
+resolve_installation() {
+  local ctx="$1" raw="$2"
+  local id="$raw"
+
+  # Already a MM installation with credentials secret
+  if kubectl --context "$ctx" -n "$SOURCE_NS" get secret "${id}-db-credentials" >/dev/null 2>&1; then
+    echo "$id"
+    return 0
+  fi
+
+  # UUID → loop + last 12 hex chars (Loop naming convention)
+  local hex last12 candidate
+  hex="$(printf '%s' "$raw" | tr -d '-' | tr '[:upper:]' '[:lower:]')"
+  if [[ "$hex" =~ ^[0-9a-f]{32}$ ]]; then
+    last12="${hex: -12}"
+    candidate="loop${last12}"
+    if kubectl --context "$ctx" -n "$SOURCE_NS" get secret "${candidate}-db-credentials" >/dev/null 2>&1; then
+      log "Resolved ${raw} → ${candidate}" >&2
+      echo "$candidate"
+      return 0
+    fi
+  fi
+
+  # Unique substring match on MM CR names
+  local matches
+  matches="$(kubectl --context "$ctx" -n "$SOURCE_NS" get mm -o jsonpath='{range .items[*]}{.metadata.name}{"\n"}{end}' 2>/dev/null \
+    | grep -Fi "$raw" || true)"
+  local count
+  count="$(printf '%s\n' "$matches" | grep -c . || true)"
+  if [[ "$count" == "1" ]]; then
+    candidate="$(printf '%s\n' "$matches" | head -1)"
+    log "Resolved ${raw} → ${candidate}" >&2
+    echo "$candidate"
+    return 0
+  fi
+  if [[ "$count" -gt 1 ]]; then
+    die "ambiguous installation '${raw}', matches: $(printf '%s' "$matches" | tr '\n' ' ')"
+  fi
+
+  die "installation not found for '${raw}' (need MM id like loop036d0e599ad3, or cloud UUID). Secret ${id}-db-credentials missing in ${ctx}/${SOURCE_NS}"
+}
+
+is_done() {
+  # Source of truth: registry status only (migrated|verified|cutover|done).
+  # done.txt is a write-only mirror and must NOT gate re-runs.
+  local installation="$1"
+  python3 "${MIGRATE_DIR}/registry_lib.py" --registry "${REGISTRY_PATH}" is-done "$installation" 2>/dev/null
+}
+
+mark_done() {
+  local installation="$1"
+  local cur
+  # do not downgrade cutover/done/verified
+  cur="$(registry_get "$installation" status 2>/dev/null || true)"
+  if [[ "$cur" == "done" || "$cur" == "cutover" || "$cur" == "verified" ]]; then
+    registry_set "$installation" "status=${cur}" || true
+    return 0
+  fi
+  registry_set "$installation" "status=migrated" || true
+}
+
+mark_failed() {
+  local installation="$1"
+  registry_set "$installation" "status=failed" || true
+}
+
+mark_cutover_done() {
+  local installation="$1"
+  registry_set "$installation" "status=done" || true
+}
+
+resolve_cnpg_host() {
+  # CNPG_PREFER_CONFIG_HOST=1 → always use CNPG_HOST (public VIP / DNAT; InternalIP may be unreachable)
+  if [[ "${CNPG_PREFER_CONFIG_HOST:-0}" == "1" ]]; then
+    echo "${CNPG_HOST}"
+    return 0
+  fi
+  local pooler="${CNPG_POOLER_RW:-pooler-cnpg-wilix-rw}"
+  local ip
+  ip="$(kubectl --context "${CNPG_KUBE_CONTEXT}" -n "${CNPG_NAMESPACE}" get pods \
+    -l "cnpg.io/poolerName=${pooler}" \
+    -o jsonpath='{.items[0].status.hostIP}' 2>/dev/null || true)"
+  if [[ -n "$ip" ]]; then
+    echo "$ip"
+  else
+    echo "${CNPG_HOST}"
+  fi
+}
+
+cnpg_primary_pod() {
+  kubectl --context "${CNPG_KUBE_CONTEXT}" -n "${CNPG_NAMESPACE}" \
+    get clusters.postgresql.cnpg.io "${CNPG_CLUSTER}" \
+    -o jsonpath='{.status.currentPrimary}'
+}
+
+cnpg_admin_psql() {
+  # Run SQL as postgres superuser inside primary pod. Args: SQL string
+  local sql="$1"
+  local pod
+  pod="$(cnpg_primary_pod)"
+  [[ -n "$pod" ]] || die "cannot resolve CNPG primary"
+  kubectl --context "${CNPG_KUBE_CONTEXT}" -n "${CNPG_NAMESPACE}" exec "$pod" -c postgres -- \
+    psql -U postgres -d postgres -v ON_ERROR_STOP=1 -c "$sql"
+}
+
+cnpg_psql_db() {
+  local db="$1" sql="$2"
+  local pod
+  pod="$(cnpg_primary_pod)"
+  kubectl --context "${CNPG_KUBE_CONTEXT}" -n "${CNPG_NAMESPACE}" exec "$pod" -c postgres -- \
+    psql -U postgres -d "$db" -v ON_ERROR_STOP=1 -c "$sql"
+}
+
+ensure_role_and_db() {
+  local user="$1" pass="$2" db="$3"
+  local quser qpass qdb
+  quser="$(sql_quote "$user")"
+  qpass="$(sql_quote "$pass")"
+  qdb="$(sql_quote "$db")"
+
+  log "Ensuring role/database ${user}/${db} on CNPG"
+  cnpg_admin_psql "
+DO \$\$
+BEGIN
+  IF NOT EXISTS (SELECT FROM pg_roles WHERE rolname = '${quser}') THEN
+    CREATE ROLE \"${quser}\" LOGIN PASSWORD '${qpass}' CREATEDB;
+  ELSE
+    ALTER ROLE \"${quser}\" WITH LOGIN PASSWORD '${qpass}' CREATEDB;
+  END IF;
+END
+\$\$;
+"
+  # Create DB if missing
+  local exists
+  exists="$(kubectl --context "${CNPG_KUBE_CONTEXT}" -n "${CNPG_NAMESPACE}" exec "$(cnpg_primary_pod)" -c postgres -- \
+    psql -U postgres -d postgres -tAc "SELECT 1 FROM pg_database WHERE datname='${qdb}'" | tr -d '[:space:]')"
+  if [[ "$exists" != "1" ]]; then
+    cnpg_admin_psql "CREATE DATABASE \"${qdb}\" OWNER \"${quser}\" ENCODING 'UTF8' LC_COLLATE 'ru_RU.UTF-8' LC_CTYPE 'ru_RU.UTF-8' TEMPLATE template0;"
+  else
+    cnpg_admin_psql "ALTER DATABASE \"${qdb}\" OWNER TO \"${quser}\";"
+  fi
+  cnpg_psql_db "$db" "CREATE EXTENSION IF NOT EXISTS pg_trgm; CREATE EXTENSION IF NOT EXISTS \"uuid-ossp\"; DO \$\$ BEGIN CREATE EXTENSION IF NOT EXISTS vector; EXCEPTION WHEN OTHERS THEN NULL; END \$\$;"
+  install_pooler_auth "$db"
+  cnpg_admin_psql "GRANT ALL PRIVILEGES ON DATABASE \"${qdb}\" TO \"${quser}\";"
+}
+
+install_pooler_auth() {
+  # Must be owned by a superuser (vizorcnpgadmin); tenant-owned SECURITY DEFINER cannot read pg_shadow.
+  local db="$1"
+  cnpg_psql_db "$db" "
+CREATE OR REPLACE FUNCTION public.user_search(uname text)
+ RETURNS TABLE(usename name, passwd text)
+ LANGUAGE sql
+ SECURITY DEFINER
+ SET search_path = pg_catalog
+AS \$\$SELECT usename, passwd FROM pg_catalog.pg_shadow WHERE usename=\$1;\$\$;
+ALTER FUNCTION public.user_search(text) OWNER TO vizorcnpgadmin;
+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;
+"
+}
+
+set_role_superuser() {
+  local user="$1" onoff="$2" # onoff: ON|OFF
+  local quser
+  quser="$(sql_quote "$user")"
+  if [[ "$onoff" == "ON" ]]; then
+    cnpg_admin_psql "ALTER ROLE \"${quser}\" WITH SUPERUSER;"
+  else
+    cnpg_admin_psql "ALTER ROLE \"${quser}\" WITH NOSUPERUSER;"
+  fi
+}
+
+# Apply CNPG endpoint/admin profile by logical cluster name (registry target_cluster).
+# Does not touch TRANSFER_* knobs. Override anytime with CNPG_CLUSTER_PROFILE=...
+apply_cnpg_cluster_profile() {
+  local name="${1:-}"
+  [[ -n "$name" ]] || return 0
+  # normalize aliases
+  case "$name" in
+    wilix|cnpg-wilix|loop-cnpg-wilix) name="loop-cnpg-wilix-1" ;;
+    lm|cnpg-lm) name="loop-cnpg-lm" ;;
+    ff|cnpg-ff) name="loop-cnpg-ff" ;;
+    lp3|lp-3|cnpg-lp3|NEW|new|cnpg-new|loop-cnpg-NEW) name="loop-cnpg-lp3" ;;
+  esac
+  case "$name" in
+    loop-cnpg-wilix-1)
+      CNPG_HOST="${CNPG_WILIX_HOST:-95.143.178.247}"
+      CNPG_PORT="${CNPG_WILIX_PORT:-31574}"
+      CNPG_HOST_RO="${CNPG_WILIX_HOST_RO:-${CNPG_HOST}}"
+      CNPG_PORT_RO="${CNPG_WILIX_PORT_RO:-31575}"
+      CNPG_SSLMODE="${CNPG_WILIX_SSLMODE:-disable}"
+      CNPG_PREFER_CONFIG_HOST="${CNPG_WILIX_PREFER_CONFIG_HOST:-0}"
+      CNPG_KUBE_CONTEXT="${CNPG_WILIX_KUBE_CONTEXT:-local}"
+      CNPG_NAMESPACE="${CNPG_WILIX_NAMESPACE:-cnpg-loop-wilix}"
+      CNPG_CLUSTER="loop-cnpg-wilix-1"
+      CNPG_ADMIN_SECRET="${CNPG_WILIX_ADMIN_SECRET:-cnpg-vizor-admin-role-secret}"
+      CNPG_POOLER_RW="pooler-cnpg-wilix-rw"
+      CNPG_POOLER_RO="pooler-cnpg-wilix-ro"
+      ;;
+    loop-cnpg-lm)
+      CNPG_HOST="${CNPG_LM_HOST:-135.106.201.40}"
+      CNPG_PORT="${CNPG_LM_PORT:-30434}"
+      CNPG_HOST_RO="${CNPG_LM_HOST_RO:-135.106.201.40}"
+      CNPG_PORT_RO="${CNPG_LM_PORT_RO:-30435}"
+      CNPG_SSLMODE="${CNPG_LM_SSLMODE:-disable}"
+      CNPG_PREFER_CONFIG_HOST="${CNPG_LM_PREFER_CONFIG_HOST:-1}"
+      CNPG_KUBE_CONTEXT="${CNPG_LM_KUBE_CONTEXT:-loop-cnpg}"
+      CNPG_NAMESPACE="${CNPG_LM_NAMESPACE:-cnpg-loop-lm}"
+      CNPG_CLUSTER="loop-cnpg-lm"
+      CNPG_ADMIN_SECRET="${CNPG_LM_ADMIN_SECRET:-cnpg-vizor-admin-role-secret}"
+      CNPG_POOLER_RW="pooler-cnpg-lm-rw"
+      CNPG_POOLER_RO="pooler-cnpg-lm-ro"
+      ;;
+    loop-cnpg-ff)
+      CNPG_HOST="${CNPG_FF_HOST:-135.106.201.38}"
+      CNPG_PORT="${CNPG_FF_PORT:-30432}"
+      CNPG_HOST_RO="${CNPG_FF_HOST_RO:-135.106.201.38}"
+      CNPG_PORT_RO="${CNPG_FF_PORT_RO:-30433}"
+      CNPG_SSLMODE="${CNPG_FF_SSLMODE:-disable}"
+      CNPG_PREFER_CONFIG_HOST="${CNPG_FF_PREFER_CONFIG_HOST:-1}"
+      CNPG_KUBE_CONTEXT="${CNPG_FF_KUBE_CONTEXT:-loop-cnpg}"
+      CNPG_NAMESPACE="${CNPG_FF_NAMESPACE:-cnpg-loop-ff}"
+      CNPG_CLUSTER="loop-cnpg-ff"
+      CNPG_ADMIN_SECRET="${CNPG_FF_ADMIN_SECRET:-cnpg-vizor-admin-role-secret}"
+      CNPG_POOLER_RW="pooler-cnpg-ff-rw"
+      CNPG_POOLER_RO="pooler-cnpg-ff-ro"
+      ;;
+    loop-cnpg-lp3)
+      CNPG_HOST="${CNPG_LP3_HOST:-135.106.201.42}"
+      CNPG_PORT="${CNPG_LP3_PORT:-30436}"
+      CNPG_HOST_RO="${CNPG_LP3_HOST_RO:-135.106.201.42}"
+      CNPG_PORT_RO="${CNPG_LP3_PORT_RO:-30437}"
+      CNPG_SSLMODE="${CNPG_LP3_SSLMODE:-disable}"
+      CNPG_PREFER_CONFIG_HOST="${CNPG_LP3_PREFER_CONFIG_HOST:-1}"
+      CNPG_KUBE_CONTEXT="${CNPG_LP3_KUBE_CONTEXT:-loop-cnpg}"
+      CNPG_NAMESPACE="${CNPG_LP3_NAMESPACE:-cnpg-loop-lp3}"
+      CNPG_CLUSTER="loop-cnpg-lp3"
+      CNPG_ADMIN_SECRET="${CNPG_LP3_ADMIN_SECRET:-cnpg-vizor-admin-role-secret}"
+      CNPG_POOLER_RW="pooler-cnpg-lp3-rw"
+      CNPG_POOLER_RO="pooler-cnpg-lp3-ro"
+      ;;
+    *)
+      die "unknown CNPG cluster profile '${name}' (loop-cnpg-wilix-1|loop-cnpg-lm|loop-cnpg-ff|loop-cnpg-lp3)"
+      ;;
+  esac
+  CNPG_CLUSTER_PROFILE="$name"
+  export CNPG_HOST CNPG_PORT CNPG_HOST_RO CNPG_PORT_RO CNPG_SSLMODE CNPG_PREFER_CONFIG_HOST \
+    CNPG_KUBE_CONTEXT CNPG_NAMESPACE CNPG_CLUSTER CNPG_ADMIN_SECRET CNPG_POOLER_RW CNPG_POOLER_RO \
+    CNPG_CLUSTER_PROFILE
+  log "CNPG profile=${CNPG_CLUSTER_PROFILE} rw=${CNPG_HOST}:${CNPG_PORT} ro=${CNPG_HOST_RO}:${CNPG_PORT_RO} admin=${CNPG_KUBE_CONTEXT}/${CNPG_NAMESPACE}/${CNPG_CLUSTER}"
+}
+
+# Pick CNPG profile: explicit CNPG_CLUSTER_PROFILE > registry.target_cluster > logs/<id>.cnpg-cluster
+# Skip when TARGET=selectel or CNPG_SKIP_REGISTRY=1.
+resolve_cnpg_cluster_profile() {
+  local installation="${1:-}"
+  local profile="${CNPG_CLUSTER_PROFILE:-}"
+  local from="env"
+
+  if [[ "${TARGET:-cnpg}" == "selectel" || "${CNPG_SKIP_REGISTRY:-0}" == "1" ]]; then
+    return 0
+  fi
+
+  if [[ -z "$profile" && -n "$installation" ]]; then
+    profile="$(registry_get "$installation" target_cluster 2>/dev/null || true)"
+    [[ -n "$profile" ]] && from="registry"
+  fi
+  if [[ -z "$profile" && -n "$installation" ]]; then
+    local f="${MIGRATE_DIR}/${LOG_DIR}/${installation}.cnpg-cluster"
+    if [[ -f "$f" ]]; then
+      profile="$(tr -d '[:space:]' <"$f")"
+      from="log"
+    fi
+  fi
+  # fall back to current CNPG_CLUSTER name if it looks like a known profile
+  if [[ -z "$profile" && -n "${CNPG_CLUSTER:-}" ]]; then
+    case "${CNPG_CLUSTER}" in
+      loop-cnpg-wilix-1|loop-cnpg-lm|loop-cnpg-ff|loop-cnpg-lp3) profile="${CNPG_CLUSTER}"; from="config" ;;
+    esac
+  fi
+  [[ -n "$profile" ]] || return 0
+  log "Resolving CNPG profile from ${from}: ${profile}"
+  apply_cnpg_cluster_profile "$profile"
+}
+
+# Resolve TARGET (cnpg|selectel) → TARGET_KIND TARGET_HOST TARGET_PORT TARGET_SSLMODE
+resolve_target() {
+  local kind="${1:-${TARGET:-cnpg}}"
+  kind="$(printf '%s' "$kind" | tr '[:upper:]' '[:lower:]')"
+  case "$kind" in
+    cnpg|cn)
+      TARGET_KIND="cnpg"
+      TARGET_HOST="$(resolve_cnpg_host)"
+      TARGET_PORT="${CNPG_PORT}"
+      TARGET_HOST_RO="${CNPG_HOST_RO:-${TARGET_HOST}}"
+      TARGET_PORT_RO="${CNPG_PORT_RO:-${TARGET_PORT}}"
+      TARGET_SSLMODE="${CNPG_SSLMODE}"
+      ;;
+    selectel|sel|managed)
+      TARGET_KIND="selectel"
+      [[ -n "${SELECTEL_HOST:-}" ]] || die "TARGET=selectel requires SELECTEL_HOST in config.env or env"
+      TARGET_HOST="${SELECTEL_HOST}"
+      TARGET_PORT="${SELECTEL_PORT:-5432}"
+      TARGET_HOST_RO="${SELECTEL_HOST_RO:-${TARGET_HOST}}"
+      TARGET_PORT_RO="${SELECTEL_PORT_RO:-${TARGET_PORT}}"
+      TARGET_SSLMODE="${SELECTEL_SSLMODE:-require}"
+      ;;
+    *)
+      die "unknown TARGET='${kind}' (use cnpg or selectel)"
+      ;;
+  esac
+  export TARGET_KIND TARGET_HOST TARGET_PORT TARGET_HOST_RO TARGET_PORT_RO TARGET_SSLMODE
+}
+
+save_installation_target() {
+  local installation="$1" kind="$2"
+  printf '%s\n' "$kind" >"${MIGRATE_DIR}/${LOG_DIR}/${installation}.target"
+  if [[ "$kind" == "cnpg" && -n "${CNPG_CLUSTER_PROFILE:-${CNPG_CLUSTER:-}}" ]]; then
+    printf '%s\n' "${CNPG_CLUSTER_PROFILE:-${CNPG_CLUSTER}}" \
+      >"${MIGRATE_DIR}/${LOG_DIR}/${installation}.cnpg-cluster"
+  fi
+}
+
+load_installation_target() {
+  local installation="$1"
+  local f="${MIGRATE_DIR}/${LOG_DIR}/${installation}.target"
+  if [[ -f "$f" ]]; then
+    tr -d '[:space:]' <"$f"
+    return 0
+  fi
+  printf '%s\n' "${TARGET:-cnpg}"
+}
+
+# Parse migrate-list / CLI extras: [target] [dsn] in any order after installation id.
+# Sets PARSE_TARGET PARSE_DSN (empty if absent).
+parse_target_dsn_args() {
+  PARSE_TARGET=""
+  PARSE_DSN=""
+  PARSE_CNPG_PROFILE=""
+  local a lower
+  for a in "$@"; do
+    [[ -z "$a" ]] && continue
+    lower="$(printf '%s' "$a" | tr '[:upper:]' '[:lower:]')"
+    case "$lower" in
+      cnpg|cn)
+        PARSE_TARGET="cnpg"
+        ;;
+      selectel|sel|managed)
+        PARSE_TARGET="selectel"
+        ;;
+      loop-cnpg-wilix-1|wilix|cnpg-wilix|loop-cnpg-wilix)
+        PARSE_TARGET="cnpg"
+        PARSE_CNPG_PROFILE="loop-cnpg-wilix-1"
+        ;;
+      loop-cnpg-lm|lm|cnpg-lm)
+        PARSE_TARGET="cnpg"
+        PARSE_CNPG_PROFILE="loop-cnpg-lm"
+        ;;
+      loop-cnpg-ff|ff|cnpg-ff)
+        PARSE_TARGET="cnpg"
+        PARSE_CNPG_PROFILE="loop-cnpg-ff"
+        ;;
+      loop-cnpg-lp3|lp3|lp-3|cnpg-lp3|loop-cnpg-new|new|cnpg-new)
+        PARSE_TARGET="cnpg"
+        PARSE_CNPG_PROFILE="loop-cnpg-lp3"
+        ;;
+      postgres://*|postgresql://*)
+        PARSE_DSN="$a"
+        ;;
+      *)
+        die "unexpected argument '${a}' (expected cnpg|selectel|loop-cnpg-lm|loop-cnpg-wilix-1|loop-cnpg-ff|loop-cnpg-lp3 and/or postgres:// DSN)"
+        ;;
+    esac
+  done
+  return 0
+}
+
+build_target_dsn() {
+  local user="$1" pass="$2" db="$3"
+  local host="${4:-${TARGET_HOST}}"
+  local port="${5:-${TARGET_PORT}}"
+  local ssl="${6:-${TARGET_SSLMODE}}"
+  local enc_pass
+  enc_pass="$(url_encode "$pass")"
+  printf 'postgres://%s:%s@%s:%s/%s?connect_timeout=10&sslmode=%s' \
+    "$user" "$enc_pass" "$host" "$port" "$db" "$ssl"
+}
+
+# RO pooler DSN for MM_SQLSETTINGS_DATASOURCEREPLICAS (falls back to RW if RO unset).
+build_target_dsn_ro() {
+  local user="$1" pass="$2" db="$3"
+  build_target_dsn "$user" "$pass" "$db" \
+    "${TARGET_HOST_RO:-${TARGET_HOST}}" \
+    "${TARGET_PORT_RO:-${TARGET_PORT}}" \
+    "${TARGET_SSLMODE}"
+}
+
+# Target-side row counts (users/channels/posts). Uses CNPG superuser or tenant→Selectel Job.
+target_counts() {
+  local ctx="$1" installation="$2" db="$3" user="$4" pass="$5"
+  local kind="${TARGET_KIND:-cnpg}"
+  if [[ "$kind" == "cnpg" ]]; then
+    local pod
+    pod="$(cnpg_primary_pod)"
+    kubectl --context "${CNPG_KUBE_CONTEXT}" -n "${CNPG_NAMESPACE}" exec "$pod" -c postgres -- \
+      psql -U postgres -d "$db" -Atc \
+      "SELECT (SELECT count(*) FROM users)||E'\\t'||(SELECT count(*) FROM channels)||E'\\t'||(SELECT count(*) FROM posts);"
+    return 0
+  fi
+  # selectel: count via short Job with tenant credentials
+  local job secret jfile
+  job="pgtcount-${installation:0:16}-$(date +%s | tail -c 4)"
+  secret="${job}-env"
+  jfile="$(mktemp)"
+  kubectl --context "$ctx" -n "$JOB_NS" create secret generic "$secret" \
+    --from-literal=DST_HOST="$TARGET_HOST" \
+    --from-literal=DST_PORT="$TARGET_PORT" \
+    --from-literal=DST_SSLMODE="$TARGET_SSLMODE" \
+    --from-literal=SRC_USER="$user" \
+    --from-literal=SRC_PASS="$pass" \
+    --from-literal=SRC_DB="$db" \
+    --dry-run=client -o yaml | kubectl --context "$ctx" -n "$JOB_NS" apply -f - >/dev/null
+  JOB_NAME="$job" JOB_NS="$JOB_NS" PG_IMAGE="$PG_IMAGE" SECRET_NAME="$secret" python3 - "$jfile" <<'PY'
+import json, os, sys
+script = r"""set -euo pipefail
+export PGSSLMODE="${DST_SSLMODE}"
+PGPASSWORD="${SRC_PASS}" psql -h "${DST_HOST}" -p "${DST_PORT}" -U "${SRC_USER}" -d "${SRC_DB}" -Atc \
+"SELECT (SELECT count(*) FROM users)||E'\t'||(SELECT count(*) FROM channels)||E'\t'||(SELECT count(*) FROM posts);"
+"""
+json.dump({
+  "apiVersion": "batch/v1", "kind": "Job",
+  "metadata": {"name": os.environ["JOB_NAME"], "namespace": os.environ["JOB_NS"]},
+  "spec": {
+    "ttlSecondsAfterFinished": 600, "backoffLimit": 1,
+    "template": {"spec": {
+      "restartPolicy": "Never",
+      "containers": [{
+        "name": "count", "image": os.environ["PG_IMAGE"],
+        "envFrom": [{"secretRef": {"name": os.environ["SECRET_NAME"]}}],
+        "command": ["bash", "-c"], "args": [script],
+      }],
+    }},
+  },
+}, open(sys.argv[1], "w"))
+PY
+  kubectl --context "$ctx" -n "$JOB_NS" apply -f "$jfile" >/dev/null
+  if wait_job "$ctx" "$JOB_NS" "$job" 600; then
+    kubectl --context "$ctx" -n "$JOB_NS" logs "job/${job}" | tail -1
+  else
+    kubectl --context "$ctx" -n "$JOB_NS" delete job "$job" --ignore-not-found --wait=false >/dev/null 2>&1 || true
+    kubectl --context "$ctx" -n "$JOB_NS" delete secret "$secret" --ignore-not-found --wait=false >/dev/null 2>&1 || true
+    rm -f "$jfile"
+    die "target count job failed for ${installation}"
+  fi
+  kubectl --context "$ctx" -n "$JOB_NS" delete job "$job" --ignore-not-found --wait=false >/dev/null 2>&1 || true
+  kubectl --context "$ctx" -n "$JOB_NS" delete secret "$secret" --ignore-not-found --wait=false >/dev/null 2>&1 || true
+  rm -f "$jfile"
+}
+
+# Wait for a Job. Returns: 0=succeeded, 1=failed, 2=timeout while still active/unknown.
+# timeout=0 (or empty/negative) → wait forever.
+wait_job() {
+  local ctx="$1" ns="$2" job="$3" timeout="${4:-0}"
+  local start now elapsed succeeded failed active phase
+  start="$(date +%s)"
+  local last_log=0
+  while true; do
+    succeeded="$(kubectl --context "$ctx" -n "$ns" get job "$job" -o jsonpath='{.status.succeeded}' 2>/dev/null || echo 0)"
+    failed="$(kubectl --context "$ctx" -n "$ns" get job "$job" -o jsonpath='{.status.failed}' 2>/dev/null || echo 0)"
+    active="$(kubectl --context "$ctx" -n "$ns" get job "$job" -o jsonpath='{.status.active}' 2>/dev/null || echo 0)"
+    if [[ "${succeeded:-0}" -ge 1 ]]; then
+      return 0
+    fi
+    if [[ "${failed:-0}" -ge 1 && "${active:-0}" -eq 0 ]]; then
+      return 1
+    fi
+    now="$(date +%s)"
+    elapsed=$((now - start))
+    if (( now - last_log >= 60 )); then
+      phase="$(kubectl --context "$ctx" -n "$ns" get pods -l "job-name=${job}" \
+        -o jsonpath='{.items[0].status.phase}' 2>/dev/null || echo unknown)"
+      if [[ "${timeout:-0}" -gt 0 ]]; then
+        log "wait_job ${job}: elapsed=${elapsed}s/timeout=${timeout}s active=${active:-0} pod=${phase}"
+      else
+        log "wait_job ${job}: elapsed=${elapsed}s (no timeout) active=${active:-0} pod=${phase}"
+      fi
+      last_log=$now
+    fi
+    if [[ "${timeout:-0}" -gt 0 && "$elapsed" -ge "$timeout" ]]; then
+      if [[ "${active:-0}" -ge 1 ]]; then
+        log "wait_job timeout after ${elapsed}s for ${job} (still active — not marking failed)"
+        return 2
+      fi
+      log "wait_job timeout after ${elapsed}s for ${job}"
+      return 1
+    fi
+    sleep 5
+  done
+}
+
+# Terminate idle backends for a tenant DB (session-pooler leftovers after MM crash).
+terminate_stale_cnpg_sessions() {
+  local db="$1" user="$2" minutes="${3:-${CUTOVER_STALE_IDLE_MINUTES:-5}}"
+  local qdb quser
+  qdb="$(sql_quote "$db")"
+  quser="$(sql_quote "$user")"
+  log "Terminating idle CNPG sessions for ${user}/${db} older than ${minutes}m"
+  cnpg_psql_db "$db" "
+SELECT pg_terminate_backend(pid)
+FROM pg_stat_activity
+WHERE datname = '${qdb}'
+  AND usename = '${quser}'
+  AND state = 'idle'
+  AND pid <> pg_backend_pid()
+  AND state_change < now() - interval '${minutes} minutes';
+DELETE FROM db_lock WHERE true;
+" 2>/dev/null || cnpg_psql_db "$db" "
+SELECT pg_terminate_backend(pid)
+FROM pg_stat_activity
+WHERE datname = '${qdb}'
+  AND usename = '${quser}'
+  AND state = 'idle'
+  AND pid <> pg_backend_pid()
+  AND state_change < now() - interval '${minutes} minutes';
+" || true
+}
+
+# Current MM CR spec.replicas (default 1 if unset).
+mm_get_replicas() {
+  local ctx="$1" installation="$2"
+  local r
+  r="$(kubectl --context "$ctx" -n "$SOURCE_NS" get mm "$installation" \
+    -o jsonpath='{.spec.replicas}' 2>/dev/null || true)"
+  if [[ -z "$r" ]]; then
+    echo 1
+  else
+    echo "$r"
+  fi
+}
+
+# Patch MM CR replicas and scale Deployment immediately (operator may lag).
+mm_set_replicas() {
+  local ctx="$1" installation="$2" replicas="$3"
+  kubectl --context "$ctx" -n "$SOURCE_NS" patch mm "$installation" --type merge \
+    -p "{\"spec\":{\"replicas\":${replicas}}}" >/dev/null
+  if kubectl --context "$ctx" -n "$SOURCE_NS" get deploy "$installation" >/dev/null 2>&1; then
+    kubectl --context "$ctx" -n "$SOURCE_NS" scale deploy "$installation" \
+      --replicas="$replicas" >/dev/null 2>&1 || true
+  fi
+}
+
+# Scale app to 0 before pg_dump so MM releases DB sessions.
+# Saves previous replica count to logs/<id>.scale0-prev and SCALE_0_PREV.
+# Idempotent if already 0. Sets SCALE_0_ACTIVE=1 when we scaled (or were already 0 under --scale-0).
+scale_mm_app_to_zero() {
+  local ctx="$1" installation="$2"
+  local prev wait_sec="${SCALE_0_WAIT_SEC:-180}" i ready
+  prev="$(mm_get_replicas "$ctx" "$installation")"
+  SCALE_0_PREV="$prev"
+  printf '%s\n' "$prev" >"${MIGRATE_DIR}/${LOG_DIR}/${installation}.scale0-prev"
+  SCALE_0_ACTIVE=1
+  export SCALE_0_PREV SCALE_0_ACTIVE
+  if [[ "$prev" -le 0 ]]; then
+    log "scale-0: ${installation} already replicas=0 — leave hibernated during transfer"
+    return 0
+  fi
+  log "scale-0: ${installation} replicas ${prev}→0 (MM CR + deploy)"
+  mm_set_replicas "$ctx" "$installation" 0
+  for i in $(seq 1 "$wait_sec"); do
+    ready="$(kubectl --context "$ctx" -n "$SOURCE_NS" get pods \
+      -l "app=mattermost,installation.mattermost.com/installation=${installation}" \
+      --field-selector=status.phase=Running \
+      -o jsonpath='{.items[*].metadata.name}' 2>/dev/null || true)"
+    if [[ -z "$ready" ]]; then
+      log "scale-0: ${installation} pods stopped (${i}s)"
+      return 0
+    fi
+    sleep 1
+  done
+  log "scale-0: warn — pods still present after ${wait_sec}s, continuing transfer anyway"
+}
+
+# Restore replicas after transfer. No-op unless SCALE_0_ACTIVE=1.
+# Skips restore while transfer Job is still running (KEEP_JOB / active).
+restore_mm_app_replicas() {
+  local ctx="${1:-${CTX:-}}" installation="${2:-${INSTALLATION:-}}"
+  [[ "${SCALE_0_ACTIVE:-0}" == "1" ]] || return 0
+  [[ -n "$ctx" && -n "$installation" ]] || return 0
+  local prev="${SCALE_0_PREV:-}"
+  if [[ -z "$prev" && -f "${MIGRATE_DIR}/${LOG_DIR}/${installation}.scale0-prev" ]]; then
+    prev="$(tr -d '[:space:]' <"${MIGRATE_DIR}/${LOG_DIR}/${installation}.scale0-prev")"
+  fi
+  [[ -n "$prev" ]] || prev=1
+  if [[ "${MIGRATE_KEEP_SCALED:-0}" == "1" ]]; then
+    log "scale-0: ${installation} остаётся 0 до cutover (было ${prev}, сохранено в ${installation}.scale0-prev)"
+    return 0
+  fi
+  if [[ "${KEEP_JOB:-0}" == "1" ]]; then
+    log "scale-0: keeping ${installation} at 0 while Job still running (was ${prev})"
+    return 0
+  fi
+  local active=0
+  if [[ -n "${JOB_NAME:-}" ]]; then
+    active="$(kubectl --context "$ctx" -n "${JOB_NS:-${SOURCE_NS}}" get job "$JOB_NAME" \
+      -o jsonpath='{.status.active}' 2>/dev/null || echo 0)"
+  fi
+  if [[ "${active:-0}" -ge 1 ]]; then
+    log "scale-0: keeping ${installation} at 0 — Job ${JOB_NAME} still active (was ${prev})"
+    return 0
+  fi
+  if [[ "$prev" -le 0 ]]; then
+    log "scale-0: ${installation} was hibernated (0) — leave at 0"
+    SCALE_0_ACTIVE=0
+    rm -f "${MIGRATE_DIR}/${LOG_DIR}/${installation}.scale0-prev"
+    return 0
+  fi
+  log "scale-0: restoring ${installation} replicas 0→${prev}"
+  mm_set_replicas "$ctx" "$installation" "$prev" || true
+  SCALE_0_ACTIVE=0
+  rm -f "${MIGRATE_DIR}/${LOG_DIR}/${installation}.scale0-prev"
+}
+
+# Scale non-current app ReplicaSets to 0 (operator can leave orphans during cutover).
+# Keeps the ReplicaSet with the highest deployment.kubernetes.io/revision among those with replicas>0.
+scale_orphan_replicasets() {
+  local ctx="$1" installation="$2"
+  local rs_json
+  rs_json="$(kubectl --context "$ctx" -n "$SOURCE_NS" get rs \
+    -l "installation.mattermost.com/installation=${installation}" -o json 2>/dev/null || echo '{}')"
+  CTX="$ctx" NS="$SOURCE_NS" python3 -c '
+import json, os, subprocess, sys
+data = json.loads(sys.argv[1])
+ctx, ns = os.environ["CTX"], os.environ["NS"]
+candidates = []
+for rs in data.get("items") or []:
+    name = rs["metadata"]["name"]
+    if "jobserver" in name:
+        continue
+    des = int(rs.get("spec", {}).get("replicas") or 0)
+    if des <= 0:
+        continue
+    rev = int((rs.get("metadata", {}).get("annotations") or {}).get("deployment.kubernetes.io/revision") or 0)
+    ready = int((rs.get("status") or {}).get("readyReplicas") or 0)
+    candidates.append((rev, ready, name, des))
+if len(candidates) <= 1:
+    sys.exit(0)
+candidates.sort(reverse=True)
+keep = candidates[0][2]
+for rev, ready, name, des in candidates[1:]:
+    print("Scaling orphan ReplicaSet %s → 0 (was %s, keep=%s)" % (name, des, keep), flush=True)
+    subprocess.run(
+        ["kubectl", "--context", ctx, "-n", ns, "scale", "rs", name, "--replicas=0"],
+        check=False, stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL,
+    )
+' "$rs_json" || true
+}
+
+# Return host:port from MM app pod DATASOURCE (first running mattermost pod).
+# Falls back to Deployment env when no Running pod (hibernated / starting).
+mm_pod_dsn_hostport() {
+  local ctx="$1" installation="$2"
+  local pod dsn
+  pod="$(kubectl --context "$ctx" -n "$SOURCE_NS" get pods \
+    -l "app=mattermost,installation.mattermost.com/installation=${installation}" \
+    --field-selector=status.phase=Running \
+    -o jsonpath='{.items[0].metadata.name}' 2>/dev/null || true)"
+  if [[ -n "$pod" ]]; then
+    dsn="$(kubectl --context "$ctx" -n "$SOURCE_NS" exec "$pod" -- printenv MM_SQLSETTINGS_DATASOURCE 2>/dev/null || true)"
+  else
+    dsn="$(kubectl --context "$ctx" -n "$SOURCE_NS" get deploy "$installation" \
+      -o jsonpath='{range .spec.template.spec.containers[0].env[?(@.name=="MM_SQLSETTINGS_DATASOURCE")]}{.value}{end}' 2>/dev/null || true)"
+    [[ -n "$dsn" ]] || return 1
+  fi
+  [[ -n "$dsn" ]] || return 1
+  printf '%s' "$dsn" | python3 -c 'import sys,urllib.parse; u=urllib.parse.urlparse(sys.stdin.read().strip());
+print("%s:%s" % (u.hostname or "", u.port or ""))' 2>/dev/null
+}
+
+mm_app_ready() {
+  local ctx="$1" installation="$2"
+  local ready
+  ready="$(kubectl --context "$ctx" -n "$SOURCE_NS" get pods \
+    -l "app=mattermost,installation.mattermost.com/installation=${installation}" \
+    -o jsonpath='{.items[0].status.containerStatuses[0].ready}' 2>/dev/null || echo false)"
+  [[ "$ready" == "true" ]]
+}
+
+mm_https_ping_ok() {
+  local endpoint="$1"
+  [[ -n "$endpoint" ]] || return 1
+  local body
+  body="$(curl -sS -m 8 "https://${endpoint}/api/v4/system/ping" 2>/dev/null || true)"
+  [[ "$body" == *'"status":"OK"'* ]] || [[ "$body" == *'"status": "OK"'* ]]
+}

db/migrate-list.sh

@@ -0,0 +1,123 @@
+#!/usr/bin/env bash
+# Run migrate-one for each line in a list file OR a registry batch.
+#
+# Usage:
+#   ./migrate-list.sh batches/foo.txt
+#   ./migrate-list.sh --batch wave1-lm
+#   PARALLELISM=4 ./migrate-list.sh --batch wave1-lm --status pending,sized
+#   ./migrate-list.sh --batch wave1-lm --scale-0   # hibernate MM during dump, restore after
+set -euo pipefail
+
+DIR="$(cd "$(dirname "${BASH_SOURCE[0]}")" && pwd)"
+# shellcheck disable=SC1091
+source "${DIR}/lib.sh"
+
+LIST=""
+BATCH=""
+# include migrating (stuck mid-run) and size_error so list is registry-driven
+STATUS="pending,sized,failed,size_error,migrating"
+SCALE_0=0
+
+while [[ $# -gt 0 ]]; do
+  case "$1" in
+    --batch) BATCH="${2:-}"; shift 2 ;;
+    --status) STATUS="${2:-}"; shift 2 ;;
+    --scale-0) SCALE_0=1; shift ;;
+    -h|--help)
+      sed -n '2,12p' "$0"
+      exit 0
+      ;;
+    *)
+      LIST="$1"
+      shift
+      ;;
+  esac
+done
+
+if [[ "$SCALE_0" == "1" ]]; then
+  export MIGRATE_SCALE_0=1
+fi
+
+WORK="$(mktemp)"
+trap 'rm -f "$WORK"' EXIT
+
+if [[ -n "$BATCH" ]]; then
+  "${DIR}/registry-export-batch.sh" "$BATCH" --status "$STATUS" -o "$WORK"
+  LIST="$WORK"
+  log "=== migrate-list batch=${BATCH} status=${STATUS} parallelism=${PARALLELISM} scale-0=${SCALE_0} ==="
+elif [[ -n "$LIST" ]]; then
+  [[ -f "$LIST" ]] || die "list file not found: $LIST"
+  # copy filtering comments
+  while IFS= read -r line || [[ -n "$line" ]]; do
+    [[ -z "$line" || "$line" =~ ^[[:space:]]*# ]] && continue
+    echo "$line" >>"${WORK}.raw"
+  done <"$LIST"
+  mv "${WORK}.raw" "$WORK"
+  log "=== migrate-list ${LIST} parallelism=${PARALLELISM} default_target=${TARGET:-cnpg} scale-0=${SCALE_0} ==="
+else
+  LIST="${LIST_PATH}"
+  [[ -f "$LIST" ]] || die "list file not found: $LIST (or pass --batch)"
+  while IFS= read -r line || [[ -n "$line" ]]; do
+    [[ -z "$line" || "$line" =~ ^[[:space:]]*# ]] && continue
+    echo "$line" >>"${WORK}.raw"
+  done <"$LIST"
+  mv "${WORK}.raw" "$WORK"
+  log "=== migrate-list ${LIST} parallelism=${PARALLELISM} default_target=${TARGET:-cnpg} scale-0=${SCALE_0} ==="
+fi
+
+# normalize to TSV ctx/installation/extras
+NORM="$(mktemp)"
+trap 'rm -f "$WORK" "$NORM"' EXIT
+while IFS= read -r line || [[ -n "$line" ]]; do
+  [[ -z "$line" || "$line" =~ ^[[:space:]]*# ]] && continue
+  # shellcheck disable=SC2086
+  set -- $line
+  ctx="${1:-}"; installation="${2:-}"
+  [[ -n "$ctx" && -n "$installation" ]] || continue
+  shift 2 2>/dev/null || true
+  printf '%s\t%s' "$ctx" "$installation" >>"$NORM"
+  for a in "$@"; do
+    printf '\t%s' "$a" >>"$NORM"
+  done
+  printf '\n' >>"$NORM"
+done <"$WORK"
+
+total="$(wc -l <"$NORM" | tr -d ' ')"
+log "Queued ${total} installations"
+[[ "$total" -gt 0 ]] || { log "nothing to migrate"; exit 0; }
+
+run_one() {
+  local ctx="$1" installation="$2"
+  shift 2
+  local extra=()
+  if [[ "$SCALE_0" == "1" ]]; then
+    extra+=(--scale-0)
+  fi
+  # do not pre-set migrating — migrate-one owns status (and may SKIP)
+  "${DIR}/migrate-one.sh" "$ctx" "$installation" "${extra[@]+"${extra[@]}"}" "$@" || log "FAILED $installation (continuing)"
+}
+
+if [[ "${PARALLELISM}" -le 1 ]]; then
+  while IFS=$'\t' read -r ctx installation a3 a4 || [[ -n "${ctx:-}" ]]; do
+    [[ -z "${ctx:-}" ]] && continue
+    args=()
+    if [[ -n "${a3:-}" ]]; then args+=("$a3"); fi
+    if [[ -n "${a4:-}" ]]; then args+=("$a4"); fi
+    run_one "$ctx" "$installation" "${args[@]+"${args[@]}"}"
+  done <"$NORM"
+else
+  while IFS=$'\t' read -r ctx installation a3 a4 || [[ -n "${ctx:-}" ]]; do
+    [[ -z "${ctx:-}" ]] && continue
+    while [[ "$(jobs -pr | wc -l | tr -d ' ')" -ge "${PARALLELISM}" ]]; do
+      sleep 5
+    done
+    args=()
+    if [[ -n "${a3:-}" ]]; then args+=("$a3"); fi
+    if [[ -n "${a4:-}" ]]; then args+=("$a4"); fi
+    run_one "$ctx" "$installation" "${args[@]+"${args[@]}"}" &
+  done <"$NORM"
+  wait || true
+fi
+
+log "=== migrate-list finished ==="
+python3 "${DIR}/registry_lib.py" --registry "${REGISTRY_PATH}" summary || true

db/migrate-one.sh

@@ -0,0 +1,516 @@
+#!/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}"

db/registry-export-batch.sh

@@ -0,0 +1,49 @@
+#!/usr/bin/env bash
+# Export registry rows as migrate-list lines.
+#
+# Usage:
+#   ./registry-export-batch.sh wave1-lm
+#   ./registry-export-batch.sh wave1-lm -o /tmp/w1.txt
+#   ./registry-export-batch.sh wave1-lm --status pending,sized
+#   ./registry-export-batch.sh --target loop-cnpg-lm --status sized
+set -euo pipefail
+
+DIR="$(cd "$(dirname "${BASH_SOURCE[0]}")" && pwd)"
+# shellcheck disable=SC1091
+source "${DIR}/lib.sh"
+
+BATCH=""
+STATUS="pending,sized,failed,size_error"
+TARGET=""
+OUT="-"
+SKIP_DONE=1
+
+while [[ $# -gt 0 ]]; do
+  case "$1" in
+    --status) STATUS="${2:-}"; shift 2 ;;
+    --target) TARGET="${2:-}"; shift 2 ;;
+    --include-done) SKIP_DONE=0; shift ;;
+    -o|--output) OUT="${2:-}"; shift 2 ;;
+    -h|--help)
+      sed -n '2,10p' "$0"
+      exit 0
+      ;;
+    *)
+      if [[ -z "$BATCH" && "$1" != --* ]]; then
+        BATCH="$1"
+        shift
+      else
+        die "unknown arg: $1"
+      fi
+      ;;
+  esac
+done
+
+args=()
+[[ -n "$BATCH" ]] && args+=(--batch "$BATCH")
+[[ -n "$STATUS" ]] && args+=(--status "$STATUS")
+[[ -n "$TARGET" ]] && args+=(--target "$TARGET")
+[[ "$SKIP_DONE" -eq 1 ]] && args+=(--skip-done)
+args+=(-o "$OUT")
+
+python3 "${DIR}/registry_lib.py" --registry "${REGISTRY_PATH}" export "${args[@]}"

db/registry_lib.py

@@ -0,0 +1,476 @@
+#!/usr/bin/env python3
+"""TSV registry helpers for Loop PG migration.
+
+Columns:
+  loop_id, domain, kube_context, yc_cluster, target_cluster,
+  size_bytes, size_pretty, batch_id, status, notes
+"""
+from __future__ import annotations
+
+import argparse
+import csv
+import fcntl
+import os
+import re
+import sys
+import tempfile
+import time
+from contextlib import contextmanager
+from pathlib import Path
+
+COLUMNS = [
+    "loop_id",
+    "domain",
+    "kube_context",
+    "yc_cluster",
+    "target_cluster",
+    "size_bytes",
+    "size_pretty",
+    "batch_id",
+    "status",
+    "notes",
+]
+
+# Fields registry-refresh may overwrite from live scan
+REFRESH_FIELDS = ("domain", "kube_context", "yc_cluster")
+
+# Fields never clobbered by refresh if already set (unless --force-*)
+PRESERVE_IF_SET = ("target_cluster", "batch_id", "status", "size_bytes", "size_pretty", "notes")
+
+STATUSES = {
+    "pending",
+    "sized",
+    "migrating",
+    "migrated",
+    "verified",
+    "cutover",
+    "done",
+    "failed",
+    "size_error",
+    "excluded",
+}
+
+DONE_LIKE = {"migrated", "verified", "cutover", "done"}
+
+
+def registry_path(migrate_dir: Path) -> Path:
+    return migrate_dir / "registry.tsv"
+
+
+def empty_row(loop_id: str = "") -> dict[str, str]:
+    return {c: "" for c in COLUMNS} | {"loop_id": loop_id, "status": "pending"}
+
+
+@contextmanager
+def registry_lock(path: Path, timeout_sec: float = 120.0):
+    """Exclusive flock for registry.tsv (and its done/failed mirrors)."""
+    path.parent.mkdir(parents=True, exist_ok=True)
+    lock_path = path.parent / f"{path.name}.lock"
+    fd = os.open(lock_path, os.O_CREAT | os.O_RDWR, 0o644)
+    deadline = time.monotonic() + timeout_sec
+    try:
+        while True:
+            try:
+                fcntl.flock(fd, fcntl.LOCK_EX | fcntl.LOCK_NB)
+                break
+            except BlockingIOError:
+                if time.monotonic() >= deadline:
+                    fcntl.flock(fd, fcntl.LOCK_EX)  # block past timeout rather than fail cutover
+                    break
+                time.sleep(0.05)
+        yield
+    finally:
+        try:
+            fcntl.flock(fd, fcntl.LOCK_UN)
+        finally:
+            os.close(fd)
+
+
+def _count_data_lines(path: Path) -> int:
+    if not path.exists():
+        return 0
+    n = 0
+    with path.open() as f:
+        for line in f:
+            if not line.strip():
+                continue
+            if line.startswith("loop_id\t") or line.strip() == "loop_id":
+                continue
+            n += 1
+    return n
+
+
+def load(path: Path) -> dict[str, dict[str, str]]:
+    """Load registry.tsv. Tolerates header missing or moved (e.g. after `sort`)."""
+    if not path.exists():
+        return {}
+    size = path.stat().st_size
+    rows: dict[str, dict[str, str]] = {}
+    with path.open(newline="") as f:
+        sample = f.readline()
+        if not sample:
+            return {}
+        # Normal file: first line is the header
+        if sample.startswith("loop_id\t") or sample.strip() == "loop_id":
+            f.seek(0)
+            reader = csv.DictReader(f, delimiter="\t")
+        else:
+            # Header missing or sorted to the bottom — parse every line as data
+            f.seek(0)
+            reader = csv.DictReader(f, fieldnames=COLUMNS, delimiter="\t")
+        for raw in reader:
+            if not raw:
+                continue
+            loop_id = (raw.get("loop_id") or "").strip()
+            if not loop_id or loop_id.startswith("#"):
+                continue
+            # stray header row in the body
+            if loop_id == "loop_id":
+                continue
+            row = empty_row(loop_id)
+            for c in COLUMNS:
+                if c in raw and raw[c] is not None:
+                    row[c] = str(raw[c]).strip()
+            rows[loop_id] = row
+    # Non-empty file that parses to nothing = corruption / wrong format
+    if size >= 500 and not rows:
+        raise RuntimeError(f"registry {path} is {size} bytes but parsed 0 rows — refuse load")
+    return rows
+
+
+def _save_unlocked(path: Path, rows: dict[str, dict[str, str]]) -> None:
+    """Atomic write via unique tempfile (safe under parallel cutover-list)."""
+    path.parent.mkdir(parents=True, exist_ok=True)
+    old_n = _count_data_lines(path)
+    new_n = len(rows)
+    # Hard guard: never collapse a full registry to a handful of rows (parallel race / empty load).
+    # Override: REGISTRY_ALLOW_SHRINK=1
+    allow_shrink = os.environ.get("REGISTRY_ALLOW_SHRINK", "").strip() in ("1", "true", "yes")
+    if not allow_shrink and old_n >= 20 and new_n < max(10, old_n // 2):
+        raise RuntimeError(
+            f"refusing to shrink registry {path} from {old_n} to {new_n} rows "
+            f"(set REGISTRY_ALLOW_SHRINK=1 to override)"
+        )
+    ordered = sorted(
+        rows.values(),
+        key=lambda r: (r.get("batch_id") or "zzz", r.get("kube_context") or "", r["loop_id"]),
+    )
+    fd, tmp_name = tempfile.mkstemp(
+        prefix=f".{path.name}.",
+        suffix=".tmp",
+        dir=str(path.parent),
+    )
+    try:
+        with os.fdopen(fd, "w", newline="") as f:
+            w = csv.DictWriter(f, fieldnames=COLUMNS, delimiter="\t", lineterminator="\n")
+            w.writeheader()
+            for r in ordered:
+                w.writerow({c: r.get(c, "") for c in COLUMNS})
+        os.replace(tmp_name, path)
+    except Exception:
+        try:
+            os.unlink(tmp_name)
+        except OSError:
+            pass
+        raise
+
+
+def save(path: Path, rows: dict[str, dict[str, str]]) -> None:
+    with registry_lock(path):
+        _save_unlocked(path, rows)
+
+
+def upsert(rows: dict[str, dict[str, str]], patch: dict[str, str], *, force: bool = False) -> dict[str, str]:
+    loop_id = patch["loop_id"]
+    cur = rows.get(loop_id) or empty_row(loop_id)
+    for k, v in patch.items():
+        if k not in COLUMNS or v is None:
+            continue
+        v = str(v).strip()
+        if k == "loop_id":
+            cur[k] = v
+            continue
+        if not force and k in PRESERVE_IF_SET and cur.get(k):
+            # allow status overwrite only when explicitly forced or empty
+            if k == "status" and not force:
+                continue
+            if k != "status":
+                continue
+        if v != "" or force:
+            cur[k] = v
+    rows[loop_id] = cur
+    return cur
+
+
+def _write_id_list(path: Path, ids: list[str]) -> None:
+    text = "\n".join(ids) + ("\n" if ids else "")
+    fd, tmp_name = tempfile.mkstemp(prefix=f".{path.name}.", suffix=".tmp", dir=str(path.parent))
+    try:
+        with os.fdopen(fd, "w") as f:
+            f.write(text)
+        os.replace(tmp_name, path)
+    except Exception:
+        try:
+            os.unlink(tmp_name)
+        except OSError:
+            pass
+        raise
+
+
+def _sync_legacy_mirrors(registry: Path, loop_id: str, status: str) -> None:
+    """done.txt / failed.txt are write-only mirrors of registry status. Caller holds registry_lock."""
+    root = registry.parent
+    done_p, fail_p = root / "done.txt", root / "failed.txt"
+
+    def _read_ids(path: Path) -> list[str]:
+        if not path.exists():
+            return []
+        return [x for x in path.read_text().split() if x]
+
+    def _add(path: Path, lid: str) -> None:
+        cur = _read_ids(path)
+        if lid not in cur:
+            cur.append(lid)
+            _write_id_list(path, sorted(cur))
+
+    def _del(path: Path, lid: str) -> None:
+        if not path.exists():
+            return
+        cur = [x for x in _read_ids(path) if x != lid]
+        _write_id_list(path, cur)
+
+    if status in DONE_LIKE:
+        _add(done_p, loop_id)
+        _del(fail_p, loop_id)
+    elif status == "failed":
+        _add(fail_p, loop_id)
+        _del(done_p, loop_id)
+    elif status:
+        _del(done_p, loop_id)
+        _del(fail_p, loop_id)
+
+
+def set_fields(path: Path, loop_id: str, fields: dict[str, str], *, force: bool = True) -> None:
+    # load→upsert→save under one lock so parallel cutover-list does not clobber rows
+    with registry_lock(path):
+        size = path.stat().st_size if path.exists() else 0
+        rows = load(path)
+        # Empty parse of a previously-large file would wipe the registry on save — abort
+        if size >= 500 and not rows:
+            raise RuntimeError(f"registry {path} unreadable ({size} bytes, 0 rows)")
+        if size >= 5000 and len(rows) < 10:
+            raise RuntimeError(
+                f"registry {path} looks truncated ({size} bytes, {len(rows)} rows) — refuse update"
+            )
+        patch = {"loop_id": loop_id, **fields}
+        # Never let force=True clear unrelated columns when caller only patches status/etc.
+        # Only apply non-empty values unless key is status (explicit empty not used).
+        clean = {"loop_id": loop_id}
+        for k, v in fields.items():
+            if k == "loop_id":
+                continue
+            if v is None:
+                continue
+            v = str(v)
+            if v == "" and k != "notes":
+                continue
+            clean[k] = v
+        upsert(rows, clean, force=force)
+        _save_unlocked(path, rows)
+        if "status" in fields:
+            _sync_legacy_mirrors(path, loop_id, fields["status"])
+
+
+def is_done_status(status: str) -> bool:
+    return status in DONE_LIKE
+
+
+# Known YC MDB cluster ids (fallback when `yc` CLI unavailable)
+KNOWN_CID_NAMES = {
+    "c9q9hoccq1gs9jf2g3r9": "LP-PG-1",
+    "c9q87tpl86oinfglqem4": "LP-PG-2",
+    "c9q6tst6nutgn8ju2p5r": "LP-PG-3",
+    "c9q9gqbgbnljg3apqkak": "LP-PG-4",
+    "c9q0a21dumlqo9u3vmm0": "LP-PG-5",
+    "c9qqfe1b3e43gkp4jju5": "LP-PG-6",
+    "c9q4h551pgq3po55f7hu": "LP-PG-FF",
+    "c9qfg20anmmknvmpqiuj": "LP-PG-LM",
+    "c9qfkheuh46d1g731bqi": "postgresql120",
+    "c9qnuaa4qqfrcj11h4rm": "4",
+    "c9qbkifpjolmls057bij": "5",
+    "c9qe5f6bo1c7isk8id9g": "6",
+}
+
+# Original shared pool → restore cluster (host rewrite after source cluster delete/move)
+# old_name → (new_name, old_cid, new_cid)
+RESTORE_REWRITES = {
+    "LP-PG-1": ("postgresql120", "c9q9hoccq1gs9jf2g3r9", "c9qfkheuh46d1g731bqi"),
+    "LP-PG-4": ("4", "c9q9gqbgbnljg3apqkak", "c9qnuaa4qqfrcj11h4rm"),
+    "LP-PG-5": ("5", "c9q0a21dumlqo9u3vmm0", "c9qbkifpjolmls057bij"),
+    "LP-PG-6": ("6", "c9qqfe1b3e43gkp4jju5", "c9qe5f6bo1c7isk8id9g"),
+}
+
+
+def rw_host(cid: str) -> str:
+    return f"c-{cid}.rw.mdb.yandexcloud.net"
+
+
+def restore_host_map() -> dict[str, tuple[str, str]]:
+    """old_fqdn → (new_fqdn, new_cluster_name)."""
+    out: dict[str, tuple[str, str]] = {}
+    for old_name, (new_name, old_cid, new_cid) in RESTORE_REWRITES.items():
+        out[rw_host(old_cid)] = (rw_host(new_cid), new_name)
+    return out
+
+
+def host_to_yc_cluster(host: str, cid_to_name: dict[str, str] | None = None) -> str:
+    host = host or ""
+    if "95.143.178.247" in host:
+        return "CNPG-wilix"
+    if "135.106.201.40" in host:
+        return "CNPG-lm"
+    if "135.106.201.38" in host:
+        return "CNPG-ff"
+    names = {**KNOWN_CID_NAMES, **(cid_to_name or {})}
+    m = re.search(r"c-([a-z0-9]+)\.rw\.mdb", host)
+    if m:
+        cid = m.group(1)
+        return names.get(cid, cid)
+    if "mdb.yandexcloud.net" in host:
+        return f"other-yc:{host.split(':')[0]}"
+    if host:
+        return host.split(":")[0]
+    return ""
+
+
+def cmd_get(args: argparse.Namespace) -> int:
+    rows = load(Path(args.registry))
+    row = rows.get(args.loop_id)
+    if not row:
+        return 1
+    if args.field:
+        print(row.get(args.field, ""))
+    else:
+        print("\t".join(row.get(c, "") for c in COLUMNS))
+    return 0
+
+
+def cmd_set(args: argparse.Namespace) -> int:
+    fields = {}
+    for item in args.field:
+        if "=" not in item:
+            raise SystemExit(f"bad field (want key=value): {item}")
+        k, v = item.split("=", 1)
+        if k not in COLUMNS or k == "loop_id":
+            raise SystemExit(f"invalid field: {k}")
+        fields[k] = v
+    set_fields(Path(args.registry), args.loop_id, fields, force=True)
+    return 0
+
+
+def cmd_export(args: argparse.Namespace) -> int:
+    rows = load(Path(args.registry))
+    out = sys.stdout
+    close = False
+    if args.output and args.output != "-":
+        out = open(args.output, "w")  # noqa: SIM115
+        close = True
+    try:
+        n = 0
+        for r in sorted(rows.values(), key=lambda x: x["loop_id"]):
+            if args.batch and r.get("batch_id") != args.batch:
+                continue
+            if args.status:
+                allowed = set(args.status.split(","))
+                if r.get("status") not in allowed:
+                    continue
+            if args.target and r.get("target_cluster") != args.target:
+                continue
+            if args.skip_done and is_done_status(r.get("status", "")):
+                continue
+            ctx = r.get("kube_context") or ""
+            lid = r.get("loop_id") or ""
+            if not ctx or not lid:
+                continue
+            # migrate-list line; optional target hint as 3rd col if cnpg/selectel
+            line = f"{ctx}  {lid}"
+            if args.with_target_hint and r.get("target_cluster"):
+                line += "  cnpg"
+            out.write(line + "\n")
+            n += 1
+        if args.output and args.output != "-":
+            print(f"exported {n} rows", file=sys.stderr)
+    finally:
+        if close:
+            out.close()
+    return 0
+
+
+def cmd_summary(args: argparse.Namespace) -> int:
+    rows = load(Path(args.registry))
+    from collections import Counter
+
+    by_status = Counter(r.get("status") or "" for r in rows.values())
+    by_batch = Counter(r.get("batch_id") or "(none)" for r in rows.values())
+    by_yc = Counter(r.get("yc_cluster") or "(none)" for r in rows.values())
+    total_bytes = 0
+    sized = 0
+    for r in rows.values():
+        b = r.get("size_bytes") or ""
+        if b.isdigit():
+            total_bytes += int(b)
+            sized += 1
+    print(f"rows={len(rows)} sized={sized} total≈{total_bytes / (1024**3):.1f} GiB")
+    print("status:", dict(by_status))
+    print("batch:", dict(sorted(by_batch.items())))
+    print("yc_cluster top:", by_yc.most_common(15))
+    return 0
+
+
+def cmd_is_done(args: argparse.Namespace) -> int:
+    rows = load(Path(args.registry))
+    row = rows.get(args.loop_id)
+    if row and is_done_status(row.get("status", "")):
+        return 0
+    return 1
+
+
+def main() -> int:
+    p = argparse.ArgumentParser(description="Loop migrate registry.tsv helpers")
+    p.add_argument("--registry", default="", help="path to registry.tsv")
+    sub = p.add_subparsers(dest="cmd", required=True)
+
+    g = sub.add_parser("get", help="print row or one field")
+    g.add_argument("loop_id")
+    g.add_argument("field", nargs="?")
+    g.set_defaults(func=cmd_get)
+
+    s = sub.add_parser("set", help="set fields: key=value …")
+    s.add_argument("loop_id")
+    s.add_argument("field", nargs="+")
+    s.set_defaults(func=cmd_set)
+
+    e = sub.add_parser("export", help="export migrate-list lines")
+    e.add_argument("--batch", default="")
+    e.add_argument("--status", default="", help="comma-separated statuses")
+    e.add_argument("--target", default="")
+    e.add_argument("--skip-done", action="store_true")
+    e.add_argument("--with-target-hint", action="store_true")
+    e.add_argument("-o", "--output", default="-")
+    e.set_defaults(func=cmd_export)
+
+    u = sub.add_parser("summary")
+    u.set_defaults(func=cmd_summary)
+
+    d = sub.add_parser("is-done", help="exit 0 if status is migrated/verified/cutover/done")
+    d.add_argument("loop_id")
+    d.set_defaults(func=cmd_is_done)
+
+    args = p.parse_args()
+    if not args.registry:
+        args.registry = str(Path(__file__).resolve().parent / "registry.tsv")
+    return args.func(args)
+
+
+if __name__ == "__main__":
+    sys.exit(main())

db/verify-list.sh

@@ -0,0 +1,157 @@
+#!/usr/bin/env bash
+# Run verify-one for each line in a list file OR a registry batch.
+#
+# Usage:
+#   ./verify-list.sh --batch wave4-lp-3
+#   ./verify-list.sh batches/foo.txt
+#   PARALLELISM=4 ./verify-list.sh --batch wave4-lp-3
+#   ./verify-list.sh --batch wave4-lp-3 --force   # also re-verify status=verified
+#
+# Without --force: ONLY registry status=migrated (batch export + per-row check).
+# With --force: also queue status=verified (re-verify).
+set -euo pipefail
+
+DIR="$(cd "$(dirname "${BASH_SOURCE[0]}")" && pwd)"
+# shellcheck disable=SC1091
+source "${DIR}/lib.sh"
+
+LIST=""
+BATCH=""
+FORCE=0
+STATUS="migrated"
+
+while [[ $# -gt 0 ]]; do
+  case "$1" in
+    --batch) BATCH="${2:-}"; shift 2 ;;
+    --status)
+      STATUS="${2:-}"
+      shift 2
+      ;;
+    --force) FORCE=1; shift ;;
+    -h|--help)
+      sed -n '2,14p' "$0"
+      exit 0
+      ;;
+    *)
+      LIST="$1"
+      shift
+      ;;
+  esac
+done
+
+if [[ "$FORCE" != "1" ]]; then
+  STATUS="migrated"
+else
+  if [[ "$STATUS" == "migrated" ]]; then
+    STATUS="migrated,verified"
+  fi
+fi
+
+registry_status() {
+  local installation="$1" st
+  st="$(registry_get "$installation" status 2>/dev/null || true)"
+  printf '%s' "$st" | tr -d '[:space:]'
+}
+
+verify_allowed() {
+  local installation="$1" st
+  st="$(registry_status "$installation")"
+  if [[ "$FORCE" == "1" ]]; then
+    case "$st" in
+      migrated|verified) return 0 ;;
+      *)
+        IFS=',' read -r -a allowed <<<"$STATUS"
+        for a in "${allowed[@]}"; do
+          [[ "$st" == "$a" ]] && return 0
+        done
+        return 1
+        ;;
+    esac
+  fi
+  [[ "$st" == "migrated" ]]
+}
+
+WORK="$(mktemp)"
+NORM="$(mktemp)"
+trap 'rm -f "$WORK" "$NORM"' EXIT
+
+if [[ -n "$BATCH" ]]; then
+  # --include-done: migrated/verified are in DONE_LIKE and would be skipped otherwise
+  "${DIR}/registry-export-batch.sh" "$BATCH" --status "$STATUS" --include-done -o "$WORK"
+  log "=== verify-list batch=${BATCH} status=${STATUS} parallelism=${PARALLELISM} force=${FORCE} ==="
+elif [[ -n "$LIST" ]]; then
+  [[ -f "$LIST" ]] || die "list file not found: $LIST"
+  while IFS= read -r line || [[ -n "$line" ]]; do
+    [[ -z "$line" || "$line" =~ ^[[:space:]]*# ]] && continue
+    echo "$line" >>"${WORK}.raw"
+  done <"$LIST"
+  mv "${WORK}.raw" "$WORK"
+  log "=== verify-list ${LIST} parallelism=${PARALLELISM} force=${FORCE} (filter status=${STATUS}) ==="
+else
+  die "usage: $0 --batch <id> | <list_file> [--force]"
+fi
+
+skipped=0
+while IFS= read -r line || [[ -n "$line" ]]; do
+  [[ -z "$line" || "$line" =~ ^[[:space:]]*# ]] && continue
+  # shellcheck disable=SC2086
+  set -- $line
+  ctx="${1:-}"; installation="${2:-}"
+  [[ -n "$ctx" && -n "$installation" ]] || continue
+  shift 2 2>/dev/null || true
+  if ! verify_allowed "$installation"; then
+    st="$(registry_status "$installation")"
+    if [[ "$FORCE" == "1" ]]; then
+      log "SKIP ${installation} (status=${st:-empty}; --force allows migrated|verified)"
+    else
+      log "SKIP ${installation} (status=${st:-empty}; need migrated, or pass --force)"
+    fi
+    skipped=$((skipped + 1))
+    continue
+  fi
+  printf '%s\t%s' "$ctx" "$installation" >>"$NORM"
+  for a in "$@"; do
+    printf '\t%s' "$a" >>"$NORM"
+  done
+  printf '\n' >>"$NORM"
+done <"$WORK"
+
+total="$(wc -l <"$NORM" | tr -d ' ')"
+log "Queued ${total} installations for verify (skipped=${skipped})"
+[[ "$total" -gt 0 ]] || { log "nothing to verify"; exit 0; }
+
+run_one() {
+  local ctx="$1" installation="$2" st
+  shift 2
+  if ! verify_allowed "$installation"; then
+    st="$(registry_status "$installation")"
+    log "SKIP ${installation} (status=${st:-empty} at run time)"
+    return 0
+  fi
+  "${DIR}/verify-one.sh" "$ctx" "$installation" "$@" || log "FAILED verify $installation (continuing)"
+}
+
+if [[ "${PARALLELISM}" -le 1 ]]; then
+  while IFS=$'\t' read -r ctx installation a3 a4 || [[ -n "${ctx:-}" ]]; do
+    [[ -z "${ctx:-}" ]] && continue
+    args=()
+    [[ -n "${a3:-}" ]] && args+=("$a3")
+    [[ -n "${a4:-}" ]] && args+=("$a4")
+    run_one "$ctx" "$installation" "${args[@]+"${args[@]}"}"
+  done <"$NORM"
+else
+  while IFS=$'\t' read -r ctx installation a3 a4 || [[ -n "${ctx:-}" ]]; do
+    [[ -z "${ctx:-}" ]] && continue
+    while [[ "$(jobs -pr | wc -l | tr -d ' ')" -ge "${PARALLELISM}" ]]; do
+      sleep 5
+    done
+    args=()
+    [[ -n "${a3:-}" ]] && args+=("$a3")
+    [[ -n "${a4:-}" ]] && args+=("$a4")
+    run_one "$ctx" "$installation" "${args[@]+"${args[@]}"}" &
+  done <"$NORM"
+  wait || true
+fi
+
+log "=== verify-list finished ==="
+python3 "${DIR}/registry_lib.py" --registry "${REGISTRY_PATH}" summary || true

db/verify-one.sh

@@ -0,0 +1,127 @@
+#!/usr/bin/env bash
+# Compare Users/Channels/Posts counts: source MDB vs target (CNPG or Selectel).
+# CNPG cluster: registry.target_cluster / logs/<id>.cnpg-cluster / CLI profile.
+# Usage: verify-one.sh <kube_context> <installation_id|uuid> [cnpg|selectel|loop-cnpg-lm|…]
+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-lm|loop-cnpg-wilix-1|loop-cnpg-ff]"
+CTX="$1"
+INSTALLATION_RAW="$2"
+shift 2
+parse_target_dsn_args "$@"
+if [[ -n "${PARSE_TARGET}" ]]; then
+  TARGET="${PARSE_TARGET}"
+fi
+if [[ -n "${PARSE_CNPG_PROFILE:-}" ]]; then
+  CNPG_CLUSTER_PROFILE="${PARSE_CNPG_PROFILE}"
+fi
+
+INSTALLATION="$(resolve_installation "$CTX" "$INSTALLATION_RAW")"
+if [[ -z "${PARSE_TARGET:-}" ]]; then
+  TARGET="$(load_installation_target "$INSTALLATION")"
+fi
+
+resolve_cnpg_cluster_profile "$INSTALLATION"
+resolve_target "${TARGET:-cnpg}"
+
+SOURCE_FILE="${DIR}/${LOG_DIR}/${INSTALLATION}.source.dsn"
+if [[ -n "${SOURCE_DSN:-}" ]]; then
+  parse_dsn "$SOURCE_DSN"
+elif [[ -f "$SOURCE_FILE" ]]; then
+  parse_dsn "$(cat "$SOURCE_FILE")"
+else
+  die "No source DSN; run migrate-one first or set SOURCE_DSN="
+fi
+
+DB_NAME="$SRC_DB"
+log "=== verify-one ${INSTALLATION} target=${TARGET_KIND} cnpg=${CNPG_CLUSTER_PROFILE:-${CNPG_CLUSTER:-?}} endpoint=${TARGET_HOST}:${TARGET_PORT} admin=${CNPG_KUBE_CONTEXT}/${CNPG_NAMESPACE}/${CNPG_CLUSTER} ==="
+log "${TARGET_KIND} counts for db=${DB_NAME}"
+tgt="$(target_counts "$CTX" "$INSTALLATION" "$DB_NAME" "$SRC_USER" "$SRC_PASS")"
+log "${TARGET_KIND} users/channels/posts: ${tgt}"
+
+JOB_NAME="pgverify-${INSTALLATION:0:18}-$(date +%s | tail -c 4)"
+SECRET_NAME="${JOB_NAME}-env"
+JOB_FILE="$(mktemp)"
+cleanup() {
+  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
+  rm -f "$JOB_FILE"
+}
+trap cleanup EXIT
+
+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}" \
+  --dry-run=client -o yaml | kubectl --context "$CTX" -n "$JOB_NS" apply -f -
+
+JOB_NAME="$JOB_NAME" JOB_NS="$JOB_NS" PG_IMAGE="$PG_IMAGE" SECRET_NAME="$SECRET_NAME" \
+python3 - "$JOB_FILE" <<'PY'
+import json, os, sys
+script = r"""set -euo pipefail
+export PGSSLMODE="${SRC_SSLMODE:-disable}"
+PGPASSWORD="${SRC_PASS}" psql -h "${SRC_HOST}" -p "${SRC_PORT}" -U "${SRC_USER}" -d "${SRC_DB}" -Atc \
+"SELECT (SELECT count(*) FROM users)||E'\t'||(SELECT count(*) FROM channels)||E'\t'||(SELECT count(*) FROM posts);"
+"""
+job = {
+  "apiVersion": "batch/v1",
+  "kind": "Job",
+  "metadata": {"name": os.environ["JOB_NAME"], "namespace": os.environ["JOB_NS"]},
+  "spec": {
+    "ttlSecondsAfterFinished": 600,
+    "backoffLimit": 1,
+    "template": {"spec": {
+      "restartPolicy": "Never",
+      "containers": [{
+        "name": "verify",
+        "image": os.environ["PG_IMAGE"],
+        "envFrom": [{"secretRef": {"name": os.environ["SECRET_NAME"]}}],
+        "command": ["bash", "-c"],
+        "args": [script],
+      }],
+    }},
+  },
+}
+json.dump(job, open(sys.argv[1], "w"))
+PY
+
+kubectl --context "$CTX" -n "$JOB_NS" apply -f "$JOB_FILE"
+
+MAX_POST_DELTA="${VERIFY_MAX_POST_DELTA:-100}"
+mark_verified() {
+  registry_set "$INSTALLATION" "status=verified" || true
+  if [[ "${TARGET_KIND}" == "cnpg" && -n "${CNPG_CLUSTER_PROFILE:-}" ]]; then
+    registry_set "$INSTALLATION" "target_cluster=${CNPG_CLUSTER_PROFILE}" || true
+  fi
+}
+
+if wait_job "$CTX" "$JOB_NS" "$JOB_NAME" 600; then
+  src="$(kubectl --context "$CTX" -n "$JOB_NS" logs "job/${JOB_NAME}" | tail -1)"
+  log "MDB  users/channels/posts: ${src}"
+  if [[ "$src" == "$tgt" ]]; then
+    log "VERIFY OK counts match exactly"
+    mark_verified
+    exit 0
+  fi
+  python3 - "$src" "$tgt" "$MAX_POST_DELTA" <<'PY'
+import sys
+src, tgt, maxd = sys.argv[1], sys.argv[2], int(sys.argv[3])
+su,sc,sp = map(int, src.split("\t"))
+tu,tc,tp = map(int, tgt.split("\t"))
+if su!=tu or sc!=tc:
+    print(f"users/channels mismatch src={src} tgt={tgt}", file=sys.stderr); sys.exit(2)
+if abs(sp-tp) > maxd:
+    print(f"posts delta {abs(sp-tp)} > {maxd}", file=sys.stderr); sys.exit(2)
+print(f"VERIFY OK users/channels match; posts delta={abs(sp-tp)} (live source)")
+PY
+  mark_verified
+  exit 0
+fi
+die "verify source job failed"