kaniop-operator 0.16.4

mod domain_appearance;
mod gateway;
mod ingress;
pub mod mail_sender;
pub mod secret;
mod service;
pub mod statefulset;
mod status;
pub mod transport;

use self::domain_appearance::reconcile_domain_appearance;
use self::mail_sender::{
    cleanup_mail_sender_in_kanidm, cleanup_mail_sender_resources, reconcile_mail_sender,
};
use super::controller::{CONTROLLER_ID, context::Context};

use self::gateway::GatewayExt;
use self::ingress::IngressExt;
use self::secret::{SECRET_TYPE_LABEL, SecretExt, SecretType};
use self::service::ServiceExt;
use self::statefulset::{
    StatefulSetApplyStrategy, StatefulSetExt, classify_statefulset_change,
    preserve_defaulted_statefulset_fields,
};
use self::status::StatusExt;
use self::status::{is_kanidm_available, is_kanidm_initialized};
use self::transport::resolve_backup_config;

use crate::controller::context::KubeOperations;
use crate::controller::{INSTANCE_LABEL, MANAGED_BY_LABEL, NAME_LABEL};
use crate::kanidm::crd::{
    Kanidm, KanidmReplicaState, KanidmStatus, KanidmUpgradeCheckResult, VersionCompatibilityResult,
};
use crate::kanidm::reconcile::statefulset::{REPLICA_LABEL, TLS_SECRET_HASH_ANNOTATION};
use crate::telemetry;

use kaniop_backup_core::crd::{KanidmBackupRepository, KanidmBackupSchedule};
use kaniop_k8s_util::client::get_output;
use kaniop_k8s_util::error::{Error, Result};
use kaniop_k8s_util::parse::parse_semver;
use kaniop_k8s_util::resources::{get_image_tag, hash_secret_data};

use std::collections::{BTreeMap, BTreeSet};
use std::fmt::Debug;
use std::sync::{Arc, LazyLock};

use futures::future::{TryJoinAll, join_all, try_join_all};
use futures::stream::{self, StreamExt};
use futures::try_join;
use k8s_openapi::NamespaceResourceScope;
use k8s_openapi::api::apps::v1::StatefulSet;
use k8s_openapi::api::core::v1::{Pod, Secret, Service};
use kube::Resource;
use kube::ResourceExt;
use kube::api::{Api, AttachParams, Patch, PatchParams, PostParams};
use kube::runtime::controller::Action;
use kube::runtime::events::{Event, EventType};
use kube::runtime::finalizer::{Error as FinalizerError, Event as Finalizer, finalizer};
use serde::{Deserialize, Serialize};
use tokio::time::{Duration, sleep};
use tracing::{Span, debug, field, info, instrument, trace, warn};

const POD_READY_WAIT_TIMEOUT_SECONDS: u64 = 30;
const POD_READY_POLL_INTERVAL_SECONDS: u64 = 2;
const STS_ROLLOUT_WAIT_TIMEOUT_SECONDS: u64 = 180;
const STS_ROLLOUT_POLL_INTERVAL_SECONDS: u64 = 2;
const STS_DELETE_WAIT_TIMEOUT_SECONDS: u64 = 30;
const STS_DELETE_POLL_INTERVAL_MILLIS: u64 = 500;
const CERT_SHOW_RETRY_ATTEMPTS: u32 = 6;
const CERT_SHOW_INITIAL_DELAY_SECONDS: u64 = 15;
const CERT_SHOW_RETRY_DELAY_SECONDS: u64 = 15;

pub const CLUSTER_LABEL: &str = "kanidm.kaniop.rs/cluster";
const KANIDM_OPERATOR_NAME: &str = "kanidms.kaniop.rs";
pub static KANIDM_FINALIZER: &str = "kanidms.kaniop.rs/finalizer";

const DEFAULT_RECONCILE_INTERVAL: Duration = Duration::from_secs(24 * 60 * 60);

static LABELS: LazyLock<BTreeMap<String, String>> = LazyLock::new(|| {
    BTreeMap::from([
        (NAME_LABEL.to_string(), "kanidm".to_string()),
        (
            MANAGED_BY_LABEL.to_string(),
            format!("kaniop-{CONTROLLER_ID}"),
        ),
    ])
});

async fn restart_statefulsets_after_replica_secret_update(
    ctx: Arc<Context>,
    sts_api: &Api<StatefulSet>,
    namespace: &str,
    statefulset_names: &[String],
    sequential: bool,
) -> Result<()> {
    // Replica cert secrets are consumed as pod env vars and rendered into config at startup.
    // Pods must restart after cert secret changes or replication keeps stale TLS material.
    let restart_futures = statefulset_names
        .iter()
        .map(|sts_name| sts_api.restart(sts_name));

    if sequential {
        let results: Vec<_> = stream::iter(restart_futures).then(|f| f).collect().await;
        if let Some(first_err) = results.into_iter().find_map(|r| r.err()) {
            return Err(Error::kube_error(
                "restart",
                "StatefulSet",
                namespace,
                "replica certificate secret update statefulsets",
                first_err,
            ));
        }
    } else {
        let results: Vec<_> = join_all(restart_futures).await;
        if let Some(first_err) = results.into_iter().find_map(|r| r.err()) {
            return Err(Error::kube_error(
                "restart",
                "StatefulSet",
                namespace,
                "replica certificate secret update statefulsets",
                first_err,
            ));
        }
    }

    let rollout_futures = statefulset_names
        .iter()
        .map(|sts_name| wait_for_sts_rollout(ctx.kaniop_ctx.client.clone(), namespace, sts_name));
    try_join_all(rollout_futures).await?;

    Ok(())
}

pub async fn reconcile_admins_secret(
    kanidm: Arc<Kanidm>,
    ctx: Arc<Context>,
    status: &KanidmStatus,
) -> Result<bool> {
    if is_kanidm_available(status.clone()) && !is_kanidm_initialized(status.clone()) {
        let admins_secret = kanidm.generate_admins_secret(ctx.clone()).await?;
        kanidm.patch(&ctx, admins_secret).await?;
        return Ok(true);
    }
    Ok(false)
}

