#!/usr/bin/env bash

set -euo pipefail

# init_cloudnative_pg.sh
# Purpose:
# - Distribute administrator ed25519 key pair to CloudNative‑PG as a Kubernetes Secret for cert auth
# - Patch CNPG cluster to enable TLS where possible
#
# Usage:
#   ./init_cloudnative_pg.sh start|stop|status|restart
#   ./init_cloudnative_pg.sh initialize          # install CNPG, create cluster, configure secrets
#   ./init_cloudnative_pg.sh update|reload       # re-apply/patch
#   ./init_cloudnative_pg.sh deploy [version]    # apply CNPG manifest and update image
#   ./init_cloudnative_pg.sh rollout             # rolling restart of CNPG pods
#
# Requirements:
# - init_openbao.sh has been run (OpenBao running as a local container)
# - $PROLE_HOME/env.sh or $HOME/.prole/env.sh defining PROLE_SERVICE
# - Optional: CNPG_MANIFEST_OVERRIDE to apply a recovery manifest instead of kustomize

# Initialize SCRIPT_DIR
SCRIPT_DIR=$(cd "$(dirname "${BASH_SOURCE[0]}")" && pwd)

# Load environment and config via prole_cfg.sh
# shellcheck disable=SC1090
source "$SCRIPT_DIR/prole_cfg.sh"

if [[ "${1:-}" == "--mode" || "${1:-}" == "-m" ]]; then
  prole_set_mode "${2:-}"
  shift 2
elif [[ "${1:-}" == --mode=* || "${1:-}" == -m=* ]]; then
  prole_set_mode "${1#*=}"
  shift
fi

if [[ -z "${PROLE_SERVICE:-}" ]]; then
  echo "ERROR: PROLE_SERVICE is not defined. Provide PROLE_HOME/env.sh or ~/.prole/env.sh" >&2
  exit 1
fi

ACTION=${1:-}
CNPG_CLUSTER_NAME=${CNPG_CLUSTER_NAME:-knoe-db}
VERSION=${2:-latest}
OPENBAO_NAME=${OPENBAO_NAME:-openbao}
REALM=${REALM:-PROLE.ORG}
DOMAIN=${DOMAIN:-prole.org}
CNPG_MANIFEST_OVERRIDE=${CNPG_MANIFEST_OVERRIDE:-}
PROLE_HOME=${PROLE_HOME:-$(cd "$SCRIPT_DIR/.." && pwd)}
BACKUP_DIR=${BACKUP_DIR:-$PROLE_HOME/prole/backup}
BACKUP_WAIT_TIMEOUT=${BACKUP_WAIT_TIMEOUT:-1800}
if [[ "${PROLE_MODE:-}" == "k3s" ]]; then
  CNPG_WAIT_TIMEOUT=${CNPG_WAIT_TIMEOUT:-900}
else
  CNPG_WAIT_TIMEOUT=${CNPG_WAIT_TIMEOUT:-300}
fi
RECOVERY_TEMPLATE="$SCRIPT_DIR/../k8s/prole/knoe-db-recovery.yaml.tpl"
BARMAN_PLUGIN_MANIFEST_URL=${BARMAN_PLUGIN_MANIFEST_URL:-}
BARMAN_PLUGIN_FALLBACK_VERSION=${BARMAN_PLUGIN_FALLBACK_VERSION:-0.9.0}
CERT_MANAGER_MANIFEST_URL=${CERT_MANAGER_MANIFEST_URL:-}
CERT_MANAGER_FALLBACK_VERSION=${CERT_MANAGER_FALLBACK_VERSION:-1.19.3}

# Protected storage requirements (k3s only)
PROLE_CNPG_STORAGE_CLASS=${PROLE_CNPG_STORAGE_CLASS:-synology-iscsi}
PROLE_PROTECTED_DATA_PATH=${PROLE_PROTECTED_DATA_PATH:-/synology/d001/data}
PROLE_PROTECTED_WAL_PATH=${PROLE_PROTECTED_WAL_PATH:-/synology/d001/wal}

if [[ "$ACTION" != "deploy" && "$ACTION" != "rollout" && "$ACTION" != "force-rollout" ]]; then
  if [[ -n "${2:-}" ]]; then
    CNPG_CLUSTER_NAME="${2}"
  fi
  VERSION=${3:-latest}
fi

# Prefer infra-managed manifests (OpenTofu/GitOps) when available so bootstrap matches Ansible.
if [[ -n "${PROLE_HOME:-}" && -d "$PROLE_HOME/deploy/opentofu/k3s/manifests/prole" ]]; then
  K8S_PROLE_DIR="$PROLE_HOME/deploy/opentofu/k3s/manifests/prole"
elif [[ -d "$SCRIPT_DIR/../deploy/opentofu/k3s/manifests/prole" ]]; then
  K8S_PROLE_DIR="$SCRIPT_DIR/../deploy/opentofu/k3s/manifests/prole"
elif [[ -d "$SCRIPT_DIR/../k8s/prole" ]]; then
  K8S_PROLE_DIR="$SCRIPT_DIR/../k8s/prole"
elif [[ -n "${PROLE_HOME:-}" && -d "$PROLE_HOME/k8s/prole" ]]; then
  K8S_PROLE_DIR="$PROLE_HOME/k8s/prole"
else
  K8S_PROLE_DIR="$SCRIPT_DIR/../k8s/prole"
fi
CNPG_MANIFEST="$K8S_PROLE_DIR/knoe-db.yaml"
BARMAN_OBJECTSTORE_MANIFEST="$K8S_PROLE_DIR/knoe-db-barman-objectstore.yaml"

# Secrets and token locations
# Use PROLE_SERVICE if writable locally, otherwise fallback to ~/.prole
if [[ -n "${PROLE_SERVICE:-}" ]] && _prole_usable_dir "${PROLE_SERVICE}/secrets" >/dev/null; then
  SECRETS_DIR="$PROLE_SERVICE/secrets"
else
  SECRETS_DIR="$HOME/.prole/etc/secrets"
fi
mkdir -p "$SECRETS_DIR" 2>/dev/null || true

# Resolving CNPG admin keys.
# We prefer names without algorithm suffixes to be more generic, matching install.py fallback strategy.
ADMIN_PRIV_ED25519="$SECRETS_DIR/admin_ed25519.key"
ADMIN_PUB_ED25519="$SECRETS_DIR/admin_ed25519.pub"
ADMIN_PRIV_GENERIC="$SECRETS_DIR/admin.key"
ADMIN_PUB_GENERIC="$SECRETS_DIR/admin.pub"
OPENBAO_TOKEN_FILE="$SECRETS_DIR/openbao-root-token"
# NAMESPACE is always derived from prole.cfg via PROLE_NAMESPACE — never from the environment
NAMESPACE="${PROLE_NAMESPACE}"
BAO_NAMESPACE="${PROLE_NAMESPACE}"
OPENBAO_NAMESPACE="${OPENBAO_NAMESPACE:-${SERVICE_NAMESPACE:-${PROLE_NAMESPACE}}}"
BAO_PATH_PREFIX="prole/${BAO_NAMESPACE}"
BAO_PATH_ADMIN="${BAO_PATH_PREFIX}/admin"
BAO_PATH_DB="${BAO_PATH_PREFIX}/db"
BAO_PATH_MONITORING="${BAO_PATH_PREFIX}/monitoring"

ensure_tools() {
  for t in kubectl curl openssl base64 jq; do
    command -v "$t" >/dev/null || { echo "Missing required tool: $t" >&2; exit 1; }
  done
}

_require_tool() {
  local t="$1"
  command -v "$t" >/dev/null 2>&1 || { echo "ERROR: Missing required tool: $t" >&2; return 1; }
}

_require_prole_protected_mount() {
  local path="$1" label="$2"
  local findmnt_bin="${PROLE_FINDMNT_BIN:-findmnt}"

  if [[ ! -d "$path" ]]; then
    echo "ERROR: Protected ${label} path missing: ${path}" >&2
    echo "Refusing CNPG deployment: Prole protected storage must be mounted at ${PROLE_PROTECTED_DATA_PATH} and ${PROLE_PROTECTED_WAL_PATH}." >&2
    return 1
  fi

  _require_tool "$findmnt_bin" || {
    echo "ERROR: '${findmnt_bin}' is required to validate protected storage mounts (k3s mode)." >&2
    return 1
  }

  local root_src path_src path_target
  root_src=$($findmnt_bin -T / -n -o SOURCE 2>/dev/null || true)
  path_src=$($findmnt_bin -T "$path" -n -o SOURCE 2>/dev/null || true)
  path_target=$($findmnt_bin -T "$path" -n -o TARGET 2>/dev/null || true)

  if [[ -z "$root_src" || -z "$path_src" || -z "$path_target" ]]; then
    echo "ERROR: Unable to determine mount backing device for '${path}' via ${findmnt_bin}." >&2
    echo "root_src='${root_src:-<empty>}' path_src='${path_src:-<empty>}' path_target='${path_target:-<empty>}'" >&2
    return 1
  fi

  if [[ "$path_target" == "/" || "$path_src" == "$root_src" ]]; then
    echo "ERROR: Protected ${label} path '${path}' is backed by the root filesystem (target='${path_target}', source='${path_src}')." >&2
    echo "Refusing CNPG deployment to prevent root/SD-backed storage from being used." >&2
    return 1
  fi
  return 0
}

ensure_prole_protected_storage() {
  if [[ "${PROLE_MODE:-}" != "k3s" ]]; then
    return 0
  fi

  echo "Validating protected Prole storage mounts (k3s mode) ..."
  _require_prole_protected_mount "$PROLE_PROTECTED_DATA_PATH" "data" || return 1
  _require_prole_protected_mount "$PROLE_PROTECTED_WAL_PATH" "WAL" || return 1
  return 0
}

validate_cnpg_manifest_storage() {
  local rendered_manifest="$1"

  if [[ "${PROLE_MODE:-}" != "k3s" ]]; then
    return 0
  fi

  if [[ ! -f "$rendered_manifest" ]]; then
    echo "ERROR: Rendered CNPG manifest not found: $rendered_manifest" >&2
    return 1
  fi

  # Guardrail: never allow default/local-path to appear in the CNPG Cluster manifest in k3s mode.
  if grep -Eq '^[[:space:]]*storageClassName:[[:space:]]*local-path([[:space:]]|$)' "$rendered_manifest"; then
    echo "ERROR: CNPG manifest references storageClassName=local-path (forbidden in k3s mode)." >&2
    return 1
  fi

  local in_cluster=0 block="" data_sc="" wal_sc="" data_role="" wal_role=""
  while IFS= read -r line || [[ -n "$line" ]]; do
    if [[ "$line" =~ ^---[[:space:]]*$ ]]; then
      in_cluster=0
      block=""
      continue
    fi
    if [[ "$line" =~ ^kind:[[:space:]]*Cluster[[:space:]]*$ ]]; then
      in_cluster=1
      block=""
      continue
    fi
    (( in_cluster )) || continue

    if [[ "$line" =~ ^[[:space:]]*storage:[[:space:]]*$ ]]; then
      block="storage"
      continue
    fi
    if [[ "$line" =~ ^[[:space:]]*walStorage:[[:space:]]*$ ]]; then
      block="wal"
      continue
    fi
    # When we leave the storage-related blocks (next top-level under spec), stop attributing matches.
    if [[ "$line" =~ ^[[:space:]]{2}[a-zA-Z0-9_-]+:[[:space:]]*$ ]]; then
      case "$line" in
        "  storage:"|"  walStorage:") ;;
        *) block="" ;;
      esac
    fi

    if [[ -n "$block" && "$line" =~ storageClassName: ]]; then
      local v
      v=$(printf '%s' "$line" | sed -E 's/^.*storageClassName:[[:space:]]*//; s/[[:space:]]+$//')
      if [[ "$block" == "storage" && -z "$data_sc" ]]; then
        data_sc="$v"
      elif [[ "$block" == "wal" && -z "$wal_sc" ]]; then
        wal_sc="$v"
      fi
    fi
    if [[ -n "$block" && "$line" =~ synology\.storage/role: ]]; then
      local r
      r=$(printf '%s' "$line" | sed -E 's/^.*synology\.storage\/role:[[:space:]]*//; s/[[:space:]]+$//')
      if [[ "$block" == "storage" && -z "$data_role" ]]; then
        data_role="$r"
      elif [[ "$block" == "wal" && -z "$wal_role" ]]; then
        wal_role="$r"
      fi
    fi
  done < "$rendered_manifest"

  if [[ "$data_sc" != "$PROLE_CNPG_STORAGE_CLASS" || "$wal_sc" != "$PROLE_CNPG_STORAGE_CLASS" ]]; then
    echo "ERROR: CNPG Cluster manifest must set storageClassName=${PROLE_CNPG_STORAGE_CLASS} for both data and walStorage (got data='${data_sc:-<missing>}', wal='${wal_sc:-<missing>}')." >&2
    echo "Refusing deployment to prevent fallback to default storage class." >&2
    return 1
  fi

  if [[ "$data_role" != "data" || "$wal_role" != "wal" ]]; then
    echo "ERROR: CNPG Cluster manifest must include explicit PV selectors for Prole storage (synology.storage/role=data and synology.storage/role=wal)." >&2
    echo "Got selector roles: data='${data_role:-<missing>}', wal='${wal_role:-<missing>}'" >&2
    return 1
  fi

  return 0
}

