prole/etc/init_cnpg_backup.sh

#!/usr/bin/env bash

set -euo pipefail

# init_cnpg_backup.sh
# Purpose:
# - Configure CloudNative-PG to backup to Garage (S3-compatible) via Barman Cloud Plugin
# - Create initial backup

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

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

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

ACTION=${1:-start}

NAMESPACE="${PROLE_NAMESPACE:-${DATABASE_NAMESPACE:-${SERVICE_NAMESPACE:-knoe-db}}}"
CNPG_CLUSTER_NAME=${CNPG_CLUSTER_NAME:-knoe-db}
CNPG_OPERATOR_NAMESPACE=${CNPG_OPERATOR_NAMESPACE:-cnpg-system}
GARAGE_NAME=${GARAGE_NAME:-garage}
SERVICE_NAMESPACE=${SERVICE_NAMESPACE:-}
GARAGE_NAMESPACE=${GARAGE_NAMESPACE:-}
GARAGE_BACKUP_BUCKET=${GARAGE_BACKUP_BUCKET:-knoe-db-backups}
GARAGE_BACKUP_KEY_NAME=${GARAGE_BACKUP_KEY_NAME:-knoe-db-backup}
GARAGE_BACKUP_SECRET_NAME=${GARAGE_BACKUP_SECRET_NAME:-knoe-db-barman-s3}
GARAGE_S3_ENDPOINT=${GARAGE_S3_ENDPOINT:-}
GARAGE_S3_REGION=${GARAGE_S3_REGION:-garage}
RUN_FIRST_BACKUP=${RUN_FIRST_BACKUP:-1}
RETENTION_POLICY=${RETENTION_POLICY:-30d}
GARAGE_LAYOUT_BOOTSTRAP_ENABLED=${GARAGE_LAYOUT_BOOTSTRAP_ENABLED:-1}
BARMAN_PLUGIN_NAME=${BARMAN_PLUGIN_NAME:-barman-cloud.cloudnative-pg.io}
BARMAN_OBJECT_NAME=${BARMAN_OBJECT_NAME:-knoe-db-barman-objectstore}
BACKUP_STATUS_TIMEOUT=${BACKUP_STATUS_TIMEOUT:-600}
BACKUP_STATUS_INTERVAL=${BACKUP_STATUS_INTERVAL:-10}
PLUGIN_READY_TIMEOUT=${PLUGIN_READY_TIMEOUT:-180}
OBJECTSTORE_READY_TIMEOUT=${OBJECTSTORE_READY_TIMEOUT:-180}
OBJECTSTORE_READY_INTERVAL=${OBJECTSTORE_READY_INTERVAL:-5}
KUBECTL_CONTEXT_OVERRIDE=${KUBECTL_CONTEXT_OVERRIDE:-${DB_CLUSTER_KUBECONTEXT:-${KUBECTL_CONTEXT:-${KUBECONTEXT:-}}}}

SCHEDULED_BACKUP_NAME=${SCHEDULED_BACKUP_NAME:-knoe-db-scheduled-backup}
SCHEDULED_BACKUP_CRON=${SCHEDULED_BACKUP_CRON:-"0 3 * * *"}
SCHEDULED_BACKUP_IMMEDIATE=${SCHEDULED_BACKUP_IMMEDIATE:-true}
SCHEDULED_BACKUP_METHOD=${SCHEDULED_BACKUP_METHOD:-plugin}
SCHEDULED_BACKUP_INCREMENTAL=${SCHEDULED_BACKUP_INCREMENTAL:-false}

PLUGIN_SOCKET_DIR=${PLUGIN_SOCKET_DIR:-/plugins}
LAST_BACKUP_NAME=""

usage() {
  cat <<USAGE
Usage: $0 [start|backup|status]

Actions:
  start                Configure Garage-backed backups and run initial backup
  backup [full|incr]   Trigger a new backup now (default: full)
  status               Show backup resources and verify at least one successful base backup
USAGE
  exit 1
}

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

select_kube_context() {
  local target_context="${KUBECTL_CONTEXT_OVERRIDE:-}"
  if [[ -z "$target_context" ]]; then
    return 0
  fi

  if kubectl config get-contexts "$target_context" >/dev/null 2>&1; then
    kubectl config use-context "$target_context" >/dev/null
  else
    echo "WARN: Requested kubecontext '$target_context' not found; continuing with current context." >&2
  fi
}

_kubectl_is_transient_error() {
  local msg="${1:-}"
  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)

  while true; do
    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
      return 1
    fi
    sleep 2
  done
}

wait_for_cnpg_webhook() {
  local timeout=${1:-180}
  local start_time elapsed
  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_OPERATOR_NAMESPACE" get endpoints cnpg-webhook-service -o jsonpath='{.subsets[*].addresses[*].ip}' 2>/dev/null || true)
    if [[ -n "$endpoints" ]]; then
      return 0
    fi
    elapsed=$(( $(date +%s) - start_time ))
    if (( elapsed > timeout )); then
      echo "ERROR: CNPG webhook service has no endpoints after ${elapsed}s." >&2
      kubectl -n "$CNPG_OPERATOR_NAMESPACE" get pods 2>/dev/null >&2 || true
      return 1
    fi
    sleep 5
  done
}

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
}

ensure_namespace() {
  if ! kubectl get namespace "$NAMESPACE" >/dev/null 2>&1; then
    echo "Creating namespace '$NAMESPACE' ..."
    kubectl create namespace "$NAMESPACE" >/dev/null 2>&1 || true
  fi
}

ensure_cluster() {
  if ! kubectl get cluster "$CNPG_CLUSTER_NAME" -n "$NAMESPACE" >/dev/null 2>&1; then
    echo "ERROR: CNPG cluster '$CNPG_CLUSTER_NAME' not found in namespace '$NAMESPACE'." >&2
    exit 1
  fi
}

barman_crd_ready() {
  kubectl get crd objectstores.barmancloud.cnpg.io >/dev/null 2>&1
}

garage_pod_exists_in_namespace() {
  local ns="${1:-}"
  local pod_name
  [[ -z "$ns" ]] && return 1
  pod_name=$(kubectl get pods -n "$ns" -l "app=$GARAGE_NAME" -o jsonpath='{.items[0].metadata.name}' 2>/dev/null || true)
  [[ -n "$pod_name" ]]
}