pub async fn reconcile_replication_secrets(
    kanidm: Arc<Kanidm>,
    ctx: Arc<Context>,
    status: &KanidmStatus,
) -> Result<bool> {
    let mut changed = false;
    let expected_secret_names = if kanidm.is_replication_enabled() {
        status
            .replica_statuses
            .iter()
            .map(|rs| kanidm.replica_secret_name(&rs.pod_name))
            .collect::<Vec<_>>()
    } else {
        Vec::new()
    };
    let replica_cert_label =
        serde_plain::to_string(&SecretType::ReplicaCert).expect("SecretType serializes");
    let deprecated_secrets = ctx
        .stores
        .secret_store
        .state()
        .into_iter()
        .filter(|secret| {
            secret.namespace() == kanidm.namespace()
                && kanidm.admins_secret_name() != secret.name_any()
                && !expected_secret_names.contains(&secret.name_any())
                && secret.metadata.labels.as_ref().is_some_and(|l| {
                    l.get(SECRET_TYPE_LABEL) == Some(&replica_cert_label)
                        && l.get(CLUSTER_LABEL) == Some(&kanidm.name_any())
                })
        })
        .collect::<Vec<_>>();
    if !deprecated_secrets.is_empty() {
        changed = true;
    }
    let secret_delete_future = deprecated_secrets
        .iter()
        .map(|secret| kanidm.delete(&ctx, secret.as_ref()))
        .collect::<TryJoinAll<_>>();
    try_join!(secret_delete_future)?;

    if kanidm.is_replication_enabled() {
        let has_single_replica = kanidm.spec.replica_groups.iter().any(|rg| rg.replicas == 1);
        let statefulset_names = status
            .replica_statuses
            .iter()
            .map(|rs| rs.statefulset_name.clone())
            .collect::<BTreeSet<_>>()
            .into_iter()
            .collect::<Vec<_>>();

        let has_pending = status
            .replica_statuses
            .iter()
            .any(|rs| rs.state == KanidmReplicaState::Pending);

        let sts_api =
            Api::<StatefulSet>::namespaced(ctx.kaniop_ctx.client.clone(), &kanidm.get_namespace());

        if has_pending {
            changed = true;
            stream::iter(
                status
                    .replica_statuses
                    .iter()
                    .filter(|rs| rs.state == KanidmReplicaState::Pending),
            )
            .then(|rs| async {
                let secret = kanidm
                    .generate_replica_secret(ctx.clone(), &rs.pod_name)
                    .await?;
                kanidm.patch(&ctx, secret).await
            })
            .collect::<Vec<_>>()
            .await
            .into_iter()
            .collect::<Result<Vec<_>>>()?;

            let restart_futures = status
                .replica_statuses
                .iter()
                .map(|rs| sts_api.restart(&rs.statefulset_name));

            if has_single_replica {
                let results: Vec<_> = stream::iter(restart_futures).then(|f| f).collect().await;
                if let Some(first_err) = results.into_iter().find_map(|r| r.err()) {
                    return Err(Error::kube_error(
                        "restart",
                        "StatefulSet",
                        kanidm.get_namespace(),
                        "pending replica statefulsets",
                        first_err,
                    ));
                }
            } else {
                let results: Vec<_> = join_all(restart_futures).await;
                if let Some(first_err) = results.into_iter().find_map(|r| r.err()) {
                    return Err(Error::kube_error(
                        "restart",
                        "StatefulSet",
                        kanidm.get_namespace(),
                        "pending replica statefulsets",
                        first_err,
                    ));
                }
            }
        }

        let has_certificate_host_invalid = status
            .replica_statuses
            .iter()
            .any(|rs| rs.state == KanidmReplicaState::CertificateHostInvalid);

        let has_certificate_expiring = status
            .replica_statuses
            .iter()
            .any(|rs| rs.state == KanidmReplicaState::CertificateExpiring);

        let has_non_ready_replicas = status
            .replica_statuses
            .iter()
            .any(|rs| rs.state != KanidmReplicaState::Ready);

        let host_invalid_replicas: Vec<_> = status
            .replica_statuses
            .iter()
            .filter(|rs| rs.state == KanidmReplicaState::CertificateHostInvalid)
            .collect();

        let cert_expiring_replicas: Vec<_> = status
            .replica_statuses
            .iter()
            .filter(|rs| rs.state == KanidmReplicaState::CertificateExpiring)
            .collect();

        if !host_invalid_replicas.is_empty() {
            changed = true;
            let restart_futures = host_invalid_replicas
                .iter()
                .map(|rs| sts_api.restart(&rs.statefulset_name));

            if has_single_replica {
                let results: Vec<_> = stream::iter(restart_futures).then(|f| f).collect().await;
                if let Some(first_err) = results.into_iter().find_map(|r| r.err()) {
                    return Err(Error::kube_error(
                        "restart",
                        "StatefulSet",
                        kanidm.get_namespace(),
                        "certificate host invalid statefulsets",
                        first_err,
                    ));
                }
            } else {
                let results: Vec<_> = join_all(restart_futures).await;
                if let Some(first_err) = results.into_iter().find_map(|r| r.err()) {
                    return Err(Error::kube_error(
                        "restart",
                        "StatefulSet",
                        kanidm.get_namespace(),
                        "certificate host invalid statefulsets",
                        first_err,
                    ));
                }
            }

            let restarted_sts_names: Vec<_> = host_invalid_replicas
                .iter()
                .map(|rs| rs.statefulset_name.clone())
                .collect::<BTreeSet<_>>()
                .into_iter()
                .collect();

            let namespace = kanidm.get_namespace();
            let rollout_futures = restarted_sts_names.iter().map(|sts_name| {
                wait_for_sts_rollout(ctx.kaniop_ctx.client.clone(), &namespace, sts_name)
            });
            try_join_all(rollout_futures).await?;

            let renew_futures = host_invalid_replicas
                .iter()
                .map(|rs| kanidm.renew_replica_cert(ctx.clone(), &rs.pod_name));
            let results: Vec<_> = join_all(renew_futures).await;
            if let Some(first_err) = results.into_iter().find_map(|r| r.err()) {
                return Err(first_err);
            }

            sleep(Duration::from_secs(CERT_SHOW_INITIAL_DELAY_SECONDS)).await;

            let cert_futures = host_invalid_replicas.iter().map(|rs| async {
                let cert = show_replica_cert_with_retries(
                    &kanidm,
                    ctx.clone(),
                    &rs.pod_name,
                    CERT_SHOW_RETRY_ATTEMPTS,
                )
                .await?;
                let secret = kanidm.build_replica_secret(cert, &rs.pod_name);
                kanidm.patch(&ctx, secret.clone()).await
            });
            try_join_all(cert_futures).await?;

            restart_statefulsets_after_replica_secret_update(
                ctx.clone(),
                &sts_api,
                &kanidm.get_namespace(),
                &statefulset_names,
                has_single_replica,
            )
            .await?;

            ctx.kaniop_ctx.release_kanidm_clients(&kanidm).await;
        }

        if !cert_expiring_replicas.is_empty() {
            changed = true;
            let renew_futures = cert_expiring_replicas
                .iter()
                .map(|rs| kanidm.renew_replica_cert(ctx.clone(), &rs.pod_name));
            let results: Vec<_> = join_all(renew_futures).await;
            if let Some(first_err) = results.into_iter().find_map(|r| r.err()) {
                return Err(first_err);
            }

            sleep(Duration::from_secs(CERT_SHOW_INITIAL_DELAY_SECONDS)).await;

            let cert_futures = cert_expiring_replicas.iter().map(|rs| async {
                let cert = show_replica_cert_with_retries(
                    &kanidm,
                    ctx.clone(),
                    &rs.pod_name,
                    CERT_SHOW_RETRY_ATTEMPTS,
                )
                .await?;
                let secret = kanidm.build_replica_secret(cert, &rs.pod_name);
                kanidm.patch(&ctx, secret.clone()).await
            });
            try_join_all(cert_futures).await?;

            restart_statefulsets_after_replica_secret_update(
                ctx.clone(),
                &sts_api,
                &kanidm.get_namespace(),
                &statefulset_names,
                has_single_replica,
            )
            .await?;

            ctx.kaniop_ctx.release_kanidm_clients(&kanidm).await;
        }

        if has_non_ready_replicas
            && !has_pending
            && !has_certificate_host_invalid
            && !has_certificate_expiring
        {
            changed = true;
            let has_single_replica = kanidm.spec.replica_groups.iter().any(|rg| rg.replicas == 1);

            let restart_futures = status
                .replica_statuses
                .iter()
                .map(|rs| sts_api.restart(&rs.statefulset_name));

            if has_single_replica {
                let results: Vec<_> = stream::iter(restart_futures).then(|f| f).collect().await;
                if let Some(first_err) = results.into_iter().find_map(|r| r.err()) {
                    return Err(Error::kube_error(
                        "restart",
                        "StatefulSet",
                        kanidm.get_namespace(),
                        "replica statefulsets",
                        first_err,
                    ));
                }
            } else {
                let results: Vec<_> = join_all(restart_futures).await;
                if let Some(first_err) = results.into_iter().find_map(|r| r.err()) {
                    return Err(Error::kube_error(
                        "restart",
                        "StatefulSet",
                        kanidm.get_namespace(),
                        "replica statefulsets",
                        first_err,
                    ));
                }
            }
        }
    }

    Ok(changed)
}

#[instrument(skip(ctx, kanidm))]
pub async fn reconcile_kanidm(kanidm: Arc<Kanidm>, ctx: Arc<Context>) -> Result<(Action, bool)> {
    let trace_id = telemetry::get_trace_id();
    Span::current().record("trace_id", field::display(&trace_id));
    let _timer = ctx
        .kaniop_ctx
        .metrics
        .reconcile_count_and_measure(&trace_id);
    info!("reconciling Kanidm");

    let status = kanidm.update_status(ctx.clone()).await.map_err(|e| {
        debug!(%e, "failed to reconcile status");
        ctx.kaniop_ctx.metrics.status_update_errors_inc();
        e
    })?;
    let kanidm_api: Api<Kanidm> =
        Api::namespaced(ctx.kaniop_ctx.client.clone(), &kanidm.get_namespace());
    let outcome = std::sync::Arc::new(std::sync::atomic::AtomicBool::new(false));
    let outcome_clone = outcome.clone();
    let action = finalizer(&kanidm_api, KANIDM_FINALIZER, kanidm, move |event| {
        let outcome = outcome_clone.clone();
        let ctx = ctx.clone();
        let status = status.clone();
        async move {
            match event {
                Finalizer::Apply(kanidm) => {
                    let (action, changed) = reconcile(kanidm, ctx, status).await?;
                    outcome.store(changed, std::sync::atomic::Ordering::Relaxed);
                    Ok(action)
                }
                Finalizer::Cleanup(kanidm) => {
                    let (action, changed) = cleanup(kanidm, ctx).await?;
                    outcome.store(changed, std::sync::atomic::Ordering::Relaxed);
                    Ok(action)
                }
            }
        }
    })
    .await
    .or_else(|e| match e {
        FinalizerError::RemoveFinalizer(kube::Error::Api(ae)) if ae.code == 404 => {
            debug!("resource already removed during finalizer cleanup");
            Ok(Action::requeue(DEFAULT_RECONCILE_INTERVAL))
        }
        _ => Err(Error::FinalizerError(
            "failed on kanidm account finalizer".to_string(),
            Box::new(e),
        )),
    })?;
    let changed = outcome.load(std::sync::atomic::Ordering::Relaxed);
    Ok((action, changed))
}

/// Hash of the TLS secret content, injected as a pod template annotation so
/// certificate renewals roll the statefulsets. `None` if the secret does not
/// exist yet.
async fn get_tls_secret_hash(kanidm: &Kanidm, ctx: &Arc<Context>) -> Result<Option<String>> {
    let secret_name = kanidm.effective_tls_secret_name();
    let secret_api =
        Api::<Secret>::namespaced(ctx.kaniop_ctx.client.clone(), &kanidm.get_namespace());
    let secret = secret_api
        .get_opt(&secret_name)
        .await
        .map_err(|e| Error::kube_error("get", "Secret", kanidm.get_namespace(), &secret_name, e))?;
    Ok(secret.as_ref().map(hash_secret_data))
}

/// Previous TLS secret hash annotation from the existing statefulsets.
///
/// Used when the TLS secret is missing: dropping the annotation would roll
/// the pods onto a missing volume and leave them stuck in ContainerCreating.
fn previous_tls_secret_hash(
    kanidm: &Kanidm,
    stateful_sets: impl IntoIterator<Item = Arc<StatefulSet>>,
) -> Option<String> {
    stateful_sets
        .into_iter()
        .filter(|sts| {
            sts.namespace() == kanidm.namespace()
                && sts
                    .metadata
                    .labels
                    .as_ref()
                    .is_some_and(|l| l.get(CLUSTER_LABEL) == Some(&kanidm.name_any()))
        })
        .find_map(|sts| {
            sts.spec
                .as_ref()?
                .template
                .metadata
                .as_ref()?
                .annotations
                .as_ref()?
                .get(TLS_SECRET_HASH_ANNOTATION)
                .cloned()
        })
}

fn is_headless_service(service: &Service) -> bool {
    service
        .spec
        .as_ref()
        .and_then(|spec| spec.cluster_ip.as_deref())
        == Some("None")
}

fn preserve_defaulted_service_fields(desired: &mut Service, current: &Service) {
    if is_headless_service(desired) != is_headless_service(current) {
        return;
    }
    let (Some(desired_spec), Some(current_spec)) = (desired.spec.as_mut(), current.spec.as_ref())
    else {
        return;
    };
    if desired_spec.cluster_ip.is_none() {
        desired_spec.cluster_ip = current_spec.cluster_ip.clone();
    }
    if desired_spec.cluster_ips.is_none() {
        desired_spec.cluster_ips = current_spec.cluster_ips.clone();
    }
}

