boatramp 0.2.3

//! Desired-workload builders: a [`BoatRampCluster`] spec → the Kubernetes objects
//! the operator owns. Pure functions (spec → object), unit-tested here; the
//! reconciler in `controller.rs` server-side-applies whatever these return.
//!
//! - **cluster mode** → a `StatefulSet` (stable identity + per-node `PVC`) + a
//!   headless `Service` (stable DNS) + a client `Service` + a `ConfigMap` +
//!   a `PodDisruptionBudget`.
//! - **stateless mode** → a `Deployment` + a client `Service` + a `ConfigMap` +
//!   an `HorizontalPodAutoscaler`.
//!
//! K2 stands the workloads up; it does **not** wire Raft membership — cluster-mode
//! pods run as standalone servers until K3 adds the `[cluster]` config + the
//! membership reconciler. That split is deliberate (a StatefulSet alone can't do
//! consensus membership — the whole reason the operator exists).

use std::collections::BTreeMap;

use k8s_openapi::api::apps::v1::{
    Deployment, DeploymentSpec, RollingUpdateStatefulSetStrategy, StatefulSet,
    StatefulSetPersistentVolumeClaimRetentionPolicy, StatefulSetSpec, StatefulSetUpdateStrategy,
};
use k8s_openapi::api::autoscaling::v2::{
    CrossVersionObjectReference, HorizontalPodAutoscaler, HorizontalPodAutoscalerSpec, MetricSpec,
    MetricTarget, ResourceMetricSource,
};
use k8s_openapi::api::core::v1::{
    ConfigMap, ConfigMapVolumeSource, Container, ContainerPort, EnvVar, EnvVarSource,
    HTTPGetAction, ObjectFieldSelector, PersistentVolumeClaim, PersistentVolumeClaimSpec, PodSpec,
    PodTemplateSpec, Probe, Service, ServicePort, ServiceSpec, TCPSocketAction, Volume,
    VolumeMount, VolumeResourceRequirements,
};
use k8s_openapi::api::policy::v1::{PodDisruptionBudget, PodDisruptionBudgetSpec};
use k8s_openapi::apimachinery::pkg::api::resource::Quantity;
use k8s_openapi::apimachinery::pkg::apis::meta::v1::{LabelSelector, ObjectMeta};
use k8s_openapi::apimachinery::pkg::util::intstr::IntOrString;
use kube::Resource;

use super::crd::{BoatRampCluster, ClusterMode};

/// The control-plane / serving port every boatramp pod listens on.
const PORT: i32 = 8080;
/// The Raft peer-mesh port (distinct from the control-plane `PORT`).
const MESH_PORT: i32 = 7000;
/// Where the config file is mounted, and the data dir.
const CONFIG_MOUNT: &str = "/etc/boatramp";
const DATA_MOUNT: &str = "/data";
/// The operator's own default image (used when the CR doesn't pin one).
const DEFAULT_IMAGE: &str = "ghcr.io/boatramp/boatramp:latest";

/// Selector/identity labels for a cluster's children.
fn labels(name: &str) -> BTreeMap<String, String> {
    [
        ("app.kubernetes.io/name".to_string(), "boatramp".to_string()),
        ("app.kubernetes.io/instance".to_string(), name.to_string()),
        (
            "app.kubernetes.io/managed-by".to_string(),
            "boatramp-operator".to_string(),
        ),
    ]
    .into()
}

/// `ObjectMeta` for a child object: named, namespaced, labelled, and
/// **owned** by the CR (so `kubectl delete brc` garbage-collects it).
fn child_meta(brc: &BoatRampCluster, name: String) -> ObjectMeta {
    ObjectMeta {
        name: Some(name),
        namespace: brc.metadata.namespace.clone(),
        labels: Some(labels(&instance(brc))),
        owner_references: brc.controller_owner_ref(&()).map(|r| vec![r]),
        ..Default::default()
    }
}

/// The CR's name (the instance every child is keyed to). Also the client
/// Service's name — the DNS the operator's membership executor reaches.
pub fn instance(brc: &BoatRampCluster) -> String {
    brc.metadata
        .name
        .clone()
        .unwrap_or_else(|| "boatramp".to_string())
}

