"""
knoe/core/ops/cloudnative_pg.py

Canonical Python owner for CloudNativePG lifecycle behavior.

Replaces etc/init_cloudnative_pg.sh runtime dispatch for the four active
call paths:

  initialize            -- install operator, apply cluster manifests, wait for readiness
  deploy                -- image-aware manifest apply + instance reconcile
  rollout               -- rolling restart via ordered pod deletion / promotion
  install_barman_plugin -- install the Barman Cloud operator plugin

Public API
----------
  ensure_operator(env, log, mode) -> None
  pin_controller(env, log, mode) -> None
  install_barman_plugin(env, log) -> None
  initialize(namespace, cluster_name, env, project_root, log, mode) -> None
  deploy(namespace, cluster_name, env, project_root, log, mode) -> None
  rollout(namespace, cluster_name, env, log) -> None
"""
from __future__ import annotations

import base64
import datetime
import os
import re
import subprocess
import tempfile
import time
from pathlib import Path

import json
import logging
from cryptography import x509
from cryptography.hazmat.primitives import hashes, serialization
from cryptography.hazmat.primitives.asymmetric.ec import (
    SECP256R1,
    generate_private_key,
)
from cryptography.hazmat.primitives.asymmetric.ed25519 import Ed25519PrivateKey
from cryptography.x509.oid import NameOID
from typing import Callable

_LOG = logging.getLogger(__name__)
_LogFn = Callable[[str], None]

_CNPG_OPERATOR_FALLBACK_VERSION = "1.28.1"
_BARMAN_PLUGIN_FALLBACK_VERSION = "0.6.0"


# ---------------------------------------------------------------------------
# Internal helpers
# ---------------------------------------------------------------------------


def _log(log: _LogFn | None, msg: str) -> None:
    if log is not None:
        log(msg)
    else:
        _LOG.info(msg)


def _kubectl(
    args: list[str],
    env: dict | None = None,
    timeout: int = 30,
    check: bool = False,
) -> subprocess.CompletedProcess:
    return subprocess.run(
        ["kubectl"] + args,
        env=env,
        capture_output=True,
        text=True,
        timeout=timeout,
        check=check,
    )


def _kubectl_ok(args: list[str], env: dict | None = None, timeout: int = 15) -> bool:
    return _kubectl(args, env=env, timeout=timeout).returncode == 0


def _release_stale_pv_claims(
    storage_class: str = "synology-iscsi",
    env: dict | None = None,
    log: _LogFn | None = None,
) -> None:
    """Clear claimRef on Released PVs so they become Available for new PVCs.

    Only touches PVs whose status.phase is 'Released' for the given storageClass.
    Idempotent — Available/Bound PVs are left untouched.
    """
    r = subprocess.run(
        ["kubectl", "get", "pv", "-o", "json"],
        text=True, capture_output=True, env=env, timeout=20,
    )
    if r.returncode != 0:
        _log(log, f"WARN: could not list PVs: {r.stderr.strip()}")
        return
    try:
        pv_list = json.loads(r.stdout or "{}")
    except Exception:
        return
    released = [
        item["metadata"]["name"]
        for item in (pv_list.get("items") or [])
        if (item.get("spec", {}).get("storageClassName") == storage_class
            and item.get("status", {}).get("phase") == "Released")
    ]
    if not released:
        return
    _log(log, f"Releasing claimRef on {len(released)} Released '{storage_class}' PV(s): {', '.join(released)}")
    for pv_name in released:
        pr = subprocess.run(
            ["kubectl", "patch", "pv", pv_name, "-p", '{"spec":{"claimRef":null}}'],
            text=True, capture_output=True, env=env, timeout=15,
        )
        if pr.returncode != 0:
            _log(log, f"WARN: could not release claimRef on PV '{pv_name}': {pr.stderr.strip()}")
        else:
            _log(log, f"Released PV '{pv_name}'.")


def _reconcile_unbound_synology_data_wal_pvs(
    env: dict | None = None,
    log: _LogFn | None = None,
) -> None:
    """Recreate unbound static Synology data/wal PVs from manifest.

    This allows nodeAffinity/layout changes in `iscsi-pvs.yaml` to take effect,
    while preserving currently bound volumes.
    """
    pv_names = [
        "synology-iscsi-d001-data",
        "synology-iscsi-d001-wal",
        "synology-iscsi-d002-data",
        "synology-iscsi-d002-wal",
        "synology-iscsi-d003-data",
        "synology-iscsi-d003-wal",
    ]

    unbound: list[str] = []
    for name in pv_names:
        r = _kubectl(["get", "pv", name, "-o", "json"], env=env, timeout=20)
        if r.returncode != 0:
            continue
        try:
            payload = json.loads(r.stdout or "{}")
        except Exception:
            continue
        phase = str((payload.get("status") or {}).get("phase") or "").strip()
        if phase != "Bound":
            unbound.append(name)

    if not unbound:
        return

    _log(log, f"Reconciling unbound Synology data/wal PVs: {', '.join(unbound)}")
    for name in unbound:
        _kubectl(["delete", "pv", name, "--ignore-not-found"], env=env, timeout=45)

    root = Path(
        (env or {}).get("PROLE_HOME")
        or (env or {}).get("PROLE_SERVICE")
        or Path(__file__).resolve().parents[4]
    )
    manifest = root / "k8s" / "prole" / "iscsi-pvs.yaml"
    if not manifest.exists():
        _log(log, f"WARN: Synology PV manifest not found: {manifest}")
        return

    r_apply = _kubectl(["apply", "-f", str(manifest)], env=env, timeout=180)
    if r_apply.returncode != 0:
        stderr = ((r_apply.stderr or "") + "\n" + (r_apply.stdout or "")).strip().lower()
        if "field is immutable" not in stderr:
            raise RuntimeError(
                f"Failed to re-apply Synology PV manifest {manifest}: "
                f"{(r_apply.stderr or r_apply.stdout or '').strip()}"
            )
        _log(log, "WARN: Immutable PV fields prevented full re-apply; continuing with recreated unbound PVs.")


def _ensure_namespace(namespace: str, env: dict | None = None) -> None:
    """Create the namespace if it does not already exist (idempotent)."""
    manifest = (
        "apiVersion: v1\n"
        "kind: Namespace\n"
        "metadata:\n"
        f"  name: {namespace}\n"
    )
    res = subprocess.run(
        ["kubectl", "apply", "-f", "-"],
        input=manifest,
        text=True,
        capture_output=True,
        env=env,
    )
    if res.returncode != 0:
        raise RuntimeError(f"Failed to ensure namespace '{namespace}':\n{res.stderr}")


def _resolve_cnpg_manifest(project_root: str | Path, env: dict | None) -> Path:
    """Resolve the CNPG cluster manifest path, mirroring the shell K8S_PROLE_DIR logic."""
    if env:
        override = (env.get("CNPG_MANIFEST_OVERRIDE") or "").strip()
        if override:
            p = Path(override)
            if p.exists():
                return p
    root = Path(project_root)
    mode = str((env or {}).get("PROLE_MODE") or "").strip()
    candidates: list[Path] = []
    if mode == "k8s":
        # GKE/prod: use the GCP-specific manifest (premium-rwo storage, no Synology selectors)
        candidates.append(root / "deploy" / "gcp" / "gke" / "knoe-db.yaml")
    candidates += [
        root / "deploy" / "opentofu" / "k3s" / "manifests" / "prole" / "knoe-db.yaml",
        root / "k8s" / "prole" / "knoe-db.yaml",
    ]
    if env:
        prole_home = (env.get("PROLE_HOME") or "").strip()
        if prole_home:
            ph = Path(prole_home)
            if mode == "k8s":
                candidates.insert(0, ph / "deploy" / "gcp" / "gke" / "knoe-db.yaml")
            candidates.append(ph / "k8s" / "prole" / "knoe-db.yaml")
    for candidate in candidates:
        if candidate.exists():
            return candidate
    return root / "k8s" / "prole" / "knoe-db.yaml"