async fn wait_for_statefulset_deletion(
    client: kube::Client,
    namespace: &str,
    name: &str,
) -> Result<()> {
    let api = Api::<StatefulSet>::namespaced(client, namespace);
    let start = std::time::Instant::now();
    let timeout = Duration::from_secs(STS_DELETE_WAIT_TIMEOUT_SECONDS);
    let poll_interval = Duration::from_millis(STS_DELETE_POLL_INTERVAL_MILLIS);

    loop {
        match api.get_opt(name).await {
            Ok(None) => return Ok(()),
            Ok(Some(_)) => {}
            Err(e) => {
                return Err(Error::kube_error(
                    "wait for deletion of",
                    "StatefulSet",
                    namespace,
                    name,
                    e,
                ));
            }
        }

        if start.elapsed() >= timeout {
            return Err(Error::ReceiveOutput(format!(
                "StatefulSet {namespace}/{name} was not deleted after {STS_DELETE_WAIT_TIMEOUT_SECONDS}s"
            )));
        }
        sleep(poll_interval).await;
    }
}

async fn reconcile_service(
    kanidm: &Kanidm,
    ctx: &Context,
    mut desired: Service,
) -> Result<Service> {
    let namespace = kanidm.get_namespace();
    let name = desired.name_any();
    let desired_headless = is_headless_service(&desired);

    if let Some(cached) = ctx
        .stores
        .service_store
        .find(|current| current.namespace() == kanidm.namespace() && current.name_any() == name)
    {
        if is_headless_service(cached.as_ref()) == desired_headless {
            preserve_defaulted_service_fields(&mut desired, cached.as_ref());
        } else {
            let service_api = Api::<Service>::namespaced(ctx.kaniop_ctx.client.clone(), &namespace);
            let current = service_api
                .get_opt(&name)
                .await
                .map_err(|e| Error::kube_error("get", "Service", &namespace, &name, e))?;

            if let Some(current) = current {
                let current_headless = is_headless_service(&current);
                if current_headless != desired_headless {
                    info!(
                        resource.name = &name,
                        resource.namespace = &namespace,
                        current_headless,
                        desired_headless,
                        "recreating Service to change headless mode"
                    );
                    kanidm.delete(ctx, &current).await?;
                    ctx.kaniop_ctx
                        .metrics
                        .reconcile_deploy_delete_create_inc("Service", "headless_transition");
                    wait_for_service_deletion(ctx.kaniop_ctx.client.clone(), &namespace, &name)
                        .await?;
                } else {
                    preserve_defaulted_service_fields(&mut desired, &current);
                }
            }
        }
    }

    kanidm.patch(ctx, desired).await
}

async fn wait_for_service_deletion(
    client: kube::Client,
    namespace: &str,
    name: &str,
) -> Result<()> {
    let api = Api::<Service>::namespaced(client, namespace);
    let start = std::time::Instant::now();
    let timeout = Duration::from_secs(STS_DELETE_WAIT_TIMEOUT_SECONDS);
    let poll_interval = Duration::from_millis(STS_DELETE_POLL_INTERVAL_MILLIS);

    loop {
        match api.get_opt(name).await {
            Ok(None) => return Ok(()),
            Ok(Some(_)) => {}
            Err(e) => {
                return Err(Error::kube_error(
                    "wait for deletion of",
                    "Service",
                    namespace,
                    name,
                    e,
                ));
            }
        }

        if start.elapsed() >= timeout {
            return Err(Error::ReceiveOutput(format!(
                "Service {namespace}/{name} was not deleted after {STS_DELETE_WAIT_TIMEOUT_SECONDS}s"
            )));
        }
        sleep(poll_interval).await;
    }
}

async fn validate_statefulset_replacement(
    api: &Api<StatefulSet>,
    desired: &StatefulSet,
    namespace: &str,
    name: &str,
) -> Result<()> {
    let mut candidate = desired.clone();
    candidate.metadata.name = None;
    candidate.metadata.generate_name = Some("kaniop-preflight-".to_string());
    candidate.metadata.namespace = None;
    candidate.metadata.resource_version = None;
    candidate.metadata.uid = None;
    candidate.metadata.creation_timestamp = None;
    candidate.metadata.generation = None;
    candidate.metadata.managed_fields = None;
    candidate.metadata.deletion_timestamp = None;
    candidate.metadata.deletion_grace_period_seconds = None;
    candidate.status = None;

    api.create(
        &PostParams {
            dry_run: true,
            ..Default::default()
        },
        &candidate,
    )
    .await
    .map(|_| ())
    .map_err(|e| {
        Error::kube_error(
            "validate replacement for",
            "StatefulSet",
            namespace,
            name,
            e,
        )
    })
}

async fn reconcile_statefulset(
    kanidm: Arc<Kanidm>,
    ctx: Arc<Context>,
    desired: StatefulSet,
) -> Result<StatefulSet> {
    let namespace = kanidm.namespace().unwrap();
    let name = desired.name_any();
    let cached_current = ctx.stores.stateful_set_store.find(|current| {
        current.namespace().as_deref() == Some(namespace.as_str()) && current.name_any() == name
    });

    // On a cold reflector cache, keep the normal create/update path GET-free. If the optimistic
    // apply fails with HTTP 422, the object may already exist with an immutable conflict that the
    // cache has not observed yet. Only 422 errors enter the live classification/recreate recovery
    // path; 429, 5xx, 403, and transport errors are returned immediately and non-destructively.
    let cache_miss_422_error = if let Some(cached_current) = cached_current {
        let mut cached_desired = desired.clone();
        preserve_defaulted_statefulset_fields(&mut cached_desired, cached_current.as_ref());
        if classify_statefulset_change(cached_current.as_ref(), &cached_desired)
            == StatefulSetApplyStrategy::Apply
        {
            return kanidm.apply(&ctx, cached_desired).await;
        }
        None
    } else {
        match kanidm.apply(&ctx, desired.clone()).await {
            Ok(applied) => return Ok(applied),
            Err(error) => {
                // Only HTTP 422 can indicate an immutable-field conflict that is safe to
                // recover from via delete/recreate. Arbitrary failures must not trigger
                // destruction.
                if is_kube_422(&error) {
                    Some(error)
                } else {
                    return Err(error);
                }
            }
        }
    };

    // Never make a destructive decision from reflector state or an SSA error alone. Re-read the
    // live object and classify it immediately before deletion; another actor or an earlier
    // reconcile may already have converged the immutable fields while the cache was catching up.
    let api = Api::<StatefulSet>::namespaced(ctx.kaniop_ctx.client.clone(), &namespace);
    let Some(live_current) = api
        .get_opt(&name)
        .await
        .map_err(|e| Error::kube_error("get", "StatefulSet", &namespace, &name, e))?
    else {
        return match cache_miss_422_error {
            Some(error) => Err(error),
            None => kanidm.apply(&ctx, desired).await,
        };
    };

    // Start from the original desired object again. Defaults copied from a stale reflector entry
    // must not survive into the live re-check and become the reason for a recreation.
    let mut live_desired = desired.clone();
    preserve_defaulted_statefulset_fields(&mut live_desired, &live_current);
    match classify_statefulset_change(&live_current, &live_desired) {
        StatefulSetApplyStrategy::Apply => match cache_miss_422_error {
            Some(error) => Err(error),
            None => kanidm.apply(&ctx, live_desired).await,
        },
        StatefulSetApplyStrategy::Recreate { .. } => {
            // An update 422 can combine an immutable conflict with independently-invalid desired
            // state. Validate the replacement as a dry-run CREATE before deleting the live object,
            // so create-time validation runs without the existing object's immutable-update check.
            validate_statefulset_replacement(&api, &live_desired, &namespace, &name).await?;

            // Re-read and reclassify immediately before destruction. Another actor or an earlier
            // reconcile may have changed the live immutable fields or PVC-retention semantics.
            let Some(latest_current) = api.get_opt(&name).await.map_err(|e| {
                Error::kube_error(
                    "re-read before deletion",
                    "StatefulSet",
                    &namespace,
                    &name,
                    e,
                )
            })?
            else {
                return kanidm.apply(&ctx, live_desired).await;
            };
            let mut latest_desired = live_desired.clone();
            preserve_defaulted_statefulset_fields(&mut latest_desired, &latest_current);

            match classify_statefulset_change(&latest_current, &latest_desired) {
                StatefulSetApplyStrategy::Apply => kanidm.apply(&ctx, latest_desired).await,
                StatefulSetApplyStrategy::Recreate { immutable_fields } => {
                    // Validate the replacement StatefulSet via a dry-run CREATE before deleting
                    // the live object. This prevents deleting on a mixed 422 (immutable conflict
                    // + independently-invalid desired state) where the replacement would still
                    // be rejected by the API.
                    validate_statefulset_replacement(&api, &latest_desired, &namespace, &name)
                        .await?;

                    info!(resource.name = %name, resource.namespace = %namespace, ?immutable_fields, "recreating StatefulSet because immutable fields changed");
                    kanidm.delete(&ctx, &latest_current).await?;
                    ctx.kaniop_ctx
                        .metrics
                        .reconcile_deploy_delete_create_inc("StatefulSet", "immutable_spec");
                    wait_for_statefulset_deletion(ctx.kaniop_ctx.client.clone(), &namespace, &name)
                        .await?;
                    kanidm.apply(&ctx, latest_desired).await
                }
            }
        }
    }
}

/// Return true only when the error structurally wraps a Kubernetes HTTP 422 response.
fn is_kube_422(error: &Error) -> bool {
    matches!(
        error,
        Error::KubeError(_, cause)
            if matches!(cause.as_ref(), kube::Error::Api(status) if status.code == 422)
    )
}

#[cfg(test)]
mod statefulset_recreate_recovery_tests {
    use super::{Error, is_kube_422};

    fn kube_api_error(code: u16) -> Error {
        Error::KubeError(
            "test error".to_string(),
            Box::new(kube::Error::Api(Box::new(kube::error::Status {
                code,
                message: "test".to_string(),
                reason: "test".to_string(),
                ..Default::default()
            }))),
        )
    }

    #[test]
    fn recreate_recovery_accepts_only_422() {
        assert!(is_kube_422(&kube_api_error(422)));
        assert!(!is_kube_422(&kube_api_error(403)));
        assert!(!is_kube_422(&kube_api_error(429)));
        assert!(!is_kube_422(&kube_api_error(500)));
    }

    #[test]
    fn recreate_recovery_rejects_429_too_many_requests() {
        assert!(!is_kube_422(&kube_api_error(429)));
    }

    #[test]
    fn recreate_recovery_rejects_5xx_server_errors() {
        assert!(!is_kube_422(&kube_api_error(500)));
        assert!(!is_kube_422(&kube_api_error(502)));
        assert!(!is_kube_422(&kube_api_error(503)));
        assert!(!is_kube_422(&kube_api_error(504)));
    }