fn image(brc: &BoatRampCluster) -> String {
    brc.spec
        .image
        .clone()
        .unwrap_or_else(|| DEFAULT_IMAGE.to_string())
}

/// The headless service's DNS name (stable per-pod identity in cluster mode).
fn headless_name(brc: &BoatRampCluster) -> String {
    format!("{}-headless", instance(brc))
}

/// The `boatramp.cfg` (RON) rendered into the ConfigMap: bind, data dir, posture.
/// K2 omits `[cluster]` — K3 adds it along with membership.
fn config_ron(brc: &BoatRampCluster) -> String {
    let posture = brc.spec.posture.as_deref().unwrap_or("multi-tenant");
    // The control-plane trust anchor: verifies admin tokens + (in cluster mode)
    // join tokens/members against it. The matching private key is wired as an env
    // var from `spec.authSecret`, never written into the ConfigMap.
    let auth = match brc.spec.root_pubkey.as_deref() {
        Some(pubkey) => format!("    auth_root_public_key: \"{pubkey}\",\n"),
        None => String::new(),
    };
    // Cluster mode: a `[cluster]` section makes `serve` run the embedded Raft node.
    // Founding/joining is driven by env (BOATRAMP_POD_NAME → ordinal-0 founds;
    // BOATRAMP_CLUSTER_JOIN → a joiner's ticket), so the config itself is uniform.
    let cluster = if brc.spec.mode == ClusterMode::Cluster {
        format!(
            "  cluster: (\n    \
               listen: \"0.0.0.0:{MESH_PORT}\",\n    \
               store_dir: \"{DATA_MOUNT}/raft\",\n  \
             ),\n"
        )
    } else {
        String::new()
    };
    format!(
        "(\n  \
           serve: (\n    \
             addr: \"0.0.0.0:{PORT}\",\n    \
             data_dir: \"{DATA_MOUNT}\",\n\
         {auth}  \
           ),\n\
         {cluster}  \
           security: ( profile: \"{posture}\" ),\n\
         )\n"
    )
}

pub fn config_map(brc: &BoatRampCluster) -> ConfigMap {
    ConfigMap {
        metadata: child_meta(brc, format!("{}-config", instance(brc))),
        data: Some([("boatramp.cfg".to_string(), config_ron(brc))].into()),
        ..Default::default()
    }
}

/// The headless Service backing StatefulSet pod DNS (`<pod>.<headless>.<ns>.svc`).
pub fn headless_service(brc: &BoatRampCluster) -> Service {
    Service {
        metadata: child_meta(brc, headless_name(brc)),
        spec: Some(ServiceSpec {
            cluster_ip: Some("None".to_string()),
            selector: Some(labels(&instance(brc))),
            // Both the control-plane port and the Raft peer-mesh port, so pods
            // reach each other's mesh over the stable per-pod DNS.
            ports: Some(vec![port("http", PORT), port("mesh", MESH_PORT)]),
            publish_not_ready_addresses: Some(true),
            ..Default::default()
        }),
        ..Default::default()
    }
}

/// The client Service (stable ClusterIP) for reaching the API / sites.
pub fn client_service(brc: &BoatRampCluster) -> Service {
    Service {
        metadata: child_meta(brc, instance(brc)),
        spec: Some(ServiceSpec {
            selector: Some(labels(&instance(brc))),
            ports: Some(vec![port("http", PORT)]),
            ..Default::default()
        }),
        ..Default::default()
    }
}

/// An env var sourced from a pod field via the downward API (e.g. the pod name).
fn downward_env(name: &str, field_path: &str) -> EnvVar {
    EnvVar {
        name: name.to_string(),
        value_from: Some(EnvVarSource {
            field_ref: Some(ObjectFieldSelector {
                field_path: field_path.to_string(),
                ..Default::default()
            }),
            ..Default::default()
        }),
        ..Default::default()
    }
}

/// An env var sourced from a Secret key, optional so a partially-provisioned
/// Secret doesn't wedge the pod (a missing key is just an unset var).
fn secret_env(name: &str, secret: &str, key: &str) -> EnvVar {
    EnvVar {
        name: name.to_string(),
        value_from: Some(EnvVarSource {
            secret_key_ref: Some(k8s_openapi::api::core::v1::SecretKeySelector {
                name: secret.to_string(),
                key: key.to_string(),
                optional: Some(true),
            }),
            ..Default::default()
        }),
        ..Default::default()
    }
}