def _resolve_knoe_db_image_tag(manifest: Path, env: dict | None) -> str:
    explicit_tag = str((env or {}).get("KNOE_DB_IMAGE_TAG") or "").strip()
    if explicit_tag:
        return explicit_tag
    explicit_tag = os.environ.get("KNOE_DB_IMAGE_TAG", "").strip()
    if explicit_tag:
        return explicit_tag

    mode_key = str((env or {}).get("PROLE_MODE") or "").strip() or "k8s"

    candidate_roots: list[Path] = []
    prole_home = str((env or {}).get("PROLE_HOME") or "").strip()
    if prole_home:
        candidate_roots.append(Path(prole_home))

    for parent in [manifest, *manifest.parents]:
        if parent in candidate_roots:
            continue
        if (parent / "conf").exists() or (parent / "knoe-db").exists() or (parent / "modes").exists():
            candidate_roots.append(parent)

    def _read(path: Path) -> str | None:
        try:
            val = path.read_text(encoding="utf-8").strip()
            return val or None
        except OSError:
            return None

    pg_version = "17.7"
    release = "43"
    for root in candidate_roots:
        mode_pg = root / "modes" / mode_key / "conf" / "postgresql" / ".version"
        mode_release = root / "modes" / mode_key / "knoe-db" / ".version"
        root_pg = root / "conf" / "postgresql" / ".version"
        root_release = root / "knoe-db" / ".version"

        if mode_pg.exists() or mode_release.exists() or root_pg.exists() or root_release.exists():
            pg_version = _read(mode_pg) or _read(root_pg) or pg_version
            release = _read(mode_release) or _read(root_release) or release
            break

    if release.isdigit():
        release = release.zfill(3)
    return f"{pg_version}-{release}"


def _apply_manifest(
    namespace: str,
    manifest: Path,
    env: dict | None,
    log: _LogFn | None,
    attempts: int = 8,
) -> None:
    """Apply a CNPG cluster manifest with retry on transient API errors."""
    if not manifest.exists():
        raise FileNotFoundError(f"CNPG manifest not found: {manifest}")

    raw_manifest = manifest.read_text(encoding="utf-8")
    rendered_manifest = raw_manifest
    artifact_registry = ""
    if "${ARTIFACT_REGISTRY}" in rendered_manifest:
        artifact_registry = str((env or {}).get("ARTIFACT_REGISTRY") or "").strip()
        if not artifact_registry:
            artifact_registry = os.environ.get("ARTIFACT_REGISTRY", "").strip()
        if not artifact_registry:
            raise RuntimeError(
                "CNPG manifest contains '${ARTIFACT_REGISTRY}' but ARTIFACT_REGISTRY is not set."
            )
        rendered_manifest = rendered_manifest.replace("${ARTIFACT_REGISTRY}", artifact_registry.rstrip("/"))

    if "${KNOE_DB_IMAGE_TAG}" in rendered_manifest:
        knoe_db_image_tag = _resolve_knoe_db_image_tag(manifest, env)
        rendered_manifest = rendered_manifest.replace("${KNOE_DB_IMAGE_TAG}", knoe_db_image_tag)

    unresolved_image_placeholder = re.search(
        r"^\s*imageName:\s*[\"']?\$\{[^}]+}[^\s\"']*",
        rendered_manifest,
        flags=re.MULTILINE,
    )
    if unresolved_image_placeholder:
        raise RuntimeError(
            "CNPG manifest contains unresolved image placeholder in imageName. "
            "Ensure deployment variables are rendered before apply."
        )

    manifest_path = manifest
    temp_manifest_path: Path | None = None
    if rendered_manifest != raw_manifest:
        with tempfile.NamedTemporaryFile(mode="w", encoding="utf-8", suffix=".yaml", delete=False) as tmp:
            tmp.write(rendered_manifest)
            temp_manifest_path = Path(tmp.name)
        manifest_path = temp_manifest_path

    _log(log, f"Applying CNPG manifest {manifest} in namespace {namespace}...")
    last_err = ""
    try:
        for i in range(1, attempts + 1):
            r = _kubectl(["-n", namespace, "apply", "-f", str(manifest_path)], env=env, timeout=60)
            if r.returncode == 0:
                if r.stdout.strip():
                    _log(log, r.stdout.strip())
                return
            last_err = (r.stderr or r.stdout or "").strip()
            transient = any(
                kw in last_err.lower()
                for kw in ("etcdserver", "connection refused", "timeout", "tls handshake", "webhook")
            )
            if transient and i < attempts:
                _log(log, f"WARN: manifest apply failed (attempt {i}/{attempts}); retrying in 5s...")
                time.sleep(5)
                continue
            break
        raise RuntimeError(f"CNPG manifest apply failed after {attempts} attempts: {last_err}")
    finally:
        if temp_manifest_path is not None:
            try:
                temp_manifest_path.unlink()
            except OSError:
                pass


def _remove_legacy_barman_object_store_when_plugin_is_enabled(
    namespace: str,
    cluster_name: str,
    env: dict | None,
    log: _LogFn | None,
) -> None:
    """Avoid CNPG validation conflict when plugin mode coexists with legacy barmanObjectStore."""
    exists = _kubectl(["-n", namespace, "get", "cluster", cluster_name, "-o", "name"], env=env, timeout=20)
    if exists.returncode != 0:
        return

    plugins = _kubectl(
        ["-n", namespace, "get", "cluster", cluster_name, "-o", "jsonpath={range .spec.plugins[*]}{.name}{\n}{end}"],
        env=env,
        timeout=20,
    )
    if plugins.returncode != 0:
        return
    plugin_names = {line.strip() for line in (plugins.stdout or "").splitlines() if line.strip()}
    plugin_enabled = "barman-cloud.cloudnative-pg.io" in plugin_names
    if not plugin_enabled:
        return

    destination = _kubectl(
        ["-n", namespace, "get", "cluster", cluster_name, "-o", "jsonpath={.spec.backup.barmanObjectStore.destinationPath}"],
        env=env,
        timeout=20,
    )
    if destination.returncode != 0 or not (destination.stdout or "").strip():
        return

    _log(log, "Removing legacy spec.backup.barmanObjectStore to keep plugin WAL archiver valid...")
    patch = json.dumps({"spec": {"backup": {"barmanObjectStore": None}}})
    patched = _kubectl(
        ["-n", namespace, "patch", "cluster", cluster_name, "--type", "merge", "-p", patch],
        env=env,
        timeout=30,
    )
    if patched.returncode != 0:
        reason = (patched.stderr or patched.stdout or "").strip()
        raise RuntimeError(
            f"Failed to clear legacy barmanObjectStore on cluster '{cluster_name}' in namespace '{namespace}': {reason}"
        )


def _wait_pod_ready(
    namespace: str, pod: str, env: dict | None, timeout: int = 300, log: _LogFn | None = None
) -> None:
    r = _kubectl(
        ["-n", namespace, "wait", "--for=condition=Ready", f"pod/{pod}", f"--timeout={timeout}s"],
        env=env,
        timeout=timeout + 10,
    )
    if r.returncode != 0:
        _log(log, f"WARN: pod {pod} did not become Ready within {timeout}s")


def _wait_cnpg_pods(
    namespace: str,
    cluster_name: str,
    env: dict | None,
    timeout: int = 900,
    log: _LogFn | None = None,
) -> None:
    """Wait for all CNPG cluster pods to reach Ready state."""
    # Determine expected pod count from cluster spec
    r = _kubectl(
        ["-n", namespace, "get", "cluster", cluster_name, "-o", "jsonpath={.spec.instances}"],
        env=env,
        timeout=15,
    )
    target = 1
    if r.returncode == 0:
        try:
            target = max(1, int((r.stdout or "1").strip() or "1"))
        except ValueError:
            target = 1

    _log(log, f"Waiting for {target} CNPG pod(s) for cluster '{cluster_name}' in '{namespace}' (timeout={timeout}s)...")

    deadline = time.monotonic() + timeout
    last_feedback = 0.0
    while True:
        now = time.monotonic()
        if now >= deadline:
            raise RuntimeError(
                f"Timed out ({timeout}s) waiting for {target} CNPG pods to be Ready "
                f"in namespace '{namespace}'"
            )

        r2 = _kubectl(
            [
                "-n", namespace, "get", "pods",
                "-l", f"cnpg.io/cluster={cluster_name}",
                "-o",
                "jsonpath={range .items[*]}{.metadata.name}={.status.conditions[?(@.type==\"Ready\")].status}\\n{end}",
            ],
            env=env,
            timeout=15,
        )
        if r2.returncode == 0:
            # kubectl jsonpath emits literal \n (backslash-n) not real newlines
            raw = (r2.stdout or "").replace("\\n", "\n")
            lines = [l for l in raw.splitlines() if "=" in l]
            ready = sum(1 for l in lines if l.split("=", 1)[1].strip() == "True")
            if ready >= target:
                _log(log, f"All {ready}/{target} CNPG pods are Ready.")
                return

            elapsed = timeout - (deadline - now)
            if elapsed - last_feedback >= 30:
                last_feedback = elapsed
                _log(log, f"  [{int(elapsed)}s/{timeout}s] {ready}/{target} CNPG pods ready...")

        time.sleep(5)