    #[test]
    fn recreate_recovery_rejects_400_and_403() {
        assert!(!is_kube_422(&kube_api_error(400)));
        assert!(!is_kube_422(&kube_api_error(403)));
    }
}

async fn reconcile(
    kanidm: Arc<Kanidm>,
    ctx: Arc<Context>,
    status: KanidmStatus,
) -> Result<(Action, bool)> {
    if kanidm
        .annotations()
        .contains_key(crate::kanidm::restore::RESTORE_ANNOTATION)
    {
        ctx.kaniop_ctx.release_kanidm_clients(&kanidm).await;
        return Ok((Action::requeue(Duration::from_secs(5)), false));
    }

    let mut changed = false;
    let admin_secret_future = reconcile_admins_secret(kanidm.clone(), ctx.clone(), &status);
    let replication_secret_futures =
        reconcile_replication_secrets(kanidm.clone(), ctx.clone(), &status);

    let sts_to_delete = {
        let expected_sts_names = kanidm
            .spec
            .replica_groups
            .iter()
            .map(|rg| kanidm.statefulset_name(&rg.name))
            .collect::<Vec<_>>();
        ctx.stores
            .stateful_set_store
            .state()
            .into_iter()
            .filter(|sts| {
                sts.namespace() == kanidm.namespace()
                    && sts
                        .metadata
                        .labels
                        .as_ref()
                        .is_some_and(|l| l.get(CLUSTER_LABEL) == Some(&kanidm.name_any()))
                    && !expected_sts_names.contains(&sts.name_any())
            })
            .collect::<Vec<_>>()
    };
    if !sts_to_delete.is_empty() {
        changed = true;
    }
    let sts_delete_futures = sts_to_delete
        .iter()
        .map(|sts| kanidm.delete(&ctx, sts.as_ref()))
        .collect::<TryJoinAll<_>>();

    let sts_futures = match kanidm.is_updatable(&status) {
        true => {
            let tls_secret_hash = get_tls_secret_hash(&kanidm, &ctx).await?.or_else(|| {
                previous_tls_secret_hash(&kanidm, ctx.stores.stateful_set_store.state())
            });

            let namespace = kanidm.get_namespace();
            let schedule_api: Api<KanidmBackupSchedule> =
                Api::namespaced(ctx.kaniop_ctx.client.clone(), &namespace);
            let schedules = schedule_api
                .list(&Default::default())
                .await
                .map_err(|e| {
                    Error::KubeError(
                        "failed to list KanidmBackupSchedules for backup configuration resolution"
                            .to_string(),
                        Box::new(e),
                    )
                })?
                .items;

            let repo_api: Api<KanidmBackupRepository> =
                Api::namespaced(ctx.kaniop_ctx.client.clone(), &namespace);
            let repositories = repo_api
                .list(&Default::default())
                .await
                .map_err(|e| {
                    Error::KubeError(
                        "failed to list KanidmBackupRepositories for backup configuration resolution"
                            .to_string(),
                        Box::new(e),
                    )
                })?
                .items;
            let backup_config = resolve_backup_config(&kanidm, &schedules, &repositories);

            kanidm
                .spec
                .replica_groups
                .iter()
                .map(|rg| {
                    let sts = kanidm.create_statefulset(
                        rg,
                        tls_secret_hash.as_deref(),
                        backup_config.as_ref(),
                    )?;
                    Ok(reconcile_statefulset(kanidm.clone(), ctx.clone(), sts))
                })
                .collect::<Result<TryJoinAll<_>, _>>()?
        }
        false => {
            let note = match status.version.as_ref() {
                Some(v) if v.compatibility_result == VersionCompatibilityResult::Incompatible => {
                    format!(
                        "Version change blocked: image version {} is not compatible with operator (uses Kanidm client v{}). Override with `.spec.disableUpgradeChecks: true` or use a compatible version.",
                        v.image_tag,
                        crate::version::KANIDM_CLIENT_VERSION
                    )
                }
                _ => "Version change blocked: upgrade pre-check failed. Override with `.spec.disableUpgradeCheck: true` or update the resource to retry.".to_string(),
            };
            let _ignore_error = ctx
                .kaniop_ctx
                .recorder
                .publish(
                    &Event {
                        type_: EventType::Warning,
                        reason: "UpgradeBlocked".to_string(),
                        note: Some(note),
                        action: "ReconcileStatefulSet".to_string(),
                        secondary: None,
                    },
                    &kanidm.object_ref(&()),
                )
                .await
                .map_err(|e| {
                    warn!(%e, "failed to publish KanidmError event");
                    Error::KubeError("failed to publish event".to_string(), Box::new(e))
                });
            try_join_all([])
        }
    };
    let service_future = reconcile_service(&kanidm, &ctx, kanidm.create_service());

    let deprecated_rg_svcs = {
        let expected_rg_svcs_names = kanidm
            .spec
            .replica_groups
            .iter()
            .filter(|rg| rg.services.is_some())
            .flat_map(|rg| (0..rg.replicas).map(|i| kanidm.replica_group_service_name(&rg.name, i)))
            .collect::<Vec<_>>();
        ctx.stores
            .service_store
            .state()
            .into_iter()
            .filter(|svc| {
                let labels = svc.metadata.labels.as_ref();
                svc.namespace() == kanidm.namespace()
                    && !expected_rg_svcs_names.contains(&svc.name_any())
                    && labels.is_some_and(|l| {
                        l.contains_key(REPLICA_LABEL)
                            && l.get(CLUSTER_LABEL) == Some(&kanidm.name_any())
                    })
            })
            .collect::<Vec<_>>()
    };
    if !deprecated_rg_svcs.is_empty() {
        changed = true;
    }
    let rg_svcs_delete_futures = deprecated_rg_svcs
        .iter()
        .map(|svc| kanidm.delete(&ctx, svc.as_ref()))
        .collect::<TryJoinAll<_>>();

    let rg_services_futures = kanidm
        .spec
        .replica_groups
        .iter()
        .filter(|rg| rg.services.is_some())
        .flat_map(|rg| {
            (0..rg.replicas).map(|i| {
                let mut svc = kanidm.create_replica_group_service(rg, i);
                if let Some(current) = ctx.stores.service_store.find(|current| {
                    current.namespace() == kanidm.namespace()
                        && current.name_any() == svc.name_any()
                }) {
                    preserve_defaulted_service_fields(&mut svc, current.as_ref());
                }
                kanidm.patch(&ctx, svc)
            })
        })
        .collect::<TryJoinAll<_>>();

    let deprecated_ingresses = {
        let names = [
            kanidm.spec.ingress.as_ref().map(|_| kanidm.name_any()),
            kanidm.generate_region_ingress_name(),
        ]
        .into_iter()
        .flatten()
        .collect::<Vec<_>>();
        ctx.stores
            .ingress_store
            .state()
            .into_iter()
            .filter(|ing| {
                ing.namespace() == kanidm.namespace()
                    && !names.contains(&ing.name_any())
                    && ing.metadata.labels == Some(kanidm.generate_labels())
            })
            .collect::<Vec<_>>()
    };
    if !deprecated_ingresses.is_empty() {
        changed = true;
    }
    let ingress_delete_futures = deprecated_ingresses
        .iter()
        .map(|ing| kanidm.delete(&ctx, ing.as_ref()))
        .collect::<TryJoinAll<_>>();

    let ingress_futures = kanidm
        .create_ingress()
        .into_iter()
        .chain(kanidm.create_region_ingress())
        .map(|ingress| kanidm.patch(&ctx, ingress))
        .collect::<TryJoinAll<_>>();

    let deprecated_http_routes = match &ctx.stores.http_route_store {
        Some(store) => {
            let expected_names = kanidm
                .spec
                .gateway
                .as_ref()
                .map(|_| vec![kanidm.name_any()])
                .unwrap_or_default();
            let routes: Vec<_> = store
                .state()
                .into_iter()
                .filter(|route| {
                    route.namespace() == kanidm.namespace()
                        && !expected_names.contains(&route.name_any())
                        && route.metadata.labels == Some(kanidm.generate_labels())
                })
                .collect();
            if !routes.is_empty() {
                changed = true;
            }
            routes
        }
        None => Vec::new(),
    };
    let http_route_delete_futures = deprecated_http_routes
        .iter()
        .map(|route| kanidm.delete(&ctx, route.as_ref()))
        .collect::<TryJoinAll<_>>();

    let http_route_futures = if ctx.stores.http_route_store.is_some() {
        kanidm
            .create_http_route()
            .into_iter()
            .map(|route| kanidm.patch(&ctx, route))
            .collect::<TryJoinAll<_>>()
    } else {
        try_join_all(Vec::new())
    };

    let deprecated_backend_tls_policies = match &ctx.stores.backend_tls_policy_store {
        Some(store) => {
            let expected_names = kanidm
                .spec
                .gateway
                .as_ref()
                .and_then(|g| g.backend_tls_policy.as_ref())
                .map(|_| vec![kanidm.name_any()])
                .unwrap_or_default();
            let policies: Vec<_> = store
                .state()
                .into_iter()
                .filter(|policy| {
                    policy.namespace() == kanidm.namespace()
                        && !expected_names.contains(&policy.name_any())
                        && policy.metadata.labels == Some(kanidm.generate_labels())
                })
                .collect();
            if !policies.is_empty() {
                changed = true;
            }
            policies
        }
        None => Vec::new(),
    };
    let backend_tls_policy_delete_futures = deprecated_backend_tls_policies
        .iter()
        .map(|policy| kanidm.delete(&ctx, policy.as_ref()))
        .collect::<TryJoinAll<_>>();

    let backend_tls_policy_futures = if ctx.stores.backend_tls_policy_store.is_some() {
        kanidm
            .create_backend_tls_policy()
            .into_iter()
            .map(|policy| kanidm.patch(&ctx, policy))
            .collect::<TryJoinAll<_>>()
    } else {
        try_join_all(Vec::new())
    };

    let (_, admin_secret_changed, replication_secrets_changed, _, _, _, _, _, _, _, _, _, _) = try_join!(
        sts_delete_futures,
        admin_secret_future,
        replication_secret_futures,
        sts_futures,
        service_future,
        rg_svcs_delete_futures,
        rg_services_futures,
        ingress_delete_futures,
        ingress_futures,
        http_route_delete_futures,
        http_route_futures,
        backend_tls_policy_delete_futures,
        backend_tls_policy_futures
    )?;

    if admin_secret_changed || replication_secrets_changed {
        changed = true;
    }

    if is_kanidm_available(status.clone()) {
        let namespace = kanidm.namespace().unwrap();
        let name = kanidm.name_any();
        match crate::controller::kanidm::KanidmClients::create_client(
            &namespace,
            &name,
            crate::controller::kanidm::KanidmUser::Admin,
            ctx.kaniop_ctx.client.clone(),
        )
        .await
        {
            Ok(system_client) => {
                if let Err(e) =
                    reconcile_domain_appearance(&kanidm, system_client, &status, ctx.clone()).await
                {
                    warn!(%e, "failed to reconcile domain appearance");
                }
            }
            Err(e) => {
                warn!(%e, "failed to create admin client for domain appearance reconciliation");
            }
        }

        match crate::controller::kanidm::KanidmClients::create_client(
            &namespace,
            &name,
            crate::controller::kanidm::KanidmUser::IdmAdmin,
            ctx.kaniop_ctx.client.clone(),
        )
        .await
        {
            Ok(kanidm_client) => {
                let (mail_sender_status, mail_sender_mutated) =
                    reconcile_mail_sender(&kanidm, kanidm_client.clone(), ctx.clone())
                        .await
                        .unwrap_or_else(|e| {
                            warn!(%e, "failed to reconcile mail sender");
                            (None, false)
                        });
                if mail_sender_mutated {
                    changed = true;
                }
                let current_mail_sender_status =
                    kanidm.status.as_ref().and_then(|s| s.mail_sender.clone());
                if mail_sender_status != current_mail_sender_status {
                    let kanidm_api: Api<Kanidm> =
                        Api::namespaced(ctx.kaniop_ctx.client.clone(), &namespace);
                    let status_patch = serde_json::json!({
                        "status": {
                            "mailSender": mail_sender_status
                        }
                    });
                    if let Err(e) = kanidm_api
                        .patch_status(&name, &PatchParams::default(), &Patch::Merge(&status_patch))
                        .await
                    {
                        warn!(%e, "failed to patch Kanidm/status for mail sender");
                    }
                }
            }
            Err(e) => {
                warn!(%e, "failed to create idm_admin client for mail sender reconciliation");
            }
        }
    }

    Ok((Action::requeue(DEFAULT_RECONCILE_INTERVAL), changed))
}