validate_cnpg_runtime_storage() {
  if [[ "${PROLE_MODE:-}" != "k3s" ]]; then
    return 0
  fi

  local sc="$PROLE_CNPG_STORAGE_CLASS"
  local cluster_json
  cluster_json=$(kubectl -n "$NAMESPACE" get cluster "$CNPG_CLUSTER_NAME" -o json 2>/dev/null || true)
  if [[ -z "$cluster_json" ]]; then
    echo "ERROR: Unable to fetch CNPG Cluster '${CNPG_CLUSTER_NAME}' for runtime storage validation." >&2
    return 1
  fi

  local data_sc wal_sc data_role wal_role
  data_sc=$(printf '%s' "$cluster_json" | jq -r '.spec.storage.pvcTemplate.storageClassName // empty')
  wal_sc=$(printf '%s' "$cluster_json" | jq -r '.spec.walStorage.pvcTemplate.storageClassName // empty')
  data_role=$(printf '%s' "$cluster_json" | jq -r '.spec.storage.pvcTemplate.selector.matchLabels["synology.storage/role"] // empty')
  wal_role=$(printf '%s' "$cluster_json" | jq -r '.spec.walStorage.pvcTemplate.selector.matchLabels["synology.storage/role"] // empty')

  if [[ "$data_sc" != "$sc" || "$wal_sc" != "$sc" ]]; then
    echo "ERROR: Live CNPG Cluster storageClassName must be '${sc}' for both data and walStorage (got data='${data_sc:-<missing>}', wal='${wal_sc:-<missing>}')." >&2
    return 1
  fi
  if [[ "$data_role" != "data" || "$wal_role" != "wal" ]]; then
    echo "ERROR: Live CNPG Cluster must select Prole PVs via selector labels (got data role='${data_role:-<missing>}', wal role='${wal_role:-<missing>}')." >&2
    return 1
  fi

  # Wait for CNPG PVCs to exist so we can validate binding and PV paths.
  local timeout=${CNPG_STORAGE_VALIDATE_TIMEOUT:-120}
  local start now elapsed pvc_json pvc_count
  start=$(date +%s)
  while true; do
    pvc_json=$(kubectl -n "$NAMESPACE" get pvc -l "cnpg.io/cluster=${CNPG_CLUSTER_NAME}" -o json 2>/dev/null || true)
    pvc_count=$(printf '%s' "$pvc_json" | jq -r '.items | length' 2>/dev/null || echo 0)
    if [[ "$pvc_count" =~ ^[0-9]+$ ]] && (( pvc_count > 0 )); then
      break
    fi
    now=$(date +%s)
    elapsed=$(( now - start ))
    if (( elapsed > timeout )); then
      echo "ERROR: Timed out (${timeout}s) waiting for CNPG PVCs to appear for cluster '${CNPG_CLUSTER_NAME}'." >&2
      return 1
    fi
    sleep 3
  done

  local bad_pvc
  bad_pvc=$(printf '%s' "$pvc_json" | jq -r --arg sc "$sc" '.items[] | select((.spec.storageClassName // "") != $sc) | .metadata.name' | head -n 1)
  if [[ -n "$bad_pvc" ]]; then
    echo "ERROR: CNPG PVC '${bad_pvc}' is not using storageClassName='${sc}'. Refusing deployment." >&2
    kubectl -n "$NAMESPACE" get pvc -l "cnpg.io/cluster=${CNPG_CLUSTER_NAME}" -o wide >&2 || true
    return 1
  fi

  local pv
  while IFS= read -r pv || [[ -n "$pv" ]]; do
    [[ -n "$pv" ]] || continue
    local pv_json pv_sc pv_path
    pv_json=$(kubectl get pv "$pv" -o json 2>/dev/null || true)
    pv_sc=$(printf '%s' "$pv_json" | jq -r '.spec.storageClassName // empty')
    pv_path=$(printf '%s' "$pv_json" | jq -r '.spec.local.path // empty')
    if [[ "$pv_sc" != "$sc" ]]; then
      echo "ERROR: Bound PV '${pv}' has storageClassName='${pv_sc:-<missing>}' (expected '${sc}')." >&2
      return 1
    fi
    if [[ -z "$pv_path" ]]; then
      echo "ERROR: Bound PV '${pv}' is missing .spec.local.path; cannot verify protected storage path." >&2
      return 1
    fi
    if [[ "$pv_path" == /var/lib/rancher/k3s/storage/* ]]; then
      echo "ERROR: Bound PV '${pv}' points into k3s local-path storage (${pv_path}). Refusing deployment." >&2
      return 1
    fi
    case "$pv_path" in
      /synology/*) ;;
      *)
        echo "ERROR: Bound PV '${pv}' path '${pv_path}' is not under /synology/. Refusing deployment." >&2
        return 1
        ;;
    esac
  done < <(printf '%s' "$pvc_json" | jq -r '.items[].spec.volumeName // empty')

  return 0
}

ensure_local_admin_keypair() {
  mkdir -p "$SECRETS_DIR" 2>/dev/null || true

  # If generic already exists, prefer it.
  if [[ -f "$ADMIN_PRIV_GENERIC" && -f "$ADMIN_PUB_GENERIC" ]]; then
    # Ensure legacy mirror exists for compatibility.
    cp "$ADMIN_PRIV_GENERIC" "$ADMIN_PRIV_ED25519" 2>/dev/null || true
    cp "$ADMIN_PUB_GENERIC" "$ADMIN_PUB_ED25519" 2>/dev/null || true
    return 0
  fi

  # If legacy exists, normalize it into generic.
  if [[ -f "$ADMIN_PRIV_ED25519" && -f "$ADMIN_PUB_ED25519" ]]; then
    cp "$ADMIN_PRIV_ED25519" "$ADMIN_PRIV_GENERIC" 2>/dev/null || true
    cp "$ADMIN_PUB_ED25519" "$ADMIN_PUB_GENERIC" 2>/dev/null || true
    chmod 0600 "$ADMIN_PRIV_GENERIC" 2>/dev/null || true
    return 0
  fi

  echo "Generating CNPG admin keypair in $SECRETS_DIR ..."
  # Attempt ed25519, fallback to rsa if not available
  if ! openssl genpkey -algorithm ED25519 -out "$ADMIN_PRIV_GENERIC" 2>/dev/null; then
    echo "WARN: ED25519 not supported by openssl, falling back to RSA 4096..." >&2
    openssl genpkey -algorithm RSA -pkeyopt rsa_keygen_bits:4096 -out "$ADMIN_PRIV_GENERIC"
  fi
  openssl pkey -in "$ADMIN_PRIV_GENERIC" -pubout -out "$ADMIN_PUB_GENERIC"
  chmod 0600 "$ADMIN_PRIV_GENERIC"

  # Mirror to legacy filenames for compatibility with other scripts.
  cp "$ADMIN_PRIV_GENERIC" "$ADMIN_PRIV_ED25519" 2>/dev/null || true
  cp "$ADMIN_PUB_GENERIC" "$ADMIN_PUB_ED25519" 2>/dev/null || true
  return 0
}

_kubectl_is_transient_error() {
  local msg="${1:-}"
  # Errors we frequently see on k3s under load / during restart.
  # These are worth retrying rather than proceeding with partial state.
  printf '%s' "$msg" | grep -Eqi \
    '(failed to download openapi|apiserver not ready|the server is currently unable to handle the request|connect: connection refused|context deadline exceeded|i/o timeout|no endpoints available for service "cnpg-webhook-service"|failed calling webhook)'
}

wait_for_apiserver_ready() {
  local timeout=${1:-120}
  local start_time elapsed
  start_time=$(date +%s)

  echo "Waiting for Kubernetes API server to be ready (timeout: ${timeout}s)..."
  while true; do
    # Prefer /readyz when available; fall back to kubectl version.
    if kubectl --request-timeout=5s get --raw='/readyz' >/dev/null 2>&1; then
      return 0
    fi
    if kubectl --request-timeout=5s version --short >/dev/null 2>&1; then
      return 0
    fi

    elapsed=$(( $(date +%s) - start_time ))
    if (( elapsed > timeout )); then
      echo "ERROR: Kubernetes API server not ready after ${elapsed}s." >&2
      echo "Check kubeconfig/context and k3s status, then retry." >&2
      return 1
    fi
    if (( elapsed % 20 < 6 && elapsed > 5 )); then
      echo "  [${elapsed}s/${timeout}s] Waiting for API server..."
    fi
    sleep 2
  done
}

ensure_namespace() {
  if [[ -z "${PROLE_NAMESPACE:-}" ]]; then
    echo "ERROR: PROLE_NAMESPACE is empty. Set NAMESPACE in conf/prole.cfg." >&2
    exit 1
  fi
  if ! kubectl get namespace "$PROLE_NAMESPACE" >/dev/null 2>&1; then
    echo "Creating namespace '$PROLE_NAMESPACE' ..."
    kubectl create namespace "$PROLE_NAMESPACE" >/dev/null 2>&1 || true
  fi
}

kubectl_apply_retry() {
  local namespace="${1:-}"
  local attempts=${2:-8}
  local sleep_s=${3:-5}
  local i out

  for ((i=1; i<=attempts; i++)); do
    if [[ -n "$namespace" ]]; then
      if out=$(kubectl apply -n "$namespace" -f - 2>&1); then
        printf '%s\n' "$out"
        return 0
      fi
    else
      if out=$(kubectl apply -f - 2>&1); then
        printf '%s\n' "$out"
        return 0
      fi
    fi

    if _kubectl_is_transient_error "$out"; then
      echo "WARN: kubectl apply failed due to transient API/webhook issue (attempt $i/$attempts); retrying..." >&2
      echo "$out" >&2
      sleep "$sleep_s"
      continue
    fi

    echo "$out" >&2
    return 1
  done

  echo "ERROR: kubectl apply failed after $attempts attempts." >&2
  echo "$out" >&2
  return 1
}

get_latest_image() {
  local pg_version_file release_file pg_version release
  if [[ -f "$SCRIPT_DIR/../conf/postgresql/.version" ]]; then
    pg_version_file="$SCRIPT_DIR/../conf/postgresql/.version"
  elif [[ -n "${PROLE_HOME:-}" && -f "$PROLE_HOME/conf/postgresql/.version" ]]; then
    pg_version_file="$PROLE_HOME/conf/postgresql/.version"
  else
    pg_version_file="$SCRIPT_DIR/../conf/postgresql/.version"
  fi

  if [[ -f "$SCRIPT_DIR/../knoe-db/.version" ]]; then
    release_file="$SCRIPT_DIR/../knoe-db/.version"
  elif [[ -n "${PROLE_HOME:-}" && -f "$PROLE_HOME/knoe-db/.version" ]]; then
    release_file="$PROLE_HOME/knoe-db/.version"
  else
    release_file="$SCRIPT_DIR/../knoe-db/.version"
  fi

  if [[ -f "$pg_version_file" ]]; then
    pg_version=$(tr -d '[:space:]' < "$pg_version_file")
  else
    pg_version="17.7"
  fi

  if [[ -f "$release_file" ]]; then
    release=$(tr -d '[:space:]' < "$release_file")
  else
    release="43"
  fi

  if [[ "$release" =~ ^[0-9]+$ ]]; then
    release=$(printf "%03d" "$release")
  fi

  echo "knoe-db:${pg_version}-${release}"
}

resolve_cnpg_image() {
  local image="$1"
  if _prole_local_registry_enabled; then
    # Prefer the in-cluster/internal registry for k3d/k3s nodes pulling images.
    # Using LOCAL_REGISTRY (e.g., localhost:5000) breaks inside cluster and may
    # also prefer IPv6 ::1, leading to connection refused. Avoid it.
    local registry=""

    _cnpg_registry_looks_like_k3d() {
      local r="${1:-}"
      [[ -n "$r" ]] || return 1
      case "$r" in
        k3d-*|*"/k3d-"*) return 0 ;;
      esac
      return 1
    }

    _cnpg_registry_is_localhostish() {
      local r="${1:-}"
      [[ -n "$r" ]] || return 1
      case "$r" in
        localhost:5000|127.0.0.1:5000|*.localhost|*.localhost:5000) return 0 ;;
      esac
      return 1
    }

    _cnpg_k3s_internal_registry() {
      # For k3s, never use k3d registry names or localhost-ish endpoints.
      local r="${LOCAL_REGISTRY_INTERNAL:-}"
      if _cnpg_registry_looks_like_k3d "$r" || _cnpg_registry_is_localhostish "$r"; then
        r=""
      fi
      if [[ -n "$r" ]]; then
        printf '%s' "$r"
        return 0
      fi

      local host=""
      if command -v _prole_host_from_url >/dev/null 2>&1; then
        host=$(_prole_host_from_url "${PROLE_K3S_SERVER:-${K3S_SERVER_URL:-}}")
      fi
      if [[ -n "${host:-}" ]]; then
        printf '%s' "${host}:5000"
        return 0
      fi

      local ns="${SERVICE_NAMESPACE:-${PROLE_NAMESPACE:-default}}"
      printf '%s' "registry.${ns}.svc.cluster.local:5000"
    }

    local _mode
    _mode=$(prole_normalize_mode "${PROLE_MODE:-${DEPLOYMENT_MODE:-}}")
    case "${_mode}" in
      k3s)
        registry=$(_cnpg_k3s_internal_registry)
        ;;
      *)
        if [[ -n "${LOCAL_REGISTRY_INTERNAL:-}" ]]; then
          registry="${LOCAL_REGISTRY_INTERNAL}"
        fi
        ;;
    esac

    if [[ -z "$registry" && -n "${LOCAL_REGISTRY:-}" ]]; then
      # Only fall back to LOCAL_REGISTRY when INTERNAL is not available
      # and it's not pointing at localhost (which is invalid for cluster pulls).
      if [[ "${LOCAL_REGISTRY}" != "localhost:5000" && "${LOCAL_REGISTRY}" != "127.0.0.1:5000" ]]; then
        registry="${LOCAL_REGISTRY}"
      fi
    fi

    if [[ -n "$registry" ]]; then
      local first="${image%%/*}"
      # If the image is unqualified (no registry), prefix it with the chosen registry
      if [[ "$image" != */* ]]; then
        image="${registry}/${image}"
      # If the first path segment has no dot/colon, it's still unqualified (e.g., knoe-db:TAG)
      elif [[ "$first" != *"."* && "$first" != *":"* ]]; then
        image="${registry}/${image}"
      fi
    fi
  fi
  printf '%s' "$image"
}

sync_manifest_image() {
  local image="$1"
  local files=()
  if [[ -n "$CNPG_MANIFEST" ]]; then
    files+=("$CNPG_MANIFEST")
  fi
  if [[ -f "$RECOVERY_TEMPLATE" ]]; then
    files+=("$RECOVERY_TEMPLATE")
  fi
  local f tmp
  for f in "${files[@]}"; do
    if [[ -f "$f" ]] && grep -qE '^[[:space:]]*imageName:' "$f"; then
      tmp=$(mktemp)
      sed -E "s|^([[:space:]]*imageName:).*|\\1 ${image}|" "$f" > "$tmp"
      mv "$tmp" "$f"
    fi
  done
}

openbao_url() {
  if [[ -n "${PROLE_OPENBAO_URL:-}" ]]; then
    echo "$PROLE_OPENBAO_URL"
    return 0
  fi
  if prole_is_in_cluster; then
    echo "http://openbao.${OPENBAO_NAMESPACE}.svc.cluster.local:8200"
    return 0
  fi

  # In k3s mode, scripts run outside the cluster must reach OpenBao via the k3s host
  # (never via localhost or kubectl port-forward).
  if [[ "${PROLE_MODE:-${DEPLOYMENT_MODE:-}}" == "k3s" ]]; then
    if command -v _prole_host_from_url >/dev/null 2>&1; then
      local host
      host=$(_prole_host_from_url "${PROLE_K3S_SERVER:-${K3S_SERVER_URL:-}}")
      if [[ -n "${host:-}" ]]; then
        # If the LoadBalancer has no external IP, k3s exposes the service via a nodePort.
        # Prefer that nodePort when available.
        if command -v kubectl >/dev/null 2>&1 && command -v jq >/dev/null 2>&1; then
          local ns node_port
          ns="${OPENBAO_NAMESPACE:-${SERVICE_NAMESPACE:-default}}"
          node_port=$(kubectl -n "$ns" get svc "${OPENBAO_NAME:-openbao}" -o json 2>/dev/null | jq -r '.spec.ports[] | select(.port==8200) | .nodePort // empty' | head -n 1)
          if [[ -n "${node_port:-}" && "${node_port:-}" != "null" ]]; then
            echo "http://${host}:${node_port}"
            return 0
          fi
        fi
        echo "http://${host}:8200"
        return 0
      fi
    fi
    echo ""
    return 0
  fi

  if curl -sS "http://127.0.0.1:8200/v1/sys/health" >/dev/null 2>&1; then
    echo "http://127.0.0.1:8200"
    return 0
  fi
  echo ""
  return 0
}

openbao_token() {
  if [[ -f "$OPENBAO_TOKEN_FILE" ]]; then
    cat "$OPENBAO_TOKEN_FILE"
  else
    echo "${OPENBAO_ROOT_TOKEN:-}"
  fi
}

fetch_openbao_secret() {
  local path="$1"
  local key="$2"
  local token url
  token=$(openbao_token)
  url=$(openbao_url)
  if [[ -z "$token" || -z "$url" ]]; then
    echo ""
    return 0
  fi
  curl -sS -H "X-Vault-Token: $token" "$url/v1/kv/data/$path" | jq -r ".data.data.\"$key\"" || echo ""
}