def _reconcile_instances(
    namespace: str,
    cluster_name: str,
    env: dict | None,
    log: _LogFn | None,
) -> None:
    """
    Cap CNPG cluster instances to the number of eligible schedulable nodes,
    then patch the cluster spec if the count has changed.

    Mirrors reconcile_cnpg_instances() from the shell script.
    """
    # Current spec instances
    r = _kubectl(
        ["-n", namespace, "get", "cluster", cluster_name, "-o", "jsonpath={.spec.instances}"],
        env=env, timeout=15,
    )
    if r.returncode != 0 or not (r.stdout or "").strip():
        return  # cluster may not exist yet
    try:
        current = max(0, int((r.stdout or "0").strip()))
    except ValueError:
        return

    desired_raw = int((env or {}).get("CNPG_INSTANCES", "3") or "3")
    desired = max(1, desired_raw)

    # Count eligible nodes (nodes with cnpg.io/node-role=db or all schedulable nodes)
    node_selector = (env or {}).get("CNPG_NODE_SELECTOR", "")
    if node_selector:
        r2 = _kubectl(["get", "nodes", "-l", node_selector, "--no-headers"], env=env, timeout=10)
    else:
        r2 = _kubectl(["get", "nodes", "--no-headers"], env=env, timeout=10)
    eligible = len([l for l in (r2.stdout or "").splitlines() if l.strip()]) if r2.returncode == 0 else 1
    eligible = max(1, eligible)

    if desired > eligible:
        desired = eligible

    if current == desired:
        return

    # Clamp maxSyncReplicas so the spec stays valid (maxSyncReplicas < instances)
    base_max_sync = int((env or {}).get("CNPG_MAX_SYNC_REPLICAS", "1") or "1")
    if desired <= 1:
        max_sync = 0
    else:
        max_sync = min(base_max_sync, desired - 1)

    _log(log, f"Reconciling CNPG instances: {current} → {desired} (eligible_nodes={eligible})")
    patch = json.dumps({"spec": {"instances": desired, "maxSyncReplicas": max_sync}})
    for attempt in range(1, 9):
        r3 = _kubectl(
            ["-n", namespace, "patch", "cluster", cluster_name, "--type", "merge", "-p", patch],
            env=env, timeout=30,
        )
        if r3.returncode == 0:
            _log(log, f"Cluster instances patched to {desired}.")
            return
        err = (r3.stderr or r3.stdout or "").strip().lower()
        transient = any(kw in err for kw in ("etcd", "connection refused", "timeout", "webhook"))
        if transient and attempt < 8:
            _log(log, f"WARN: patch failed (attempt {attempt}/8); retrying in 5s...")
            time.sleep(5)
        else:
            _log(log, f"WARN: could not patch CNPG instances: {err}")
            return


def _controller_manifest_url(version: str) -> str:
    minor = ".".join(version.split(".")[:2])
    return (
        f"https://raw.githubusercontent.com/cloudnative-pg/cloudnative-pg"
        f"/release-{minor}/releases/cnpg-{version}.yaml"
    )


def _controller_deployment_json(env: dict | None = None) -> dict | None:
    r = _kubectl(
        ["-n", "cnpg-system", "get", "deploy", "cnpg-controller-manager", "-o", "json"],
        env=env,
        timeout=30,
    )
    if r.returncode != 0 or not (r.stdout or "").strip():
        return None
    try:
        return json.loads(r.stdout)
    except Exception:
        return None


def _controller_is_healthy(env: dict | None = None) -> bool:
    dep = _controller_deployment_json(env)
    if not dep:
        return False
    spec = dep.get("spec") or {}
    status = dep.get("status") or {}
    desired = int(spec.get("replicas") or 1)
    available = int(status.get("availableReplicas") or 0)
    return desired > 0 and available >= desired


def _reconcile_controller_replicasets(
    env: dict | None = None,
    log: _LogFn | None = None,
    aggressive_cleanup: bool = False,
) -> None:
    dep = _controller_deployment_json(env)
    if not dep:
        return

    dep_meta = dep.get("metadata") or {}
    dep_uid = str(dep_meta.get("uid") or "").strip()
    dep_annotations = dep_meta.get("annotations") or {}
    try:
        dep_revision = int(dep_annotations.get("deployment.kubernetes.io/revision") or "0")
    except Exception:
        dep_revision = 0

    r = _kubectl(
        [
            "-n",
            "cnpg-system",
            "get",
            "rs",
            "-l",
            "app.kubernetes.io/name=cloudnative-pg",
            "-o",
            "json",
        ],
        env=env,
        timeout=45,
    )
    if r.returncode != 0 or not (r.stdout or "").strip():
        return
    try:
        payload = json.loads(r.stdout)
    except Exception:
        return

    def _rs_revision(item: dict) -> int:
        ann = (item.get("metadata") or {}).get("annotations") or {}
        try:
            return int(ann.get("deployment.kubernetes.io/revision") or "0")
        except Exception:
            return 0

    owned: list[dict] = []
    for item in payload.get("items") or []:
        meta = item.get("metadata") or {}
        owners = meta.get("ownerReferences") or []
        for owner in owners:
            if str(owner.get("kind") or "") != "Deployment":
                continue
            if str(owner.get("name") or "") != "cnpg-controller-manager":
                continue
            owner_uid = str(owner.get("uid") or "").strip()
            if dep_uid and owner_uid and owner_uid != dep_uid:
                continue
            owned.append(item)
            break

    active = [i for i in owned if int((i.get("spec") or {}).get("replicas") or 0) > 0]
    if len(active) <= 1:
        return

    stale: list[dict]
    if dep_revision > 0:
        stale = [i for i in active if _rs_revision(i) < dep_revision]
    else:
        active_sorted = sorted(
            active,
            key=lambda i: (
                _rs_revision(i),
                str((i.get("metadata") or {}).get("creationTimestamp") or ""),
            ),
        )
        stale = active_sorted[:-1]

    for rs in stale:
        rs_meta = rs.get("metadata") or {}
        rs_name = str(rs_meta.get("name") or "").strip()
        if not rs_name:
            continue
        _log(log, f"Scaling stale CNPG controller replicaset/{rs_name} to 0")
        _kubectl(
            ["-n", "cnpg-system", "scale", f"rs/{rs_name}", "--replicas=0"],
            env=env,
            timeout=60,
        )

        if aggressive_cleanup:
            template_hash = str((rs_meta.get("labels") or {}).get("pod-template-hash") or "").strip()
            if template_hash:
                _kubectl(
                    [
                        "-n",
                        "cnpg-system",
                        "delete",
                        "pod",
                        "-l",
                        f"app.kubernetes.io/name=cloudnative-pg,pod-template-hash={template_hash}",
                        "--ignore-not-found",
                        "--wait=false",
                    ],
                    env=env,
                    timeout=60,
                )


def _wait_controller_rollout(
    env: dict | None = None,
    timeout_seconds: int = 180,
) -> bool:
    r = _kubectl(
        [
            "-n",
            "cnpg-system",
            "rollout",
            "status",
            "deploy/cnpg-controller-manager",
            f"--timeout={timeout_seconds}s",
        ],
        env=env,
        timeout=timeout_seconds + 10,
    )
    return r.returncode == 0


# ---------------------------------------------------------------------------
# Public API
# ---------------------------------------------------------------------------