async fn cleanup(kanidm: Arc<Kanidm>, ctx: Arc<Context>) -> Result<(Action, bool)> {
    debug!("cleanup");

    let mut changed = false;
    let mail_resources_cleaned = cleanup_mail_sender_resources(&kanidm, &ctx).await?;
    if mail_resources_cleaned {
        changed = true;
    }

    let namespace = kanidm.namespace().unwrap();
    let name = kanidm.name_any();

    if let Ok(kanidm_client) = crate::controller::kanidm::KanidmClients::create_client(
        &namespace,
        &name,
        crate::controller::kanidm::KanidmUser::IdmAdmin,
        ctx.kaniop_ctx.client.clone(),
    )
    .await
    {
        let kanidm_cleanup_done =
            cleanup_mail_sender_in_kanidm(&kanidm_client, &name, &ctx.kaniop_ctx.metrics).await?;
        if kanidm_cleanup_done {
            changed = true;
        }
    }

    ctx.kaniop_ctx.release_kanidm_clients(&kanidm).await;
    Ok((Action::requeue(DEFAULT_RECONCILE_INTERVAL), changed))
}

async fn wait_for_sts_rollout(client: kube::Client, namespace: &str, sts_name: &str) -> Result<()> {
    let sts_api = Api::<StatefulSet>::namespaced(client, namespace);
    let start = std::time::Instant::now();
    let timeout = Duration::from_secs(STS_ROLLOUT_WAIT_TIMEOUT_SECONDS);
    let poll_interval = Duration::from_secs(STS_ROLLOUT_POLL_INTERVAL_SECONDS);

    loop {
        let sts = sts_api.get(sts_name).await.map_err(|e| {
            Error::KubeError(
                format!("failed to get StatefulSet {namespace}/{sts_name}"),
                Box::new(e),
            )
        })?;

        if sts.metadata.deletion_timestamp.is_some() {
            return Err(Error::ReceiveOutput(format!(
                "StatefulSet {namespace}/{sts_name} is being deleted"
            )));
        }

        let spec = sts.spec.as_ref().ok_or_else(|| {
            Error::MissingData(format!("StatefulSet {namespace}/{sts_name} has no spec"))
        })?;

        let status = match sts.status {
            Some(ref s) => s,
            None => {
                if start.elapsed() >= timeout {
                    return Err(Error::ReceiveOutput(format!(
                        "StatefulSet {namespace}/{sts_name} has no status after {STS_ROLLOUT_WAIT_TIMEOUT_SECONDS}s"
                    )));
                }
                sleep(poll_interval).await;
                continue;
            }
        };

        let replicas = spec.replicas.unwrap_or(0);

        if replicas == 0 {
            return Ok(());
        }

        let updated_replicas = status.updated_replicas.unwrap_or(0);
        let current_revision = status.current_revision.as_deref().unwrap_or("");
        let update_revision = status.update_revision.as_deref().unwrap_or("");
        let ready_replicas = status.ready_replicas.unwrap_or(0);

        if updated_replicas == replicas
            && current_revision == update_revision
            && ready_replicas == replicas
        {
            return Ok(());
        }

        if start.elapsed() >= timeout {
            return Err(Error::ReceiveOutput(format!(
                "StatefulSet {namespace}/{sts_name} rollout not complete after {STS_ROLLOUT_WAIT_TIMEOUT_SECONDS}s (updated={updated_replicas}/{replicas}, ready={ready_replicas}/{replicas})"
            )));
        }

        sleep(poll_interval).await;
    }
}

async fn show_replica_cert_with_retries(
    kanidm: &Kanidm,
    ctx: Arc<Context>,
    pod_name: &str,
    max_attempts: u32,
) -> std::result::Result<String, Error> {
    let mut last_error = None;
    for attempt in 0..max_attempts {
        if attempt > 0 {
            sleep(Duration::from_secs(CERT_SHOW_RETRY_DELAY_SECONDS)).await;
        }
        match kanidm.show_replica_cert(ctx.clone(), pod_name).await {
            Ok(cert) => return Ok(cert),
            Err(e) => {
                info!(
                    "show-replication-certificate attempt {}/{} failed for {pod_name}: {e}",
                    attempt + 1,
                    max_attempts
                );
                last_error = Some(e);
            }
        }
    }
    Err(last_error.unwrap_or_else(|| {
        Error::ReceiveOutput(format!(
            "failed to get certificate for {pod_name} after {max_attempts} attempts"
        ))
    }))
}

impl Kanidm {
    // Convenience methods that handle context and operator name
    pub async fn delete<K>(&self, ctx: &Context, resource: &K) -> Result<()>
    where
        K: Resource<Scope = NamespaceResourceScope>
            + Serialize
            + Clone
            + std::fmt::Debug
            + for<'de> Deserialize<'de>,
        <K as kube::Resource>::DynamicType: Default,
        <K as Resource>::Scope: std::marker::Sized,
    {
        self.kube_delete(
            ctx.kaniop_ctx.client.clone(),
            &ctx.kaniop_ctx.metrics,
            resource,
        )
        .await
    }

    /// Server-side apply without any generic delete/recreate fallback.
    /// Resource-specific reconcilers use this when they own lifecycle decisions.
    pub async fn apply<K>(&self, ctx: &Context, resource: K) -> Result<K>
    where
        K: Resource<Scope = NamespaceResourceScope>
            + Serialize
            + Clone
            + std::fmt::Debug
            + for<'de> Deserialize<'de>,
        <K as kube::Resource>::DynamicType: Default,
        <K as Resource>::Scope: std::marker::Sized,
    {
        self.kube_apply(
            ctx.kaniop_ctx.client.clone(),
            resource,
            KANIDM_OPERATOR_NAME,
        )
        .await
    }

    pub async fn patch<K>(&self, ctx: &Context, resource: K) -> Result<K>
    where
        K: Resource<Scope = NamespaceResourceScope>
            + Serialize
            + Clone
            + std::fmt::Debug
            + for<'de> Deserialize<'de>,
        <K as kube::Resource>::DynamicType: Default,
        <K as Resource>::Scope: std::marker::Sized,
    {
        self.kube_patch(
            ctx.kaniop_ctx.client.clone(),
            &ctx.kaniop_ctx.metrics,
            resource,
            KANIDM_OPERATOR_NAME,
        )
        .await
    }

    #[inline]
    fn generate_resource_labels(&self) -> BTreeMap<String, String> {
        LABELS
            .clone()
            .into_iter()
            .chain([
                (INSTANCE_LABEL.to_string(), self.name_any()),
                (CLUSTER_LABEL.to_string(), self.name_any()),
            ])
            .collect()
    }

    #[inline]
    fn generate_labels(&self) -> BTreeMap<String, String> {
        self.generate_resource_labels()
    }

    #[inline]
    fn get_tls_secret_name(&self) -> String {
        format!("{}-tls", self.name_any())
    }

    /// TLS secret name mounted by the statefulsets: explicit `tlsSecretName`,
    /// falling back to the ingress TLS secret and then to the default name.
    pub(crate) fn effective_tls_secret_name(&self) -> String {
        self.spec.tls_secret_name.clone().unwrap_or_else(|| {
            self.spec
                .ingress
                .as_ref()
                .and_then(|i| i.tls_secret_name.clone())
                .unwrap_or_else(|| self.get_tls_secret_name())
        })
    }

    #[inline]
    fn get_namespace(&self) -> String {
        // safe unwrap: Kanidm is namespaced scoped
        self.namespace().unwrap()
    }

    #[inline]
    fn is_replication_enabled(&self) -> bool {
        self.spec.replica_groups.len() > 1
            || self
                .spec
                .replica_groups
                .first()
                .is_some_and(|rg| rg.replicas > 1)
            || !self.spec.external_replication_nodes.is_empty()
    }