resolve_garage_namespace_and_endpoint() {
  if [[ -z "$GARAGE_NAMESPACE" ]]; then
    if [[ -n "$SERVICE_NAMESPACE" ]] && garage_pod_exists_in_namespace "$SERVICE_NAMESPACE"; then
      GARAGE_NAMESPACE="$SERVICE_NAMESPACE"
    elif garage_pod_exists_in_namespace "knoe-system"; then
      GARAGE_NAMESPACE="knoe-system"
    elif garage_pod_exists_in_namespace "$NAMESPACE"; then
      GARAGE_NAMESPACE="$NAMESPACE"
    elif [[ -n "$SERVICE_NAMESPACE" ]]; then
      GARAGE_NAMESPACE="$SERVICE_NAMESPACE"
    else
      GARAGE_NAMESPACE="$NAMESPACE"
    fi
  fi

  if [[ -z "$GARAGE_S3_ENDPOINT" ]]; then
    GARAGE_S3_ENDPOINT="http://$GARAGE_NAME.$GARAGE_NAMESPACE.svc.cluster.local:3900"
  fi
}

get_garage_pod() {
  kubectl get pods -n "$GARAGE_NAMESPACE" -l "app=$GARAGE_NAME" -o jsonpath='{.items[0].metadata.name}' 2>/dev/null || true
}

garage_exec() {
  local pod
  pod=$(get_garage_pod)
  if [[ -z "$pod" ]]; then
    echo "ERROR: Garage pod not found in namespace '$GARAGE_NAMESPACE'." >&2
    exit 1
  fi
  kubectl exec -n "$GARAGE_NAMESPACE" "$pod" -- /garage "$@"
}

bootstrap_garage_layout_if_needed() {
  if [[ "${GARAGE_LAYOUT_BOOTSTRAP_ENABLED}" != "1" ]]; then
    return 0
  fi

  local init_script="$SCRIPT_DIR/init_garage_store.sh"
  if [[ ! -f "$init_script" ]]; then
    echo "WARN: Garage bootstrap helper not found: $init_script" >&2
    return 0
  fi

  echo "Garage layout still not applied; invoking Garage bootstrap recovery ..."
  if [[ -n "${KNOE_MODE:-}" ]]; then
    if ! bash "$init_script" --mode "$KNOE_MODE" start; then
      echo "WARN: Garage bootstrap recovery failed (continuing wait loop)." >&2
    fi
  else
    if ! bash "$init_script" start; then
      echo "WARN: Garage bootstrap recovery failed (continuing wait loop)." >&2
    fi
  fi
}

parse_key_output() {
  local output="$1"
  local access_key secret_key
  access_key=$(echo "$output" | sed -nE 's/^(Access key ID|Key ID):[[:space:]]+//p' | head -n1)
  secret_key=$(echo "$output" | sed -nE 's/^(Secret access key|Secret key):[[:space:]]+//p' | head -n1)
  if [[ -z "$access_key" || -z "$secret_key" ]]; then
    return 1
  fi
  printf "%s\n%s" "$access_key" "$secret_key"
}

ensure_garage_ready() {
  echo "Checking Garage readiness ..."
  local i status_out
  local bootstrap_attempted=0
  for i in {1..30}; do
    if status_out=$(garage_exec status 2>/dev/null); then
      # If layout is applied, DataAvail should eventually show something or at least the node should be healthy.
      # A better check is 'layout show' to see if current layout version > 0
      local layout_out version
      layout_out=$(garage_exec layout show 2>/dev/null | grep -v "INFO" || true)
      version=$(echo "$layout_out" | awk -F: '/Current cluster layout version/ {gsub(/[[:space:]]/,"",$2); print $2; exit}' || true)
      if [[ -n "$version" && "$version" -gt 0 ]]; then
         echo "Garage layout version $version is applied and ready."
         return 0
      fi
    fi
    if [[ "$bootstrap_attempted" -eq 0 && "$i" -ge 6 ]]; then
      bootstrap_garage_layout_if_needed
      bootstrap_attempted=1
    fi
    echo "Waiting for Garage layout to be applied... ($i/30)"
    sleep 5
  done
  echo "ERROR: Garage not ready (layout not applied) after 30 attempts." >&2
  exit 1
}

ensure_garage_bucket_and_key() {
  ensure_garage_ready
  echo "Ensuring Garage bucket and access key for backups ..."
  local key_info parsed access_key secret_key
  if key_info=$(garage_exec key info --show-secret "$GARAGE_BACKUP_KEY_NAME" 2>/dev/null); then
    :
  else
    key_info=$(garage_exec key create "$GARAGE_BACKUP_KEY_NAME")
  fi

  if ! parsed=$(parse_key_output "$key_info"); then
    echo "ERROR: Unable to parse Garage key output." >&2
    echo "$key_info" >&2
    exit 1
  fi
  access_key=$(echo "$parsed" | sed -n '1p')
  secret_key=$(echo "$parsed" | sed -n '2p')

  if ! garage_exec bucket info "$GARAGE_BACKUP_BUCKET" >/dev/null 2>&1; then
    garage_exec bucket create "$GARAGE_BACKUP_BUCKET"
  fi

  garage_exec bucket allow --read --write --owner --key "$GARAGE_BACKUP_KEY_NAME" "$GARAGE_BACKUP_BUCKET" || true

  echo "Creating/updating Kubernetes secret '$GARAGE_BACKUP_SECRET_NAME' ..."
  kubectl create secret generic "$GARAGE_BACKUP_SECRET_NAME" -n "$NAMESPACE" \
    --from-literal=ACCESS_KEY_ID="$access_key" \
    --from-literal=SECRET_ACCESS_KEY="$secret_key" \
    --from-literal=REGION="$GARAGE_S3_REGION" \
    --dry-run=client -o yaml | kubectl_apply_retry "$NAMESPACE"
}