def ensure_operator(
    env: dict | None = None,
    log: _LogFn | None = None,
    mode: str = "",
    allow_reapply_if_unhealthy: bool = False,
    aggressive_cleanup: bool = False,
) -> None:
    """Install the CloudNativePG operator if not already present, then wait for readiness."""
    if _kubectl_ok(["-n", "cnpg-system", "get", "deploy", "cnpg-controller-manager"], env=env):
        _reconcile_controller_replicasets(env=env, log=log, aggressive_cleanup=aggressive_cleanup)
        if _controller_is_healthy(env=env):
            _log(log, "CloudNative-PG operator already healthy; skipping operator re-apply.")
            if mode == "k3s":
                _tune_operator_for_k3s(env, log)
                _wait_controller_rollout(env=env, timeout_seconds=300)
            return

        _log(log, "CloudNative-PG operator exists but is not healthy; waiting for recovery...")
        if _wait_controller_rollout(env=env, timeout_seconds=180):
            if mode == "k3s":
                _tune_operator_for_k3s(env, log)
                _wait_controller_rollout(env=env, timeout_seconds=300)
            return

        if not allow_reapply_if_unhealthy:
            raise RuntimeError(
                "CNPG operator deployment exists but is not ready; "
                "refusing disruptive re-apply in non-initialize flow."
            )
        _log(log, "CNPG operator unhealthy after wait; attempting controlled re-apply...")

    version = ""
    if env:
        version = (env.get("CNPG_OPERATOR_VERSION") or env.get("CNPG_VERSION") or "").strip()
    if not version or version == "latest":
        version = _CNPG_OPERATOR_FALLBACK_VERSION
    url = _controller_manifest_url(version)
    _log(log, f"Installing CloudNative-PG operator version {version}...")
    r = _kubectl(["apply", "--server-side", "-f", url], env=env, timeout=120)
    if r.returncode != 0:
        raise RuntimeError(f"Failed to install CNPG operator: {(r.stderr or r.stdout or '').strip()}")

    if _kubectl_ok(["-n", "cnpg-system", "get", "deploy", "cnpg-controller-manager"], env=env, timeout=20):
        _reconcile_controller_replicasets(env=env, log=log, aggressive_cleanup=aggressive_cleanup)
        if not _wait_controller_rollout(env=env, timeout_seconds=180):
            raise RuntimeError("CNPG controller rollout did not complete after operator apply")
        if mode == "k3s":
            _tune_operator_for_k3s(env, log)
            if not _wait_controller_rollout(env=env, timeout_seconds=300):
                raise RuntimeError("CNPG controller rollout did not complete after k3s tuning")


def _tune_operator_for_k3s(env: dict | None, log: _LogFn | None) -> None:
    """Relax CNPG controller probes/resources for single-node k3s environments."""
    _log(log, "Tuning CNPG operator deployment probes/resources for k3s...")
    patch = json.dumps({
        "spec": {
            "template": {
                "spec": {
                    "tolerations": [
                        {
                            "key": "node-role.kubernetes.io/control-plane",
                            "operator": "Exists",
                            "effect": "NoSchedule",
                        },
                        {
                            "key": "node-role.kubernetes.io/master",
                            "operator": "Exists",
                            "effect": "NoSchedule",
                        },
                    ],
                    "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},
                        }
                    ]
                }
            }
        }
    })
    _kubectl(
        ["-n", "cnpg-system", "patch", "deploy", "cnpg-controller-manager", "--type", "merge", "-p", patch],
        env=env, timeout=30,
    )


def _clear_controller_pin(env: dict | None = None, log: _LogFn | None = None) -> None:
    """Remove stale hard nodeSelector pinning for controller-manager."""
    _log(log, "Clearing CNPG controller hard nodeSelector pin (if present)...")
    patch = json.dumps({"spec": {"template": {"spec": {"nodeSelector": None}}}})
    _kubectl(
        ["-n", "cnpg-system", "patch", "deployment", "cnpg-controller-manager", "--type", "merge", "-p", patch],
        env=env,
        timeout=30,
    )


def pin_controller(
    env: dict | None = None,
    log: _LogFn | None = None,
    mode: str = "",
) -> None:
    """Pin the CNPG controller manager to storage-designated nodes (k3s) or as configured."""
    selector = (env or {}).get("CNPG_CONTROLLER_NODE_SELECTOR", "").strip()

    if not selector and mode == "k3s":
        r = _kubectl(["get", "nodes", "-l", "storage=primary", "--no-headers"], env=env, timeout=10)
        if r.returncode == 0 and (r.stdout or "").strip():
            selector = "storage=primary"

    if not selector:
        _clear_controller_pin(env=env, log=log)
        _reconcile_controller_replicasets(env=env, log=log)
        return

    if "=" not in selector:
        _log(log, f"WARN: CNPG_CONTROLLER_NODE_SELECTOR must be key=value (got '{selector}'). Skipping pin.")
        return

    key, value = selector.split("=", 1)
    if not _kubectl_ok(["get", "nodes", "-l", f"{key}={value}", "--no-headers"], env=env):
        _log(log, f"WARN: No nodes match CNPG_CONTROLLER_NODE_SELECTOR={selector}. Skipping pin.")
        return

    _log(log, f"Pinning cnpg-controller-manager to nodes with {selector}...")
    patch = json.dumps({"spec": {"template": {"spec": {"nodeSelector": {key: value}}}}})
    _kubectl(
        ["-n", "cnpg-system", "patch", "deployment", "cnpg-controller-manager",
         "--type", "merge", "-p", patch],
        env=env, timeout=30,
    )
    _reconcile_controller_replicasets(env=env, log=log)


def install_barman_plugin(
    env: dict | None = None,
    log: _LogFn | None = None,
) -> None:
    """Install the Barman Cloud operator plugin, retrying on cert-manager webhook errors."""
    url = ((env or {}).get("BARMAN_PLUGIN_MANIFEST_URL") or "").strip()
    if not url:
        tag = _resolve_barman_plugin_tag(env, log)
        url = f"https://github.com/cloudnative-pg/plugin-barman-cloud/releases/download/{tag}/manifest.yaml"

    _log(log, f"Installing Barman Cloud plugin from {url}...")
    _CERT_WEBHOOK_PATTERNS = (
        "webhook.cert-manager.io",
        "cert-manager",
        "x509",
        "certificate signed by unknown authority",
        "tls:",
    )
    _cert_manager_recovered = False
    _last_error: str = ""
    _apply_ok = False
    for _attempt in range(1, 7):  # up to 6 attempts; first triggers cert-manager recovery
        r = subprocess.run(
            ["kubectl", "apply", "-f", url],
            env=env, capture_output=True, text=True, timeout=120,
        )
        if r.returncode == 0:
            if r.stdout.strip():
                _log(log, r.stdout.strip())
            _apply_ok = True
            break
        combined = ((r.stdout or "") + (r.stderr or "")).lower()
        _last_error = (r.stderr or r.stdout or "").strip()
        is_cert_issue = any(pat in combined for pat in _CERT_WEBHOOK_PATTERNS)
        if is_cert_issue:
            if not _cert_manager_recovered:
                _log(log, "WARN: cert-manager webhook error; ensuring cert-manager and retrying...")
                ca_ready = _ensure_cert_manager_for_barman(env=env, log=log)
                _cert_manager_recovered = True
                if ca_ready:
                    _log(log, "cert-manager recovered; webhook CA bundle ready.")
                else:
                    _log(log, "WARN: cert-manager restarted but CA bundle not confirmed; proceeding with retries...")
            else:
                wait_s = min(20 * _attempt, 60)
                _log(log, f"Webhook CA not yet trusted (attempt {_attempt}); retrying in {wait_s} s...")
                time.sleep(wait_s)
        else:
            raise RuntimeError(
                f"Failed to apply Barman Cloud plugin manifest: {_last_error}"
            )
    if not _apply_ok:
        raise RuntimeError(
            f"Failed to apply Barman Cloud plugin after cert-manager restart: {_last_error}"
        )

    # Wait for barman-cloud deployment if it was created
    r3 = _kubectl(["-n", "cnpg-system", "get", "deploy", "barman-cloud"], env=env, timeout=15)
    if r3.returncode == 0:
        timeout_s = int((env or {}).get("BARMAN_DEPLOY_TIMEOUT", "300") or "300")
        _kubectl(
            ["-n", "cnpg-system", "rollout", "status", "deploy/barman-cloud", f"--timeout={timeout_s}s"],
            env=env, timeout=timeout_s + 10,
        )

    # Wait for cert-manager to issue the Barman TLS secrets that the pod mounts
    tls_timeout = int((env or {}).get("BARMAN_TLS_TIMEOUT", "300") or "300")
    if not _wait_for_barman_tls_secrets(env=env, log=log, timeout=tls_timeout):
        _log(log, f"WARN: Barman Cloud TLS secrets not ready after {tls_timeout}s; bootstrapping self-signed fallback...")
        _bootstrap_barman_tls_secrets(env=env, log=log)