    #[inline]
    fn is_updatable(&self, status: &KanidmStatus) -> bool {
        status
            .version
            .as_ref()
            .map(|v| {
                if v.compatibility_result == VersionCompatibilityResult::Incompatible {
                    return false;
                }
                let current_tag = get_image_tag(&self.spec.image).unwrap_or_default();
                let versions = (parse_semver(&current_tag), parse_semver(&v.image_tag));
                match v.upgrade_check_result {
                    KanidmUpgradeCheckResult::Passed => {
                        if let (Some((_, minor, _)), Some((_, v_minor, _))) = versions {
                            minor <= v_minor + 1
                        } else {
                            true
                        }
                    }
                    KanidmUpgradeCheckResult::Failed => {
                        if let (Some((major, minor, patch)), Some((v_major, v_minor, v_patch))) =
                            versions
                        {
                            major == v_major && minor == v_minor && patch >= v_patch
                        } else {
                            v.image_tag == current_tag
                        }
                    }
                }
            })
            .unwrap_or(true)
    }

    async fn is_pod_ready(ctx: Arc<Context>, namespace: &str, pod_name: &str) -> Result<bool> {
        let pod_api = Api::<Pod>::namespaced(ctx.kaniop_ctx.client.clone(), namespace);
        let pod = pod_api.get(pod_name).await.map_err(|e| {
            Error::KubeError(
                format!("failed to get pod {namespace}/{pod_name}"),
                Box::new(e),
            )
        })?;
        Ok(pod
            .status
            .as_ref()
            .and_then(|s| s.conditions.as_ref())
            .is_some_and(|conditions| {
                conditions
                    .iter()
                    .any(|c| c.type_ == "Ready" && c.status == "True")
            }))
    }

    async fn wait_for_pod_ready(ctx: Arc<Context>, namespace: &str, pod_name: &str) -> Result<()> {
        let start = std::time::Instant::now();
        let timeout = Duration::from_secs(POD_READY_WAIT_TIMEOUT_SECONDS);
        let poll_interval = Duration::from_secs(POD_READY_POLL_INTERVAL_SECONDS);

        loop {
            match Self::is_pod_ready(ctx.clone(), namespace, pod_name).await {
                Ok(true) => return Ok(()),
                Ok(false) => {}
                Err(e) => {
                    if !e.is_retryable() {
                        return Err(e);
                    }
                    trace!(%e, "pod readiness check failed, retrying");
                }
            }

            if start.elapsed() >= timeout {
                return Err(Error::ReceiveOutput(format!(
                    "pod {namespace}/{pod_name} not ready after {POD_READY_WAIT_TIMEOUT_SECONDS}s"
                )));
            }

            sleep(poll_interval).await;
        }
    }

    async fn exec<I, T>(&self, ctx: Arc<Context>, pod_name: &str, command: I) -> Result<String>
    where
        I: IntoIterator<Item = T> + Debug,
        T: Into<String>,
    {
        let namespace = &self.get_namespace();
        trace!(
            resource.name = &pod_name,
            resource.namespace = &namespace,
            ?command,
            "pod exec"
        );
        let pod = Api::<Pod>::namespaced(ctx.kaniop_ctx.client.clone(), namespace);
        let attached = pod
            .exec(
                pod_name,
                command,
                &AttachParams::default().container("kanidm"),
            )
            .await
            .map_err(|e| {
                Error::KubeError(
                    format!("failed to exec pod {namespace}/{pod_name}"),
                    Box::new(e),
                )
            })?;
        get_output(attached).await
    }

    async fn exec_with_wait<I, T>(
        &self,
        ctx: Arc<Context>,
        pod_name: &str,
        command: I,
    ) -> Result<String>
    where
        I: IntoIterator<Item = T> + Debug,
        T: Into<String>,
    {
        let namespace = &self.get_namespace();
        Self::wait_for_pod_ready(ctx.clone(), namespace, pod_name).await?;
        self.exec(ctx, pod_name, command).await
    }

    async fn exec_any<I, T>(&self, ctx: Arc<Context>, command: I) -> Result<String>
    where
        I: IntoIterator<Item = T> + Debug,
        T: Into<String>,
    {
        // TODO: if replicas > 1 and replicas initialized, exec on pod available
        let first_replica_group = self
            .spec
            .replica_groups
            .first()
            .ok_or_else(|| Error::MissingData("no replica groups configured".to_string()))?;
        let sts_name = self.statefulset_name(&first_replica_group.name);
        let pod_name = format!("{sts_name}-0");
        self.exec(ctx, &pod_name, command).await
    }

    async fn exec_any_with_wait<I, T>(&self, ctx: Arc<Context>, command: I) -> Result<String>
    where
        I: IntoIterator<Item = T> + Debug,
        T: Into<String>,
    {
        let first_replica_group = self
            .spec
            .replica_groups
            .first()
            .ok_or_else(|| Error::MissingData("no replica groups configured".to_string()))?;
        let sts_name = self.statefulset_name(&first_replica_group.name);
        let pod_name = format!("{sts_name}-0");
        self.exec_with_wait(ctx, &pod_name, command).await
    }
}

#[cfg(test)]
mod test {
    use super::statefulset::{StatefulSetExt, TLS_SECRET_HASH_ANNOTATION};
    use super::{CLUSTER_LABEL, Kanidm, previous_tls_secret_hash, reconcile_kanidm};

    use crate::controller::State;
    use crate::kanidm::controller::context::{Context, Stores};
    use crate::kanidm::crd::KanidmStatus;

    use kaniop_k8s_util::error::Result;
    use kaniop_k8s_util::resources::hash_secret_data;

    use std::collections::BTreeMap;
    use std::sync::Arc;

    use http::{Request, Response};
    use k8s_openapi::ByteString;
    use k8s_openapi::api::apps::v1::{StatefulSet, StatefulSetSpec};
    use k8s_openapi::api::core::v1::{PodTemplateSpec, Secret, Service};
    use k8s_openapi::api::networking::v1::Ingress;
    use kube::api::ObjectMeta;
    use kube::runtime::reflector::store::Writer;
    use kube::{Client, Resource, ResourceExt, client::Body};
    use opentelemetry::metrics::MeterProvider;
    use serde_json::json;

    impl Kanidm {
        /// A normal test kanidm with a given status
        pub fn test() -> Self {
            let mut e = Kanidm::new(
                "test",
                serde_json::from_value(json!({
                    "domain": "idm.example.com",
                    "replicaGroups": [
                        {
                            "name": "default",
                            "replicas": 1
                        }
                    ]
                }))
                .unwrap(),
            );
            e.meta_mut().namespace = Some("default".into());
            // Pre-populate finalizer so reconcile callback runs immediately
            // (skips the "add finalizer + wait for next reconcile" step)
            e.meta_mut().finalizers = Some(vec![super::KANIDM_FINALIZER.to_string()]);
            e
        }

        pub fn with_ingress(mut self) -> Self {
            self.spec.ingress = Some(serde_json::from_value(json!({})).unwrap());
            self
        }

        /// Modify kanidm replicas
        pub fn with_replicas(mut self, replicas: i32) -> Self {
            self.spec.replica_groups[0].replicas = replicas;
            self
        }

        /// Modify kanidm to set a deletion timestamp
        pub fn needs_delete(mut self) -> Self {
            use k8s_openapi::apimachinery::pkg::apis::meta::v1::Time;
            use k8s_openapi::jiff::Timestamp;
            let now = Timestamp::from_second(1491138632).unwrap(); // 2017-04-02 12:50:32 UTC
            self.meta_mut().deletion_timestamp = Some(Time(now));
            self
        }

        /// Modify a kanidm to have an expected status
        pub fn with_status(mut self, status: KanidmStatus) -> Self {
            self.status = Some(status);
            self
        }
    }

    // We wrap tower_test::mock::Handle
    type ApiServerHandle = tower_test::mock::Handle<Request<Body>, Response<Body>>;
    pub struct ApiServerVerifier(ApiServerHandle);

    /// Scenarios we test for in ApiServerVerifier
    pub enum Scenario {
        Create(Kanidm),
        CreateWithTwoReplicas(Kanidm),
        CreateWithIngress(Kanidm),
        CreateWithIngressWithTwoReplicas(Kanidm),
        CreateWithTlsSecret(Kanidm, Box<Secret>),
    }

    pub async fn timeout_after_1s(handle: tokio::task::JoinHandle<()>) {
        tokio::time::timeout(std::time::Duration::from_secs(1), handle)
            .await
            .expect("timeout on mock apiserver")
            .expect("scenario succeeded")
    }