fn port(name: &str, p: i32) -> ServicePort {
    ServicePort {
        name: Some(name.to_string()),
        port: p,
        target_port: Some(IntOrString::Int(p)),
        ..Default::default()
    }
}

/// The pod's container — shared by the StatefulSet and Deployment. Runs
/// `serve --config <mounted cfg>`, probes `/healthz` (liveness) + `/readyz`
/// (readiness), and learns its own pod name via the downward API (for K3).
fn container(brc: &BoatRampCluster) -> Container {
    Container {
        name: "boatramp".to_string(),
        image: Some(image(brc)),
        // The operator controls the image via the CR's `spec.image` (an explicit
        // version), so image changes roll through a pod-spec change, not a re-pull.
        // Own the field explicitly (`IfNotPresent`): otherwise the apiserver's
        // `:latest`-era `Always` default sticks, and a pinned/loaded image (or a
        // `k3d image import`) is needlessly re-pulled.
        image_pull_policy: Some("IfNotPresent".to_string()),
        // Set the entrypoint explicitly so the pod works regardless of whether the
        // image declares one (`args` alone need an image ENTRYPOINT).
        command: Some(vec!["boatramp".to_string()]),
        args: Some({
            let mut args = vec![
                "serve".to_string(),
                "--config".to_string(),
                format!("{CONFIG_MOUNT}/boatramp.cfg"),
            ];
            // Cluster mode serves its control plane over RPK-TLS (`--tls rpk`) so
            // the operator (and joining pods) can pin it against the root anchor and
            // fetch the root-signed bootstrap attestation the dynamic join needs —
            // every pod holds the root private key (from `authSecret`) to self-attest.
            if brc.spec.mode == ClusterMode::Cluster {
                args.push("--tls".to_string());
                args.push("rpk".to_string());
            }
            args
        }),
        ports: Some({
            let mut ports = vec![ContainerPort {
                name: Some("http".to_string()),
                container_port: PORT,
                ..Default::default()
            }];
            if brc.spec.mode == ClusterMode::Cluster {
                ports.push(ContainerPort {
                    name: Some("mesh".to_string()),
                    container_port: MESH_PORT,
                    ..Default::default()
                });
            }
            ports
        }),
        env: Some(
            vec![
                // The pod's own name (downward API): the operator's ordinal-0 pod
                // founds the cluster; every other ordinal joins. The node *identity*
                // is still derived from the mesh key — this only designates the founder.
                downward_env("BOATRAMP_POD_NAME", "metadata.name"),
                // The single-use join ticket the operator's executor rolls into a
                // Secret for joining pods. Optional: absent for the founder / before
                // the first AddLearner.
                {
                    let (secret, key) = super::executor::join_env_source(brc);
                    EnvVar {
                        name: "BOATRAMP_CLUSTER_JOIN".to_string(),
                        value_from: Some(EnvVarSource {
                            secret_key_ref: Some(k8s_openapi::api::core::v1::SecretKeySelector {
                                name: secret,
                                key: key.to_string(),
                                optional: Some(true),
                            }),
                            ..Default::default()
                        }),
                        ..Default::default()
                    }
                },
            ]
            .into_iter()
            // Auth from `spec.authSecret`: the root private key (the founder signs
            // join tokens / attestations / members with it) + an optional single-use
            // bootstrap secret (to mint the first admin token). Both optional at the
            // Secret level so a partially-provisioned Secret doesn't wedge the pod.
            .chain(brc.spec.auth_secret.as_deref().into_iter().flat_map(|s| {
                [
                    secret_env("BOATRAMP_AUTH_ROOT_PRIVATE_KEY", s, "root-private-key"),
                    secret_env("BOATRAMP_BOOTSTRAP_SECRET", s, "bootstrap-secret"),
                ]
            }))
            // Cluster mode: the pod advertises its **own** stable DNS as the mesh
            // address the leader dials (the default `0.0.0.0` bind isn't dialable).
            // Composed with k8s `$(VAR)` substitution from the pod name + namespace.
            .chain(
                (brc.spec.mode == ClusterMode::Cluster)
                    .then(|| {
                        vec![
                            downward_env("POD_NAMESPACE", "metadata.namespace"),
                            EnvVar {
                                name: "BOATRAMP_CLUSTER_ADVERTISE_ADDR".to_string(),
                                value: Some(format!(
                        "https://$(BOATRAMP_POD_NAME).{}.$(POD_NAMESPACE).svc:{MESH_PORT}",
                        headless_name(brc)
                    )),
                                ..Default::default()
                            },
                        ]
                    })
                    .into_iter()
                    .flatten(),
            )
            .collect(),
        ),
        // Cluster mode serves the control plane over RPK-TLS (RFC 7250 raw public
        // keys), which the kubelet's HTTP prober can't speak — so probe the port
        // with a TCP socket instead. A cluster node binds its listener only *after*
        // it has founded/joined and is serving, so "port open" is exactly the
        // right readiness gate (and it drives StatefulSet OrderedReady sequencing).
        // Stateless mode is plain HTTP → the real `/healthz` + `/readyz` endpoints.
        liveness_probe: Some(if brc.spec.mode == ClusterMode::Cluster {
            tcp_probe()
        } else {
            http_probe("/healthz")
        }),
        readiness_probe: Some(if brc.spec.mode == ClusterMode::Cluster {
            tcp_probe()
        } else {
            http_probe("/readyz")
        }),
        volume_mounts: Some(vec![
            VolumeMount {
                name: "config".to_string(),
                mount_path: CONFIG_MOUNT.to_string(),
                read_only: Some(true),
                ..Default::default()
            },
            VolumeMount {
                name: "data".to_string(),
                mount_path: DATA_MOUNT.to_string(),
                ..Default::default()
            },
        ]),
        ..Default::default()
    }
}