def _ensure_cert_manager_for_barman(
    env: dict | None = None,
    log: _LogFn | None = None,
) -> None:
    def _exists(kind: str, name: str) -> bool:
        return _kubectl(["-n", "cert-manager", "get", kind, name], env=env, timeout=20).returncode == 0

    need_bootstrap = not all(
        [
            _exists("service", "cert-manager-webhook"),
            _exists("deployment", "cert-manager"),
            _exists("deployment", "cert-manager-webhook"),
            _exists("deployment", "cert-manager-cainjector"),
        ]
    )

    if need_bootstrap:
        _log(log, "[CNPG] cert-manager resources missing; bootstrapping cert-manager...")
        root = Path(
            (env or {}).get("PROLE_HOME")
            or (env or {}).get("PROLE_SERVICE")
            or Path(__file__).resolve().parents[4]
        )
        init_script = root / "etc" / "init_certmgr.sh"
        if init_script.exists():
            cmd = ["bash", str(init_script)]
            mode = str((env or {}).get("PROLE_MODE") or "").strip()
            if mode:
                cmd.extend(["--mode", mode])
            cfg_hint = (
                str((env or {}).get("PROLE_CONFIG_PATH") or "").strip()
                or str((env or {}).get("PROLE_CFG_PATH") or "").strip()
                or str((env or {}).get("PROLE_CFG") or "").strip()
            )
            if cfg_hint:
                cmd.extend(["-c", cfg_hint])
            cmd.append("start")
            r = subprocess.run(
                cmd,
                env=env,
                capture_output=True,
                text=True,
                timeout=600,
            )
            if r.returncode != 0:
                raise RuntimeError(
                    "Failed to bootstrap cert-manager via init_certmgr.sh: "
                    f"{(r.stderr or r.stdout or '').strip()}"
                )
        else:
            _kubectl(["get", "namespace", "cert-manager"], env=env, timeout=20)
            _kubectl(["create", "namespace", "cert-manager"], env=env, timeout=20)
            certmgr_version = str((env or {}).get("CERTMGR_VERSION") or "v1.14.6").strip()
            certmgr_url = (
                "https://github.com/cert-manager/cert-manager/releases/download/"
                f"{certmgr_version}/cert-manager.yaml"
            )
            r = subprocess.run(
                ["kubectl", "apply", "-f", certmgr_url],
                env=env,
                capture_output=True,
                text=True,
                timeout=180,
            )
            if r.returncode != 0:
                raise RuntimeError(
                    f"Failed to bootstrap cert-manager from {certmgr_url}: "
                    f"{(r.stderr or r.stdout or '').strip()}"
                )

    for deploy in ["cert-manager", "cert-manager-webhook", "cert-manager-cainjector"]:
        if _exists("deployment", deploy):
            _kubectl(["-n", "cert-manager", "rollout", "restart", f"deploy/{deploy}"], env=env, timeout=45)
    for deploy in ["cert-manager-webhook", "cert-manager", "cert-manager-cainjector"]:
        if _exists("deployment", deploy):
            _kubectl(
                ["-n", "cert-manager", "rollout", "status", f"deploy/{deploy}", "--timeout=240s"],
                env=env,
                timeout=250,
            )

    _log(log, "Waiting for cert-manager webhook CA bundle to be injected...")
    if _wait_for_webhook_ca_bundle(env=env, log=log, timeout=300):
        return True
    _log(log, "WARN: cert-manager webhook CA bundle not detected within 300 s; attempting manual CA injection...")
    if _try_inject_ca_bundle_manually(env=env, log=log):
        _log(log, "Manual CA bundle injection succeeded.")
        return True
    _log(log, "WARN: Manual CA bundle injection also failed; cert-manager webhook may not be fully trusted.")
    return False


def _try_inject_ca_bundle_manually(
    env: dict | None = None,
    log: _LogFn | None = None,
) -> bool:
    """Manually patch the caBundle into the cert-manager ValidatingWebhookConfiguration.

    Reads the CA certificate from the ``cert-manager-webhook-ca`` secret (created
    by cert-manager itself) and patches it directly into the webhook configuration
    when the cainjector is too slow or has not yet reconciled.

    Returns True if the patch succeeded and the caBundle is now non-empty.
    """
    for secret_name in ("cert-manager-webhook-ca", "cert-manager-cainjector-leader-election"):
        r = subprocess.run(
            [
                "kubectl", "-n", "cert-manager", "get", "secret", secret_name,
                "-o", "jsonpath={.data.tls\\.crt}",
            ],
            env=env, capture_output=True, text=True, timeout=15,
        )
        ca_bundle = (r.stdout or "").strip()
        if r.returncode == 0 and ca_bundle:
            break
    else:
        _log(log, "WARN: cert-manager CA secret not found; skipping manual caBundle patch.")
        return False

    _log(log, "Patching cert-manager-webhook ValidatingWebhookConfiguration with CA bundle...")
    patch_json = json.dumps(
        [{"op": "replace", "path": "/webhooks/0/clientConfig/caBundle", "value": ca_bundle}]
    )
    r2 = subprocess.run(
        [
            "kubectl", "patch", "validatingwebhookconfiguration", "cert-manager-webhook",
            "--type=json", f"-p={patch_json}",
        ],
        env=env, capture_output=True, text=True, timeout=20,
    )
    if r2.returncode != 0:
        _log(log, f"WARN: Failed to patch caBundle: {(r2.stderr or r2.stdout or '').strip()}")
        return False
    # Brief wait for API server to propagate the patch
    time.sleep(3)
    return _wait_for_webhook_ca_bundle(env=env, log=log, timeout=30)


def _wait_for_webhook_ca_bundle(
    env: dict | None = None,
    log: _LogFn | None = None,
    timeout: int = 300,
) -> bool:
    """Poll until the cert-manager ValidatingWebhookConfiguration has a CA bundle injected.

    Returns True if the caBundle field becomes non-empty within *timeout* seconds,
    False otherwise.
    """
    deadline = time.time() + timeout
    interval = 5
    while time.time() < deadline:
        r = subprocess.run(
            [
                "kubectl", "get", "validatingwebhookconfiguration",
                "cert-manager-webhook",
                "-o", "jsonpath={.webhooks[0].clientConfig.caBundle}",
            ],
            env=env, capture_output=True, text=True, timeout=15,
        )
        if r.returncode == 0 and r.stdout.strip():
            return True
        time.sleep(interval)
    return False


def _wait_for_barman_tls_secrets(
    env: dict | None = None,
    log: _LogFn | None = None,
    timeout: int = 180,
) -> bool:
    """Poll until cert-manager has issued both Barman Cloud TLS secrets in cnpg-system.

    The Barman Cloud plugin manifest declares two Certificate resources
    (``barman-cloud-client`` and ``barman-cloud-server``).  cert-manager
    populates the corresponding secrets asynchronously; the pod cannot start
    until both secrets exist and have non-empty ``tls.crt`` / ``tls.key`` data.

    Returns True when all four fields are present, False on timeout.
    """
    _log(log, f"Waiting for Barman Cloud TLS secrets in cnpg-system (timeout: {timeout}s)...")
    deadline = time.time() + timeout
    interval = 5
    while time.time() < deadline:
        all_ready = True
        for secret in ("barman-cloud-client-tls", "barman-cloud-server-tls"):
            for field in ("tls.crt", "tls.key"):
                r = subprocess.run(
                    [
                        "kubectl", "-n", "cnpg-system", "get", "secret", secret,
                        "-o", f"jsonpath={{.data.{field.replace('.', '\\.')}}}",
                    ],
                    env=env, capture_output=True, text=True, timeout=15,
                )
                if r.returncode != 0 or not (r.stdout or "").strip():
                    all_ready = False
                    break
            if not all_ready:
                break
        if all_ready:
            _log(log, "Barman Cloud TLS secrets are available.")
            return True
        time.sleep(interval)
    _log(log, f"WARN: Barman Cloud TLS secrets not ready after {timeout}s.")
    return False