    impl ApiServerVerifier {
        /// Tests only get to run specific scenarios that has matching handlers
        ///
        /// This setup makes it easy to handle multiple requests by chaining handlers together.
        ///
        /// NB: If the controller is making more calls than we are handling in the scenario,
        /// you then typically see a `KubeError(Service(Closed(())))` from the reconciler.
        ///
        /// You should await the `JoinHandle` (with a timeout) from this function to ensure that the
        /// scenario runs to completion (i.e. all expected calls were responded to),
        /// using the timeout to catch missing api calls to Kubernetes.
        pub fn run(self, scenario: Scenario) -> tokio::task::JoinHandle<()> {
            tokio::spawn(async move {
                // moving self => one scenario per test
                match scenario {
                    Scenario::Create(kanidm) => {
                        self.handle_kanidm_get(kanidm.clone())
                            .await
                            .unwrap()
                            .handle_kanidm_status_patch(kanidm.clone())
                            .await
                            .unwrap()
                            .handle_tls_secret_get_not_found(kanidm.clone())
                            .await
                            .unwrap()
                            .handle_backup_schedule_list()
                            .await
                            .unwrap()
                            .handle_backup_repository_list()
                            .await
                            .unwrap()
                            .handle_statefulset_patch(kanidm.clone(), None)
                            .await
                            .unwrap()
                            .handle_service_patch(kanidm.clone())
                            .await
                    }
                    Scenario::CreateWithTwoReplicas(kanidm) => {
                        self.handle_kanidm_get(kanidm.clone())
                            .await
                            .unwrap()
                            .handle_kanidm_status_patch(kanidm.clone())
                            .await
                            .unwrap()
                            .handle_tls_secret_get_not_found(kanidm.clone())
                            .await
                            .unwrap()
                            .handle_backup_schedule_list()
                            .await
                            .unwrap()
                            .handle_backup_repository_list()
                            .await
                            .unwrap()
                            .handle_statefulset_patch(kanidm.clone(), None)
                            .await
                            .unwrap()
                            .handle_service_patch(kanidm.clone())
                            .await
                    }
                    Scenario::CreateWithIngress(kanidm) => {
                        self.handle_kanidm_get(kanidm.clone())
                            .await
                            .unwrap()
                            .handle_kanidm_status_patch(kanidm.clone())
                            .await
                            .unwrap()
                            .handle_tls_secret_get_not_found(kanidm.clone())
                            .await
                            .unwrap()
                            .handle_backup_schedule_list()
                            .await
                            .unwrap()
                            .handle_backup_repository_list()
                            .await
                            .unwrap()
                            .handle_statefulset_patch(kanidm.clone(), None)
                            .await
                            .unwrap()
                            .handle_service_patch(kanidm.clone())
                            .await
                            .unwrap()
                            .handle_ingress_patch(kanidm.clone())
                            .await
                    }
                    Scenario::CreateWithIngressWithTwoReplicas(kanidm) => {
                        self.handle_kanidm_get(kanidm.clone())
                            .await
                            .unwrap()
                            .handle_kanidm_status_patch(kanidm.clone())
                            .await
                            .unwrap()
                            .handle_tls_secret_get_not_found(kanidm.clone())
                            .await
                            .unwrap()
                            .handle_backup_schedule_list()
                            .await
                            .unwrap()
                            .handle_backup_repository_list()
                            .await
                            .unwrap()
                            .handle_statefulset_patch(kanidm.clone(), None)
                            .await
                            .unwrap()
                            .handle_service_patch(kanidm.clone())
                            .await
                            .unwrap()
                            .handle_ingress_patch(kanidm.clone())
                            .await
                    }
                    Scenario::CreateWithTlsSecret(kanidm, secret) => {
                        let expected_hash = hash_secret_data(&secret);
                        self.handle_kanidm_get(kanidm.clone())
                            .await
                            .unwrap()
                            .handle_kanidm_status_patch(kanidm.clone())
                            .await
                            .unwrap()
                            .handle_tls_secret_get(kanidm.clone(), *secret)
                            .await
                            .unwrap()
                            .handle_backup_schedule_list()
                            .await
                            .unwrap()
                            .handle_backup_repository_list()
                            .await
                            .unwrap()
                            .handle_statefulset_patch(kanidm.clone(), Some(expected_hash))
                            .await
                            .unwrap()
                            .handle_service_patch(kanidm.clone())
                            .await
                    }
                }
                .expect("scenario completed without errors");
            })
        }

        async fn handle_kanidm_get(mut self, kanidm: Kanidm) -> Result<Self> {
            let (request, send) = self.0.next_request().await.expect("service not called");
            assert_eq!(request.method(), http::Method::GET);
            let expected_uri_prefix = format!(
                "/apis/kaniop.rs/v1beta1/namespaces/default/kanidms/{}",
                kanidm.name_any()
            );
            let uri = request.uri().to_string();
            assert!(
                uri.starts_with(&expected_uri_prefix),
                "expected uri to start with {}, got {}",
                expected_uri_prefix,
                uri
            );
            let response = serde_json::to_vec(&kanidm).unwrap();
            send.send_response(Response::builder().body(Body::from(response)).unwrap());
            Ok(self)
        }

        async fn handle_kanidm_status_patch(mut self, kanidm: Kanidm) -> Result<Self> {
            let (request, send) = self.0.next_request().await.expect("service not called");
            assert_eq!(request.method(), http::Method::PATCH);
            assert_eq!(
                request.uri().to_string(),
                format!(
                    "/apis/kaniop.rs/v1beta1/namespaces/default/kanidms/{}/status?",
                    kanidm.name_any()
                )
            );

            let req_body = request.into_body().collect_bytes().await.unwrap();
            let json: serde_json::Value =
                serde_json::from_slice(&req_body).expect("patch object is json");
            let status: KanidmStatus = serde_json::from_value(json.get("status").unwrap().clone())
                .expect("valid kanidm status");
            let mut kanidm = Kanidm::test();
            kanidm.status = Some(status.clone());
            let response = serde_json::to_vec(&kanidm).unwrap();
            // pass through kanidm "patch accepted"
            send.send_response(Response::builder().body(Body::from(response)).unwrap());
            Ok(self)
        }

        async fn handle_tls_secret_get_not_found(mut self, kanidm: Kanidm) -> Result<Self> {
            let (request, send) = self.0.next_request().await.expect("service not called");
            assert_eq!(request.method(), http::Method::GET);
            let expected_uri_prefix = format!(
                "/api/v1/namespaces/default/secrets/{}",
                kanidm.effective_tls_secret_name()
            );
            let uri = request.uri().to_string();
            assert!(
                uri.starts_with(&expected_uri_prefix),
                "expected uri to start with {expected_uri_prefix}, got {uri}"
            );
            let status = serde_json::json!({
                "kind": "Status",
                "apiVersion": "v1",
                "metadata": {},
                "status": "Failure",
                "message": "secret not found",
                "reason": "NotFound",
                "code": 404
            });
            send.send_response(
                Response::builder()
                    .status(404)
                    .body(Body::from(serde_json::to_vec(&status).unwrap()))
                    .unwrap(),
            );
            Ok(self)
        }

        async fn handle_tls_secret_get(mut self, kanidm: Kanidm, secret: Secret) -> Result<Self> {
            let (request, send) = self.0.next_request().await.expect("service not called");
            assert_eq!(request.method(), http::Method::GET);
            let expected_uri_prefix = format!(
                "/api/v1/namespaces/default/secrets/{}",
                kanidm.effective_tls_secret_name()
            );
            let uri = request.uri().to_string();
            assert!(
                uri.starts_with(&expected_uri_prefix),
                "expected uri to start with {expected_uri_prefix}, got {uri}"
            );
            let response = serde_json::to_vec(&secret).unwrap();
            send.send_response(Response::builder().body(Body::from(response)).unwrap());
            Ok(self)
        }

        async fn handle_backup_schedule_list(mut self) -> Result<Self> {
            let (request, send) = self.0.next_request().await.expect("service not called");
            assert_eq!(request.method(), http::Method::GET);
            assert!(
                request
                    .uri()
                    .to_string()
                    .contains("/apis/kaniop.rs/v1alpha1/namespaces/default/kanidmbackupschedules"),
                "expected backup schedule list request, got {}",
                request.uri()
            );
            let list = json!({
                "apiVersion": "kaniop.rs/v1alpha1",
                "kind": "KanidmBackupScheduleList",
                "metadata": {},
                "items": []
            });
            let response = serde_json::to_vec(&list).unwrap();
            send.send_response(Response::builder().body(Body::from(response)).unwrap());
            Ok(self)
        }

        async fn handle_backup_repository_list(mut self) -> Result<Self> {
            let (request, send) = self.0.next_request().await.expect("service not called");
            assert_eq!(request.method(), http::Method::GET);
            assert!(
                request.uri().to_string().contains(
                    "/apis/kaniop.rs/v1alpha1/namespaces/default/kanidmbackuprepositories"
                ),
                "expected backup repository list request, got {}",
                request.uri()
            );
            let list = json!({
                "apiVersion": "kaniop.rs/v1alpha1",
                "kind": "KanidmBackupRepositoryList",
                "metadata": {},
                "items": []
            });
            let response = serde_json::to_vec(&list).unwrap();
            send.send_response(Response::builder().body(Body::from(response)).unwrap());
            Ok(self)
        }

        async fn handle_statefulset_patch(
            mut self,
            kanidm: Kanidm,
            expected_tls_hash: Option<String>,
        ) -> Result<Self> {
            for rg in kanidm.spec.replica_groups.iter() {
                let (request, send) = self.0.next_request().await.expect("service not called");
                assert_eq!(request.method(), http::Method::PATCH);
                assert_eq!(
                    request.uri().to_string(),
                    format!(
                        "/apis/apps/v1/namespaces/default/statefulsets/{}?&force=true&fieldManager=kanidms.kaniop.rs",
                        kanidm.statefulset_name(&rg.name)
                    )
                );
                let req_body = request.into_body().collect_bytes().await.unwrap();
                let json: serde_json::Value =
                    serde_json::from_slice(&req_body).expect("patch object is json");
                let statefulset: StatefulSet =
                    serde_json::from_value(json).expect("valid statefulset");
                assert_eq!(
                    statefulset.clone().spec.unwrap().replicas.unwrap(),
                    rg.replicas
                );
                let tls_hash_annotation = statefulset
                    .spec
                    .as_ref()
                    .unwrap()
                    .template
                    .metadata
                    .as_ref()
                    .unwrap()
                    .annotations
                    .as_ref()
                    .and_then(|a| a.get(TLS_SECRET_HASH_ANNOTATION))
                    .cloned();
                assert_eq!(tls_hash_annotation, expected_tls_hash);
                let response = serde_json::to_vec(&statefulset).unwrap();
                // pass through kanidm "patch accepted"
                send.send_response(Response::builder().body(Body::from(response)).unwrap());
            }

            Ok(self)
        }

        async fn handle_service_patch(mut self, kanidm: Kanidm) -> Result<Self> {
            let (request, send) = self.0.next_request().await.expect("service not called");
            assert_eq!(request.method(), http::Method::PATCH);
            assert_eq!(
                request.uri().to_string(),
                format!(
                    "/api/v1/namespaces/default/services/{}?&force=true&fieldManager=kanidms.kaniop.rs",
                    kanidm.name_any()
                )
            );

            let req_body = request.into_body().collect_bytes().await.unwrap();
            let json: serde_json::Value =
                serde_json::from_slice(&req_body).expect("patch object is json");
            let service: Service = serde_json::from_value(json).expect("valid service");
            let response = serde_json::to_vec(&service).unwrap();
            // pass through kanidm "patch accepted"
            send.send_response(Response::builder().body(Body::from(response)).unwrap());
            Ok(self)
        }