apply_barman_object_store() {
  if ! barman_crd_ready; then
    echo "ERROR: Barman Cloud plugin CRD not found (objectstores.barmancloud.cnpg.io). Install the plugin first." >&2
    exit 1
  fi

  echo "Applying Barman Cloud ObjectStore '$BARMAN_OBJECT_NAME' ..."
  kubectl_apply_retry "$NAMESPACE" <<OBJECTSTORE
apiVersion: barmancloud.cnpg.io/v1
kind: ObjectStore
metadata:
  name: $BARMAN_OBJECT_NAME
spec:
  retentionPolicy: $RETENTION_POLICY
  configuration:
    destinationPath: s3://$GARAGE_BACKUP_BUCKET/
    endpointURL: $GARAGE_S3_ENDPOINT
    s3Credentials:
      accessKeyId:
        name: $GARAGE_BACKUP_SECRET_NAME
        key: ACCESS_KEY_ID
      secretAccessKey:
        name: $GARAGE_BACKUP_SECRET_NAME
        key: SECRET_ACCESS_KEY
      region:
        name: $GARAGE_BACKUP_SECRET_NAME
        key: REGION
    wal:
      compression: gzip
    data:
      compression: gzip
OBJECTSTORE
}

objectstore_spec_matches_expected() {
  local destination endpoint retention
  local access_key_secret access_key_name secret_key_secret secret_key_name
  local region_secret region_name

  destination=$(kubectl -n "$NAMESPACE" get objectstore "$BARMAN_OBJECT_NAME" -o jsonpath='{.spec.configuration.destinationPath}' 2>/dev/null || true)
  endpoint=$(kubectl -n "$NAMESPACE" get objectstore "$BARMAN_OBJECT_NAME" -o jsonpath='{.spec.configuration.endpointURL}' 2>/dev/null || true)
  retention=$(kubectl -n "$NAMESPACE" get objectstore "$BARMAN_OBJECT_NAME" -o jsonpath='{.spec.retentionPolicy}' 2>/dev/null || true)
  access_key_secret=$(kubectl -n "$NAMESPACE" get objectstore "$BARMAN_OBJECT_NAME" -o jsonpath='{.spec.configuration.s3Credentials.accessKeyId.name}' 2>/dev/null || true)
  access_key_name=$(kubectl -n "$NAMESPACE" get objectstore "$BARMAN_OBJECT_NAME" -o jsonpath='{.spec.configuration.s3Credentials.accessKeyId.key}' 2>/dev/null || true)
  secret_key_secret=$(kubectl -n "$NAMESPACE" get objectstore "$BARMAN_OBJECT_NAME" -o jsonpath='{.spec.configuration.s3Credentials.secretAccessKey.name}' 2>/dev/null || true)
  secret_key_name=$(kubectl -n "$NAMESPACE" get objectstore "$BARMAN_OBJECT_NAME" -o jsonpath='{.spec.configuration.s3Credentials.secretAccessKey.key}' 2>/dev/null || true)
  region_secret=$(kubectl -n "$NAMESPACE" get objectstore "$BARMAN_OBJECT_NAME" -o jsonpath='{.spec.configuration.s3Credentials.region.name}' 2>/dev/null || true)
  region_name=$(kubectl -n "$NAMESPACE" get objectstore "$BARMAN_OBJECT_NAME" -o jsonpath='{.spec.configuration.s3Credentials.region.key}' 2>/dev/null || true)

  [[ "$destination" == "s3://$GARAGE_BACKUP_BUCKET/" ]] || return 1
  [[ "$endpoint" == "$GARAGE_S3_ENDPOINT" ]] || return 1
  [[ "$retention" == "$RETENTION_POLICY" ]] || return 1
  [[ "$access_key_secret" == "$GARAGE_BACKUP_SECRET_NAME" ]] || return 1
  [[ "$access_key_name" == "ACCESS_KEY_ID" ]] || return 1
  [[ "$secret_key_secret" == "$GARAGE_BACKUP_SECRET_NAME" ]] || return 1
  [[ "$secret_key_name" == "SECRET_ACCESS_KEY" ]] || return 1
  [[ "$region_secret" == "$GARAGE_BACKUP_SECRET_NAME" ]] || return 1
  [[ "$region_name" == "REGION" ]] || return 1
  return 0
}

ensure_barman_object_store_config() {
  local attempts=${OBJECTSTORE_RECONCILE_ATTEMPTS:-3}
  local i

  for (( i=1; i<=attempts; i++ )); do
    apply_barman_object_store
    if objectstore_spec_matches_expected; then
      return 0
    fi
    echo "WARN: ObjectStore '$BARMAN_OBJECT_NAME' spec does not match expected Garage configuration (attempt ${i}/${attempts}); reapplying ..."
    sleep 2
  done

  echo "ERROR: ObjectStore '$BARMAN_OBJECT_NAME' does not match expected Garage configuration after ${attempts} attempts." >&2
  kubectl -n "$NAMESPACE" get objectstore "$BARMAN_OBJECT_NAME" -o yaml 2>/dev/null >&2 || true
  return 1
}