def _bootstrap_barman_tls_secrets(
    env: dict | None = None,
    log: _LogFn | None = None,
) -> None:
    """Generate and apply self-signed TLS secrets for the Barman Cloud plugin.

    Fallback for environments where cert-manager cannot issue the Certificate
    resources in time (e.g. GKE Autopilot with slow cert-manager scheduling).
    Generates EC P-256 self-signed certs and applies them as
    ``kubernetes.io/tls`` secrets directly to ``cnpg-system``, unblocking
    the barman-cloud pod's volume mounts.

    Secrets managed (both ``kubernetes.io/tls`` type):
    - ``barman-cloud-server-tls``
    - ``barman-cloud-client-tls``
    """
    namespace = "cnpg-system"
    server_secret = "barman-cloud-server-tls"
    client_secret = "barman-cloud-client-tls"

    def _secret_has_tls(name: str) -> bool:
        for field in ("tls.crt", "tls.key"):
            r = subprocess.run(
                [
                    "kubectl", "-n", namespace, "get", "secret", name,
                    "-o", f"jsonpath={{.data.{field.replace('.', '\\.')}}}",
                ],
                env=env, capture_output=True, text=True, timeout=15,
            )
            if r.returncode != 0 or not (r.stdout or "").strip():
                return False
        return True

    if _secret_has_tls(server_secret) and _secret_has_tls(client_secret):
        _log(log, "Barman Cloud TLS secrets already present — skipping bootstrap.")
        return

    _log(log, "Bootstrapping Barman Cloud TLS secrets in cnpg-system (cert-manager fallback)...")

    now = datetime.datetime.now(datetime.timezone.utc)
    one_year = datetime.timedelta(days=365)

    ca_key = generate_private_key(SECP256R1())
    ca_name = x509.Name([x509.NameAttribute(NameOID.COMMON_NAME, "barman-cloud CA")])
    ca_cert = (
        x509.CertificateBuilder()
        .subject_name(ca_name)
        .issuer_name(ca_name)
        .public_key(ca_key.public_key())
        .serial_number(x509.random_serial_number())
        .not_valid_before(now)
        .not_valid_after(now + one_year)
        .add_extension(x509.SubjectKeyIdentifier.from_public_key(ca_key.public_key()), critical=False)
        .add_extension(x509.AuthorityKeyIdentifier.from_issuer_public_key(ca_key.public_key()), critical=False)
        .add_extension(x509.BasicConstraints(ca=True, path_length=None), critical=True)
        .sign(ca_key, hashes.SHA256())
    )

    def _pem(obj) -> str:
        return obj.public_bytes(serialization.Encoding.PEM).decode()

    def _key_pem(k) -> str:
        return k.private_bytes(
            serialization.Encoding.PEM,
            serialization.PrivateFormat.TraditionalOpenSSL,
            serialization.NoEncryption(),
        ).decode()

    def _make_cert(cn: str, sans: list[str]):
        key = generate_private_key(SECP256R1())
        name = x509.Name([x509.NameAttribute(NameOID.COMMON_NAME, cn)])
        cert = (
            x509.CertificateBuilder()
            .subject_name(name)
            .issuer_name(ca_name)
            .public_key(key.public_key())
            .serial_number(x509.random_serial_number())
            .not_valid_before(now)
            .not_valid_after(now + one_year)
            .add_extension(
                x509.SubjectAlternativeName([x509.DNSName(d) for d in sans]),
                critical=False,
            )
            .add_extension(x509.SubjectKeyIdentifier.from_public_key(key.public_key()), critical=False)
            .add_extension(x509.AuthorityKeyIdentifier.from_issuer_public_key(ca_key.public_key()), critical=False)
            .sign(ca_key, hashes.SHA256())
        )
        return cert, key

    def _apply_tls_secret(name: str, cert_pem: str, key_pem: str) -> None:
        manifest = (
            "apiVersion: v1\n"
            "kind: Secret\n"
            "type: kubernetes.io/tls\n"
            "metadata:\n"
            f"  name: {name}\n"
            f"  namespace: {namespace}\n"
            "data:\n"
            f"  tls.crt: {base64.b64encode(cert_pem.encode()).decode()}\n"
            f"  tls.key: {base64.b64encode(key_pem.encode()).decode()}\n"
        )
        res = subprocess.run(
            ["kubectl", "apply", "-n", namespace, "-f", "-"],
            input=manifest, text=True, capture_output=True, env=env,
        )
        if res.returncode != 0:
            raise RuntimeError(f"Failed to apply Barman TLS secret '{name}': {res.stderr}")
        _log(log, f"Applied Barman TLS secret '{name}' in '{namespace}'.")

    srv_cert, srv_key = _make_cert(
        f"barman-cloud.{namespace}.svc",
        [
            "barman-cloud",
            f"barman-cloud.{namespace}.svc",
            f"barman-cloud.{namespace}.svc.cluster.local",
        ],
    )
    _apply_tls_secret(server_secret, _pem(srv_cert), _key_pem(srv_key))

    cli_cert, cli_key = _make_cert(
        "barman-cloud-client",
        ["barman-cloud-client"],
    )
    _apply_tls_secret(client_secret, _pem(cli_cert), _key_pem(cli_key))


def _resolve_barman_plugin_tag(env: dict | None, log: _LogFn | None) -> str:
    """Fetch the latest Barman Cloud plugin release tag from GitHub, with fallback."""
    fallback = ((env or {}).get("BARMAN_PLUGIN_FALLBACK_VERSION") or _BARMAN_PLUGIN_FALLBACK_VERSION).strip()
    try:
        import urllib.request as _req
        import json as _json
        with _req.urlopen(
            "https://api.github.com/repos/cloudnative-pg/plugin-barman-cloud/releases/latest",
            timeout=10,
        ) as resp:
            data = _json.loads(resp.read())
            tag = (data.get("tag_name") or "").strip()
            if tag and tag != "null":
                return tag
    except Exception as exc:
        _log(log, f"WARN: could not fetch latest Barman plugin version: {exc}; using v{fallback}")
    return f"v{fallback}"


def _bootstrap_db_user_secrets(
    namespace: str,
    env: dict | None = None,
    log: _LogFn | None = None,
) -> None:
    """Create knoe-db-user and knoe-db-superuser secrets from DB_PASSWORD if missing.

    Uses ``env["DB_PASSWORD"]`` and ``env.get("KNOE_DB_USER", "postgres")`` as the
    credentials source.  Skips silently if the secrets already exist or if no
    password is available in env.
    """
    pw = ((env or {}).get("DB_PASSWORD") or "").strip()
    if not pw or pw.startswith("${"):
        _log(log, "WARN: DB_PASSWORD not resolved in env; skipping db-user secret bootstrap.")
        return

    # CNPG requires the secret username to match the cluster owner field (owner: prole).
    # KNOE_DB_USER is the admin/superuser login name, not the cluster owner.
    db_owner = ((env or {}).get("CNPG_DB_OWNER") or "prole").strip()

    def _apply_if_missing(secret_name: str, username: str, password: str) -> None:
        if _kubectl_ok(["-n", namespace, "get", "secret", secret_name], env=env):
            return
        _log(log, f"Creating missing secret '{secret_name}' in '{namespace}'...")
        manifest_res = subprocess.run(
            [
                "kubectl", "create", "secret", "generic", secret_name,
                "-n", namespace,
                f"--from-literal=username={username}",
                f"--from-literal=password={password}",
                "--dry-run=client", "-o", "yaml",
            ],
            text=True, capture_output=True, env=env,
        )
        if manifest_res.returncode != 0:
            raise RuntimeError(f"Failed to render secret '{secret_name}':\n{manifest_res.stderr}")
        apply_res = subprocess.run(
            ["kubectl", "apply", "-n", namespace, "-f", "-"],
            input=manifest_res.stdout, text=True, capture_output=True, env=env,
        )
        if apply_res.returncode != 0:
            raise RuntimeError(f"Failed to apply secret '{secret_name}':\n{apply_res.stderr}")
        _log(log, f"Created secret '{secret_name}' in '{namespace}'.")

    _apply_if_missing("knoe-db-user", db_owner, pw)
    _apply_if_missing("knoe-db-superuser", "postgres", pw)