resolve_db_password() {
  local db_pw="${DB_PASSWORD:-}"
  if [[ -z "$db_pw" || "$db_pw" == '${OPENBAO:'* || "$db_pw" == '${PROLE_SECRET:'* ]]; then
    local fetched_db
    fetched_db=$(fetch_openbao_secret "$BAO_PATH_DB" "password")
    if [[ -n "$fetched_db" && "$fetched_db" != "null" ]]; then
      db_pw="$fetched_db"
    fi
  fi
  printf '%s' "$db_pw"
}


ensure_cnpg_operator() {
  if kubectl get deployment -n cnpg-system cnpg-controller-manager >/dev/null 2>&1; then
    kubectl -n cnpg-system rollout status deploy/cnpg-controller-manager --timeout=180s || true
    wait_for_cnpg_webhook 180 || true
    if [[ "${PROLE_MODE:-}" == "k3s" ]]; then
      tune_cnpg_operator_for_k3s || true
      kubectl -n cnpg-system rollout status deploy/cnpg-controller-manager --timeout=300s || true
      wait_for_cnpg_webhook 300 || true
    fi
    return 0
  fi

  local target_version minor_version yaml_url
  target_version="${CNPG_OPERATOR_VERSION:-${CNPG_VERSION:-${CNPG_OPERATOR_FALLBACK_VERSION:-1.28.1}}}"
  if [[ -z "$target_version" || "$target_version" == "latest" ]]; then
    target_version=$(get_latest_cnpg_version)
  fi
  minor_version=$(echo "$target_version" | cut -d. -f1,2)
  yaml_url="https://raw.githubusercontent.com/cloudnative-pg/cloudnative-pg/release-${minor_version}/releases/cnpg-${target_version}.yaml"

  echo "Installing CloudNative-PG operator version ${target_version} ..."
  kubectl apply --server-side -f "$yaml_url"

  if kubectl get deployment -n cnpg-system cnpg-controller-manager >/dev/null 2>&1; then
    kubectl -n cnpg-system rollout status deploy/cnpg-controller-manager --timeout=180s || true
    wait_for_cnpg_webhook 180 || true
    if [[ "${PROLE_MODE:-}" == "k3s" ]]; then
      tune_cnpg_operator_for_k3s || true
      kubectl -n cnpg-system rollout status deploy/cnpg-controller-manager --timeout=300s || true
      wait_for_cnpg_webhook 300 || true
    fi
  fi
}

tune_cnpg_operator_for_k3s() {
  # k3s on a single node can experience short API/server or scheduling stalls under load.
  # CNPG's default probes are very aggressive (timeoutSeconds=1), which can cause flapping
  # readiness and webhook endpoints disappearing mid-apply.
  echo "Tuning CNPG operator deployment probes/resources for k3s ..."
  kubectl -n cnpg-system patch deploy cnpg-controller-manager --type merge -p '
  {
    "spec": {
      "template": {
        "spec": {
          "containers": [
            {
              "name": "manager",
              "resources": {
                "requests": {"cpu": "250m", "memory": "512Mi"},
                "limits": {"cpu": "500m", "memory": "1Gi"}
              },
              "livenessProbe": {"timeoutSeconds": 5, "failureThreshold": 6},
              "readinessProbe": {"timeoutSeconds": 5, "failureThreshold": 6},
              "startupProbe": {"timeoutSeconds": 5, "failureThreshold": 60}
            }
          ]
        }
      }
    }
  }' >/dev/null 2>&1 || return 1
}

get_latest_barman_plugin_version() {
  local version tag
  tag=$(curl -s --connect-timeout 5 --max-time 10 "https://api.github.com/repos/cloudnative-pg/plugin-barman-cloud/releases/latest" | jq -r '.tag_name' || echo "")
  if [[ -z "$tag" || "$tag" == "null" ]]; then
    echo "v${BARMAN_PLUGIN_FALLBACK_VERSION}"
    return 0
  fi
  echo "$tag"
}

resolve_barman_plugin_manifest_url() {
  if [[ -n "$BARMAN_PLUGIN_MANIFEST_URL" ]]; then
    echo "$BARMAN_PLUGIN_MANIFEST_URL"
    return 0
  fi
  local tag
  tag=$(get_latest_barman_plugin_version)
  echo "https://github.com/cloudnative-pg/plugin-barman-cloud/releases/download/${tag}/manifest.yaml"
}

cert_manager_ready() {
  if ! kubectl get crd certificates.cert-manager.io >/dev/null 2>&1; then
    return 1
  fi
  if ! kubectl -n cert-manager get deploy cert-manager >/dev/null 2>&1; then
    return 1
  fi
  return 0
}

resolve_control_plane_selector() {
  local node=""
  node=$(kubectl get nodes -l "node-role.kubernetes.io/control-plane" -o jsonpath='{.items[0].metadata.name}' 2>/dev/null || true)
  if [[ -n "$node" ]]; then
    printf 'kubernetes.io/hostname=%s' "$node"
  fi
}

deployment_node_selector_value() {
  local ns="$1"
  local dep="$2"
  local key="$3"
  kubectl -n "$ns" get deployment "$dep" -o jsonpath="{.spec.template.spec.nodeSelector['${key}']}" 2>/dev/null || true
}

ensure_deployment_node_selector() {
  local ns="$1"
  local dep="$2"
  local key="$3"
  local value="$4"

  if ! kubectl -n "$ns" get deployment "$dep" >/dev/null 2>&1; then
    return 0
  fi

  local current
  current=$(deployment_node_selector_value "$ns" "$dep" "$key")
  if [[ "$current" == "$value" ]]; then
    return 0
  fi

  kubectl -n "$ns" patch deployment "$dep" --type merge \
    -p "{\"spec\":{\"template\":{\"spec\":{\"nodeSelector\":{\"${key}\":\"${value}\"}}}}}" >/dev/null 2>&1 || true
}

pin_cert_manager() {
  local selector="${CERT_MANAGER_NODE_SELECTOR:-}"
  if [[ -z "$selector" && "${PROLE_MODE:-}" == "k3s" ]]; then
    selector="$(resolve_control_plane_selector)"
  fi
  if [[ -z "$selector" ]]; then
    return 0
  fi
  local key value
  key="${selector%%=*}"
  value="${selector#*=}"
  if [[ -z "$key" || -z "$value" ]]; then
    echo "WARN: CERT_MANAGER_NODE_SELECTOR must be key=value (got '$selector'). Skipping pin." >&2
    return 0
  fi
  if [[ -z "$(kubectl get nodes -l "${key}=${value}" --no-headers 2>/dev/null)" ]]; then
    echo "WARN: No nodes match CERT_MANAGER_NODE_SELECTOR=${selector}. Skipping pin." >&2
    return 0
  fi
  echo "Pinning cert-manager deployments to nodes with ${selector} ..."
  for dep in cert-manager cert-manager-webhook cert-manager-cainjector; do
    ensure_deployment_node_selector cert-manager "$dep" "$key" "$value"
  done
}

pin_barman_cloud() {
  local selector="${BARMAN_NODE_SELECTOR:-}"
  if [[ -z "$selector" && "${PROLE_MODE:-}" == "k3s" ]]; then
    selector="$(resolve_control_plane_selector)"
  fi
  if [[ -z "$selector" ]]; then
    return 0
  fi
  local key value
  key="${selector%%=*}"
  value="${selector#*=}"
  if [[ -z "$key" || -z "$value" ]]; then
    echo "WARN: BARMAN_NODE_SELECTOR must be key=value (got '$selector'). Skipping pin." >&2
    return 0
  fi
  if [[ -z "$(kubectl get nodes -l "${key}=${value}" --no-headers 2>/dev/null)" ]]; then
    echo "WARN: No nodes match BARMAN_NODE_SELECTOR=${selector}. Skipping pin." >&2
    return 0
  fi
  echo "Pinning barman-cloud deployment to nodes with ${selector} ..."
  ensure_deployment_node_selector cnpg-system barman-cloud "$key" "$value"
}

get_latest_cert_manager_version() {
  local tag
  tag=$(curl -s --connect-timeout 5 --max-time 10 "https://api.github.com/repos/cert-manager/cert-manager/releases/latest" | jq -r '.tag_name' || echo "")
  if [[ -z "$tag" || "$tag" == "null" ]]; then
    echo "v${CERT_MANAGER_FALLBACK_VERSION}"
    return 0
  fi
  echo "$tag"
}

resolve_cert_manager_manifest_url() {
  if [[ -n "$CERT_MANAGER_MANIFEST_URL" ]]; then
    echo "$CERT_MANAGER_MANIFEST_URL"
    return 0
  fi
  local tag
  tag=$(get_latest_cert_manager_version)
  echo "https://github.com/cert-manager/cert-manager/releases/download/${tag}/cert-manager.yaml"
}

ensure_cert_manager() {
  if cert_manager_ready; then
    pin_cert_manager
    if kubectl -n cert-manager get deploy cert-manager >/dev/null 2>&1; then
      kubectl -n cert-manager rollout status deploy/cert-manager --timeout=180s || true
      kubectl -n cert-manager rollout status deploy/cert-manager-webhook --timeout=180s || true
      kubectl -n cert-manager rollout status deploy/cert-manager-cainjector --timeout=180s || true
    fi
    return 0
  fi

  local cm_url
  cm_url=$(resolve_cert_manager_manifest_url)
  echo "Installing cert-manager from ${cm_url} ..."
  kubectl apply -f "$cm_url"
  pin_cert_manager

  if kubectl -n cert-manager get deploy cert-manager >/dev/null 2>&1; then
    kubectl -n cert-manager rollout status deploy/cert-manager --timeout=180s || true
    kubectl -n cert-manager rollout status deploy/cert-manager-webhook --timeout=180s || true
    kubectl -n cert-manager rollout status deploy/cert-manager-cainjector --timeout=180s || true
  fi
}

wait_for_barman_crd() {
  local timeout=${1:-120}
  local start_time
  start_time=$(date +%s)
  echo "Waiting for Barman Cloud CRD (timeout: ${timeout}s)..."
  while true; do
    if kubectl get crd objectstores.barmancloud.cnpg.io >/dev/null 2>&1; then
      echo "Barman Cloud CRD is available."
      return 0
    fi
    local elapsed=$(( $(date +%s) - start_time ))
    if (( elapsed > timeout )); then
      echo "Barman Cloud CRD not ready after ${elapsed}s." >&2
      return 1
    fi
    if (( elapsed % 30 < 6 && elapsed > 5 )); then
      echo "  [${elapsed}s/${timeout}s] Still waiting for Barman Cloud CRD..."
    fi
    sleep 5
  done
}

wait_for_barman_tls_secrets() {
  local timeout=${1:-180}
  local start_time
  start_time=$(date +%s)
  echo "Waiting for Barman Cloud TLS secrets (timeout: ${timeout}s)..."
  while true; do
    local client_crt client_key server_crt server_key
    client_crt=$(kubectl -n cnpg-system get secret barman-cloud-client-tls -o jsonpath='{.data.tls\.crt}' 2>/dev/null || true)
    client_key=$(kubectl -n cnpg-system get secret barman-cloud-client-tls -o jsonpath='{.data.tls\.key}' 2>/dev/null || true)
    server_crt=$(kubectl -n cnpg-system get secret barman-cloud-server-tls -o jsonpath='{.data.tls\.crt}' 2>/dev/null || true)
    server_key=$(kubectl -n cnpg-system get secret barman-cloud-server-tls -o jsonpath='{.data.tls\.key}' 2>/dev/null || true)
    if [[ -n "$client_crt" && -n "$client_key" && -n "$server_crt" && -n "$server_key" ]]; then
      echo "Barman Cloud TLS secrets are available."
      return 0
    fi
    local elapsed=$(( $(date +%s) - start_time ))
    if (( elapsed > timeout )); then
      echo "Barman Cloud TLS secrets not ready after ${elapsed}s." >&2
      return 1
    fi
    if (( elapsed % 30 < 6 && elapsed > 5 )); then
      echo "  [${elapsed}s/${timeout}s] Still waiting for Barman Cloud TLS secrets..."
    fi
    sleep 5
  done
}

ensure_barman_plugin() {
  ensure_cert_manager

  local plugin_url
  plugin_url=$(resolve_barman_plugin_manifest_url)
  echo "Installing Barman Cloud plugin from ${plugin_url} ..."
  local apply_out=""
  if ! apply_out=$(kubectl apply -f "$plugin_url" 2>&1); then
    echo "$apply_out" >&2
    if echo "$apply_out" | grep -qi "webhook.cert-manager.io"; then
      echo "WARN: cert-manager webhook error detected; restarting cert-manager components and retrying..." >&2
      kubectl -n cert-manager rollout restart deploy/cert-manager deploy/cert-manager-webhook deploy/cert-manager-cainjector >/dev/null 2>&1 || true
      kubectl -n cert-manager rollout status deploy/cert-manager-webhook --timeout=180s >/dev/null 2>&1 || true
      kubectl -n cert-manager rollout status deploy/cert-manager --timeout=180s >/dev/null 2>&1 || true
      kubectl -n cert-manager rollout status deploy/cert-manager-cainjector --timeout=180s >/dev/null 2>&1 || true
      kubectl apply -f "$plugin_url" || true
    fi
  else
    printf '%s\n' "$apply_out"
  fi

  if ! wait_for_barman_crd 120; then
    echo "WARN: Barman Cloud ObjectStore CRD not ready after install." >&2
  fi

  if ! wait_for_barman_tls_secrets 180; then
    echo "WARN: Barman Cloud TLS secrets not ready after install." >&2
  fi

  pin_barman_cloud

  if kubectl -n cnpg-system get deploy barman-cloud >/dev/null 2>&1; then
    kubectl -n cnpg-system rollout status deploy/barman-cloud --timeout=180s || true
  fi
}

pin_cnpg_controller() {
  local selector="${CNPG_CONTROLLER_NODE_SELECTOR:-}"
  if [[ -z "$selector" && "${PROLE_MODE:-}" == "k3s" ]]; then
    if [[ -n "$(kubectl get nodes -l "storage=primary" --no-headers 2>/dev/null)" ]]; then
      selector="storage=primary"
    else
      local cp_node=""
      cp_node=$(kubectl get nodes -l "node-role.kubernetes.io/control-plane" -o jsonpath='{.items[0].metadata.name}' 2>/dev/null || true)
      if [[ -n "$cp_node" ]]; then
        selector="kubernetes.io/hostname=${cp_node}"
      fi
    fi
  fi
  if [[ -z "$selector" ]]; then
    return 0
  fi
  local key value
  key="${selector%%=*}"
  value="${selector#*=}"
  if [[ -z "$key" || -z "$value" ]]; then
    echo "WARN: CNPG_CONTROLLER_NODE_SELECTOR must be key=value (got '$selector'). Skipping pin." >&2
    return 0
  fi
  if [[ -z "$(kubectl get nodes -l "${key}=${value}" --no-headers 2>/dev/null)" ]]; then
    echo "WARN: No nodes match CNPG_CONTROLLER_NODE_SELECTOR=${selector}. Skipping pin." >&2
    return 0
  fi
  echo "Pinning cnpg-controller-manager to nodes with ${selector} ..."
  kubectl -n cnpg-system patch deployment cnpg-controller-manager --type merge \
    -p "{\"spec\":{\"template\":{\"spec\":{\"nodeSelector\":{\"${key}\":\"${value}\"}}}}}" >/dev/null 2>&1 || true
}