ensure_barman_plugin_config() {
  local plugin_names plugin_present barman_obj_name plugin_enabled plugin_wal_archiver
  plugin_names=$(kubectl -n "$NAMESPACE" get cluster "$CNPG_CLUSTER_NAME" -o jsonpath='{.spec.plugins[*].name}' 2>/dev/null || true)
  plugin_present=$(printf '%s\n' "$plugin_names" | tr ' ' '\n' | grep -F "$BARMAN_PLUGIN_NAME" || true)
  if [[ -n "$plugin_present" ]]; then
    # Plugin name present — verify it's fully active for WAL archiving and object store routing.
    barman_obj_name=$(kubectl -n "$NAMESPACE" get cluster "$CNPG_CLUSTER_NAME" \
      -o jsonpath="{range .spec.plugins[?(@.name==\"$BARMAN_PLUGIN_NAME\")]}{.parameters.barmanObjectName}{end}" 2>/dev/null || true)
    plugin_enabled=$(kubectl -n "$NAMESPACE" get cluster "$CNPG_CLUSTER_NAME" \
      -o jsonpath="{range .spec.plugins[?(@.name==\"$BARMAN_PLUGIN_NAME\")]}{.enabled}{end}" 2>/dev/null || true)
    plugin_wal_archiver=$(kubectl -n "$NAMESPACE" get cluster "$CNPG_CLUSTER_NAME" \
      -o jsonpath="{range .spec.plugins[?(@.name==\"$BARMAN_PLUGIN_NAME\")]}{.isWALArchiver}{end}" 2>/dev/null || true)

    if [[ "$barman_obj_name" == "$BARMAN_OBJECT_NAME" && "${plugin_enabled,,}" == "true" && "${plugin_wal_archiver,,}" == "true" ]]; then
      return 0
    fi
    echo "Updating Barman plugin entry: ensuring enabled=true, isWALArchiver=true, and barmanObjectName=$BARMAN_OBJECT_NAME ..."
    local idx
    idx=$(kubectl -n "$NAMESPACE" get cluster "$CNPG_CLUSTER_NAME" \
      -o json 2>/dev/null \
      | python3 -c "import sys,json; items=json.load(sys.stdin)['spec'].get('plugins',[]); print(next((i for i,p in enumerate(items) if p.get('name')=='barman-cloud.cloudnative-pg.io'),-1))")
    if [[ "$idx" -ge 0 ]]; then
      kubectl patch cluster "$CNPG_CLUSTER_NAME" -n "$NAMESPACE" --type json -p "[
        {
          \"op\": \"replace\",
          \"path\": \"/spec/plugins/$idx\",
          \"value\": {
            \"enabled\": true,
            \"name\": \"$BARMAN_PLUGIN_NAME\",
            \"isWALArchiver\": true,
            \"parameters\": {\"barmanObjectName\": \"$BARMAN_OBJECT_NAME\"}
          }
        }
      ]"
      return 0
    fi
    # Fallback: append a correctly configured entry
  fi

  local plugins_json
  plugins_json=$(kubectl -n "$NAMESPACE" get cluster "$CNPG_CLUSTER_NAME" -o jsonpath='{.spec.plugins}' 2>/dev/null || true)

  echo "Configuring CNPG to use Barman Cloud plugin '$BARMAN_PLUGIN_NAME' ..."
  if [[ -z "$plugins_json" || "$plugins_json" == "null" || "$plugins_json" == "[]" ]]; then
    kubectl patch cluster "$CNPG_CLUSTER_NAME" -n "$NAMESPACE" --type merge -p "{
      \"spec\": {
        \"plugins\": [
          {
            \"enabled\": true,
            \"name\": \"$BARMAN_PLUGIN_NAME\",
            \"isWALArchiver\": true,
            \"parameters\": {\"barmanObjectName\": \"$BARMAN_OBJECT_NAME\"}
          }
        ]
      }
    }"
  else
    kubectl patch cluster "$CNPG_CLUSTER_NAME" -n "$NAMESPACE" --type json -p "[
      {
        \"op\": \"add\",
        \"path\": \"/spec/plugins/-\",
        \"value\": {
          \"enabled\": true,
          \"name\": \"$BARMAN_PLUGIN_NAME\",
          \"isWALArchiver\": true,
          \"parameters\": {\"barmanObjectName\": \"$BARMAN_OBJECT_NAME\"}
        }
      }
    ]"
  fi
}

ensure_cluster_backup_config() {
  # The Barman Cloud plugin is configured via:
  # - `Cluster.spec.plugins[]` (for WAL archiving)
  # - `Backup`/`ScheduledBackup` resources (method=plugin + pluginConfiguration)
  # Retention is configured in `ObjectStore.spec.retentionPolicy`.
  #
  # Historical versions of this script used to patch `Cluster.spec.backup.*`.
  # That configuration is not used by the plugin and can lead to warnings.
  # Clean up only the legacy fields (if present) and otherwise leave `spec.backup` untouched.

  local cur_retention cur_plugin cur_barman_object cur_barman_objectstore
  cur_retention=$(kubectl -n "$NAMESPACE" get cluster "$CNPG_CLUSTER_NAME" -o jsonpath='{.spec.backup.retentionPolicy}' 2>/dev/null || true)
  cur_plugin=$(kubectl -n "$NAMESPACE" get cluster "$CNPG_CLUSTER_NAME" -o jsonpath='{.spec.backup.pluginConfiguration.name}' 2>/dev/null || true)
  cur_barman_object=$(kubectl -n "$NAMESPACE" get cluster "$CNPG_CLUSTER_NAME" -o jsonpath='{.spec.backup.pluginConfiguration.parameters.barmanObjectName}' 2>/dev/null || true)
  cur_barman_objectstore=$(kubectl -n "$NAMESPACE" get cluster "$CNPG_CLUSTER_NAME" -o jsonpath='{.spec.backup.barmanObjectStore}' 2>/dev/null || true)

  if [[ -z "$cur_retention" && -z "$cur_plugin" && -z "$cur_barman_object" && -z "$cur_barman_objectstore" ]]; then
    return 0
  fi

  echo "Cleaning up legacy Cluster.spec.backup fields (plugin uses ObjectStore + spec.plugins instead) ..."
  local json_patch="[]"
  if [[ -n "$cur_retention" ]]; then
    json_patch='[{"op":"remove","path":"/spec/backup/retentionPolicy"}]'
  fi
  if [[ -n "$cur_plugin" || -n "$cur_barman_object" ]]; then
    if [[ "$json_patch" == "[]" ]]; then
      json_patch='[{"op":"remove","path":"/spec/backup/pluginConfiguration"}]'
    else
      json_patch='[{"op":"remove","path":"/spec/backup/retentionPolicy"},{"op":"remove","path":"/spec/backup/pluginConfiguration"}]'
    fi
  fi
  if [[ -n "$cur_barman_objectstore" ]]; then
    # Remove built-in barmanObjectStore config when using the plugin.
    if [[ "$json_patch" == "[]" ]]; then
      json_patch='[{"op":"remove","path":"/spec/backup/barmanObjectStore"}]'
    else
      json_patch=$(echo "$json_patch" | sed 's/]$/, {"op":"remove","path":"\/spec\/backup\/barmanObjectStore"}]/')
    fi
  fi

  if [[ "$json_patch" != "[]" ]]; then
    kubectl patch cluster "$CNPG_CLUSTER_NAME" -n "$NAMESPACE" --type json -p "$json_patch" >/dev/null 2>&1 || true
  fi
}