def _bootstrap_cnpg_admin_key(
    namespace: str,
    env: dict | None = None,
    log: _LogFn | None = None,
) -> None:
    """Generate and apply cnpg-admin-key secret (admin.key + admin.pub) if missing.

    Uses ED25519 if available; falls back to RSA-4096.  The secret is
    create-if-missing: existing keys are never rotated.
    """
    secret_name = "cnpg-admin-key"
    if _kubectl_ok(["-n", namespace, "get", "secret", secret_name], env=env):
        return

    _log(log, f"Creating missing secret '{secret_name}' in '{namespace}'...")

    # Generate ED25519 keypair; fall back to RSA-4096 if the runtime lacks it.
    try:
        priv_key = Ed25519PrivateKey.generate()
        priv_pem = priv_key.private_bytes(
            serialization.Encoding.PEM,
            serialization.PrivateFormat.PKCS8,
            serialization.NoEncryption(),
        ).decode()
        pub_pem = priv_key.public_key().public_bytes(
            serialization.Encoding.PEM,
            serialization.PublicFormat.SubjectPublicKeyInfo,
        ).decode()
    except Exception:  # pragma: no cover — runtime without ED25519 support
        from cryptography.hazmat.primitives.asymmetric import rsa as _rsa
        priv_key_rsa = _rsa.generate_private_key(public_exponent=65537, key_size=4096)
        priv_pem = priv_key_rsa.private_bytes(
            serialization.Encoding.PEM,
            serialization.PrivateFormat.PKCS8,
            serialization.NoEncryption(),
        ).decode()
        pub_pem = priv_key_rsa.public_key().public_bytes(
            serialization.Encoding.PEM,
            serialization.PublicFormat.SubjectPublicKeyInfo,
        ).decode()

    manifest_res = subprocess.run(
        [
            "kubectl", "create", "secret", "generic", secret_name,
            "-n", namespace,
            f"--from-literal=admin.key={priv_pem}",
            f"--from-literal=admin.pub={pub_pem}",
            "--dry-run=client", "-o", "yaml",
        ],
        text=True, capture_output=True, env=env,
    )
    if manifest_res.returncode != 0:
        raise RuntimeError(f"Failed to render secret '{secret_name}':\n{manifest_res.stderr}")
    apply_res = subprocess.run(
        ["kubectl", "apply", "-n", namespace, "-f", "-"],
        input=manifest_res.stdout, text=True, capture_output=True, env=env,
    )
    if apply_res.returncode != 0:
        raise RuntimeError(f"Failed to apply secret '{secret_name}':\n{apply_res.stderr}")
    _log(log, f"Created secret '{secret_name}' in '{namespace}'.")


def bootstrap_cnpg_tls_secrets(
    namespace: str,
    cluster_name: str,
    env: dict | None = None,
    log: _LogFn | None = None,
) -> None:
    """Generate and apply self-signed CA + TLS secrets for CNPG if missing.

    Creates two Kubernetes secrets (skipped if already present):
    - ``{cluster_name}-ca``  : Opaque, keys ``ca.crt`` and ``ca.key``
    - ``{cluster_name}-tls`` : kubernetes.io/tls, keys ``tls.crt`` and ``tls.key``

    The CA uses CN=``{cluster_name} CNPG CA`` and is self-signed (EC P-256, SHA-256).
    The server cert is signed by that CA and carries SANs matching the standard
    CNPG service names for the cluster.
    """
    ca_secret = f"{cluster_name}-ca"
    tls_secret = f"{cluster_name}-tls"

    # Ensure namespace exists before any secret operations
    _ensure_namespace(namespace, env=env)

    ca_exists = _kubectl_ok(["-n", namespace, "get", "secret", ca_secret], env=env)
    tls_exists = _kubectl_ok(["-n", namespace, "get", "secret", tls_secret], env=env)
    if ca_exists and tls_exists:
        _log(log, f"CNPG TLS secrets already present in '{namespace}' — skipping bootstrap.")
        return

    _log(log, f"Bootstrapping CNPG TLS secrets in namespace '{namespace}'...")

    now = datetime.datetime.now(datetime.timezone.utc)
    one_year = datetime.timedelta(days=365)

    # --- CA ---
    ca_key = generate_private_key(SECP256R1())
    ca_name = x509.Name([x509.NameAttribute(NameOID.COMMON_NAME, f"{cluster_name} CNPG CA")])
    ca_cert = (
        x509.CertificateBuilder()
        .subject_name(ca_name)
        .issuer_name(ca_name)
        .public_key(ca_key.public_key())
        .serial_number(x509.random_serial_number())
        .not_valid_before(now)
        .not_valid_after(now + one_year)
        .add_extension(x509.SubjectKeyIdentifier.from_public_key(ca_key.public_key()), critical=False)
        .add_extension(x509.AuthorityKeyIdentifier.from_issuer_public_key(ca_key.public_key()), critical=False)
        .add_extension(x509.BasicConstraints(ca=True, path_length=None), critical=True)
        .sign(ca_key, hashes.SHA256())
    )

    # --- Server cert ---
    srv_key = generate_private_key(SECP256R1())
    san_dns = [
        cluster_name,
        f"{cluster_name}-rw",
        f"{cluster_name}-rw.{namespace}.svc",
        f"{cluster_name}-r",
        f"{cluster_name}-ro",
    ]
    primary_srv_cn = f"{cluster_name}.{namespace}.svc"
    srv_cn = primary_srv_cn
    if len(srv_cn) > 64:
        for candidate in (f"{cluster_name}-rw", cluster_name, "cnpg-server"):
            if len(candidate) <= 64:
                srv_cn = candidate
                break
        _log(
            log,
            "CNPG TLS subject CN exceeds X.509 limit; "
            f"using fallback CN '{srv_cn}' instead of '{primary_srv_cn}'.",
        )

    srv_name = x509.Name([x509.NameAttribute(NameOID.COMMON_NAME, srv_cn)])
    srv_cert = (
        x509.CertificateBuilder()
        .subject_name(srv_name)
        .issuer_name(ca_name)
        .public_key(srv_key.public_key())
        .serial_number(x509.random_serial_number())
        .not_valid_before(now)
        .not_valid_after(now + one_year)
        .add_extension(
            x509.SubjectAlternativeName([x509.DNSName(d) for d in san_dns]),
            critical=False,
        )
        .add_extension(x509.SubjectKeyIdentifier.from_public_key(srv_key.public_key()), critical=False)
        .add_extension(x509.AuthorityKeyIdentifier.from_issuer_public_key(ca_key.public_key()), critical=False)
        .sign(ca_key, hashes.SHA256())
    )

    def _pem(obj) -> str:
        return obj.public_bytes(serialization.Encoding.PEM).decode()

    def _key_pem(k) -> str:
        return k.private_bytes(
            serialization.Encoding.PEM,
            serialization.PrivateFormat.TraditionalOpenSSL,
            serialization.NoEncryption(),
        ).decode()

    def _apply_secret(manifest: str) -> None:
        res = subprocess.run(
            ["kubectl", "apply", "-n", namespace, "-f", "-"],
            input=manifest,
            text=True,
            capture_output=True,
            env=env,
        )
        if res.returncode != 0:
            raise RuntimeError(f"kubectl apply failed:\n{res.stderr}")

    ca_crt_pem = _pem(ca_cert)
    ca_key_pem = _key_pem(ca_key)

    # Build CA secret manifest (Opaque)
    ca_manifest = (
        "apiVersion: v1\n"
        "kind: Secret\n"
        "metadata:\n"
        f"  name: {ca_secret}\n"
        f"  namespace: {namespace}\n"
        "data:\n"
        f"  ca.crt: {base64.b64encode(ca_crt_pem.encode()).decode()}\n"
        f"  ca.key: {base64.b64encode(ca_key_pem.encode()).decode()}\n"
    )
    _apply_secret(ca_manifest)
    _log(log, f"Applied secret '{ca_secret}' in '{namespace}'.")

    tls_crt_pem = _pem(srv_cert)
    tls_key_pem = _key_pem(srv_key)

    # Build TLS secret manifest (kubernetes.io/tls)
    tls_manifest = (
        "apiVersion: v1\n"
        "kind: Secret\n"
        "type: kubernetes.io/tls\n"
        "metadata:\n"
        f"  name: {tls_secret}\n"
        f"  namespace: {namespace}\n"
        "data:\n"
        f"  tls.crt: {base64.b64encode(tls_crt_pem.encode()).decode()}\n"
        f"  tls.key: {base64.b64encode(tls_key_pem.encode()).decode()}\n"
    )
    _apply_secret(tls_manifest)
    _log(log, f"Applied secret '{tls_secret}' in '{namespace}'.")