fn http_probe(path: &str) -> Probe {
    Probe {
        http_get: Some(HTTPGetAction {
            path: Some(path.to_string()),
            port: IntOrString::Int(PORT),
            ..Default::default()
        }),
        period_seconds: Some(10),
        ..Default::default()
    }
}

/// A TCP-socket probe on the control-plane port — for cluster mode, whose RPK-TLS
/// listener the kubelet's HTTP prober can't speak. The listener binds only once
/// the node is founded/joined and serving, so an open port is a sound readiness
/// signal.
fn tcp_probe() -> Probe {
    Probe {
        tcp_socket: Some(TCPSocketAction {
            port: IntOrString::Int(PORT),
            ..Default::default()
        }),
        period_seconds: Some(10),
        ..Default::default()
    }
}

/// The pod template, minus the `data` volume (the StatefulSet supplies a PVC
/// claim template; the Deployment supplies an `emptyDir`).
fn pod_template(brc: &BoatRampCluster, data_volume: Option<Volume>) -> PodTemplateSpec {
    let mut volumes = vec![Volume {
        name: "config".to_string(),
        config_map: Some(ConfigMapVolumeSource {
            name: format!("{}-config", instance(brc)),
            ..Default::default()
        }),
        ..Default::default()
    }];
    volumes.extend(data_volume);
    PodTemplateSpec {
        metadata: Some(ObjectMeta {
            labels: Some(labels(&instance(brc))),
            ..Default::default()
        }),
        spec: Some(PodSpec {
            containers: vec![container(brc)],
            volumes: Some(volumes),
            ..Default::default()
        }),
    }
}

fn selector(brc: &BoatRampCluster) -> LabelSelector {
    LabelSelector {
        match_labels: Some(labels(&instance(brc))),
        ..Default::default()
    }
}