ensure_scheduled_backup() {
  local incremental="${SCHEDULED_BACKUP_INCREMENTAL,,}"
  echo "Ensuring ScheduledBackup '$SCHEDULED_BACKUP_NAME' ..."
  kubectl_apply_retry "$NAMESPACE" <<SCHEDULEDBACKUP
apiVersion: postgresql.cnpg.io/v1
kind: ScheduledBackup
metadata:
  name: $SCHEDULED_BACKUP_NAME
spec:
  schedule: "$SCHEDULED_BACKUP_CRON"
  immediate: $SCHEDULED_BACKUP_IMMEDIATE
  backupOwnerReference: self
  cluster:
    name: $CNPG_CLUSTER_NAME
  method: $SCHEDULED_BACKUP_METHOD
  pluginConfiguration:
    name: $BARMAN_PLUGIN_NAME
    parameters:
      barmanObjectName: $BARMAN_OBJECT_NAME
$(if [[ "$incremental" == "true" || "$incremental" == "1" || "$incremental" == "yes" ]]; then echo "      backupType: incremental"; fi)
SCHEDULEDBACKUP
}

wait_for_plugin_ready() {
  local start_time now plugin_names deployment_rows ns ready replicas
  local plugin_deploy_ready plugin_ns svc_ns svc_name client_secret server_secret plugin_port
  start_time=$(date +%s)
  while true; do
    plugin_deploy_ready=0
    plugin_ns=""
    plugin_names=$(kubectl -n "$NAMESPACE" get cluster "$CNPG_CLUSTER_NAME" -o jsonpath='{.spec.plugins[*].name}' 2>/dev/null || true)
    if printf '%s\n' "$plugin_names" | tr ' ' '\n' | grep -Fxq "$BARMAN_PLUGIN_NAME"; then
      deployment_rows=$(kubectl get deployment -A -l app.kubernetes.io/name=barman-cloud -o jsonpath='{range .items[*]}{.metadata.namespace}{"\t"}{.status.readyReplicas}{"\t"}{.status.replicas}{"\n"}{end}' 2>/dev/null || true)
      if [[ -z "$deployment_rows" ]]; then
        deployment_rows=$(kubectl get deployment -A -o jsonpath='{range .items[?(@.metadata.name=="barman-cloud")]}{.metadata.namespace}{"\t"}{.status.readyReplicas}{"\t"}{.status.replicas}{"\n"}{end}' 2>/dev/null || true)
      fi

      while IFS=$'\t' read -r ns ready replicas; do
        [[ -z "$ns" ]] && continue
        ready=${ready:-0}
        replicas=${replicas:-0}
        if (( ready >= 1 && replicas >= 1 )); then
          plugin_deploy_ready=1
          plugin_ns="$ns"
          break
        fi
      done <<< "$deployment_rows"

      if (( plugin_deploy_ready == 1 )); then
        # The Barman Cloud plugin is deployed as a standalone CNPG-I plugin.
        # In this model, there is no Unix socket expected inside the database pod.
        # Instead, CNPG discovers the plugin via a Kubernetes Service annotated with
        # cnpg.io/pluginClientSecret, cnpg.io/pluginServerSecret and cnpg.io/pluginPort.
        #
        # We validate that registration artifact exists and that the referenced TLS
        # secrets are present. The subsequent backup operation will fail if the plugin
        # is still not usable.

        svc_ns="${plugin_ns:-$CNPG_OPERATOR_NAMESPACE}"
        svc_name=$(kubectl -n "$svc_ns" get svc -l "cnpg.io/pluginName=$BARMAN_PLUGIN_NAME" -o jsonpath='{.items[0].metadata.name}' 2>/dev/null || true)
        if [[ -z "$svc_name" ]]; then
          # Fallback for older manifests that don't set the cnpg.io/pluginName label
          svc_name=$(kubectl -n "$svc_ns" get svc barman-cloud -o jsonpath='{.metadata.name}' 2>/dev/null || true)
        fi

        if [[ -n "$svc_name" ]]; then
          client_secret=$(kubectl -n "$svc_ns" get svc "$svc_name" -o jsonpath='{.metadata.annotations.cnpg\.io/pluginClientSecret}' 2>/dev/null || true)
          server_secret=$(kubectl -n "$svc_ns" get svc "$svc_name" -o jsonpath='{.metadata.annotations.cnpg\.io/pluginServerSecret}' 2>/dev/null || true)
          plugin_port=$(kubectl -n "$svc_ns" get svc "$svc_name" -o jsonpath='{.metadata.annotations.cnpg\.io/pluginPort}' 2>/dev/null || true)

          if [[ -n "$client_secret" && -n "$server_secret" && -n "$plugin_port" ]]; then
            if kubectl -n "$svc_ns" get secret "$client_secret" >/dev/null 2>&1 && kubectl -n "$svc_ns" get secret "$server_secret" >/dev/null 2>&1; then
              return 0
            fi
          fi
        fi
      fi
    fi

    now=$(date +%s)
    if (( now - start_time >= PLUGIN_READY_TIMEOUT )); then
      echo "ERROR: Timed out waiting for CNPG plugin '$BARMAN_PLUGIN_NAME' to become available." >&2
      return 1
    fi

    echo "Waiting for CNPG plugin '$BARMAN_PLUGIN_NAME' to become available ..."
    sleep 5
  done
}