def initialize(
    namespace: str,
    cluster_name: str,
    env: dict | None = None,
    project_root: str | Path = ".",
    log: _LogFn | None = None,
    mode: str = "",
) -> None:
    """
    Full CNPG initialization pipeline.

    Steps (matching init_cloudnative_pg.sh initialize):
      1. Ensure the CNPG operator is installed and ready
      2. Pin the controller to storage nodes (k3s)
      3. Install the Barman Cloud plugin
      4. Preflight: verify required secrets exist
      5. Apply the CNPG cluster manifest
      6. Wait for cluster pods to become Ready
    """
    _log(log, f"Initializing CNPG cluster '{cluster_name}' in namespace '{namespace}'...")

    ensure_operator(
        env=env,
        log=log,
        mode=mode,
        allow_reapply_if_unhealthy=True,
        aggressive_cleanup=True,
    )
    pin_controller(env=env, log=log, mode=mode)
    install_barman_plugin(env=env, log=log)

    # Bootstrap TLS secrets (create-if-missing; no rotation of existing certs)
    bootstrap_cnpg_tls_secrets(namespace=namespace, cluster_name=cluster_name, env=env, log=log)

    # Bootstrap DB user secrets from env if missing (create-if-missing only)
    _bootstrap_db_user_secrets(namespace=namespace, env=env, log=log)

    # Bootstrap admin keypair secret (create-if-missing; no rotation)
    _bootstrap_cnpg_admin_key(namespace=namespace, env=env, log=log)

    # Preflight: required secrets must exist before applying the cluster
    missing = []
    for secret in ["knoe-db-user", f"{cluster_name}-tls", f"{cluster_name}-ca"]:
        if not _kubectl_ok(["-n", namespace, "get", "secret", secret], env=env):
            missing.append(secret)
    if missing:
        raise RuntimeError(
            f"Required secrets missing in namespace '{namespace}': {', '.join(missing)}. "
            "The CNPG cluster cannot start without these secrets."
        )
    _log(log, "Pre-flight check passed: all required secrets present.")
    # Release any Released PVs so new PVCs can bind (idempotent, k3s only)
    if mode == "k3s":
        _reconcile_unbound_synology_data_wal_pvs(env=env, log=log)
        _release_stale_pv_claims(env=env, log=log)
    _remove_legacy_barman_object_store_when_plugin_is_enabled(
        namespace=namespace,
        cluster_name=cluster_name,
        env=env,
        log=log,
    )
    manifest = _resolve_cnpg_manifest(project_root, env)
    _apply_manifest(namespace, manifest, env, log)

    wait_timeout = int((env or {}).get("CNPG_WAIT_TIMEOUT", "900") or "900")
    _wait_cnpg_pods(namespace, cluster_name, env, timeout=wait_timeout, log=log)

    _log(log, f"Initialization complete for CNPG cluster '{cluster_name}'.")


def deploy(
    namespace: str,
    cluster_name: str,
    env: dict | None = None,
    project_root: str | Path = ".",
    log: _LogFn | None = None,
    mode: str = "",
) -> None:
    """
    Image-aware CNPG deploy pipeline.

    Steps (matching init_cloudnative_pg.sh deploy_cluster):
      1. Ensure the CNPG operator is installed and ready
      2. Pin the controller to storage nodes (k3s)
      3. Install the Barman Cloud plugin
      4. Apply the CNPG cluster manifest
      5. Reconcile instance count to match eligible node count
    """
    _log(log, f"Deploying CNPG cluster '{cluster_name}' in namespace '{namespace}'...")

    ensure_operator(
        env=env,
        log=log,
        mode=mode,
        allow_reapply_if_unhealthy=False,
        aggressive_cleanup=False,
    )
    pin_controller(env=env, log=log, mode=mode)
    install_barman_plugin(env=env, log=log)

    _remove_legacy_barman_object_store_when_plugin_is_enabled(
        namespace=namespace,
        cluster_name=cluster_name,
        env=env,
        log=log,
    )
    manifest = _resolve_cnpg_manifest(project_root, env)
    _apply_manifest(namespace, manifest, env, log)
    _reconcile_instances(namespace, cluster_name, env, log)

    _log(log, f"Deploy complete for CNPG cluster '{cluster_name}'.")


def rollout(
    namespace: str,
    cluster_name: str,
    env: dict | None = None,
    log: _LogFn | None = None,
) -> None:
    """
    Rolling restart of a CNPG cluster via ordered pod deletion.

    Steps (matching init_cloudnative_pg.sh rollout_cluster):
      1. Identify the current primary pod
      2. Delete replica pods one-by-one, waiting for readiness
      3. Promote a replica to primary (if kubectl-cnpg plugin is available)
      4. Delete and recreate the old primary pod
    """
    _log(log, f"Starting rollout for CNPG cluster '{cluster_name}' in namespace '{namespace}'...")

    # Identify current primary
    r = _kubectl(
        ["-n", namespace, "get", "cluster", cluster_name, "-o", "jsonpath={.status.currentPrimary}"],
        env=env, timeout=15,
    )
    primary = (r.stdout or "").strip()
    if not primary:
        raise RuntimeError(f"Could not identify primary instance for cluster '{cluster_name}'")
    _log(log, f"Primary instance: {primary}")

    # List all cluster pods
    r2 = _kubectl(
        ["-n", namespace, "get", "pods", "-l", f"cnpg.io/cluster={cluster_name}",
         "-o", "jsonpath={.items[*].metadata.name}"],
        env=env, timeout=15,
    )
    pods = (r2.stdout or "").strip().split()
    if not pods:
        raise RuntimeError(f"No pods found for cluster '{cluster_name}'")

    has_plugin = _kubectl_ok(["cnpg", "version"], env=env, timeout=10)

    # Delete non-primary pods first
    new_primary: str | None = None
    for pod in pods:
        if pod == primary:
            continue
        _log(log, f"Recreating non-primary pod: {pod}...")
        _kubectl(["-n", namespace, "delete", "pod", pod], env=env, timeout=60)
        _wait_pod_ready(namespace, pod, env, timeout=300, log=log)
        if new_primary is None:
            new_primary = pod

    # Promote a replica if the cnpg plugin is available
    if new_primary and has_plugin:
        _log(log, f"Promoting {new_primary}...")
        _kubectl(["cnpg", "promote", cluster_name, new_primary, "-n", namespace], env=env, timeout=30)
        deadline = time.monotonic() + 300
        while time.monotonic() < deadline:
            r3 = _kubectl(
                ["-n", namespace, "get", "cluster", cluster_name,
                 "-o", "jsonpath={.status.currentPrimary}"],
                env=env, timeout=10,
            )
            if (r3.stdout or "").strip() == new_primary:
                _log(log, f"{new_primary} is now the primary.")
                break
            time.sleep(5)
    elif new_primary and not has_plugin:
        _log(log, "WARN: kubectl cnpg plugin not available; skipping explicit promotion.")

    # Delete the old primary last
    _log(log, f"Recreating old primary pod: {primary}...")
    _kubectl(["-n", namespace, "delete", "pod", "-n", namespace, primary], env=env, timeout=60)
    _wait_pod_ready(namespace, primary, env, timeout=300, log=log)

    if has_plugin:
        _log(log, "Rollout complete. Final cluster status:")
        _kubectl(["cnpg", "status", cluster_name, "-n", namespace], env=env, timeout=30)
    else:
        _log(log, "Rollout complete.")