/// The cluster-mode StatefulSet: stable identity + a per-node data PVC.
///
/// `roll_partition` is the `RollingUpdate` partition the operator controls
/// (quorum-aware upgrades, K4): only pods with an ordinal `>=` the partition are
/// updated, so the operator **pauses** the rollout by setting it to `replicas`
/// when the cluster has no quorum margin, and to `0` to let it proceed one pod at
/// a time (highest ordinal first).
pub fn stateful_set(brc: &BoatRampCluster, roll_partition: i32) -> StatefulSet {
    let storage = brc
        .spec
        .storage
        .clone()
        .unwrap_or_else(|| "10Gi".to_string());
    let pvc = PersistentVolumeClaim {
        metadata: ObjectMeta {
            name: Some("data".to_string()),
            ..Default::default()
        },
        spec: Some(PersistentVolumeClaimSpec {
            access_modes: Some(vec!["ReadWriteOnce".to_string()]),
            resources: Some(VolumeResourceRequirements {
                requests: Some([("storage".to_string(), Quantity(storage))].into()),
                ..Default::default()
            }),
            ..Default::default()
        }),
        ..Default::default()
    };
    StatefulSet {
        metadata: child_meta(brc, instance(brc)),
        spec: Some(StatefulSetSpec {
            replicas: Some(brc.spec.replicas as i32),
            service_name: Some(headless_name(brc)),
            selector: selector(brc),
            // The data volume comes from `volume_claim_templates`, not the pod spec.
            template: pod_template(brc, None),
            volume_claim_templates: Some(vec![pvc]),
            // Quorum-aware rolling upgrade: the operator advances/pauses this
            // partition based on the cluster's roll margin (K4).
            update_strategy: Some(StatefulSetUpdateStrategy {
                type_: Some("RollingUpdate".to_string()),
                rolling_update: Some(RollingUpdateStatefulSetStrategy {
                    partition: Some(roll_partition),
                    ..Default::default()
                }),
            }),
            // **Retain** a node's PVC on scale-down or StatefulSet delete — a Raft
            // voter's durable log/state must never be reclaimed automatically (it
            // would lose the vote + data); reclaiming is an explicit operator step.
            persistent_volume_claim_retention_policy: Some(
                StatefulSetPersistentVolumeClaimRetentionPolicy {
                    when_deleted: Some("Retain".to_string()),
                    when_scaled: Some("Retain".to_string()),
                },
            ),
            ..Default::default()
        }),
        ..Default::default()
    }
}

/// The stateless-mode Deployment: no per-pod identity, an ephemeral data dir
/// (state lives in the shared/replicated KV).
pub fn deployment(brc: &BoatRampCluster) -> Deployment {
    let data = Volume {
        name: "data".to_string(),
        empty_dir: Some(Default::default()),
        ..Default::default()
    };
    Deployment {
        metadata: child_meta(brc, instance(brc)),
        spec: Some(DeploymentSpec {
            replicas: Some(brc.spec.replicas as i32),
            selector: selector(brc),
            template: pod_template(brc, Some(data)),
            ..Default::default()
        }),
        ..Default::default()
    }
}

/// Keep a majority available during voluntary disruptions (cluster mode).
pub fn pod_disruption_budget(brc: &BoatRampCluster) -> PodDisruptionBudget {
    // Tolerate losing a minority: `min_available = floor(n/2) + 1` keeps quorum.
    let min_available = (brc.spec.replicas / 2) + 1;
    PodDisruptionBudget {
        metadata: child_meta(brc, instance(brc)),
        spec: Some(PodDisruptionBudgetSpec {
            min_available: Some(IntOrString::Int(min_available as i32)),
            selector: Some(selector(brc)),
            ..Default::default()
        }),
        ..Default::default()
    }
}

/// Autoscale the stateless Deployment on CPU (1..=`replicas`×4, target 75%).
pub fn hpa(brc: &BoatRampCluster) -> HorizontalPodAutoscaler {
    HorizontalPodAutoscaler {
        metadata: child_meta(brc, instance(brc)),
        spec: Some(HorizontalPodAutoscalerSpec {
            scale_target_ref: CrossVersionObjectReference {
                api_version: Some("apps/v1".to_string()),
                kind: "Deployment".to_string(),
                name: instance(brc),
            },
            min_replicas: Some(brc.spec.replicas.max(1) as i32),
            max_replicas: (brc.spec.replicas.max(1) * 4) as i32,
            metrics: Some(vec![MetricSpec {
                type_: "Resource".to_string(),
                resource: Some(ResourceMetricSource {
                    name: "cpu".to_string(),
                    target: MetricTarget {
                        type_: "Utilization".to_string(),
                        average_utilization: Some(75),
                        ..Default::default()
                    },
                }),
                ..Default::default()
            }]),
            ..Default::default()
        }),
        ..Default::default()
    }
}