wait_for_barman_plugin_infra_ready() {
  # Ensure the plugin controller and its registration artifacts (Service annotations + TLS secrets)
  # are present *before* patching the CNPG Cluster to depend on the plugin.
  local start_time now
  local deployment_rows ns ready replicas
  local plugin_deploy_ready plugin_ns svc_ns svc_name client_secret server_secret plugin_port
  start_time=$(date +%s)

  while true; do
    plugin_deploy_ready=0
    plugin_ns=""

    deployment_rows=$(kubectl get deployment -A -l app.kubernetes.io/name=barman-cloud -o jsonpath='{range .items[*]}{.metadata.namespace}{"\t"}{.status.readyReplicas}{"\t"}{.status.replicas}{"\n"}{end}' 2>/dev/null || true)
    if [[ -z "$deployment_rows" ]]; then
      deployment_rows=$(kubectl get deployment -A -o jsonpath='{range .items[?(@.metadata.name=="barman-cloud")]}{.metadata.namespace}{"\t"}{.status.readyReplicas}{"\t"}{.status.replicas}{"\n"}{end}' 2>/dev/null || true)
    fi

    while IFS=$'\t' read -r ns ready replicas; do
      [[ -z "$ns" ]] && continue
      ready=${ready:-0}
      replicas=${replicas:-0}
      if (( ready >= 1 && replicas >= 1 )); then
        plugin_deploy_ready=1
        plugin_ns="$ns"
        break
      fi
    done <<< "$deployment_rows"

    if (( plugin_deploy_ready == 1 )); then
      svc_ns="${plugin_ns:-$CNPG_OPERATOR_NAMESPACE}"
      svc_name=$(kubectl -n "$svc_ns" get svc -l "cnpg.io/pluginName=$BARMAN_PLUGIN_NAME" -o jsonpath='{.items[0].metadata.name}' 2>/dev/null || true)
      if [[ -z "$svc_name" ]]; then
        svc_name=$(kubectl -n "$svc_ns" get svc barman-cloud -o jsonpath='{.metadata.name}' 2>/dev/null || true)
      fi

      if [[ -n "$svc_name" ]]; then
        client_secret=$(kubectl -n "$svc_ns" get svc "$svc_name" -o jsonpath='{.metadata.annotations.cnpg\.io/pluginClientSecret}' 2>/dev/null || true)
        server_secret=$(kubectl -n "$svc_ns" get svc "$svc_name" -o jsonpath='{.metadata.annotations.cnpg\.io/pluginServerSecret}' 2>/dev/null || true)
        plugin_port=$(kubectl -n "$svc_ns" get svc "$svc_name" -o jsonpath='{.metadata.annotations.cnpg\.io/pluginPort}' 2>/dev/null || true)

        if [[ -n "$client_secret" && -n "$server_secret" && -n "$plugin_port" ]]; then
          if kubectl -n "$svc_ns" get secret "$client_secret" >/dev/null 2>&1 && kubectl -n "$svc_ns" get secret "$server_secret" >/dev/null 2>&1; then
            return 0
          fi
        fi
      fi
    fi

    now=$(date +%s)
    if (( now - start_time >= PLUGIN_READY_TIMEOUT )); then
      echo "ERROR: Timed out waiting for Barman plugin infrastructure to become ready." >&2
      return 1
    fi

    echo "Waiting for Barman plugin infrastructure (deployment + service registration) ..."
    sleep 5
  done
}

continuous_archiving_condition_line() {
  kubectl -n "$NAMESPACE" get cluster "$CNPG_CLUSTER_NAME" \
    -o jsonpath='{range .status.conditions[?(@.type=="ContinuousArchiving")]}{.status}{"\t"}{.reason}{"\t"}{.message}{"\n"}{end}' 2>/dev/null || true
}

wait_for_continuous_archiving_ready() {
  local timeout=${1:-$PLUGIN_READY_TIMEOUT}
  local start_time now elapsed
  local reconcile_count=0
  local reconcile_limit=${PLUGIN_BACKUP_RETRY_LIMIT:-6}

  start_time=$(date +%s)
  while true; do
    local ca_line ca_status ca_reason ca_message ca_status_lc ca_reason_lc ca_message_lc
    ca_line=$(continuous_archiving_condition_line)
    ca_status=""
    ca_reason=""
    ca_message=""

    if [[ -n "$ca_line" ]]; then
      IFS=$'\t' read -r ca_status ca_reason ca_message <<< "$ca_line"
    fi

    ca_status_lc=$(echo "${ca_status:-}" | tr '[:upper:]' '[:lower:]')
    ca_reason_lc=$(echo "${ca_reason:-}" | tr '[:upper:]' '[:lower:]')
    ca_message_lc=$(echo "${ca_message:-}" | tr '[:upper:]' '[:lower:]')

    if [[ "$ca_status_lc" == "true" ]]; then
      return 0
    fi

    if [[ "$ca_reason_lc" == *"continuousarchivingfailing"* || "$ca_message_lc" == *"wal archive plugin is not available"* || "$ca_message_lc" == *"plugin is not available"* ]]; then
      if (( reconcile_count < reconcile_limit )); then
        reconcile_count=$((reconcile_count + 1))
        echo "WARN: ContinuousArchiving reports plugin unavailable. Reconciling plugin config (${reconcile_count}/${reconcile_limit}) ..."
        ensure_barman_plugin_config || true
        wait_for_plugin_ready || true
      fi
    fi

    now=$(date +%s)
    elapsed=$((now - start_time))
    if (( elapsed >= timeout )); then
      echo "ERROR: Timed out waiting for ContinuousArchiving to become ready (status=${ca_status:-<unset>} reason=${ca_reason:-<unset>} message=${ca_message:-<unset>})." >&2
      return 1
    fi

    sleep 5
  done
}

wait_for_objectstore_ready() {
  local timeout=${1:-$OBJECTSTORE_READY_TIMEOUT}
  local start_time now elapsed
  start_time=$(date +%s)

  echo "Waiting for ObjectStore '$BARMAN_OBJECT_NAME' to be Ready (timeout: ${timeout}s)..."
  while true; do
    local ready_cond phase
    ready_cond=$(kubectl -n "$NAMESPACE" get objectstore "$BARMAN_OBJECT_NAME" \
      -o jsonpath='{range .status.conditions[?(@.type=="Ready")]}{.status}{end}' 2>/dev/null || true)
    phase=$(kubectl -n "$NAMESPACE" get objectstore "$BARMAN_OBJECT_NAME" \
      -o jsonpath='{.status.phase}' 2>/dev/null || true)

    # barman-cloud v0.11+ signals readiness via serverRecoveryWindow rather than
    # a Ready condition; accept either form.
    local recovery_window
    recovery_window=$(kubectl -n "$NAMESPACE" get objectstore "$BARMAN_OBJECT_NAME" \
      -o jsonpath='{.status.serverRecoveryWindow}' 2>/dev/null || true)
    if [[ "$ready_cond" == "True" || "$ready_cond" == "true" || "$phase" == "Ready" || "$phase" == "ready" || -n "$recovery_window" ]]; then
      echo "ObjectStore '$BARMAN_OBJECT_NAME' is Ready."
      return 0
    fi

    now=$(date +%s)
    elapsed=$((now - start_time))

    # Some plugin/controller versions do not currently set Ready/phase status on
    # ObjectStore. In that case, continue only when the ObjectStore exists,
    # barman-cloud controller is available, and the ObjectStore spec still matches
    # the expected Garage-backed configuration.
    local object_exists barman_available generation
    object_exists=$(kubectl -n "$NAMESPACE" get objectstore "$BARMAN_OBJECT_NAME" -o name 2>/dev/null || true)
    barman_available=$(kubectl -n "$CNPG_OPERATOR_NAMESPACE" get deploy barman-cloud -o jsonpath='{.status.availableReplicas}' 2>/dev/null || true)
    generation=$(kubectl -n "$NAMESPACE" get objectstore "$BARMAN_OBJECT_NAME" -o jsonpath='{.metadata.generation}' 2>/dev/null || true)
    if [[ -n "$object_exists" && "$barman_available" =~ ^[1-9][0-9]*$ && "$generation" =~ ^[0-9]+$ && $elapsed -ge 30 ]] && objectstore_spec_matches_expected; then
      echo "WARN: ObjectStore status fields are not populated, but barman-cloud is available and ObjectStore spec matches expected Garage configuration; continuing."
      return 0
    fi

    if (( elapsed >= timeout )); then
      echo "ERROR: Timed out waiting for ObjectStore '$BARMAN_OBJECT_NAME' to become Ready." >&2
      kubectl -n "$NAMESPACE" get objectstore "$BARMAN_OBJECT_NAME" -o yaml 2>/dev/null >&2 || true
      return 1
    fi

    sleep "$OBJECTSTORE_READY_INTERVAL"
  done
}