wait_for_cnpg_webhook() {
  local timeout=${1:-120}
  local start_time
  start_time=$(date +%s)

  echo "Waiting for CNPG webhook service endpoints to be ready (timeout: ${timeout}s)..."
  while true; do
    local endpoints
    endpoints=$(kubectl -n cnpg-system get endpoints cnpg-webhook-service -o jsonpath='{.subsets[*].addresses[*].ip}' 2>/dev/null || true)
    if [[ -n "$endpoints" ]]; then
      echo "CNPG webhook service has endpoints."
      return 0
    fi
    local elapsed=$(( $(date +%s) - start_time ))
    if (( elapsed > timeout )); then
      echo "WARN: CNPG webhook endpoints not ready after ${elapsed}s." >&2
      kubectl -n cnpg-system get pods 2>/dev/null >&2 || true
      return 1
    fi
    if (( elapsed % 30 < 6 && elapsed > 5 )); then
      echo "  [${elapsed}s/${timeout}s] Waiting for CNPG webhook endpoints..."
      kubectl -n cnpg-system get pods --no-headers 2>/dev/null | sed 's/^/    /' || true
    fi
    sleep 5
  done
}

apply_barman_objectstore_if_present() {
  if [[ ! -f "$BARMAN_OBJECTSTORE_MANIFEST" ]]; then
    return 0
  fi
  if kubectl get crd objectstores.barmancloud.cnpg.io >/dev/null 2>&1; then
    if [[ -n "${SERVICE_NAMESPACE:-}" && "${SERVICE_NAMESPACE}" != "${NAMESPACE}" ]]; then
      local endpoint
      endpoint="http://garage.${SERVICE_NAMESPACE}.svc.cluster.local:3900"
      prole_render_manifest "$BARMAN_OBJECTSTORE_MANIFEST" \
        | sed -E "s|^[[:space:]]*endpointURL:.*|    endpointURL: ${endpoint}|" \
        | kubectl_apply_retry "$NAMESPACE"
    else
      prole_render_manifest "$BARMAN_OBJECTSTORE_MANIFEST" | kubectl_apply_retry "$NAMESPACE"
    fi
  else
    echo "WARN: Barman Cloud ObjectStore CRD not found; skipping $BARMAN_OBJECTSTORE_MANIFEST."
  fi
}

detect_root_manifest_namespace() {
  # Extract the root-level `metadata.namespace` for a single YAML document.
  # Note: This intentionally ignores nested `metadata:` blocks (e.g., templates).
  local file="$1"
  awk '
    BEGIN{in_meta=0}
    /^metadata:[[:space:]]*$/ {in_meta=1; next}
    in_meta && /^[[:space:]]{2}namespace:[[:space:]]*/ {
      sub(/^[[:space:]]{2}namespace:[[:space:]]*/, "")
      gsub(/\r/, "")
      print
      exit
    }
    in_meta && /^[^[:space:]]/ {in_meta=0}
  ' "$file" | head -n 1
}

apply_manifest_split_by_doc() {
  # Apply a potentially multi-document YAML manifest, routing each document
  # to either $NAMESPACE (when no namespace is specified) or to its declared
  # `metadata.namespace`.
  local file="$1"
  local attempts=${2:-8}

  local tmpdir
  tmpdir=$(mktemp -d -t prole-manifest-docs.XXXXXX)

  awk -v dir="$tmpdir" '
    BEGIN{i=0; f=sprintf("%s/doc-%03d.yaml", dir, i)}
    /^---[[:space:]]*$/ {i++; f=sprintf("%s/doc-%03d.yaml", dir, i); next}
    {print > f}
  ' "$file"

  local doc doc_ns apply_ns
  for doc in "$tmpdir"/doc-*.yaml; do
    [[ -f "$doc" ]] || continue
    # Skip empty/whitespace-only docs
    if ! grep -q '[^[:space:]]' "$doc"; then
      continue
    fi

    doc_ns=$(detect_root_manifest_namespace "$doc")
    apply_ns="$NAMESPACE"
    if [[ -n "$doc_ns" ]]; then
      apply_ns="$doc_ns"
      if ! kubectl get namespace "$apply_ns" >/dev/null 2>&1; then
        echo "Creating namespace '$apply_ns' for resources in '$(basename "$file")' ..."
        kubectl create namespace "$apply_ns" >/dev/null 2>&1 || true
      fi
    fi

    if ! kubectl_apply_retry "$apply_ns" "$attempts" <"$doc"; then
      rm -rf "$tmpdir"
      return 1
    fi
  done

  rm -rf "$tmpdir"
  return 0
}

apply_prole_manifest_file() {
  local file="$1"
  local attempts=${KUBECTL_APPLY_RETRIES:-8}
  local i output=""

  local tmp
  tmp=$(mktemp -t prole-manifest.XXXXXX)
  prole_render_manifest "$file" >"$tmp"

  for ((i=1; i<=attempts; i++)); do
    if output=$(kubectl apply -n "$NAMESPACE" -f "$tmp" 2>&1); then
      printf '%s\n' "$output"
      rm -f "$tmp"
      return 0
    fi

    if [[ "${PROLE_MODE:-}" == "k3d" && "$(basename "$file")" == "garage-statefulset.yaml" ]] \
      && echo "$output" | grep -q "updates to statefulset spec"; then
      echo "WARN: Garage StatefulSet immutable in k3d; skipping apply."
      rm -f "$tmp"
      return 0
    fi

    # Some manifests intentionally contain resources in other namespaces
    # (e.g., dashboard/supabase ingress). If we force `-n $NAMESPACE`, kubectl
    # errors with a namespace mismatch. In that case, split the manifest into
    # documents and apply each doc with its correct namespace.
    if printf '%s' "$output" | grep -Eqi 'namespace from the provided object.*does not match the namespace'; then
      echo "WARN: Namespace mismatch applying '$(basename "$file")' with -n '$NAMESPACE'; applying per-document namespaces instead." >&2
      apply_manifest_split_by_doc "$tmp" "$attempts"
      local rc=$?
      rm -f "$tmp"
      return $rc
    fi

    if _kubectl_is_transient_error "$output"; then
      echo "WARN: Failed to apply '$(basename "$file")' due to transient API/webhook issue (attempt $i/$attempts); retrying..." >&2
      echo "$output" >&2
      sleep 5
      continue
    fi

    echo "$output" >&2
    rm -f "$tmp"
    return 1
  done

  echo "ERROR: Failed to apply '$(basename "$file")' after $attempts attempts." >&2
  echo "$output" >&2
  rm -f "$tmp"
  return 1
}

_prole_k3s_registry_host_for_cache() {
  # Best-effort host:port for the k3s registry, used as an optional cache source in k3d mode.
  # Intentionally avoids k3d registry names and localhost-ish endpoints.
  local r="${PROLE_K3S_REGISTRY:-${K3S_REGISTRY_HOST:-${LOCAL_REGISTRY_INTERNAL:-${LOCAL_REGISTRY:-}}}}"

  r="${r#http://}"
  r="${r#https://}"

  case "${r}" in
    "")
      r="myrddin.prole.org:5000"
      ;;
    localhost:5000|127.0.0.1:5000|*.localhost|*.localhost:5000)
      r="myrddin.prole.org:5000"
      ;;
    k3d-*|*/k3d-*)
      r="myrddin.prole.org:5000"
      ;;
  esac

  printf '%s' "$r"
}

_docker_build_knoe_db_image() {
  # Build the knoe-db image with BuildKit and optional cache source.
  # Args: <tag> <context_dir> [cache_ref]
  local tag="$1"
  local context_dir="$2"
  local cache_ref="${3:-}"

  local ssh_args=()
  if [[ -n "${SSH_AUTH_SOCK:-}" ]]; then
    ssh_args+=(--ssh default)
  fi

  if docker buildx version >/dev/null 2>&1; then
    # Prefer an isolated docker-container builder to avoid host snapshot/cache corruption.
    local builder_name="${PROLE_BUILDX_BUILDER:-prole-buildkit}"
    if ! docker buildx inspect "$builder_name" >/dev/null 2>&1; then
      docker buildx create --name "$builder_name" --driver docker-container --use >/dev/null 2>&1 || true
    else
      docker buildx use "$builder_name" >/dev/null 2>&1 || true
    fi

    local cache_args=()
    if [[ -n "$cache_ref" ]]; then
      cache_args+=(--cache-from "type=registry,ref=${cache_ref}")
    fi

    if docker buildx build \
      --pull \
      --tag "$tag" \
      --load \
      "${ssh_args[@]}" \
      "${cache_args[@]}" \
      "$context_dir"; then
      return 0
    fi
  fi

  # Fallback to docker build (still BuildKit-enabled); cache-from here only works if the cache image is present locally.
  local cache_from_args=()
  if [[ -n "$cache_ref" ]]; then
    docker pull "$cache_ref" >/dev/null 2>&1 || true
    cache_from_args+=(--cache-from "$cache_ref")
  fi

  DOCKER_BUILDKIT=1 docker build \
    --pull \
    "${ssh_args[@]}" \
    "${cache_from_args[@]}" \
    -t "$tag" \
    "$context_dir"
}

_push_to_k3s_registry() {
  local image="$1"
  local push_host="${LOCAL_REGISTRY:-${LOCAL_REGISTRY_INTERNAL:-myrddin.prole.org:5000}}"
  local plain_image="${image##*/}"

  # Normalize push_host (strip scheme if provided)
  push_host="${push_host#http://}"
  push_host="${push_host#https://}"

  # The in-cluster registry is deployed with hostPort:5000 (binds to 0.0.0.0:5000)
  # on the control-plane node.  When running on that same host, localhost:5000 is
  # the most reliable path — no DNS resolution, no firewall, no port-forward needed.
  # Build an ordered list of candidate endpoints to try.
  local -a _push_candidates=()
  _push_candidates+=("$push_host")
  # Add localhost:5000 as a candidate when push_host is not already localhost-ish.
  case "$push_host" in
    localhost:*|127.0.0.1:*|0.0.0.0:*) ;;
    *) _push_candidates+=("localhost:5000") ;;
  esac

  if command -v skopeo >/dev/null 2>&1; then
    for _cand in "${_push_candidates[@]}"; do
      # Probe the registry endpoint before attempting a (potentially slow) push.
      if command -v curl >/dev/null 2>&1; then
        if ! curl -k -fsS -m 2 "https://${_cand}/v2/" >/dev/null 2>&1 \
            && ! curl -fsS -m 2 "http://${_cand}/v2/" >/dev/null 2>&1; then
          echo "  WARN: Registry endpoint '${_cand}' is not reachable; trying next candidate ..." >&2
          continue
        fi
      fi
      echo "  Pushing to k3s registry at '${_cand}' using skopeo ..."
      if skopeo copy --dest-tls-verify=false docker-daemon:"$image" docker://"${_cand}/${plain_image}"; then
        echo "  ✓ Image pushed to registry at '${_cand}'."
        return 0
      fi
      echo "  WARN: skopeo push to '${_cand}' failed; trying next candidate ..." >&2
    done
  fi

  # Fallback: direct node import via SSH (no port-forward needed).
  if _import_image_to_k3s_nodes "$image"; then
    return 0
  fi

  # Fallback to docker push if skopeo is missing or fails (might fail if daemon not configured)
  for _cand in "${_push_candidates[@]}"; do
    local push_ref="${_cand}/${plain_image}"
    docker tag "$image" "$push_ref" 2>/dev/null || true
    if docker push "$push_ref" 2>/dev/null; then
      echo "  ✓ Image pushed to registry at '${_cand}' via docker push."
      return 0
    fi
  done

  echo "ERROR: Failed to push image '$image' to k3s registry at '${push_host}'." >&2
  return 1
}

# Push image to local registry (via LOCAL_REGISTRY host address) with k3d import fallback.
_push_to_k3d_registry() {
  local image="$1"
  local cluster_name="$2"
  local push_host="${LOCAL_REGISTRY:-localhost:5000}"
  local plain_image="${image##*/}"  # strip registry prefix, e.g. knoe-db:18-088

  if [[ -n "$push_host" ]]; then
    local push_ref="${push_host}/${plain_image}"
    docker tag "$image" "$push_ref" 2>/dev/null || true
    if docker push "$push_ref" 2>/dev/null; then
      echo "  ✓ Image pushed to registry at '${push_host}'."
      return 0
    fi
    echo "  WARN: push to '${push_host}' failed; falling back to k3d image import ..." >&2
  fi

  if k3d image import "$image" -c "$cluster_name" 2>/dev/null; then
    echo "  ✓ Image '$image' imported directly into k3d cluster '$cluster_name'."
    return 0
  fi
  echo "ERROR: Failed to push or import image '$image'." >&2
  return 1
}