#[cfg(test)]
mod tests {
    use super::*;
    use crate::operator::crd::{BoatRampClusterSpec, ClusterMode};

    fn cluster(name: &str, mode: ClusterMode, replicas: u32) -> BoatRampCluster {
        let mut brc = BoatRampCluster::new(
            name,
            BoatRampClusterSpec {
                mode,
                replicas,
                image: None,
                storage: None,
                posture: None,
                admin_token_secret: None,
                root_pubkey: None,
                auth_secret: None,
            },
        );
        brc.metadata.namespace = Some("tenant-a".to_string());
        brc.metadata.uid = Some("uid-123".to_string());
        brc
    }

    #[test]
    fn statefulset_has_pvc_stable_dns_probes_and_owner_ref() {
        let brc = cluster("db", ClusterMode::Cluster, 3);
        // A paused rollout (partition == replicas) — the operator's K4 knob.
        let sts = stateful_set(&brc, 3);
        let spec = sts.spec.unwrap();
        assert_eq!(spec.replicas, Some(3));
        assert_eq!(spec.service_name.as_deref(), Some("db-headless"));
        // A per-node PVC (the reason cluster mode is a StatefulSet).
        assert_eq!(spec.volume_claim_templates.as_ref().unwrap().len(), 1);
        // K4: the operator-controlled rolling-upgrade partition.
        assert_eq!(
            spec.update_strategy
                .as_ref()
                .and_then(|u| u.rolling_update.as_ref())
                .and_then(|r| r.partition),
            Some(3)
        );
        // K4: a Raft voter's PVC is never auto-reclaimed (data + vote safety).
        let retain = spec
            .persistent_volume_claim_retention_policy
            .as_ref()
            .unwrap();
        assert_eq!(retain.when_deleted.as_deref(), Some("Retain"));
        assert_eq!(retain.when_scaled.as_deref(), Some("Retain"));
        // Cluster mode serves RPK-TLS, so probes are TCP-socket (see
        // `cluster_pods_serve_rpk_tls_and_probe_by_tcp` for the full contract).
        let c = &spec.template.spec.unwrap().containers[0];
        assert!(c.liveness_probe.as_ref().unwrap().tcp_socket.is_some());
        assert!(c.readiness_probe.as_ref().unwrap().tcp_socket.is_some());
        // Owned by the CR ⇒ garbage-collected with it.
        let owners = sts.metadata.owner_references.unwrap();
        assert_eq!(owners[0].kind, "BoatRampCluster");
        assert_eq!(owners[0].controller, Some(true));
    }

    #[test]
    fn cluster_pods_serve_rpk_tls_and_probe_by_tcp() {
        // Cluster mode: the control plane is `--tls rpk` (so the operator + joiners
        // can pin it), and — because the kubelet can't HTTP-probe an RPK-TLS port —
        // liveness/readiness are TCP-socket probes on the control-plane port.
        let brc = cluster("db", ClusterMode::Cluster, 3);
        let c = stateful_set(&brc, 0)
            .spec
            .unwrap()
            .template
            .spec
            .unwrap()
            .containers
            .remove(0);
        let args = c.args.unwrap();
        assert_eq!(
            args,
            vec![
                "serve".to_string(),
                "--config".to_string(),
                format!("{CONFIG_MOUNT}/boatramp.cfg"),
                "--tls".to_string(),
                "rpk".to_string(),
            ]
        );
        assert!(c.readiness_probe.as_ref().unwrap().tcp_socket.is_some());
        assert!(c.readiness_probe.as_ref().unwrap().http_get.is_none());
        assert!(c.liveness_probe.as_ref().unwrap().tcp_socket.is_some());

        // Stateless mode stays plain HTTP: no `--tls rpk`, real `/readyz` probe.
        let web = cluster("web", ClusterMode::Stateless, 2);
        let wc = deployment(&web)
            .spec
            .unwrap()
            .template
            .spec
            .unwrap()
            .containers
            .remove(0);
        assert!(!wc.args.unwrap().contains(&"rpk".to_string()));
        let rp = wc.readiness_probe.unwrap();
        assert!(rp.tcp_socket.is_none());
        assert_eq!(rp.http_get.unwrap().path.as_deref(), Some("/readyz"));
    }