has_successful_base_backup() {
  local rows name phase method backup_type phase_lc method_lc backup_type_lc
  rows=$(kubectl get backup -n "$NAMESPACE" -o jsonpath="{range .items[?(@.spec.cluster.name=='$CNPG_CLUSTER_NAME')]}{.metadata.name}{\"\t\"}{.status.phase}{\"\t\"}{.spec.method}{\"\t\"}{.spec.pluginConfiguration.parameters.backupType}{\"\n\"}{end}" 2>/dev/null || true)

  if [[ -z "$rows" ]]; then
    return 1
  fi

  while IFS=$'\t' read -r name phase method backup_type; do
    [[ -z "$name" ]] && continue
    phase_lc=$(echo "${phase:-}" | tr '[:upper:]' '[:lower:]')
    method_lc=$(echo "${method:-}" | tr '[:upper:]' '[:lower:]')
    backup_type_lc=$(echo "${backup_type:-}" | tr '[:upper:]' '[:lower:]')

    if [[ "$phase_lc" != "completed" && "$phase_lc" != "succeeded" ]]; then
      continue
    fi

    if [[ "$method_lc" == "plugin" ]] && [[ "$backup_type_lc" == "incremental" || "$backup_type_lc" == "incr" ]]; then
      continue
    fi

    return 0
  done <<< "$rows"

  return 1
}

wait_for_successful_base_backup() {
  local start_time now elapsed
  local plugin_retry_count=0
  local plugin_retry_limit=${PLUGIN_BACKUP_RETRY_LIMIT:-6}
  local plugin_retry_interval=${PLUGIN_BACKUP_RETRY_INTERVAL:-20}
  start_time=$(date +%s)
  while true; do
    if [[ -n "$LAST_BACKUP_NAME" ]]; then
      local phase err
      phase=$(kubectl -n "$NAMESPACE" get backup "$LAST_BACKUP_NAME" -o jsonpath='{.status.phase}' 2>/dev/null || true)
      err=$(kubectl -n "$NAMESPACE" get backup "$LAST_BACKUP_NAME" -o jsonpath='{.status.error}' 2>/dev/null || true)
      case "${phase:-}" in
        Completed|Succeeded|completed|succeeded)
          return 0
          ;;
        Failed|failed)
          local err_lc
          err_lc=$(echo "${err:-}" | tr '[:upper:]' '[:lower:]')
          if [[ "$err_lc" == *"requested plugin is not available"* || "$err_lc" == *"wal archive plugin is not available"* || "$err_lc" == *"plugin is not available"* ]] && (( plugin_retry_count < plugin_retry_limit )); then
            plugin_retry_count=$((plugin_retry_count + 1))
            echo "WARN: Backup '$LAST_BACKUP_NAME' failed because plugin is not yet available. Retry ${plugin_retry_count}/${plugin_retry_limit} ..."
            wait_for_plugin_ready || true
            local retry_sleep=$((plugin_retry_interval * plugin_retry_count))
            if (( retry_sleep > 120 )); then
              retry_sleep=120
            fi
            if (( retry_sleep > 0 )); then
              echo "Waiting ${retry_sleep}s before retrying backup trigger ..."
              sleep "$retry_sleep"
            fi
            LAST_BACKUP_NAME=""
            trigger_backup
            sleep 5
            continue
          fi
          echo "ERROR: Backup '$LAST_BACKUP_NAME' failed: ${err:-<no error provided>}" >&2
          return 1
          ;;
      esac
    fi

    local ca_line ca_status ca_reason ca_message ca_status_lc ca_reason_lc ca_message_lc
    ca_line=$(continuous_archiving_condition_line)
    ca_status=""
    ca_reason=""
    ca_message=""
    if [[ -n "$ca_line" ]]; then
      IFS=$'\t' read -r ca_status ca_reason ca_message <<< "$ca_line"
    fi

    ca_status_lc=$(echo "${ca_status:-}" | tr '[:upper:]' '[:lower:]')
    ca_reason_lc=$(echo "${ca_reason:-}" | tr '[:upper:]' '[:lower:]')
    ca_message_lc=$(echo "${ca_message:-}" | tr '[:upper:]' '[:lower:]')
    if [[ "$ca_status_lc" != "true" && ( "$ca_reason_lc" == *"continuousarchivingfailing"* || "$ca_message_lc" == *"wal archive plugin is not available"* || "$ca_message_lc" == *"plugin is not available"* ) ]] && (( plugin_retry_count < plugin_retry_limit )); then
      plugin_retry_count=$((plugin_retry_count + 1))
      echo "WARN: ContinuousArchiving is failing due to plugin availability. Retry ${plugin_retry_count}/${plugin_retry_limit} ..."
      ensure_barman_plugin_config || true
      wait_for_plugin_ready || true
      wait_for_continuous_archiving_ready "$PLUGIN_READY_TIMEOUT" || true
      LAST_BACKUP_NAME=""
      trigger_backup
      sleep 5
      continue
    fi

    if has_successful_base_backup; then
      return 0
    fi

    now=$(date +%s)
    elapsed=$((now - start_time))
    if (( elapsed >= BACKUP_STATUS_TIMEOUT )); then
      return 1
    fi

    echo "Waiting for a successful base backup ... (${elapsed}s/${BACKUP_STATUS_TIMEOUT}s)"
    sleep "$BACKUP_STATUS_INTERVAL"
  done
}