# Pre-flight: ensure the knoe-db image is available in the k3d cluster before
# the CNPG operator ever tries to pull it, avoiding ErrImagePull backoff loops.
# Steps: containerd cache → Docker daemon (registry tag) → Docker daemon (plain tag)
#        → tar import → docker build + push/import.
_ensure_knoe_db_image() {
  if [[ "${PROLE_MODE:-}" != "k3d" && "${PROLE_MODE:-}" != "k3s" ]]; then
    return 0
  fi

  local image_override="${CNPG_IMAGE:-${KNOE_DB_IMAGE:-}}"
  local image=""
  if [[ -n "$image_override" ]]; then
    image="$image_override"
  else
    if [[ "$VERSION" == "latest" || -z "$VERSION" ]]; then
      image=$(get_latest_image)
    else
      image="knoe-db:$VERSION"
    fi
  fi
  image=$(resolve_cnpg_image "$image")
  local knoe_db_dir="${PROLE_HOME:-$SCRIPT_DIR/..}/knoe-db"
  local plain_image="${image##*/}"  # e.g. knoe-db:18-088
  local k3s_cache_ref=""

  if [[ "${PROLE_MODE:-}" == "k3s" ]]; then
    echo "Pre-flight: ensuring image '$image' is available in k3s registry/import path ..."
  else
    # Auto-detect active k3d cluster name
    local cluster_name="${K3D_CLUSTER_NAME:-}"
    if [[ -z "$cluster_name" ]]; then
      cluster_name=$(k3d cluster list --no-headers 2>/dev/null | awk '{print $1}' | head -1)
    fi
    cluster_name="${cluster_name:-knoe-dev-cluster}"

    # In k3d mode, optionally consult the k3s registry as a cache/source of truth before building.
    local k3s_host
    k3s_host=$(_prole_k3s_registry_host_for_cache)
    if [[ -n "${k3s_host:-}" ]]; then
      k3s_cache_ref="${k3s_host}/${plain_image}"
    fi

    echo "Pre-flight: verifying image '$image' is available in k3d cluster '$cluster_name' ..."

    # Step 1: check if already present in k3d containerd with matching digest
    local containerd_digest local_digest containerd_sha
    containerd_digest=$(docker exec "k3d-${cluster_name}-server-0" \
      ctr images ls -q 2>/dev/null | grep -F "$image" | head -1 || true)
    if [[ -n "$containerd_digest" ]]; then
      local_digest=$(docker inspect --format='{{index .RepoDigests 0}}' "$image" 2>/dev/null \
        | awk -F@ '{print $2}' || true)
      containerd_sha=$(docker exec "k3d-${cluster_name}-server-0" \
        ctr images ls 2>/dev/null | grep -F "$image" | awk '{print $3}' | head -1 || true)
      if [[ -z "$local_digest" || "$containerd_sha" == "$local_digest" ]]; then
        echo "  ✓ Image '$image' already in k3d containerd (digest match); no import needed."
        return 0
      fi
      echo "  Image '$image' in containerd but digest mismatch (local: ${local_digest:-unknown}, containerd: ${containerd_sha:-unknown}); re-importing ..."
      docker exec "k3d-${cluster_name}-server-0" ctr images rm "$image" 2>/dev/null || true
    fi
  fi

  # Step 2a: registry-tagged image in local Docker daemon → push + import
  if docker image inspect "$image" >/dev/null 2>&1; then
    echo "  Image '$image' found in Docker daemon; pushing to registry ..."
    if [[ "${PROLE_MODE:-}" == "k3s" ]]; then
      _push_to_k3s_registry "$image"
    else
      _push_to_k3d_registry "$image" "$cluster_name"
    fi
    return $?
  fi

  # Step 2b: plain-tagged image in local Docker daemon → tag + push + import
  if docker image inspect "$plain_image" >/dev/null 2>&1; then
    echo "  Plain image '$plain_image' found in Docker daemon; tagging as '$image' and pushing ..."
    docker tag "$plain_image" "$image"
    if [[ "${PROLE_MODE:-}" == "k3s" ]]; then
      _push_to_k3s_registry "$image"
    else
      _push_to_k3d_registry "$image" "$cluster_name"
    fi
    return $?
  fi

  # Step 3: look for a matching tar in the docker-import directory
  local prole_data="${PROLE_DATA:-$HOME/.prole/data}"
  local docker_import_dir="${DOCKER_IMPORT_DIR:-${prole_data}/docker-import}"
  local name_part="${plain_image%%:*}"  # e.g. knoe-db
  local tag_part="${plain_image##*:}"   # e.g. 18-088
  local found_tar=""
  if [[ -d "$docker_import_dir" ]]; then
    for _t in "$docker_import_dir"/*.tar; do
      [[ -f "$_t" ]] || continue
      local _bn
      _bn=$(basename "$_t" .tar)
      if [[ "$_bn" == *"$name_part"* && "$_bn" == *"$tag_part"* ]]; then
        found_tar="$_t"
        break
      fi
    done
  fi
  if [[ -n "$found_tar" ]]; then
    echo "  Loading tar '$(basename "$found_tar")' into Docker daemon ..."
    docker load -i "$found_tar"
    docker tag "$plain_image" "$image" 2>/dev/null || true
    if [[ "${PROLE_MODE:-}" == "k3s" ]]; then
      _push_to_k3s_registry "$image"
    else
      _push_to_k3d_registry "$image" "$cluster_name"
    fi
    return $?
  fi

  # Step 3b (k3d only): try to reuse a previously-built image from the k3s registry.
  if [[ "${PROLE_MODE:-}" == "k3d" && -n "${k3s_cache_ref:-}" && "${k3s_cache_ref}" != "${image}" ]]; then
    if docker image inspect "$k3s_cache_ref" >/dev/null 2>&1 || docker pull "$k3s_cache_ref" >/dev/null 2>&1; then
      echo "  ✓ Found '$plain_image' in k3s registry (${k3s_cache_ref}); re-tagging for k3d as '$image' ..."
      docker tag "$k3s_cache_ref" "$image" 2>/dev/null || true
      _push_to_k3d_registry "$image" "$cluster_name"
      return $?
    fi
    echo "  (cache) '$plain_image' not present in k3s registry (${k3s_cache_ref}); building locally ..."
  fi

  # Step 4: image not found anywhere — build from source then push + import
  if [[ ! -f "$knoe_db_dir/Dockerfile" ]]; then
    echo "ERROR: Dockerfile not found in '$knoe_db_dir'; cannot build knoe-db image." >&2
    return 1
  fi
  if [[ "${PROLE_MODE:-}" == "k3s" ]]; then
    echo "  Image '$image' not found in Docker daemon or docker-import dir."
  else
    echo "  Image '$image' not found in k3d, Docker daemon, or docker-import dir."
  fi
  echo "  Building knoe-db image from '$knoe_db_dir' ..."
  if ! _docker_build_knoe_db_image "$plain_image" "$knoe_db_dir" "${k3s_cache_ref:-}"; then
    echo "ERROR: docker build failed for image '$plain_image'." >&2
    return 1
  fi
  docker tag "$plain_image" "$image"
  echo "  Build complete. Pushing '$image' to registry ..."
  if [[ "${PROLE_MODE:-}" == "k3s" ]]; then
    _push_to_k3s_registry "$image"
  else
    _push_to_k3d_registry "$image" "$cluster_name"
  fi
  return $?
}

cleanup_unintended_cnpg_services() {
  # Historical manifest bug: we used to create a CNPG-managed Service named `knoe-db-001`
  # as `type: LoadBalancer`. In k3s this spawns `svclb-knoe-db-001` pods, and it can
  # interfere with CNPG startup/reconciliation. Ensure it is removed if it exists.
  local svc_name="knoe-db-001"
  if kubectl -n "$NAMESPACE" get svc "$svc_name" >/dev/null 2>&1; then
    echo "Removing unintended Service '$svc_name' from namespace '$NAMESPACE' ..."
    kubectl -n "$NAMESPACE" delete svc "$svc_name" --ignore-not-found >/dev/null 2>&1 || true
  fi
}

ensure_prole_stack_resources() {
  echo "Applying CloudNative-PG cluster and related resources ..."
  wait_for_apiserver_ready 180
  ensure_prole_protected_storage
  cleanup_unintended_cnpg_services || true
  local image_override="${CNPG_IMAGE:-${KNOE_DB_IMAGE:-}}"
  local image=""
  if [[ -n "$image_override" ]]; then
    image="$image_override"
  else
    if [[ "$VERSION" == "latest" || -z "$VERSION" ]]; then
      image=$(get_latest_image)
    else
      image="knoe-db:$VERSION"
    fi
  fi
  image=$(resolve_cnpg_image "$image")
  sync_manifest_image "$image"
  if [[ -n "$CNPG_MANIFEST_OVERRIDE" ]]; then
    if [[ ! -f "$CNPG_MANIFEST_OVERRIDE" ]]; then
      echo "ERROR: CNPG_MANIFEST_OVERRIDE not found: $CNPG_MANIFEST_OVERRIDE" >&2
      return 1
    fi
    local dir file
    dir="$K8S_PROLE_DIR"
    for file in "$dir"/*.yaml; do
      case "$(basename "$file")" in
        knoe-db.yaml|kustomization.yaml|supabase-*.yaml|knoe-db-barman-objectstore.yaml|ingress.yaml)
          continue
          ;;
        openbao-statefulset.yaml|openbao-service.yaml)
          if [[ -n "${SERVICE_NAMESPACE:-}" && "${SERVICE_NAMESPACE}" != "${NAMESPACE}" ]]; then
            continue
          fi
          ;;
        garage-*.yaml|grafana-*.yaml|prometheus-*.yaml)
          if [[ -n "${SERVICE_NAMESPACE:-}" && "${SERVICE_NAMESPACE}" != "${NAMESPACE}" ]]; then
            continue
          fi
          ;;
        kong-*.yaml)
          if [[ -n "${SERVICE_NAMESPACE:-}" && "${SERVICE_NAMESPACE}" != "${NAMESPACE}" ]]; then
            continue
          fi
          ;;
      esac
      apply_prole_manifest_file "$file"
    done
    # Apply ingress.yaml without -n flag so each document targets its own namespace
    if [[ -f "$dir/ingress.yaml" ]]; then
      # Ensure referenced namespaces exist before applying multi-namespace ingress
      for _ing_ns in $(grep -E '^\s+namespace:' "$dir/ingress.yaml" | awk '{print $2}' | sort -u); do
        if ! kubectl get namespace "$_ing_ns" >/dev/null 2>&1; then
          echo "Creating namespace '$_ing_ns' for ingress resource ..."
          kubectl create namespace "$_ing_ns" 2>/dev/null || true
        fi
      done
      prole_render_manifest "$dir/ingress.yaml" | kubectl_apply_retry "" || true
    fi
    apply_barman_objectstore_if_present
    apply_cnpg_cluster_manifest "$CNPG_MANIFEST_OVERRIDE"
    validate_cnpg_runtime_storage
    reconcile_cnpg_instances
  else
    local dir file
    dir="$K8S_PROLE_DIR"
    for file in "$dir"/*.yaml; do
      case "$(basename "$file")" in
        knoe-db.yaml|kustomization.yaml|supabase-*.yaml|knoe-db-barman-objectstore.yaml|ingress.yaml)
          continue
          ;;
        openbao-statefulset.yaml|openbao-service.yaml)
          if [[ -n "${SERVICE_NAMESPACE:-}" && "${SERVICE_NAMESPACE}" != "${NAMESPACE}" ]]; then
            continue
          fi
          ;;
        garage-*.yaml|grafana-*.yaml|prometheus-*.yaml)
          if [[ -n "${SERVICE_NAMESPACE:-}" && "${SERVICE_NAMESPACE}" != "${NAMESPACE}" ]]; then
            continue
          fi
          ;;
        kong-*.yaml)
          if [[ -n "${SERVICE_NAMESPACE:-}" && "${SERVICE_NAMESPACE}" != "${NAMESPACE}" ]]; then
            continue
          fi
          ;;
      esac
      apply_prole_manifest_file "$file"
    done
    # Apply ingress.yaml without -n flag so each document targets its own namespace
    if [[ -f "$dir/ingress.yaml" ]]; then
      # Ensure referenced namespaces exist before applying multi-namespace ingress
      for _ing_ns in $(grep -E '^\s+namespace:' "$dir/ingress.yaml" | awk '{print $2}' | sort -u); do
        if ! kubectl get namespace "$_ing_ns" >/dev/null 2>&1; then
          echo "Creating namespace '$_ing_ns' for ingress resource ..."
          kubectl create namespace "$_ing_ns" 2>/dev/null || true
        fi
      done
      prole_render_manifest "$dir/ingress.yaml" | kubectl_apply_retry "" || true
    fi
    apply_barman_objectstore_if_present
    apply_cnpg_cluster_manifest "$CNPG_MANIFEST"
    validate_cnpg_runtime_storage
    reconcile_cnpg_instances
  fi

  # Ensure prole-index-html exists for prole deployment readiness probe
  if ! kubectl get configmap prole-index-html -n "$NAMESPACE" >/dev/null 2>&1; then
    echo "Creating prole-index-html configmap..."
    printf "<html><body><h1>Prole</h1></body></html>" > /tmp/index.html
    kubectl create configmap prole-index-html --from-file=index.html=/tmp/index.html -n "$NAMESPACE" --dry-run=client -o yaml \
      | kubectl_apply_retry "$NAMESPACE"
    rm /tmp/index.html
  fi

  # Ensure prole-nginx-tls exists (self-signed for dev)
  if ! kubectl get secret prole-nginx-tls -n "$NAMESPACE" >/dev/null 2>&1; then
    echo "Generating self-signed prole-nginx-tls for development..."
    openssl req -x509 -nodes -days 365 -newkey rsa:2048 \
      -keyout /tmp/nginx-tls.key -out /tmp/nginx-tls.crt \
      -subj "/CN=prole.org" >/dev/null 2>&1
    kubectl create secret tls prole-nginx-tls --key /tmp/nginx-tls.key --cert /tmp/nginx-tls.crt -n "$NAMESPACE" --dry-run=client -o yaml \
      | kubectl_apply_retry "$NAMESPACE"
    rm /tmp/nginx-tls.key /tmp/nginx-tls.crt
  fi
}

_coerce_uint() {
  local raw="${1:-}" default="${2:-0}"
  if [[ "$raw" =~ ^[0-9]+$ ]]; then
    echo "$raw"
  else
    echo "$default"
  fi
}

_ready_schedulable_nodes_count() {
  kubectl get nodes --no-headers 2>/dev/null \
    | awk '$2 ~ /^Ready/ && $2 !~ /SchedulingDisabled/ {c++} END {print c+0}'
}

_ready_schedulable_db_nodes_count() {
  local key="${CNPG_NODE_ROLE_LABEL_KEY:-prole.org/node-role}"
  local val="${CNPG_NODE_ROLE_LABEL_VALUE:-db}"
  kubectl get nodes -l "${key}=${val}" --no-headers 2>/dev/null \
    | awk '$2 ~ /^Ready/ && $2 !~ /SchedulingDisabled/ {c++} END {print c+0}'
}

reconcile_cnpg_instances() {
  # Cap instances to cluster capacity to avoid overloading a single node during degraded startup.
  # Inputs are taken from the same sources Ansible uses today:
  # - node labels (e.g., `prole.org/node-role=db`)
  # - optional-workloads guard (`OPTIONAL_WORKLOADS_MIN_READY_SCHEDULABLE_NODES`)

  local current
  current=$(kubectl -n "$NAMESPACE" get cluster "$CNPG_CLUSTER_NAME" -o jsonpath='{.spec.instances}' 2>/dev/null || true)
  [[ -z "$current" ]] && return 0
  current=$(_coerce_uint "$current" 0)

  local min_required
  min_required=$(_coerce_uint "${OPTIONAL_WORKLOADS_MIN_READY_SCHEDULABLE_NODES:-2}" 2)
  if (( min_required < 1 )); then
    min_required=1
  fi

  local ready_nodes eligible_db_nodes
  ready_nodes=$(_coerce_uint "$(_ready_schedulable_nodes_count)" 0)
  eligible_db_nodes=$(_coerce_uint "$(_ready_schedulable_db_nodes_count)" 0)
  if (( eligible_db_nodes < 1 )); then
    echo "ERROR: No Ready, schedulable CNPG DB nodes found (label ${CNPG_NODE_ROLE_LABEL_KEY:-prole.org/node-role}=${CNPG_NODE_ROLE_LABEL_VALUE:-db})." >&2
    echo "Refusing to force CNPG pods onto an arbitrary node during bootstrap." >&2
    return 1
  fi

  local desired
  desired=$(_coerce_uint "${CNPG_INSTANCES:-3}" 3)
  if (( desired < 1 )); then
    desired=1
  fi
  if (( desired > eligible_db_nodes )); then
    desired=$eligible_db_nodes
  fi

  # Degraded cluster: keep instances at 1 until enough nodes are Ready+schedulable.
  if (( ready_nodes < min_required )); then
    desired=1
  fi

  if (( desired < 1 )); then
    desired=1
  fi

  if (( current == desired )); then
    return 0
  fi

  echo "Reconciling CNPG cluster '$CNPG_CLUSTER_NAME' instances=${desired} (was ${current}; ready_nodes=${ready_nodes}; eligible_db_nodes=${eligible_db_nodes}) ..."
  local attempts=${KUBECTL_PATCH_RETRIES:-8}
  local i out
  for ((i=1; i<=attempts; i++)); do
    if out=$(kubectl -n "$NAMESPACE" patch cluster "$CNPG_CLUSTER_NAME" --type merge -p "{\"spec\":{\"instances\":${desired}}}" 2>&1); then
      printf '%s\n' "$out"
      return 0
    fi
    if _kubectl_is_transient_error "$out"; then
      echo "WARN: Failed to patch CNPG instances due to transient API issue (attempt $i/$attempts); retrying..." >&2
      echo "$out" >&2
      sleep 5
      continue
    fi
    echo "$out" >&2
    return 1
  done

  echo "ERROR: Failed to reconcile instances=${desired} after $attempts attempts." >&2
  echo "$out" >&2
  return 1
}

apply_cnpg_cluster_manifest() {
  local manifest="$1"
  local attempts=${CNPG_APPLY_RETRIES:-6}
  local i out

  for ((i=1; i<=attempts; i++)); do
    local tmp
    tmp=$(mktemp -t prole-cnpg.XXXXXX)
    prole_render_manifest "$manifest" >"$tmp"
    if ! validate_cnpg_manifest_storage "$tmp"; then
      rm -f "$tmp"
      return 1
    fi
    if out=$(kubectl apply -n "$NAMESPACE" -f "$tmp" 2>&1); then
      printf '%s\n' "$out"
      rm -f "$tmp"
      return 0
    fi
    rm -f "$tmp"

    if _kubectl_is_transient_error "$out" || echo "$out" | grep -q "cnpg-webhook-service"; then
      echo "CNPG apply failed due to transient API/webhook issue (attempt $i/$attempts). Retrying..." >&2
      echo "$out" >&2
      sleep 5
      continue
    fi

    echo "$out" >&2
    return 1
  done

  echo "ERROR: Failed to apply CNPG manifest after $attempts attempts." >&2
  echo "$out" >&2
  return 1
}

wait_for_cnpg_pods() {
  local timeout=${1:-${CNPG_WAIT_TIMEOUT:-900}}
  local start_time
  start_time=$(date +%s)

  local target_pods="${CNPG_TARGET_PODS:-}"
  if [[ -z "$target_pods" ]]; then
    target_pods=$(kubectl -n "$NAMESPACE" get cluster "$CNPG_CLUSTER_NAME" -o jsonpath='{.spec.instances}' 2>/dev/null || true)
  fi
  if [[ -z "$target_pods" || ! "$target_pods" =~ ^[0-9]+$ ]]; then
    target_pods=1
  fi

  echo "Waiting for $target_pods CNPG pods for cluster '$CNPG_CLUSTER_NAME' in namespace '$NAMESPACE' to be Ready..."
  echo "  Timeout: ${timeout}s"
  local last_feedback=0
  local feedback_interval=30  # print detailed status every 30s
  local _image_pull_remediated=""  # set after first k3d import attempt to avoid loops
  while true; do
    local now elapsed
    now=$(date +%s)
    elapsed=$(( now - start_time ))

    local pods
    pods=$(kubectl -n "$NAMESPACE" get pods -l "cnpg.io/cluster=$CNPG_CLUSTER_NAME" --no-headers 2>/dev/null || true)

    if [[ -n "$pods" ]]; then
      # Check for Error or CrashLoopBackOff (terminal failures)
      if echo "$pods" | grep -E "Error|CrashLoopBackOff" >/dev/null; then
        echo "ERROR: Some CNPG pods are in Error or CrashLoopBackOff state:" >&2
        echo "$pods" | grep -E "Error|CrashLoopBackOff" >&2
        return 1
      fi

      # Check for image pull failures — attempt remediation in k3d/k3s mode
      if echo "$pods" | grep -E "ErrImagePull|ImagePullBackOff" >/dev/null; then
        if [[ "${PROLE_MODE:-}" == "k3d" || "${PROLE_MODE:-}" == "k3s" ]]; then
          if [[ -z "$_image_pull_remediated" ]]; then
            _image_pull_remediated=1
            local fail_pod fail_image
            fail_pod=$(echo "$pods" | grep -E "ErrImagePull|ImagePullBackOff" | awk '{print $1}' | head -1)
            # Try containers then initContainers
            fail_image=$(kubectl -n "$NAMESPACE" get pod "$fail_pod" \
              -o jsonpath='{.spec.containers[0].image}' 2>/dev/null || true)
            if [[ -z "$fail_image" ]]; then
              fail_image=$(kubectl -n "$NAMESPACE" get pod "$fail_pod" \
                -o jsonpath='{.spec.initContainers[0].image}' 2>/dev/null || true)
            fi
            echo "WARN: Pod '$fail_pod' cannot pull image '${fail_image:-unknown}' (ErrImagePull/ImagePullBackOff)." >&2
            import_dir="${DOCKER_IMPORT_DIR:-${PROLE_DATA:+${PROLE_DATA}/docker-import}}"
            # Re-run full image ensure logic to push/import the failing image
            echo "  Attempting image remediation via _ensure_knoe_db_image ..." >&2
            _ensure_knoe_db_image >&2 || true
            echo "  Remediation complete; resuming wait ..." >&2
          fi
          # Continue the wait loop — do not return 1
        else
          echo "ERROR: Image pull failure in non-k3d mode — cannot auto-recover:" >&2
          echo "$pods" | grep -E "ErrImagePull|ImagePullBackOff" >&2
          return 1
        fi
      fi

      # Count Ready pods by name (exclude initdb)
      local ready_pods
      ready_pods=$(kubectl -n "$NAMESPACE" get pods -l "cnpg.io/cluster=$CNPG_CLUSTER_NAME" -o jsonpath='{range .items[*]}{.status.conditions[?(@.type=="Ready")].status}{"\n"}{end}' 2>/dev/null | grep "True" | wc -l | xargs)

      if [[ "$ready_pods" -ge "$target_pods" ]]; then
        echo "All $ready_pods/$target_pods pods are Ready."
        # By default, pod readiness is sufficient. Connectivity checks can be enabled explicitly.
        # This avoids unnecessary timeouts/resets during initial bootstrap when CNPG may still be
        # initializing even though pods are already Ready.
        if kubectl cnpg version >/dev/null 2>&1; then
          local require_psql="${CNPG_REQUIRE_PSQL:-false}"
          if [[ "$require_psql" == "1" || "$require_psql" == "true" || "$require_psql" == "True" || "$require_psql" == "yes" || "$require_psql" == "YES" ]]; then
            echo "Verifying database connectivity via 'kubectl cnpg psql'..."
            if kubectl cnpg -n "$NAMESPACE" psql "$CNPG_CLUSTER_NAME" -- -c "SELECT 1" >/dev/null 2>&1; then
              echo "Database connection verified."
              return 0
            else
              echo "  [${elapsed}s/${timeout}s] Pods are Ready but database connection failed — waiting for database to accept connections..."
            fi
          else
            # Best-effort connectivity check (warn-only)
            if ! kubectl cnpg -n "$NAMESPACE" psql "$CNPG_CLUSTER_NAME" -- -c "SELECT 1" >/dev/null 2>&1; then
              echo "WARN: CNPG pods are Ready but 'kubectl cnpg psql' is not yet available; continuing." >&2
            fi
            return 0
          fi
        else
          # Fallback if no cnpg plugin: just return 0 if pods are ready
          return 0
        fi
      fi

      # Periodic detailed feedback
      if (( now - last_feedback >= feedback_interval )); then
        last_feedback=$now
        echo ""
        echo "  [${elapsed}s/${timeout}s] Waiting — ${ready_pods:-0}/${target_pods} pods Ready"
        echo "  Pod status:"
        echo "$pods" | sed 's/^/    /'

        # Show cluster status if available
        local cluster_phase
        cluster_phase=$(kubectl -n "$NAMESPACE" get cluster "$CNPG_CLUSTER_NAME" -o jsonpath='{.status.phase}' 2>/dev/null || true)
        if [[ -n "$cluster_phase" ]]; then
          echo "  Cluster phase: ${cluster_phase}"
        fi

        # Show recent events (last 3)
        local events
        events=$(kubectl -n "$NAMESPACE" get events \
          --sort-by='.lastTimestamp' \
          --field-selector="involvedObject.name=${CNPG_CLUSTER_NAME}" \
          -o custom-columns='TIME:.lastTimestamp,TYPE:.type,REASON:.reason,MESSAGE:.message' \
          --no-headers 2>/dev/null | tail -3 || true)
        if [[ -n "$events" ]]; then
          echo "  Recent cluster events:"
          echo "$events" | sed 's/^/    /'
        fi
      fi
    else
      # No pods yet — give feedback
      if (( now - last_feedback >= feedback_interval )); then
        last_feedback=$now
        echo "  [${elapsed}s/${timeout}s] No CNPG pods found yet for cluster '$CNPG_CLUSTER_NAME' — waiting for operator to create them..."
        local cluster_phase
        cluster_phase=$(kubectl -n "$NAMESPACE" get cluster "$CNPG_CLUSTER_NAME" -o jsonpath='{.status.phase}' 2>/dev/null || true)
        if [[ -n "$cluster_phase" ]]; then
          echo "  Cluster phase: ${cluster_phase}"
        fi
        # If stuck in "Unable to create required cluster objects", show conditions and events for diagnosis
        if [[ "$cluster_phase" == *"Unable to create"* ]]; then
          echo "  Cluster conditions:"
          kubectl -n "$NAMESPACE" get cluster "$CNPG_CLUSTER_NAME" \
            -o jsonpath='{range .status.conditions[*]}    {.type}: {.status} — {.message}{"\n"}{end}' 2>/dev/null || true
          echo "  Recent events:"
          kubectl -n "$NAMESPACE" get events --sort-by='.lastTimestamp' \
            --field-selector="involvedObject.name=${CNPG_CLUSTER_NAME}" \
            -o custom-columns='TYPE:.type,REASON:.reason,MESSAGE:.message' \
            --no-headers 2>/dev/null | tail -5 | sed 's/^/    /' || true
        fi
      fi
    fi

    if (( elapsed > timeout )); then
      echo ""
      echo "ERROR: Timed out after ${elapsed}s waiting for $target_pods CNPG pods for cluster '$CNPG_CLUSTER_NAME' in namespace '$NAMESPACE'." >&2
      echo "Final pod status:" >&2
      kubectl -n "$NAMESPACE" get pods -l "cnpg.io/cluster=$CNPG_CLUSTER_NAME" 2>/dev/null >&2 || true
      echo "Recent events:" >&2
      kubectl -n "$NAMESPACE" get events --sort-by='.lastTimestamp' \
        --field-selector="involvedObject.name=${CNPG_CLUSTER_NAME}" \
        -o custom-columns='TIME:.lastTimestamp,TYPE:.type,REASON:.reason,MESSAGE:.message' \
        --no-headers 2>/dev/null | tail -5 >&2 || true
      return 1
    fi
    sleep 5
  done
}

cluster_exists() {
  kubectl -n "$NAMESPACE" get cluster "$CNPG_CLUSTER_NAME" >/dev/null 2>&1
}

cluster_has_pods() {
  kubectl -n "$NAMESPACE" get pods -l "cnpg.io/cluster=$CNPG_CLUSTER_NAME" --no-headers 2>/dev/null | grep -q .
}

cluster_has_ready_pods() {
  kubectl -n "$NAMESPACE" get pods -l "cnpg.io/cluster=$CNPG_CLUSTER_NAME" \
    -o jsonpath='{.items[?(@.status.conditions[?(@.type=="Ready")].status=="True")].metadata.name}' 2>/dev/null | grep -q .
}

latest_backup_name() {
  kubectl -n "$NAMESPACE" get backup \
    --sort-by=.metadata.creationTimestamp \
    -o jsonpath='{range .items[*]}{.metadata.name}{"\n"}{end}' 2>/dev/null | tail -n 1
}

latest_completed_backup_name() {
  kubectl -n "$NAMESPACE" get backup \
    --sort-by=.metadata.creationTimestamp \
    -o jsonpath='{range .items[*]}{.metadata.name}{"|"}{.status.phase}{"\n"}{end}' 2>/dev/null | \
    awk -F'|' '{p=tolower($2); if (p=="completed" || p=="succeeded") {name=$1}} END {print name}'
}

wait_for_backup() {
  local backup_name="$1"
  local start_time now phase phase_lc
  start_time=$(date +%s)

  while true; do
    phase=$(kubectl -n "$NAMESPACE" get backup "$backup_name" -o jsonpath='{.status.phase}' 2>/dev/null || true)
    phase_lc=$(printf '%s' "$phase" | tr '[:upper:]' '[:lower:]')

    case "$phase_lc" in
      completed|succeeded)
        echo "Backup $backup_name completed."
        return 0
        ;;
      failed|error)
        echo "Backup $backup_name failed (phase=$phase)." >&2
        return 1
        ;;
    esac

    now=$(date +%s)
    if (( now - start_time > BACKUP_WAIT_TIMEOUT )); then
      echo "Timed out waiting for backup $backup_name." >&2
      return 1
    fi

    echo "Waiting for backup $backup_name to complete (phase=${phase:-unknown}) ..."
    sleep 10
  done
}

run_garage_backup() {
  if [[ ! -x "$SCRIPT_DIR/init_cnpg_backup.sh" ]]; then
    echo "WARN: init_cnpg_backup.sh not found; skipping Garage backup." >&2
    return 1
  fi

  echo "Running Garage backup via init_cnpg_backup.sh ..."
  if ! "$SCRIPT_DIR/init_cnpg_backup.sh" start; then
    echo "Garage backup script failed." >&2
    return 1
  fi

  sleep 2
  local backup_name
  backup_name=$(latest_backup_name)
  if [[ -z "$backup_name" ]]; then
    echo "No backup resource detected after triggering backup." >&2
    return 1
  fi

  wait_for_backup "$backup_name"
}

get_primary_pod() {
  local primary
  primary=$(kubectl -n "$NAMESPACE" get cluster "$CNPG_CLUSTER_NAME" -o jsonpath='{.status.currentPrimary}' 2>/dev/null || true)
  if [[ -z "$primary" ]]; then
    primary=$(kubectl -n "$NAMESPACE" get pods -l "cnpg.io/cluster=$CNPG_CLUSTER_NAME" -o jsonpath='{.items[0].metadata.name}' 2>/dev/null || true)
  fi
  printf '%s' "$primary"
}

decode_b64() {
  local data="$1"
  if [[ -z "$data" ]]; then
    return 1
  fi
  printf '%s' "$data" | base64 -d 2>/dev/null
}

resolve_db_credentials() {
  local user_b64 pass_b64 user pass

  user_b64=$(kubectl -n "$NAMESPACE" get secret knoe-db-user -o jsonpath='{.data.username}' 2>/dev/null || true)
  pass_b64=$(kubectl -n "$NAMESPACE" get secret knoe-db-user -o jsonpath='{.data.password}' 2>/dev/null || true)
  user=$(decode_b64 "$user_b64" || true)
  pass=$(decode_b64 "$pass_b64" || true)

  if [[ -z "$user" || -z "$pass" ]]; then
    user_b64=$(kubectl -n "$NAMESPACE" get secret knoe-db-superuser -o jsonpath='{.data.username}' 2>/dev/null || true)
    pass_b64=$(kubectl -n "$NAMESPACE" get secret knoe-db-superuser -o jsonpath='{.data.password}' 2>/dev/null || true)
    user=$(decode_b64 "$user_b64" || true)
    pass=$(decode_b64 "$pass_b64" || true)
  fi

  if [[ -z "$user" || -z "$pass" ]]; then
    return 1
  fi

  printf '%s\n%s' "$user" "$pass"
}

pgdump_local() {
  local pod user pass db_name dump_dir dump_file timestamp
  pod=$(get_primary_pod)
  if [[ -z "$pod" ]]; then
    echo "ERROR: No CNPG pod available for pg_dump." >&2
    return 1
  fi

  local creds
  if ! creds=$(resolve_db_credentials); then
    echo "ERROR: Unable to resolve database credentials for pg_dump." >&2
    return 1
  fi
  user=$(printf '%s' "$creds" | sed -n '1p')
  pass=$(printf '%s' "$creds" | sed -n '2p')

  db_name=${KNOE_DB_NAME:-knoe-db}
  dump_dir="$BACKUP_DIR"
  mkdir -p "$dump_dir"

  timestamp=$(date +%Y%m%d%H%M%S)
  dump_file="$dump_dir/${CNPG_CLUSTER_NAME}-pgdump-${timestamp}.dump"

  echo "Running pg_dump against pod $pod (db=$db_name) ..."
  if kubectl -n "$NAMESPACE" exec "$pod" -c postgres -- env PGPASSWORD="$pass" \
      pg_dump -U "$user" -d "$db_name" -Fc > "$dump_file"; then
    echo "pg_dump saved to $dump_file"
    return 0
  fi

  echo "pg_dump failed; removing partial file." >&2
  rm -f "$dump_file"
  return 1
}

attempt_backup_if_active() {
  if ! cluster_exists; then
    return 0
  fi

  if ! cluster_has_pods; then
    echo "CNPG cluster '$CNPG_CLUSTER_NAME' exists but no pods detected; skipping backup." >&2
    return 0
  fi

  if ! cluster_has_ready_pods; then
    echo "CNPG cluster '$CNPG_CLUSTER_NAME' exists but no ready pods detected; skipping backup." >&2
    return 0
  fi

  echo "CNPG cluster '$CNPG_CLUSTER_NAME' detected; attempting Garage backup ..."
  if run_garage_backup; then
    return 0
  fi

  echo "Garage backup failed; attempting local pg_dump ..." >&2
  if pgdump_local; then
    return 0
  fi

  echo "WARN: Both Garage backup and local pg_dump failed." >&2
  return 0
}

wait_for_pod_ready() {
  local pod="$1"
  if ! kubectl -n "$NAMESPACE" wait --for=condition=Ready pod "$pod" --timeout=300s >/dev/null 2>&1; then
    echo "WARN: pod $pod did not become Ready within timeout." >&2
    return 1
  fi
  return 0
}

rollout_cluster() {
  ensure_tools
  ensure_namespace
  echo "Starting manual recreate rollout for $CNPG_CLUSTER_NAME in $NAMESPACE..."

  local has_cnpg_plugin="0"
  if kubectl cnpg version >/dev/null 2>&1; then
    has_cnpg_plugin="1"
    kubectl cnpg status "$CNPG_CLUSTER_NAME" -n "$NAMESPACE" || true
  fi

  local primary
  primary=$(kubectl -n "$NAMESPACE" get cluster "$CNPG_CLUSTER_NAME" -o jsonpath='{.status.currentPrimary}' 2>/dev/null || true)
  if [[ -z "$primary" ]]; then
    echo "ERROR: Could not identify primary instance." >&2
    return 1
  fi
  echo "Primary instance: $primary"

  local instances
  instances=$(kubectl -n "$NAMESPACE" get pods -l "cnpg.io/cluster=$CNPG_CLUSTER_NAME" -o jsonpath='{.items[*].metadata.name}' 2>/dev/null || true)
  if [[ -z "$instances" ]]; then
    echo "ERROR: No pods found for cluster $CNPG_CLUSTER_NAME." >&2
    return 1
  fi

  local pod
  for pod in $instances; do
    if [[ "$pod" != "$primary" ]]; then
      echo "Recreating non-primary pod: $pod..."
      kubectl delete pod -n "$NAMESPACE" "$pod"
      wait_for_pod_ready "$pod" || true
      if [[ "$has_cnpg_plugin" == "1" ]]; then
        kubectl cnpg status "$CNPG_CLUSTER_NAME" -n "$NAMESPACE" || true
      fi
    fi
  done

  local new_primary=""
  for pod in $instances; do
    if [[ "$pod" != "$primary" ]]; then
      new_primary="$pod"
      break
    fi
  done

  if [[ -n "$new_primary" && "$has_cnpg_plugin" == "1" ]]; then
    echo "Promoting $new_primary..."
    kubectl cnpg promote "$CNPG_CLUSTER_NAME" "$new_primary" -n "$NAMESPACE" || true
    echo "Waiting for $new_primary to become primary..."
    for _ in {1..60}; do
      local current_primary
      current_primary=$(kubectl -n "$NAMESPACE" get cluster "$CNPG_CLUSTER_NAME" -o jsonpath='{.status.currentPrimary}' 2>/dev/null || true)
      if [[ "$current_primary" == "$new_primary" ]]; then
        echo "$new_primary is now the primary."
        break
      fi
      sleep 5
    done
  else
    if [[ -n "$new_primary" ]]; then
      echo "WARN: kubectl cnpg plugin not available; skipping explicit promotion."
    fi
  fi

  echo "Recreating the old primary pod: $primary..."
  kubectl delete pod -n "$NAMESPACE" "$primary"
  wait_for_pod_ready "$primary" || true

  if [[ "$has_cnpg_plugin" == "1" ]]; then
    echo "Rollout complete. Final cluster status:"
    kubectl cnpg status "$CNPG_CLUSTER_NAME" -n "$NAMESPACE" || true
  fi
}

force_rollout() {
  echo "Attempting force rollout of CNPG pods ..."
  rollout_cluster
}

reset_and_reinit() {
  local backup_name tmp_manifest
  backup_name=$(latest_completed_backup_name)

  echo "Resetting CNPG cluster '$CNPG_CLUSTER_NAME' in namespace '$NAMESPACE' ..."
  kubectl -n "$NAMESPACE" delete cluster "$CNPG_CLUSTER_NAME" --ignore-not-found
  kubectl -n "$NAMESPACE" wait --for=delete pod -l "cnpg.io/cluster=$CNPG_CLUSTER_NAME" --timeout=180s >/dev/null 2>&1 || true

  recycle_released_prole_iscsi_pvs || true

  if [[ -n "$backup_name" && -f "$RECOVERY_TEMPLATE" ]]; then
    tmp_manifest=$(mktemp)
    sed "s/{{BACKUP_NAME}}/${backup_name}/g" "$RECOVERY_TEMPLATE" > "$tmp_manifest"
    echo "Re-initializing from backup $backup_name ..."
    CNPG_MANIFEST_OVERRIDE="$tmp_manifest" ensure_prole_stack_resources
    rm -f "$tmp_manifest"
  else
    if [[ -n "$backup_name" && ! -f "$RECOVERY_TEMPLATE" ]]; then
      echo "WARN: Recovery template not found: $RECOVERY_TEMPLATE" >&2
    fi
    if [[ -z "$backup_name" ]]; then
      echo "No completed backups found; starting fresh initialization." >&2
    fi
    ensure_prole_stack_resources
  fi

  wait_for_cnpg_pods "${CNPG_WAIT_TIMEOUT}"
}

recycle_released_prole_iscsi_pvs() {
  # StorageClass `synology-iscsi` uses `Retain` PV reclaim policy. After a CNPG reset, PVs can remain
  # in `Released` with a stale `claimRef`, which prevents new PVCs (same names) from binding.
  # In that case, CNPG init jobs remain `Pending` with "didn't find available persistent volumes".
  echo "Recycling Released synology-iscsi PVs (clearing stale claimRefs) for namespace '$NAMESPACE' ..."
  local pvs
  pvs=$(kubectl get pv -o json \
    | jq -r --arg ns "$NAMESPACE" '.items[]
      | select(.spec.storageClassName == "synology-iscsi")
      | select(.status.phase == "Released")
      | select((.spec.claimRef.namespace // "") == $ns)
      | .metadata.name' 2>/dev/null || true)

  if [[ -z "${pvs:-}" ]]; then
    return 0
  fi

  local pv
  for pv in $pvs; do
    echo "  - Clearing claimRef on PV: $pv"
    kubectl patch pv "$pv" --type json -p '[{"op":"remove","path":"/spec/claimRef"}]' >/dev/null 2>&1 || true
  done
}

# Resolve OpenBao URL: prefer explicit env, then in-cluster service.
# In k3s mode, out-of-cluster runs must use the k3s host (no localhost/port-forward).
bao_service_url() {
  if [[ -n "${PROLE_OPENBAO_URL:-}" ]]; then
    echo "$PROLE_OPENBAO_URL"
    return 0
  fi
  if prole_is_in_cluster; then
    local ns="${OPENBAO_NAMESPACE:-${SERVICE_NAMESPACE:-default}}"
    echo "http://$OPENBAO_NAME.$ns.svc.cluster.local:8200"
    return 0
  fi

  if [[ "${PROLE_MODE:-${DEPLOYMENT_MODE:-}}" == "k3s" ]]; then
    if command -v _prole_host_from_url >/dev/null 2>&1; then
      local host
      host=$(_prole_host_from_url "${PROLE_K3S_SERVER:-${K3S_SERVER_URL:-}}")
      if [[ -n "${host:-}" ]]; then
        # If the LoadBalancer has no external IP, k3s exposes the service via a nodePort.
        # Prefer that nodePort when available.
        if command -v kubectl >/dev/null 2>&1 && command -v jq >/dev/null 2>&1; then
          local ns node_port
          ns="${OPENBAO_NAMESPACE:-${SERVICE_NAMESPACE:-default}}"
          node_port=$(kubectl -n "$ns" get svc "${OPENBAO_NAME:-openbao}" -o json 2>/dev/null | jq -r '.spec.ports[] | select(.port==8200) | .nodePort // empty' | head -n 1)
          if [[ -n "${node_port:-}" && "${node_port:-}" != "null" ]]; then
            echo "http://${host}:${node_port}"
            return 0
          fi
        fi
        echo "http://${host}:8200"
        return 0
      fi
    fi
    echo ""
    return 0
  fi

  if curl -sS --connect-timeout 2 "http://127.0.0.1:8200/v1/sys/health" >/dev/null 2>&1; then
    echo "http://127.0.0.1:8200"
    return 0
  fi
  echo ""
}

fetch_admin_keys_and_db_pass_from_bao_or_local() {
  local priv_b64 pub_b64
  priv_b64=$(fetch_openbao_secret "$BAO_PATH_ADMIN" "admin_private_key_b64")
  pub_b64=$(fetch_openbao_secret "$BAO_PATH_ADMIN" "admin_public_key_b64")
  if [[ -n "${priv_b64:-}" && "${priv_b64:-}" != "null" && -n "${pub_b64:-}" && "${pub_b64:-}" != "null" ]]; then
    echo "Attempting to read admin key pair from OpenBao kv/$BAO_PATH_ADMIN ..."

    # Prefer the generic filenames.
    printf "%s" "$priv_b64" | base64 -d >"$ADMIN_PRIV_GENERIC"
    printf "%s" "$pub_b64" | base64 -d >"$ADMIN_PUB_GENERIC"
    chmod 0600 "$ADMIN_PRIV_GENERIC"

    # Mirror to legacy path for compatibility.
    cp "$ADMIN_PRIV_GENERIC" "$ADMIN_PRIV_ED25519" 2>/dev/null || true
    cp "$ADMIN_PUB_GENERIC" "$ADMIN_PUB_ED25519" 2>/dev/null || true
  fi

  local db_pass db_user
  db_pass=$(resolve_db_password)
  db_user="${KNOE_DB_USER:-prole}"
  if [[ -n "${db_pass:-}" && "${db_pass:-}" != "null" ]]; then
    echo "Ensuring database user secret 'knoe-db-user' ..."
    kubectl create secret generic knoe-db-user -n "$NAMESPACE" \
      --from-literal=username="$db_user" \
      --from-literal=password="$db_pass" \
      --dry-run=client -o yaml | kubectl_apply_retry "$NAMESPACE"

    echo "Ensuring database superuser secret 'knoe-db-superuser' ..."
    kubectl create secret generic knoe-db-superuser -n "$NAMESPACE" \
      --from-literal=username=postgres \
      --from-literal=password="$db_pass" \
      --dry-run=client -o yaml | kubectl_apply_retry "$NAMESPACE"
  fi

  # No env fallback: if OpenBao is unreachable and secrets are missing, fail clearly

  if ! kubectl -n "$NAMESPACE" get secret knoe-db-user >/dev/null 2>&1; then
    echo "ERROR: 'knoe-db-user' secret is missing in namespace '$NAMESPACE' and could not be resolved from OpenBao or local DB_PASSWORD." >&2
    return 1
  fi

  ensure_local_admin_keypair
  if [[ -f "$ADMIN_PRIV_GENERIC" && -f "$ADMIN_PUB_GENERIC" ]]; then
    echo "Using local admin key pair at $SECRETS_DIR"
    return 0
  fi

  echo "ERROR: Could not obtain admin key pair from OpenBao and could not generate local keys." >&2
  return 1
}

apply_cnpg_admin_secret() {
  echo "Creating/updating Secret cnpg-admin-key ..."
  kubectl create secret generic cnpg-admin-key -n "$NAMESPACE" \
    --from-file=admin.key="$ADMIN_PRIV_GENERIC" \
    --from-file=admin.pub="$ADMIN_PUB_GENERIC" \
    --dry-run=client -o yaml | kubectl_apply_retry "$NAMESPACE"
}

detect_and_reprovision_unencrypted_cluster() {
  local encryption_enabled="${AT_REST_ENCRYPTION_ENABLED:-false}"
  if [[ "$encryption_enabled" != "true" && "$encryption_enabled" != "True" && "$encryption_enabled" != "1" ]]; then
    return 0
  fi

  if ! kubectl -n "$NAMESPACE" get cluster "$CNPG_CLUSTER_NAME" >/dev/null 2>&1; then
    return 0
  fi

  local server_tls
  server_tls=$(kubectl -n "$NAMESPACE" get cluster "$CNPG_CLUSTER_NAME" -o jsonpath='{.spec.certificates.serverTLSSecret}' 2>/dev/null || echo "")
  if [[ -n "$server_tls" ]]; then
    echo "Existing cluster '$CNPG_CLUSTER_NAME' already has TLS configured ($server_tls); no conversion needed."
    return 0
  fi

  echo "WARN: Existing cluster '$CNPG_CLUSTER_NAME' does not have TLS/at-rest-encryption configured."
  echo "At-rest encryption is now enabled; the cluster must be reprovisioned."
  echo "NOTE: External full backup and restore from barman backup is not yet available."
  echo "      Removing existing cluster to allow re-creation with encryption enabled ..."

  attempt_backup_if_active

  kubectl -n "$NAMESPACE" delete cluster "$CNPG_CLUSTER_NAME" --ignore-not-found
  kubectl -n "$NAMESPACE" wait --for=delete pod -l "cnpg.io/cluster=$CNPG_CLUSTER_NAME" --timeout=180s >/dev/null 2>&1 || true
  echo "Existing unencrypted cluster removed. Will be re-created with TLS and encryption."
}

validate_ca_secret() {
  local secret_name="$1"
  local ns="$2"
  if ! kubectl -n "$ns" get secret "$secret_name" >/dev/null 2>&1; then
    return 1
  fi
  local crt key
  crt=$(kubectl -n "$ns" get secret "$secret_name" -o jsonpath='{.data.ca\.crt}' 2>/dev/null || true)
  key=$(kubectl -n "$ns" get secret "$secret_name" -o jsonpath='{.data.ca\.key}' 2>/dev/null || true)
  if [[ -z "$crt" || -z "$key" ]]; then
    echo "WARN: CA secret '$secret_name' exists but is missing ca.crt or ca.key data." >&2
    return 1
  fi
  if ! echo "$crt" | base64 -d 2>/dev/null | openssl x509 -noout 2>/dev/null; then
    echo "WARN: CA secret '$secret_name' has invalid certificate data." >&2
    return 1
  fi
  # CNPG machinery expects EC PRIVATE KEY (ECDSA P-256), not RSA or PKCS#8
  local key_header
  key_header=$(echo "$key" | base64 -d 2>/dev/null | head -1)
  if [[ "$key_header" != *"BEGIN EC PRIVATE KEY"* ]]; then
    echo "WARN: CA secret '$secret_name' key is not EC format (found: $key_header); CNPG requires EC PRIVATE KEY." >&2
    return 1
  fi
  return 0
}

validate_tls_secret() {
  local secret_name="$1"
  local ns="$2"
  if ! kubectl -n "$ns" get secret "$secret_name" >/dev/null 2>&1; then
    return 1
  fi
  local crt key
  crt=$(kubectl -n "$ns" get secret "$secret_name" -o jsonpath='{.data.tls\.crt}' 2>/dev/null || true)
  key=$(kubectl -n "$ns" get secret "$secret_name" -o jsonpath='{.data.tls\.key}' 2>/dev/null || true)
  if [[ -z "$crt" || -z "$key" ]]; then
    echo "WARN: TLS secret '$secret_name' exists but is missing tls.crt or tls.key data." >&2
    return 1
  fi
  if ! echo "$crt" | base64 -d 2>/dev/null | openssl x509 -noout 2>/dev/null; then
    echo "WARN: TLS secret '$secret_name' has invalid certificate data." >&2
    return 1
  fi
  # CNPG machinery expects EC PRIVATE KEY (ECDSA P-256), not RSA or PKCS#8
  local key_header
  key_header=$(echo "$key" | base64 -d 2>/dev/null | head -1)
  if [[ "$key_header" != *"BEGIN EC PRIVATE KEY"* ]]; then
    echo "WARN: TLS secret '$secret_name' key is not EC format (found: $key_header); CNPG requires EC PRIVATE KEY." >&2
    return 1
  fi
  return 0
}

generate_tls_if_missing() {
  local ca_secret_name="${CNPG_CLUSTER_NAME}-ca"
  local tls_secret_name="${CNPG_CLUSTER_NAME}-tls"
  local need_ca=0
  local need_tls=0

  if validate_ca_secret "$ca_secret_name" "$NAMESPACE"; then
    echo "CA secret $ca_secret_name already exists and is valid."
  else
    if kubectl -n "$NAMESPACE" get secret "$ca_secret_name" >/dev/null 2>&1; then
      echo "Removing invalid CA secret '$ca_secret_name' ..."
      kubectl -n "$NAMESPACE" delete secret "$ca_secret_name" --ignore-not-found
    fi
    need_ca=1
  fi

  if validate_tls_secret "$tls_secret_name" "$NAMESPACE"; then
    echo "TLS secret $tls_secret_name already exists and is valid."
  else
    if kubectl -n "$NAMESPACE" get secret "$tls_secret_name" >/dev/null 2>&1; then
      echo "Removing invalid TLS secret '$tls_secret_name' ..."
      kubectl -n "$NAMESPACE" delete secret "$tls_secret_name" --ignore-not-found
    fi
    need_tls=1
  fi

  if [[ "$need_ca" -eq 0 && "$need_tls" -eq 0 ]]; then
    echo "TLS secrets already present and valid; skipping generation."
    return 0
  fi

  local TMPD
  TMPD=$(mktemp -d)

  if [[ "$need_ca" -eq 1 ]]; then
    echo "Generating self-signed CA (EC P-256) for CNPG ..."
    openssl ecparam -name prime256v1 -genkey -noout -out "$TMPD/ca.key"
    openssl req -x509 -new -key "$TMPD/ca.key" -out "$TMPD/ca.crt" -days 3650 -subj "/CN=Prole CNPG CA"
    kubectl -n "$NAMESPACE" create secret generic "$ca_secret_name" \
      --from-file=ca.crt="$TMPD/ca.crt" \
      --from-file=ca.key="$TMPD/ca.key" \
      --dry-run=client -o yaml | kubectl_apply_retry "$NAMESPACE"
  else
    # Extract existing CA for signing the server certificate
    kubectl -n "$NAMESPACE" get secret "$ca_secret_name" -o jsonpath='{.data.ca\.crt}' | base64 -d > "$TMPD/ca.crt"
    kubectl -n "$NAMESPACE" get secret "$ca_secret_name" -o jsonpath='{.data.ca\.key}' | base64 -d > "$TMPD/ca.key"
  fi

  if [[ "$need_tls" -eq 1 ]]; then
    echo "Generating server TLS certificate for CNPG ($tls_secret_name) ..."
    openssl ecparam -name prime256v1 -genkey -noout -out "$TMPD/tls.key"
    openssl req -new -key "$TMPD/tls.key" -out "$TMPD/tls.csr" \
      -subj "/CN=${CNPG_CLUSTER_NAME}.${NAMESPACE}.svc"
    openssl x509 -req -in "$TMPD/tls.csr" -CA "$TMPD/ca.crt" -CAkey "$TMPD/ca.key" \
      -CAcreateserial -out "$TMPD/tls.crt" -days 365 \
      -extfile <(printf "subjectAltName=DNS:%s,DNS:%s-rw,DNS:%s-rw.%s.svc,DNS:%s-r,DNS:%s-ro" \
        "$CNPG_CLUSTER_NAME" "$CNPG_CLUSTER_NAME" "$CNPG_CLUSTER_NAME" "$NAMESPACE" \
        "$CNPG_CLUSTER_NAME" "$CNPG_CLUSTER_NAME")
    kubectl -n "$NAMESPACE" create secret tls "$tls_secret_name" \
      --cert="$TMPD/tls.crt" \
      --key="$TMPD/tls.key" \
      --dry-run=client -o yaml | kubectl_apply_retry "$NAMESPACE"
  fi

  rm -rf "$TMPD"
}

# Resolve latest CNPG version from GitHub if possible, fallback to a sensible default.
get_latest_cnpg_version() {
  local version
  version=$(curl -s "https://api.github.com/repos/cloudnative-pg/cloudnative-pg/releases/latest" | jq -r '.tag_name' | sed 's/^v//' || echo "")
  if [[ -z "$version" || "$version" == "null" ]]; then
    echo "1.27.0"
  else
    echo "$version"
  fi
}

initialize() {
  ensure_tools
  ensure_namespace
  wait_for_apiserver_ready 180
  attempt_backup_if_active
  ensure_cnpg_operator
  pin_cnpg_controller
  ensure_barman_plugin
  echo "Using namespace: $NAMESPACE"
  
  if ! fetch_admin_keys_and_db_pass_from_bao_or_local; then
    echo "ERROR: Failed to fetch/ensure database secrets and admin keys." >&2
    return 1
  fi
  apply_cnpg_admin_secret
  detect_and_reprovision_unencrypted_cluster
  generate_tls_if_missing

  # Pre-flight: verify required secrets exist before deploying cluster
  local _missing_secrets=""
  if ! kubectl -n "$NAMESPACE" get secret knoe-db-user >/dev/null 2>&1; then
    _missing_secrets="${_missing_secrets} knoe-db-user"
  fi
  if ! kubectl -n "$NAMESPACE" get secret "${CNPG_CLUSTER_NAME}-tls" >/dev/null 2>&1; then
    _missing_secrets="${_missing_secrets} ${CNPG_CLUSTER_NAME}-tls"
  fi
  if ! kubectl -n "$NAMESPACE" get secret "${CNPG_CLUSTER_NAME}-ca" >/dev/null 2>&1; then
    _missing_secrets="${_missing_secrets} ${CNPG_CLUSTER_NAME}-ca"
  fi
  if [[ -n "$_missing_secrets" ]]; then
    echo "ERROR: Required secrets missing in namespace '$NAMESPACE':${_missing_secrets}" >&2
    echo "The CNPG cluster cannot start without these secrets." >&2
    return 1
  fi
  echo "Pre-flight check passed: all required secrets present."

  if ! _ensure_knoe_db_image; then
    echo "ERROR: Pre-flight image check failed; aborting cluster initialization." >&2
    return 1
  fi

  ensure_prole_stack_resources

  if ! wait_for_cnpg_pods "${CNPG_WAIT_TIMEOUT}"; then
    if cluster_has_pods; then
      echo "WARN: CNPG pods exist but did not become ready; attempting force rollout ..." >&2
      if force_rollout; then
        if ! wait_for_cnpg_pods "${CNPG_WAIT_TIMEOUT}"; then
          echo "WARN: Force rollout did not recover CNPG; attempting reset and re-init ..." >&2
          if ! reset_and_reinit; then
            return 1
          fi
        fi
      else
        echo "WARN: Force rollout failed; attempting reset and re-init ..." >&2
        if ! reset_and_reinit; then
          return 1
        fi
      fi
    else
      echo "WARN: No CNPG pods found — this is a new installation." >&2
      echo "The CNPG operator was unable to create cluster pods." >&2
      echo "Check operator logs:  kubectl -n cnpg-system logs -l app.kubernetes.io/name=cloudnative-pg" >&2
      echo "Check cluster status: kubectl -n $NAMESPACE describe cluster $CNPG_CLUSTER_NAME" >&2
      return 1
    fi
  fi

  echo "Ensuring CNPG continuous backups (ObjectStore + Cluster backup config + ScheduledBackup) ..."
  "$SCRIPT_DIR/init_cnpg_backup.sh" start

  echo "Initialization complete for CNPG + cert artifacts."
  prole_register_port_forward "postgres" "${NAMESPACE:-default}" "svc/${CNPG_CLUSTER_NAME}-rw" "5432" "5432" "0.0.0.0" "TCP" "PostgreSQL"
}

update_reload() {
  initialize
}

deploy_cluster() {
  ensure_tools
  ensure_namespace
  ensure_cnpg_operator
  pin_cnpg_controller
  ensure_barman_plugin

  ensure_prole_protected_storage

  local image
  if [[ "$VERSION" == "latest" || -z "$VERSION" ]]; then
    image=$(get_latest_image)
  else
    image="knoe-db:$VERSION"
  fi
  image=$(resolve_cnpg_image "$image")
  sync_manifest_image "$image"

  generate_tls_if_missing

  echo "Deploying $image to cluster $CNPG_CLUSTER_NAME in namespace $NAMESPACE..."
  _ensure_knoe_db_image || true
  local manifest
  manifest="$CNPG_MANIFEST"
  if [[ -n "$CNPG_MANIFEST_OVERRIDE" ]]; then
    manifest="$CNPG_MANIFEST_OVERRIDE"
  fi
  if [[ ! -f "$manifest" ]]; then
    echo "ERROR: CNPG manifest not found at $manifest" >&2
    return 1
  fi
  apply_cnpg_cluster_manifest "$manifest"
  validate_cnpg_runtime_storage
  reconcile_cnpg_instances

  local current_image
  current_image=$(kubectl -n "$NAMESPACE" get cluster "$CNPG_CLUSTER_NAME" -o jsonpath='{.spec.imageName}' 2>/dev/null || echo "")
  if [[ -n "$current_image" && "$current_image" != "$image" ]]; then
    echo "Patching cluster to use image '$image' (was '$current_image')..."
    kubectl -n "$NAMESPACE" patch cluster "$CNPG_CLUSTER_NAME" --type merge -p "{\"spec\": {\"imageName\": \"$image\"}}"
    if kubectl cnpg version >/dev/null 2>&1; then
      kubectl cnpg restart "$CNPG_CLUSTER_NAME" -n "$NAMESPACE" || true
    fi
  fi
}

preflight_image() {
  ensure_tools
  ensure_namespace
  _ensure_knoe_db_image
}

case "$ACTION" in
  preflight-image|image-preflight|preflight_image)
    preflight_image
    ;;
  recreate)
    ensure_tools
    "$0" delete "$CNPG_CLUSTER_NAME"
    # Clean up PVCs left behind by the CNPG operator (not in static manifests)
    echo "Cleaning up PVCs in namespace '$NAMESPACE' ..."
    kubectl -n "$NAMESPACE" delete pvc --all --ignore-not-found 2>/dev/null || true
    # Wait briefly for pods to fully terminate before re-creating
    echo "Waiting for pods to terminate in namespace '$NAMESPACE' ..."
    _wait_term=0
    while kubectl -n "$NAMESPACE" get pods --no-headers 2>/dev/null | grep -qv '^No resources'; do
      sleep 3
      _wait_term=$(( _wait_term + 3 ))
      if (( _wait_term >= 60 )); then
        echo "WARN: Pods still present after 60s; proceeding anyway." >&2
        break
      fi
    done
    "$0" create "$CNPG_CLUSTER_NAME"
    ;;
  create)
    ensure_tools
    ensure_namespace
    initialize
    ;;
  delete)
    ensure_tools
    echo "Deleting all resources for '$CNPG_CLUSTER_NAME' ..."
    # Delete each manifest individually, mirroring the apply pattern:
    # ingress.yaml contains multi-namespace resources and must be deleted without -n.
    for _del_f in "$K8S_PROLE_DIR"/*.yaml; do
      _del_base=$(basename "$_del_f")
      case "$_del_base" in
        kustomization.yaml|ingress.yaml) continue ;;
        *) kubectl delete -n "$NAMESPACE" -f "$_del_f" --ignore-not-found 2>&1 \
             | grep -v "^Error from server (NotFound)" || true ;;
      esac
    done
    if [[ -f "$K8S_PROLE_DIR/ingress.yaml" ]]; then
      kubectl delete -f "$K8S_PROLE_DIR/ingress.yaml" --ignore-not-found 2>&1 \
        | grep -v "^Error from server (NotFound)" || true
    fi
    ;;
  start)
    ensure_tools
    ensure_namespace
    ensure_cnpg_operator
    pin_cnpg_controller
    ensure_barman_plugin
    if [[ ! -f "$CNPG_MANIFEST" ]]; then
      echo "ERROR: CNPG manifest not found at $CNPG_MANIFEST" >&2
      exit 1
    fi
    echo "Starting CloudNative-PG cluster from $CNPG_MANIFEST in namespace $NAMESPACE..."
    prole_render_manifest "$CNPG_MANIFEST" | kubectl apply -n "$NAMESPACE" -f -
    reconcile_cnpg_instances
    ;;
  stop)
    ensure_tools
    if [[ ! -f "$CNPG_MANIFEST" ]]; then
      echo "ERROR: CNPG manifest not found at $CNPG_MANIFEST" >&2
      exit 1
    fi
    echo "Stopping CloudNative-PG cluster using $CNPG_MANIFEST ..."
    kubectl delete -f "$CNPG_MANIFEST" --ignore-not-found
    ;;
  status)
    ensure_tools
    echo "--- CloudNative-PG Cluster Status ($CNPG_CLUSTER_NAME) ---"
    if kubectl -n "$NAMESPACE" get cluster "$CNPG_CLUSTER_NAME" >/dev/null 2>&1; then
      kubectl -n "$NAMESPACE" get cluster "$CNPG_CLUSTER_NAME"
      echo ""
      echo "CNPG Plugin Status:"
      if kubectl cnpg version >/dev/null 2>&1; then
        kubectl cnpg status "$CNPG_CLUSTER_NAME" -n "$NAMESPACE"
      else
        echo "Note: 'kubectl cnpg' plugin not found; skipping detailed status."
      fi
    else
      echo "Cluster '$CNPG_CLUSTER_NAME' not found in namespace '$NAMESPACE'."
    fi
    ;;
  restart)
    ensure_tools
    "$0" stop
    "$0" start
    ;;
  initialize)
    if ! initialize; then
      exit 1
    fi
    ;;
  update|reload)
    if ! update_reload; then
      exit 1
    fi
    ;;
  deploy)
    deploy_cluster
    ;;
  rollout|force-rollout)
    rollout_cluster
    ;;
  install-barman-plugin)
    ensure_tools
    ensure_cnpg_operator
    pin_cnpg_controller
    ensure_barman_plugin
    ;;
  *)
    echo "Usage: $0 {create|delete|recreate|start|stop|status|restart|initialize|update|reload|deploy|rollout|install-barman-plugin} [cluster] [version]" >&2
    exit 2
    ;;
esac