    #[test]
    fn headless_service_is_headless_and_publishes_not_ready() {
        let svc = headless_service(&cluster("db", ClusterMode::Cluster, 3));
        let spec = svc.spec.unwrap();
        assert_eq!(spec.cluster_ip.as_deref(), Some("None"));
        // Pods must resolve before /readyz passes so peers can find each other.
        assert_eq!(spec.publish_not_ready_addresses, Some(true));
    }

    #[test]
    fn pdb_keeps_a_quorum_majority() {
        // 5 nodes → min_available 3 (tolerate losing 2); 3 → 2; 1 → 1.
        for (n, want) in [(1, 1), (3, 2), (5, 3)] {
            let pdb = pod_disruption_budget(&cluster("db", ClusterMode::Cluster, n));
            let min = pdb.spec.unwrap().min_available.unwrap();
            assert_eq!(min, IntOrString::Int(want), "n={n}");
        }
    }

    #[test]
    fn stateless_mode_is_a_deployment_with_hpa_and_ephemeral_data() {
        let brc = cluster("web", ClusterMode::Stateless, 2);
        let dep = deployment(&brc);
        let tspec = dep.spec.unwrap().template.spec.unwrap();
        // No PVC; data is ephemeral (state is in the shared KV).
        let data_vol = tspec
            .volumes
            .unwrap()
            .into_iter()
            .find(|v| v.name == "data")
            .unwrap();
        assert!(data_vol.empty_dir.is_some());
        let hpa = hpa(&brc);
        let hspec = hpa.spec.unwrap();
        assert_eq!(hspec.scale_target_ref.kind, "Deployment");
        assert_eq!(hspec.min_replicas, Some(2));
        assert_eq!(hspec.max_replicas, 8);
    }

    #[test]
    fn config_map_carries_posture_and_bind() {
        let mut brc = cluster("db", ClusterMode::Cluster, 3);
        brc.spec.posture = Some("single-tenant".to_string());
        let cm = config_map(&brc);
        let cfg = &cm.data.unwrap()["boatramp.cfg"];
        assert!(cfg.contains("profile: \"single-tenant\""));
        assert!(cfg.contains("0.0.0.0:8080"));
    }

    #[test]
    fn cluster_config_and_auth_are_wired_and_parse() {
        let mut brc = cluster("db", ClusterMode::Cluster, 3);
        brc.spec.root_pubkey = Some("es256:03a1".to_string());
        brc.spec.auth_secret = Some("db-auth".to_string());
        // The rendered config carries a `[cluster]` section + the root anchor and
        // round-trips through the real `serve` loader (so pods enter cluster mode).
        let cfg = &config_map(&brc).data.unwrap()["boatramp.cfg"];
        let parsed = crate::config::ServerConfig::parse(cfg).expect("valid boatramp.cfg");
        assert!(parsed.cluster.is_some(), "cluster mode config present");
        assert_eq!(
            parsed.serve.unwrap().auth_root_public_key.as_deref(),
            Some("es256:03a1")
        );

        // The StatefulSet pod is wired to sign (root private key) + advertise its
        // own dialable DNS as the mesh address.
        let sts = stateful_set(&brc, 0);
        let env = sts
            .spec
            .unwrap()
            .template
            .spec
            .unwrap()
            .containers
            .remove(0)
            .env
            .unwrap();
        let names: Vec<&str> = env.iter().map(|e| e.name.as_str()).collect();
        assert!(names.contains(&"BOATRAMP_AUTH_ROOT_PRIVATE_KEY"));
        assert!(names.contains(&"BOATRAMP_BOOTSTRAP_SECRET"));
        let advertise = env
            .iter()
            .find(|e| e.name == "BOATRAMP_CLUSTER_ADVERTISE_ADDR")
            .and_then(|e| e.value.as_deref())
            .unwrap();
        assert_eq!(
            advertise,
            "https://$(BOATRAMP_POD_NAME).db-headless.$(POD_NAMESPACE).svc:7000"
        );
    }
}