        async fn handle_ingress_patch(mut self, kanidm: Kanidm) -> Result<Self> {
            let (request, send) = self.0.next_request().await.expect("service not called");
            assert_eq!(request.method(), http::Method::PATCH);
            assert_eq!(
                request.uri().to_string(),
                format!(
                    "/apis/networking.k8s.io/v1/namespaces/default/ingresses/{}?&force=true&fieldManager=kanidms.kaniop.rs",
                    kanidm.name_any()
                )
            );

            let req_body = request.into_body().collect_bytes().await.unwrap();
            let json: serde_json::Value =
                serde_json::from_slice(&req_body).expect("patch object is json");
            let ingress: Ingress = serde_json::from_value(json).expect("valid service");
            let response = serde_json::to_vec(&ingress).unwrap();
            // pass through kanidm "patch accepted"
            send.send_response(Response::builder().body(Body::from(response)).unwrap());
            Ok(self)
        }
    }

    pub fn get_test_context() -> (Arc<Context>, ApiServerVerifier) {
        let (mock_service, handle) = tower_test::mock::pair::<Request<Body>, Response<Body>>();
        let mock_client = Client::new(mock_service, "default");
        let stores = Stores {
            stateful_set_store: Writer::default().as_reader(),
            service_store: Writer::default().as_reader(),
            ingress_store: Writer::default().as_reader(),
            secret_store: Writer::default().as_reader(),
            http_route_store: Some(Writer::default().as_reader()),
            backend_tls_policy_store: Some(Writer::default().as_reader()),
            deployment_store: Writer::default().as_reader(),
            config_map_store: Writer::default().as_reader(),
        };
        let controller_id = "test";

        // Create a test meter for metrics
        let provider = opentelemetry_sdk::metrics::SdkMeterProvider::builder().build();
        let meter = provider.meter("test");
        let metrics = crate::metrics::Metrics::new(&meter, &[controller_id]);

        let state = State::new(
            metrics,
            Writer::default().as_reader(),
            Writer::default().as_reader(),
            None,
        );
        let ctx = Arc::new(Context::new(
            state.to_context(mock_client, controller_id),
            stores,
        ));
        (ctx, ApiServerVerifier(handle))
    }

    #[tokio::test]
    async fn kanidm_create() {
        let (testctx, fakeserver) = get_test_context();
        let kanidm = Kanidm::test();
        let mocksrv = fakeserver.run(Scenario::Create(kanidm.clone()));
        reconcile_kanidm(Arc::new(kanidm), testctx)
            .await
            .expect("reconciler");
        timeout_after_1s(mocksrv).await;
    }

    #[tokio::test]
    async fn kanidm_create_with_two_replicas() {
        let (testctx, fakeserver) = get_test_context();
        let kanidm = Kanidm::test().with_replicas(2);
        let mocksrv = fakeserver.run(Scenario::CreateWithTwoReplicas(kanidm.clone()));
        reconcile_kanidm(Arc::new(kanidm), testctx)
            .await
            .expect("reconciler");
        timeout_after_1s(mocksrv).await;
    }

    #[tokio::test]
    async fn kanidm_create_with_ingress() {
        let (testctx, fakeserver) = get_test_context();
        let kanidm = Kanidm::test().with_ingress();
        let mocksrv = fakeserver.run(Scenario::CreateWithIngress(kanidm.clone()));
        reconcile_kanidm(Arc::new(kanidm), testctx)
            .await
            .expect("reconciler");
        timeout_after_1s(mocksrv).await;
    }

    #[tokio::test]
    async fn kanidm_create_with_ingress_with_two_replicas() {
        let (testctx, fakeserver) = get_test_context();
        let kanidm = Kanidm::test().with_ingress().with_replicas(2);
        let mocksrv = fakeserver.run(Scenario::CreateWithIngressWithTwoReplicas(kanidm.clone()));
        reconcile_kanidm(Arc::new(kanidm), testctx)
            .await
            .expect("reconciler");
        timeout_after_1s(mocksrv).await;
    }

    fn statefulset_of(cluster: &str, namespace: &str, tls_hash: Option<&str>) -> Arc<StatefulSet> {
        Arc::new(StatefulSet {
            metadata: ObjectMeta {
                name: Some(format!("{cluster}-default")),
                namespace: Some(namespace.to_string()),
                labels: Some(BTreeMap::from([(
                    CLUSTER_LABEL.to_string(),
                    cluster.to_string(),
                )])),
                ..Default::default()
            },
            spec: Some(StatefulSetSpec {
                template: PodTemplateSpec {
                    metadata: Some(ObjectMeta {
                        annotations: tls_hash.map(|h| {
                            BTreeMap::from([(
                                TLS_SECRET_HASH_ANNOTATION.to_string(),
                                h.to_string(),
                            )])
                        }),
                        ..Default::default()
                    }),
                    ..Default::default()
                },
                ..Default::default()
            }),
            ..Default::default()
        })
    }

    #[test]
    fn test_previous_tls_secret_hash() {
        let kanidm = Kanidm::test();

        // annotation carried over from an existing statefulset
        let stss = vec![statefulset_of("test", "default", Some("abc123"))];
        assert_eq!(
            previous_tls_secret_hash(&kanidm, stss),
            Some("abc123".to_string())
        );

        // no annotation on the existing statefulset
        let stss = vec![statefulset_of("test", "default", None)];
        assert_eq!(previous_tls_secret_hash(&kanidm, stss), None);

        // statefulsets from another cluster or namespace are ignored
        let stss = vec![
            statefulset_of("other", "default", Some("abc123")),
            statefulset_of("test", "other-ns", Some("abc123")),
        ];
        assert_eq!(previous_tls_secret_hash(&kanidm, stss), None);
    }

    #[tokio::test]
    async fn kanidm_create_with_tls_secret() {
        let (testctx, fakeserver) = get_test_context();
        let kanidm = Kanidm::test();
        let secret = Secret {
            metadata: ObjectMeta {
                name: Some(kanidm.effective_tls_secret_name()),
                namespace: Some("default".to_string()),
                ..Default::default()
            },
            data: Some(BTreeMap::from([
                ("tls.crt".to_string(), ByteString(b"cert".to_vec())),
                ("tls.key".to_string(), ByteString(b"key".to_vec())),
            ])),
            ..Default::default()
        };
        let mocksrv = fakeserver.run(Scenario::CreateWithTlsSecret(
            kanidm.clone(),
            Box::new(secret),
        ));
        reconcile_kanidm(Arc::new(kanidm), testctx)
            .await
            .expect("reconciler");
        timeout_after_1s(mocksrv).await;
    }

    #[tokio::test]
    async fn replication_secret_gc_skips_non_replica_secrets() {
        use super::reconcile_replication_secrets;
        use super::secret::{SECRET_TYPE_LABEL, SecretType};

        let kanidm = Kanidm::test();
        let status = kanidm.status.clone().unwrap_or_default();

        let replica_cert_label =
            serde_plain::to_string(&SecretType::ReplicaCert).expect("SecretType serializes");

        let replica_cert_secret = Secret {
            metadata: ObjectMeta {
                name: Some(format!("{}-default-0-cert", kanidm.name_any())),
                namespace: Some("default".into()),
                labels: Some(BTreeMap::from([
                    (CLUSTER_LABEL.to_string(), kanidm.name_any()),
                    (SECRET_TYPE_LABEL.to_string(), replica_cert_label.clone()),
                ])),
                ..Default::default()
            },
            ..Default::default()
        };

        let mail_sender_secret = Secret {
            metadata: ObjectMeta {
                name: Some(format!("{}-mail-sender-config", kanidm.name_any())),
                namespace: Some("default".into()),
                labels: Some(BTreeMap::from([(
                    CLUSTER_LABEL.to_string(),
                    kanidm.name_any(),
                )])),
                ..Default::default()
            },
            ..Default::default()
        };

        let (mock_service, mut handle) = tower_test::mock::pair::<Request<Body>, Response<Body>>();
        let mock_client = Client::new(mock_service, "default");

        let mut secret_writer = Writer::default();
        secret_writer.apply_watcher_event(&kube::runtime::watcher::Event::Apply(
            replica_cert_secret.clone(),
        ));
        secret_writer.apply_watcher_event(&kube::runtime::watcher::Event::Apply(
            mail_sender_secret.clone(),
        ));

        let stores = Stores {
            stateful_set_store: Writer::default().as_reader(),
            service_store: Writer::default().as_reader(),
            ingress_store: Writer::default().as_reader(),
            secret_store: secret_writer.as_reader(),
            http_route_store: Some(Writer::default().as_reader()),
            backend_tls_policy_store: Some(Writer::default().as_reader()),
            deployment_store: Writer::default().as_reader(),
            config_map_store: Writer::default().as_reader(),
        };

        let provider = opentelemetry_sdk::metrics::SdkMeterProvider::builder().build();
        let meter = provider.meter("test");
        let metrics = crate::metrics::Metrics::new(&meter, &["test"]);
        let state = State::new(
            metrics,
            Writer::default().as_reader(),
            Writer::default().as_reader(),
            None,
        );
        let ctx = Arc::new(Context::new(state.to_context(mock_client, "test"), stores));

        let mock_srv = tokio::spawn(async move {
            let (request, send) = handle.next_request().await.expect("expected a request");
            assert_eq!(request.method(), http::Method::DELETE);
            let uri = request.uri().to_string();
            assert!(
                uri.contains("-default-0-cert"),
                "expected delete for replica cert secret, got {uri}"
            );
            let not_found = serde_json::json!({
                "kind": "Status",
                "apiVersion": "v1",
                "metadata": {},
                "status": "Failure",
                "message": "not found",
                "reason": "NotFound",
                "code": 404
            });
            send.send_response(
                Response::builder()
                    .status(404)
                    .body(Body::from(serde_json::to_vec(&not_found).unwrap()))
                    .unwrap(),
            );

            if let Ok(Some((req, _))) =
                tokio::time::timeout(std::time::Duration::from_millis(200), handle.next_request())
                    .await
            {
                panic!("unexpected second request: {} {}", req.method(), req.uri());
            }
        });

        let changed = reconcile_replication_secrets(Arc::new(kanidm), ctx, &status)
            .await
            .expect("reconcile succeeds despite 404 on delete");
        assert!(
            changed,
            "expected changed=true since a deprecated secret was found"
        );

        tokio::time::timeout(std::time::Duration::from_secs(1), mock_srv)
            .await
            .expect("timeout waiting for mock server")
            .expect("mock server task succeeded");
    }
}