trigger_backup() {
  local backup_type="${1:-full}"
  local backup_name
  backup_name="${CNPG_CLUSTER_NAME}-backup-$(date +%Y%m%d%H%M%S)"
  LAST_BACKUP_NAME="$backup_name"
  echo "Triggering ${backup_type} backup $backup_name ..."
  if [[ "$backup_type" == "incremental" || "$backup_type" == "incr" ]]; then
    kubectl_apply_retry "$NAMESPACE" <<BACKUP
apiVersion: postgresql.cnpg.io/v1
kind: Backup
metadata:
  name: $backup_name
spec:
  method: plugin
  pluginConfiguration:
    name: $BARMAN_PLUGIN_NAME
    parameters:
      barmanObjectName: $BARMAN_OBJECT_NAME
      backupType: incremental
  cluster:
    name: $CNPG_CLUSTER_NAME
BACKUP
  else
    kubectl_apply_retry "$NAMESPACE" <<BACKUP
apiVersion: postgresql.cnpg.io/v1
kind: Backup
metadata:
  name: $backup_name
spec:
  method: plugin
  pluginConfiguration:
    name: $BARMAN_PLUGIN_NAME
    parameters:
      barmanObjectName: $BARMAN_OBJECT_NAME
  cluster:
    name: $CNPG_CLUSTER_NAME
BACKUP
  fi
}

status() {
  ensure_tools
  select_kube_context
  echo "CNPG backup status for cluster '$CNPG_CLUSTER_NAME' (namespace: '$NAMESPACE')"

  local backup_spec
  backup_spec=$(kubectl -n "$NAMESPACE" get cluster "$CNPG_CLUSTER_NAME" -o jsonpath='{.spec.backup}' 2>/dev/null || true)
  if [[ -z "$backup_spec" || "$backup_spec" == "null" ]]; then
    echo "Cluster spec.backup: MISSING"
  else
    local rp plugin obj
    rp=$(kubectl -n "$NAMESPACE" get cluster "$CNPG_CLUSTER_NAME" -o jsonpath='{.spec.backup.retentionPolicy}' 2>/dev/null || true)
    plugin=$(kubectl -n "$NAMESPACE" get cluster "$CNPG_CLUSTER_NAME" -o jsonpath='{.spec.backup.pluginConfiguration.name}' 2>/dev/null || true)
    obj=$(kubectl -n "$NAMESPACE" get cluster "$CNPG_CLUSTER_NAME" -o jsonpath='{.spec.backup.pluginConfiguration.parameters.barmanObjectName}' 2>/dev/null || true)
    echo "Cluster spec.backup: present (retentionPolicy=${rp:-<unset>} plugin=${plugin:-<unset>} barmanObjectName=${obj:-<unset>})"
  fi

  local ca_line
  ca_line=$(continuous_archiving_condition_line)
  if [[ -n "$ca_line" ]]; then
    echo "ContinuousArchiving condition:"
    printf '%s' "$ca_line" | awk -F$'\t' '{print "- status=" $1 (length($2)?" reason=" $2:"") (length($3)?" message=" $3:"")}'
  else
    echo "ContinuousArchiving condition: <not reported>"
  fi

  echo "ScheduledBackup resources:"
  kubectl get scheduledbackup -n "$NAMESPACE" 2>/dev/null || true

  echo "Backup resources:"
  kubectl get backup -n "$NAMESPACE" 2>/dev/null | grep "$CNPG_CLUSTER_NAME" || true

  if has_successful_base_backup; then
    echo "Backup status: OK (at least one successful base backup is available)."
    return 0
  fi

  echo "ERROR: No successful base backup found for cluster '$CNPG_CLUSTER_NAME'." >&2
  return 1
}

case "$ACTION" in
  start)
    ensure_tools
    select_kube_context
    ensure_namespace
    wait_for_apiserver_ready 180
    ensure_cluster
    # Fast-path: skip re-initialization when continuous archiving is already healthy
    # and at least one successful base backup exists. All the wait_for_* loops below
    # are idempotent but slow — no need to re-run them on every installer pass.
    _ca_status_fp=$(kubectl -n "$NAMESPACE" get cluster "$CNPG_CLUSTER_NAME" \
      -o jsonpath='{range .status.conditions[?(@.type=="ContinuousArchiving")]}{.status}{end}' \
      2>/dev/null || true)
    if [[ "$_ca_status_fp" == "True" ]] && has_successful_base_backup; then
      echo "[CNPG BACKUP] Continuous archiving healthy and base backup exists — skipping re-initialization."
      exit 0
    fi
    resolve_garage_namespace_and_endpoint
    ensure_garage_bucket_and_key
    wait_for_barman_plugin_infra_ready
    ensure_barman_object_store_config
    wait_for_objectstore_ready
    ensure_cluster_backup_config
    ensure_barman_plugin_config
    wait_for_plugin_ready
    wait_for_continuous_archiving_ready
    ensure_scheduled_backup
    if [[ "$RUN_FIRST_BACKUP" == "1" ]]; then
      wait_for_cnpg_webhook 300
      trigger_backup
      if ! wait_for_successful_base_backup; then
        echo "ERROR: Timed out waiting for a successful base backup." >&2
        status || true
        exit 1
      fi
      status
    else
      status || true
    fi
    ;;
  backup)
    ensure_tools
    select_kube_context
    ensure_namespace
    wait_for_apiserver_ready 180
    ensure_cluster
    resolve_garage_namespace_and_endpoint
    wait_for_barman_plugin_infra_ready
    ensure_barman_object_store_config
    wait_for_objectstore_ready
    ensure_cluster_backup_config
    ensure_barman_plugin_config
    wait_for_plugin_ready
    wait_for_continuous_archiving_ready
    wait_for_cnpg_webhook 300
    trigger_backup "${2:-full}"
    ;;
  status)
    select_kube_context
    wait_for_apiserver_ready 60 || true
    status
    ;;
  *)
    usage
    ;;
esac