kopiur_api/

expand.rs

//! Selector expansion: turning one `SnapshotPolicy` source into N concrete
//! per-PVC backups (#346).
//!
//! # The model
//!
//! A `SnapshotPolicy` source is exactly one of `pvc`, `nfs`, or `pvcSelector`.
//! The first two name a single thing; the third matches many. Kopiur's whole
//! data model is built on **one `Snapshot` CR = one mover Job = one kopia
//! source path = one kopia manifest**, owned via a finalizer — retention,
//! restore, the catalog and the deletion policy all rest on that 1:1. So a
//! selector is expanded into N *ordinary* `Snapshot` CRs, one per matched PVC,
//! each of which is then indistinguishable from a hand-written single-PVC
//! backup.
//!
//! Expansion happens **once, at mint time**, in whichever component creates the
//! CR — a `SnapshotSchedule` fire or `kubectl kopiur snapshot now`. A `Snapshot`
//! never expands itself: a CR minting sibling CRs would re-enter the same
//! reconciler for each child and break the one-shot `run_decision` discipline,
//! and a `SnapshotPolicy` that minted invocations would collapse the
//! recipe/invocation/schedule split the project is built around.
//!
//! # What was here before
//!
//! Nothing. `pvcSelector` was schema-valid, admission-accepted, documented on
//! six pages and shipped as `deploy/examples/04-multi-pvc-selector.yaml` — and
//! no expansion code existed anywhere, so `build_backup_run` hit its
//! `_ =>` arm and returned `invariant violated … This is likely a bug in
//! kopiur`. That is #346.
//!
//! # The restore side (#443)
//!
//! Expansion is only half the story. A selector policy writes N kopia sources,
//! one per member path; a RESTORE against that policy must read exactly ONE of
//! them, chosen by the PVC it is filling. [`restore_source_path`] is that
//! inverse: it re-derives a member's path from the policy's own
//! `sourcePathStrategy`, so the restore's kopia filter names the same path the
//! backup wrote. Without it the filter is `user@host:` with no path, which
//! matches EVERY member — and the newest of them wins, so one PVC could be
//! filled with another PVC's data.
//!
//! It lives here, next to [`EffectiveSource::kopia_source_path`] and
//! [`strategy_for`], on purpose: the backup path and the restore path are then
//! literally the same code, and the two strings cannot drift.

use std::collections::BTreeMap;

use crate::error::ValidationError;
use crate::snapshot::{PvcTargetRef, SnapshotSourceGroup, SnapshotSourceRef, SnapshotSourceTarget};
use crate::snapshot_policy::SnapshotPolicy;
use crate::snapshot_policy::{self, Source, SourcePathStrategy};
use kube::ResourceExt;

/// Max name length for a `Snapshot` CR produced by expansion.
///
/// **63, not 253.** The CR's name becomes the mover `Job`'s name, and
/// `io::cleanup_staged_source` finds that Job's pods by the
/// `batch.kubernetes.io/job-name` **label value**, which Kubernetes caps at 63
/// bytes. A longer name silently breaks the pvc-protection release step and
/// wedges staged-PVC teardown. Fan-out makes long names routine, so this is
/// enforced here rather than discovered in production.
const MAX_CHILD_NAME: usize = 63;

/// The marker segment that makes a fanned-out name unambiguous.
const FANOUT_MARKER: &str = "-pvc-";

// --- the effective source for one run --------------------------------------

/// The single source one `Snapshot` actually backs up, after resolving
/// `spec.source` (a fanned-out child) against the policy's `sources[]`.
// `PartialEq` only, never `Eq`: `stream` carries a `StreamExec`, which embeds
// k8s-openapi's `LabelSelector` — a `PartialEq`-only type (api-conventions §3).
#[derive(Debug, Clone, PartialEq)]
pub struct EffectiveSource {
    /// Index into `policy.spec.sources` these knobs came from.
    pub index: usize,
    /// The concrete PVC, when this is a PVC-shaped source.
    pub pvc: Option<PvcTargetRef>,
    /// The NFS export path, when this is an NFS source.
    pub nfs_path: Option<String>,
    /// The stream producer, when this is a `stream` source. Mounts nothing: the mover
    /// execs a command and pipes its stdout into kopia, so no `pvc`/`nfs_path` is set
    /// and `read_only` is meaningless (see [`EffectiveSource::read_only`]).
    pub stream: Option<crate::snapshot_policy::StreamSource>,
    /// `sourcePathOverride` from the governing source.
    pub source_path_override: Option<String>,
    /// Whether the mount is read-only.
    pub read_only: bool,
}

impl EffectiveSource {
    /// The kopia source path (and mount path) for this run.
    ///
    /// `sourcePathOverride` wins. Otherwise a PVC yields `/pvc/<name>` or
    /// `/pvc/<namespace>/<name>` depending on the governing source's
    /// [`SourcePathStrategy`], and an NFS source yields its export path.
    ///
    /// The `pvc`-before-`stream` order is safe because the combination is
    /// UNREPRESENTABLE, not because PVC is the better guess: a `Source` is
    /// exactly-one-of, and [`effective_source_in`] refuses a PVC pin against a
    /// source that is not PVC-shaped (see the comment there for what the divergence
    /// used to cost). Every other constructor sets `stream: None` wherever it sets
    /// `pvc`. Anything that starts building an `EffectiveSource` with both must
    /// decide which path is authoritative FIRST — the two consumers
    /// ([`identity_source_path`], which records the manifest, and
    /// `snapshot::build`, which fills the work spec) do not read this function the
    /// same way.
    pub fn kopia_source_path(&self, strategy: SourcePathStrategy) -> Option<String> {
        if let Some(o) = &self.source_path_override {
            return Some(o.clone());
        }
        if let Some(p) = &self.pvc {
            return Some(match strategy {
                SourcePathStrategy::PvcName => format!("/pvc/{}", p.name),
                SourcePathStrategy::PvcNamespacedName => {
                    format!("/pvc/{}/{}", p.namespace, p.name)
                }
            });
        }
        // A stream source records `/stream/<fileName>` — deliberately a different
        // root from `/pvc/...` so a streamed artifact can never share a kopia
        // identity with a volume backup.
        if let Some(stream) = &self.stream {
            return Some(crate::snapshot_policy::stream_source_path(stream));
        }
        self.nfs_path.clone()
    }
}

/// Resolve which source this `Snapshot` covers.
///
/// * `pin: None` — the ordinary single-source case: `sources[0]` governs, and
///   its own `pvc`/`nfs` is the target. Byte-for-byte the old behavior.
/// * `pin: Some(_)` — a fanned-out child: `sourceIndex` selects the governing
///   source's knobs, and the pinned `target` is the concrete PVC.
///
/// An out-of-range `sourceIndex` (the policy shrank mid-run) is a named,
/// terminal error. It is emphatically NOT a fallback to `sources[0]`: silently
/// backing up a different volume than the one the CR names is the worst
/// possible failure mode for a backup operator.
pub fn effective_source(
    policy: &SnapshotPolicy,
    pin: Option<&SnapshotSourceRef>,
) -> Result<EffectiveSource, ValidationError> {
    effective_source_in(
        &policy.spec.sources,
        &policy.namespace().unwrap_or_default(),
        &policy.name_any(),
        pin,
    )
}

/// [`effective_source`] over a bare `sources[]` + namespace + name.
///
/// The webhook validates a `SnapshotPolicySpec` it has no full object for, so the
/// derivation has to be reachable without one. `effective_source` is a thin
/// wrapper over this, which keeps the two on ONE code path — a spec-level
/// derivation that drifted from the object-level one is exactly the class of bug
/// [`identity_source_path_in`] exists to make impossible.
pub fn effective_source_in(
    sources: &[Source],
    policy_namespace: &str,
    policy_name: &str,
    pin: Option<&SnapshotSourceRef>,
) -> Result<EffectiveSource, ValidationError> {
    let index = pin.map(|p| p.source_index as usize).unwrap_or(0);
    let Some(source) = sources.get(index) else {
        return Err(ValidationError::InvalidFieldValue {
            field: "spec.source.sourceIndex".to_string(),
            reason: format!(
                "is {index} but SnapshotPolicy `{policy_name}` now has {} source(s) (it was \
                 edited). Fix: delete this Snapshot and let the schedule re-fire",
                sources.len()
            ),
        });
    };
    let read_only = snapshot_policy::source_read_only(source);
    let common = |pvc: Option<PvcTargetRef>| EffectiveSource {
        index,
        pvc,
        nfs_path: source.nfs.as_ref().map(|n| n.path.clone()),
        stream: source.stream.clone(),
        source_path_override: source.source_path_override.clone(),
        read_only,
    };
    match pin.map(|p| &p.target) {
        // A fanned-out child names its own PVC; the policy source it came from
        // is a selector and has no `pvc` of its own.
        //
        // The governing source must be PVC-SHAPED for that to mean anything, and the
        // check is here rather than downstream so the bad state is unrepresentable:
        // an `EffectiveSource` carrying BOTH a `pvc` target and a `stream` producer
        // has two different kopia paths, and the two consumers pick different ones.
        // `kopia_source_path` checks `pvc` first, so `identity_source_path` (hence the
        // mover's `--override-source`, hence the RECORDED manifest) said `/pvc/data`,
        // while `snapshot::build`'s stream arm derives `/stream/<fileName>` for the
        // work spec — and verification and every `fromPolicy` restore look at the
        // stream path, which no snapshot was ever written under. The quick verify then
        // fails TERMINALLY on a policy whose snapshots are perfectly real.
        //
        // Nothing upstream rejects the input: `validate::snapshot` does not validate
        // `Snapshot.spec.source` at all, and the webhook's pin refusals only fire for a
        // selector policy MISSING a pin and for multi-repo membership. So a
        // hand-written `spec.source: {sourceIndex: 0, target: {pvc: …}}` against a
        // `stream` (or `nfs`) policy is admitted, and this is where it stops.
        //
        // Exhaustive over the wire form rather than an `is_some()` test, so a fifth
        // source form must decide whether a PVC pin means anything for it before it
        // compiles. A malformed source propagates its own error instead of silently
        // accepting the pin.
        Some(SnapshotSourceTarget::Pvc(t)) => match snapshot_policy::source_shape(source)? {
            // Both PVC-shaped: a selector's fanned-out member (the ordinary case), and
            // a plain `pvc:` source addressed by an explicit pin (the pre-existing
            // one-shot shape, unchanged).
            snapshot_policy::SourceShape::PvcSelector(_) | snapshot_policy::SourceShape::Pvc(_) => {
                Ok(common(Some(t.clone())))
            }
            // A stream source's bytes come from a command's stdout and an NFS source's
            // from an export; neither has a PVC to pin, and neither can be MOUNTED as
            // the pinned claim either.
            shape @ (snapshot_policy::SourceShape::Stream(_)
            | snapshot_policy::SourceShape::Nfs(_)) => Err(ValidationError::InvalidFieldValue {
                field: "spec.source.target.pvc".to_string(),
                reason: format!(
                    "pins PVC `{}/{}`, but SnapshotPolicy `{policy_name}` sources[{index}] is a \
                     `{}` source with no PVC. Fix: drop `spec.source`, or point this Snapshot at \
                     a policy whose sources[{index}] is a PVC source",
                    t.namespace,
                    t.name,
                    shape.kind_str(),
                ),
            }),
        },
        None => Ok(common(source.pvc.as_ref().map(|p| PvcTargetRef {
            // A non-selector `pvc:` source is always same-namespace.
            namespace: policy_namespace.to_string(),
            name: p.name.clone(),
        }))),
    }
}

/// **THE** kopia source path a `SnapshotPolicy`'s governing source records —
/// the single derivation every identity site must use (#451).
///
/// # Why this function exists
///
/// Six places resolve a policy's kopia identity: the backup mover
/// (`snapshot::build::resolve_identity_for`), the policy reconciler's
/// `config_identity` / `config_identity_for_path` / `resolve_config_identity`,
/// the verification reconciler's `verify_identity`, and the webhook's
/// `identity_collision::resolve_policy_identity`. Each of them used to re-derive
/// the path inline from `sources.first()`'s `pvc`/`nfs` — a pattern that is
/// *correct only for the two source kinds that existed when it was written*.
///
/// A `stream` source has neither a `pvc` nor an `nfs`, so five of the six
/// resolved a PATHLESS (or `/data`) identity while the backup side recorded
/// `/stream/<fileName>`. Every consequence was silent: a `fromPolicy` `Restore`
/// looked for `user@host:/data` and reported `SnapshotNotFound`; quick
/// verification verified a path with no manifests and **exited 0**; the
/// collision webhook collapsed every stream policy onto one pathless identity,
/// raising false collisions and missing real ones; and `identities_match`
/// compares source paths field-by-field, so a discovered stream snapshot could
/// never be adopted.
///
/// Routing every site through here — and removing the raw `pvc`/`nfs` pieces
/// from [`crate::IdentityInputs`], so no caller *can* re-derive a path — makes a
/// seventh site structurally unable to drift.
///
/// # Semantics
///
/// Exactly [`EffectiveSource::kopia_source_path`] under the governing source's
/// [`strategy_for`], which is byte-identical to the old inline derivation for
/// every `pvc`/`nfs`/selector/override shape and additionally correct for
/// `stream`. `Ok(None)` is kopia's identity-only `username@hostname` form:
/// a selector source with no fan-out pin, or a **zero-source** legacy policy,
/// which is deliberately tolerated rather than turned into a terminal error
/// (admission forbids it; a hand-patched object may still carry it, and a
/// working restore must not break on upgrade).
pub fn identity_source_path(
    policy: &SnapshotPolicy,
    pin: Option<&SnapshotSourceRef>,
) -> Result<Option<String>, ValidationError> {
    identity_source_path_in(
        &policy.spec.sources,
        &policy.namespace().unwrap_or_default(),
        &policy.name_any(),
        pin,
    )
}

/// The kopia source path ONE policy source records, by index — the per-source
/// form of [`identity_source_path`], for the `status.resolved.sources[]` mirror.
///
/// Same derivation, so a human reading status sees exactly what kopia stores:
/// `/pvc/<name>`, the NFS export path, `/stream/<fileName>`, or an explicit
/// `sourcePathOverride`. `None` only for a selector source, whose path is
/// per-member and therefore not a property of the source alone.
pub fn source_kopia_path(source: &Source, index: usize, policy_namespace: &str) -> Option<String> {
    let eff = EffectiveSource {
        index,
        pvc: source.pvc.as_ref().map(|p| PvcTargetRef {
            // A non-selector `pvc:` source is always same-namespace.
            namespace: policy_namespace.to_string(),
            name: p.name.clone(),
        }),
        nfs_path: source.nfs.as_ref().map(|n| n.path.clone()),
        stream: source.stream.clone(),
        source_path_override: source.source_path_override.clone(),
        read_only: snapshot_policy::source_read_only(source),
    };
    eff.kopia_source_path(strategy_for(source))
}

/// [`identity_source_path`] over a bare `sources[]` (the webhook's spec-only shape).
pub fn identity_source_path_in(
    sources: &[Source],
    policy_namespace: &str,
    policy_name: &str,
    pin: Option<&SnapshotSourceRef>,
) -> Result<Option<String>, ValidationError> {
    // Zero-source legacy tolerance, checked BEFORE `effective_source_in` so it
    // stays a pathless identity rather than the out-of-range error. A pin on a
    // zero-source policy is still the real error it has always been.
    if sources.is_empty() && pin.is_none() {
        return Ok(None);
    }
    let eff = effective_source_in(sources, policy_namespace, policy_name, pin)?;
    let strategy = sources.get(eff.index).map(strategy_for).unwrap_or_default();
    Ok(eff.kopia_source_path(strategy))
}

/// The `sourcePathStrategy` governing a source.
///
/// Deliberately only consulted for **selector-expanded** sources. A plain
/// `pvc:` source keeps `/pvc/<name>` unconditionally: changing the path of an
/// existing single-PVC policy would re-identify its kopia source and orphan
/// every manifest it has ever taken.
pub fn strategy_for(source: &Source) -> SourcePathStrategy {
    if source.pvc_selector.is_some() {
        source
            .source_path_strategy
            .unwrap_or(SourcePathStrategy::PvcName)
    } else {
        SourcePathStrategy::PvcName
    }
}

// --- the restore side: which member path fills THIS pvc (#443) ---------------

/// Where a restore's kopia source path came from — the result of
/// [`restore_source_path`].
///
/// The provenance is kept (rather than collapsed to an `Option<String>`) because
/// the three cases mean different things to a human reading `status`, and to the
/// reconciler: an [`Override`](Self::Override) is what the user asked for and is
/// never second-guessed; a [`PolicySource`](Self::PolicySource) reproduces the
/// pre-#443 behavior byte-for-byte; and a
/// [`DerivedFromTarget`](Self::DerivedFromTarget) is the new per-PVC derivation
/// that makes a fan-out restore read the right member.
#[derive(Debug, Clone, PartialEq, Eq)]
pub enum RestoreSourcePath {
    /// `source.fromPolicy.sourcePath` — the user named the path explicitly.
    Override(String),
    /// The governing plain (`pvc:`/`nfs`) source's own path.
    ///
    /// `None` is the identity-only form (kopia `username@hostname`, matching any
    /// path): a policy with **zero** sources, which admission forbids but a
    /// hand-patched legacy object may still carry. Tolerated exactly as
    /// `config_identity` tolerates it — a working restore must not become a
    /// terminal error on an upgrade.
    PolicySource(Option<String>),
    /// The selector `sourcePathStrategy` applied to the PVC being restored — the
    /// same code path the backup used, so the strings cannot drift.
    DerivedFromTarget(String),
}

impl RestoreSourcePath {
    /// The path to filter kopia snapshots by, if any. Exhaustive.
    pub fn path(&self) -> Option<&str> {
        match self {
            Self::Override(p) | Self::DerivedFromTarget(p) => Some(p),
            Self::PolicySource(p) => p.as_deref(),
        }
    }

    /// Consume into the owned path, if any. Exhaustive.
    pub fn into_path(self) -> Option<String> {
        match self {
            Self::Override(p) | Self::DerivedFromTarget(p) => Some(p),
            Self::PolicySource(p) => p,
        }
    }
}

/// A total classification of one `SnapshotPolicy` source, **for the restore-path
/// derivation only**.
///
/// Deliberately a DIFFERENT type from the wire-form resolver
/// [`snapshot_policy::SourceShape`](crate::snapshot_policy::SourceShape), which
/// answers "which one of the mutually-exclusive keys is set?" and borrows each
/// form's whole sub-object for the backup side. This one answers the narrower
/// restore question "can this source's path be derived from the target PVC?",
/// so it carries only the fields that decision needs and adds an `Invalid` arm
/// the wire-form resolver reports as an error instead. Keep the two in step:
/// every form added there needs an arm here.
///
/// The point of the enum is that "not a selector" is not one thing: a plain
/// `pvc:` source can match the restore target by NAME (and then contributes its
/// own path, override included), while `nfs`, `stream` and a malformed source
/// contribute nothing at all. Matched exhaustively, so a fifth source shape has
/// to decide what it means for a restore before it compiles.
#[derive(Debug, Clone, PartialEq, Eq)]
enum SourceShape<'a> {
    /// A plain `pvc:` source, with the PVC name it addresses (always in the
    /// policy's own namespace).
    Pvc { name: &'a str },
    /// A `pvcSelector` source: the knobs that decide each member's path.
    Selector {
        /// The strategy this selector derives member paths with.
        strategy: SourcePathStrategy,
        /// A `sourcePathOverride` that would apply to EVERY member — which is
        /// exactly what makes the selector non-per-PVC.
        source_path_override: Option<&'a str>,
    },
    /// An `nfs` export: never addressed by a PVC target.
    Nfs,
    /// A `stream` source: its bytes come from a command's stdout and its kopia
    /// path is `/stream/<fileName>`, which no PVC target can derive. Never
    /// addressed by a PVC target, exactly like [`SourceShape::Nfs`] — but named
    /// rather than folded into [`SourceShape::Invalid`], because a stream source
    /// is perfectly VALID and reporting it as malformed would send a future
    /// reader hunting a spec bug that isn't there.
    Stream,
    /// The source is not a singular, recognized wire form: none of
    /// `pvc`/`pvcSelector`/`nfs`/`stream` is set, or several are — i.e. exactly the
    /// cases [`snapshot_policy::source_shape`] reports as a
    /// [`ValidationError`]. Admission (the CRD's own CEL rule AND the webhook)
    /// forbids both; a hand-patched object, or one written against an older CRD
    /// schema, may carry either.
    ///
    /// Contributes nothing to a derivation, and fails the derivation CLOSED when it is
    /// the governing source — see [`restore_source_path`] step (3). Guessing which of
    /// several set forms "wins" is how a malformed policy came to resolve a real path
    /// holding ANOTHER volume's data.
    Invalid,
}

/// Classify one source. Pure and total.
///
/// **Driven off [`snapshot_policy::source_shape`], not off the raw `Option` fields.**
/// That is what makes the two classifiers compile-coupled: a fifth wire form adds a
/// [`snapshot_policy::SourceShape`] variant, this `match` stops being exhaustive, and
/// the build fails until someone decides what the new form means for a restore. The
/// old `match (&source.pvc, &source.nfs, &source.stream)` could not do that — a fifth
/// `Source` FIELD simply folded into [`SourceShape::Invalid`], which
/// [`restore_source_path`] then answered for by accident. (A hardcoded list in a test
/// cannot close that gap either: the test's inputs would just never name the new form.)
///
/// A malformed source — no form set, or several — becomes [`SourceShape::Invalid`]
/// rather than whichever field an ad-hoc precedence happened to check first.
fn source_shape(source: &Source) -> SourceShape<'_> {
    match snapshot_policy::source_shape(source) {
        Ok(snapshot_policy::SourceShape::PvcSelector(_)) => SourceShape::Selector {
            strategy: strategy_for(source),
            source_path_override: source.source_path_override.as_deref(),
        },
        Ok(snapshot_policy::SourceShape::Pvc(p)) => SourceShape::Pvc { name: &p.name },
        Ok(snapshot_policy::SourceShape::Nfs(_)) => SourceShape::Nfs,
        Ok(snapshot_policy::SourceShape::Stream(_)) => SourceShape::Stream,
        // Nothing set, or several set. Either way there is no honest answer, and
        // `restore_source_path` step (3) fails closed rather than deriving from it.
        Err(_) => SourceShape::Invalid,
    }
}

/// The kopia source path a restore of `target` should read from `policy` (#443).
///
/// This is the cross-volume fix. `RestoreSelector.source_path: None` becomes the
/// kopia filter `username@hostname:` — an EMPTY path, which matches every member
/// path of a selector policy — so before this, restoring one PVC of a multi-PVC
/// policy took the newest snapshot of *any* member and could fill a volume with
/// another volume's data.
///
/// The rule, in order:
///
/// 1. `override_` (`source.fromPolicy.sourcePath`) wins outright.
/// 2. A plain `pvc:` source addressing EXACTLY this target (same name, same
///    namespace) ⇒ that source's own path. An exact match beats every
///    derivation AND the first-source fallback.
/// 3. The policy has **no selector sources** ⇒ [`RestoreSourcePath::PolicySource`]
///    of `sources[0]`'s own path — byte-identical to what `config_identity` +
///    `resolve_identity` produced before this function existed, for the plain
///    `pvc:`, `nfs` and `sourcePathOverride` shapes alike (and `None` for a
///    zero-source legacy object).
/// 4. Every selector source agrees on `(sourcePathStrategy, sourcePathOverride)`
///    **and that override is `None`** ⇒
///    [`RestoreSourcePath::DerivedFromTarget`], built through
///    [`EffectiveSource::kopia_source_path`] — the same call the backup side
///    makes, so the two strings cannot drift.
/// 5. Anything else ⇒ a named error telling the user to set
///    `fromPolicy.sourcePath`.
///
/// **(2) is deliberately ahead of (3).** Validation admits N plain `pvc:`
/// sources on one policy (`validate::snapshot` only requires "at least one"),
/// but today only the FIRST is ever captured: [`expand_sources`] returns `None`
/// unless some source carries a `pvcSelector`, so a selector-free policy mints
/// one unpinned child and `effective_source(policy, None)` resolves index 0.
/// (A pre-existing backup-side limitation, tracked separately — not something
/// this function can fix.) That is precisely why the fallback must not answer
/// for every target: `sources: [pvc: a, pvc: b]` restoring into PVC `b` used to
/// resolve `/pvc/a`, a path that is real but holds ANOTHER volume's data, and
/// filled `b` with it under a green `Completed`. With the exact match first, `b`
/// resolves `/pvc/b` — never written — so the restore fails honestly with
/// `SnapshotNotFound`, or comes up empty under `Continue`. The same applies to
/// `[nfs, pvc: a]` restoring `a`, which used to read the NFS export path.
///
/// Putting the exact match first is byte-identical for every SINGLE-source
/// shape: a lone plain `pvc:` source whose name equals the target builds the
/// same `EffectiveSource` (same index, same `PvcTargetRef`, same override, same
/// strategy — `strategy_for` is `PvcName` for any non-selector source) and so
/// the same string; an `nfs` source, a differently-named target and a
/// cross-namespace target all miss (2) and fall through to (3) untouched.
///
/// **A target matching NO plain source still falls back to `sources[0]`** under
/// (3) — e.g. `[pvc: a, pvc: b]` restoring into a PVC named `c` reads `/pvc/a`.
/// That is deliberate, not an oversight: it is the pre-#443 answer, it is what
/// makes "restore this policy's data into a differently-named scratch volume"
/// keep working, and (5) is reserved for the shapes where a path genuinely
/// cannot be derived (disagreeing selectors, a flattening selector override) —
/// not for a multi-source policy where `sources[0]` is a defined, if arbitrary,
/// answer. Set `fromPolicy.sourcePath` to name the member you want.
///
/// A selector carrying `sourcePathOverride: Some(o)` is deliberately NOT
/// per-PVC: `kopia_source_path` returns the override before it ever looks at the
/// PVC, so every member was backed up under the one path `o` and no derivation
/// can tell them apart. That falls to (5) rather than silently returning `o`,
/// because "which member is this?" genuinely has no answer.
///
/// Matching is by **strategy rule, not by claimant labels**: a `target.pvc` has
/// no labels to match against, and a claimant's labels may have changed since the
/// backup was taken. This function is therefore pure over the policy + target
/// alone.
///
/// The namespace in `target` is the TARGET's namespace. For a cross-namespace
/// `target.pvcRef` under a `pvcNamespacedName` strategy that derives
/// `/pvc/<target-ns>/<name>`, which may be a path the repository never saw — use
/// the override there.
///
/// ```
/// # use kopiur_api::expand::{restore_source_path, RestoreSourcePath};
/// # use kopiur_api::snapshot::PvcTargetRef;
/// # use kopiur_api::SnapshotPolicy;
/// let policy: SnapshotPolicy = serde_json::from_value(serde_json::json!({
///     "apiVersion": "kopiur.home-operations.com/v1alpha1",
///     "kind": "SnapshotPolicy",
///     "metadata": { "name": "app", "namespace": "db" },
///     "spec": {
///         "repository": { "name": "r" },
///         "sources": [{
///             "pvcSelector": { "matchLabels": { "app": "web" } },
///             "sourcePathStrategy": "PvcName",
///         }],
///     },
/// }))
/// .unwrap();
/// let target = PvcTargetRef { namespace: "db".into(), name: "data-1".into() };
/// assert_eq!(
///     restore_source_path(&policy, None, &target).unwrap(),
///     RestoreSourcePath::DerivedFromTarget("/pvc/data-1".into())
/// );
/// ```
pub fn restore_source_path(
    policy: &SnapshotPolicy,
    override_: Option<&str>,
    target: &PvcTargetRef,
) -> Result<RestoreSourcePath, ValidationError> {
    // (1) An explicit override is never second-guessed.
    if let Some(o) = override_ {
        return Ok(RestoreSourcePath::Override(o.to_string()));
    }

    let policy_ns = policy.namespace().unwrap_or_default();
    let shapes: Vec<SourceShape<'_>> = policy.spec.sources.iter().map(source_shape).collect();

    // (2) An exact plain-`pvc:` match, BEFORE the first-source fallback: a policy
    // may carry several plain `pvc:` sources, and `sources[0]` would then be
    // another volume's path — wrong-but-real data, rather than the honest
    // `SnapshotNotFound` the unwritten path yields. Same namespace is required:
    // a plain source always addresses the POLICY's namespace, so a same-named
    // PVC in another namespace is a different volume.
    if policy_ns == target.namespace
        && let Some(index) = shapes.iter().position(|s| match s {
            SourceShape::Pvc { name } => *name == target.name,
            SourceShape::Selector { .. }
            | SourceShape::Nfs
            | SourceShape::Stream
            | SourceShape::Invalid => false,
        })
    {
        let source = &policy.spec.sources[index];
        let eff = EffectiveSource {
            index,
            pvc: Some(target.clone()),
            nfs_path: None,
            // Matched by its own plain `pvc:` source, so PVC-shaped by construction.
            stream: None,
            source_path_override: source.source_path_override.clone(),
            read_only: snapshot_policy::source_read_only(source),
        };
        return Ok(RestoreSourcePath::PolicySource(
            eff.kopia_source_path(strategy_for(source)),
        ));
    }

    // (3) No selector anywhere: the pre-#443 answer, verbatim (including the
    // zero-source legacy tolerance, where `sources.first()` is `None`). Reached
    // only when (2) did not match, so a target that names one of the policy's
    // plain sources never lands here.
    let has_selector = shapes
        .iter()
        .any(|s| matches!(s, SourceShape::Selector { .. }));
    if !has_selector {
        let Some(first) = policy.spec.sources.first() else {
            return Ok(RestoreSourcePath::PolicySource(None));
        };
        // …but only when `sources[0]` is a source this build can classify. A
        // MALFORMED governing source is the #451 failure mode reached from the
        // other end: it is not a selector, so it lands here, and
        // `effective_source` hands back an all-`None` source whose
        // `kopia_source_path` is `None` — the PATHLESS identity `user@host:`,
        // which kopia reads as a relative filesystem path and matches against
        // EVERY member of the repository. A multi-set source (say `pvc` beside
        // `pvcSelector`) was worse still under the old field-order precedence:
        // it answered with `sources[0]`'s own `/pvc/<name>`, a real path holding
        // another volume's data, under a green `Completed`. Fail closed instead.
        //
        // Deliberately scoped to the GOVERNING source, not to any source: a
        // malformed sibling alongside a working selector is irrelevant to (4)'s
        // derivation and must not break a restore that has an honest answer
        // (`restore_path_ignores_nfs_and_malformed_siblings_of_a_selector`). And
        // a ZERO-source legacy policy keeps its tolerance above — it has no
        // source to be malformed.
        if shapes.first() == Some(&SourceShape::Invalid) {
            return Err(malformed_source_path(policy));
        }
        let eff = effective_source(policy, None)?;
        return Ok(RestoreSourcePath::PolicySource(
            eff.kopia_source_path(strategy_for(first)),
        ));
    }

    // (4) Every selector agrees, and none of them flattens its members onto one
    // shared path.
    let mut agreed: Option<(SourcePathStrategy, Option<&str>)> = None;
    for shape in &shapes {
        let (strategy, over) = match shape {
            SourceShape::Selector {
                strategy,
                source_path_override,
            } => (*strategy, *source_path_override),
            SourceShape::Pvc { .. }
            | SourceShape::Nfs
            | SourceShape::Stream
            | SourceShape::Invalid => continue,
        };
        match agreed {
            None => agreed = Some((strategy, over)),
            Some(prev) if prev == (strategy, over) => {}
            Some(_) => return Err(ambiguous_source_path(policy)),
        }
    }
    match agreed {
        Some((strategy, None)) => {
            let eff = EffectiveSource {
                // The index is inert here: the path comes from the target PVC and
                // the agreed strategy, not from the source's own `pvc`/`nfs`.
                index: 0,
                pvc: Some(target.clone()),
                nfs_path: None,
                // A selector-derived path is PVC-shaped; a stream source never
                // participates in a selector derivation.
                stream: None,
                source_path_override: None,
                read_only: true,
            };
            // A PVC-bearing `EffectiveSource` with no override always yields a
            // path, so the `None` arm is unreachable — but returning the fail-
            // closed error rather than unwrapping keeps the function total.
            eff.kopia_source_path(strategy)
                .map(RestoreSourcePath::DerivedFromTarget)
                .ok_or_else(|| ambiguous_source_path(policy))
        }
        // (5) A shared `sourcePathOverride`, or (defensively) no selector at all
        // after the `has_selector` check.
        Some((_, Some(_))) | None => Err(ambiguous_source_path(policy)),
    }
}

/// The fail-closed error for the malformed-governing-source case of
/// [`restore_source_path`] step (3): the source does not set exactly one of the four
/// forms, so no path can be derived from it — what / why / fix.
fn malformed_source_path(policy: &SnapshotPolicy) -> ValidationError {
    ValidationError::InvalidFieldValue {
        field: "spec.source.fromPolicy.sourcePath".to_string(),
        reason: format!(
            "SnapshotPolicy `{}` has a source that does not set exactly one of \
             pvc/pvcSelector/nfs/stream, so the path to restore is unknown. Fix: repair the \
             policy's sources[], or set source.fromPolicy.sourcePath",
            policy.name_any()
        ),
    }
}

/// The fail-closed error for (5) of [`restore_source_path`] — what / why / fix.
fn ambiguous_source_path(policy: &SnapshotPolicy) -> ValidationError {
    ValidationError::InvalidFieldValue {
        field: "spec.source.fromPolicy.sourcePath".to_string(),
        reason: format!(
            "SnapshotPolicy `{}` selector sources have no single per-PVC path (they differ in \
             sourcePathStrategy/sourcePathOverride, or share one sourcePathOverride), so \
             kopiur cannot tell which volume to restore. Fix: set source.fromPolicy.sourcePath \
             (e.g. /pvc/<name>)",
            policy.name_any()
        ),
    }
}

/// The deterministic name of the populate mover `Job` for one claiming PVC.
///
/// `<restore>-populate-<h8>`, capped at [`MAX_CHILD_NAME`] (63) — a Job name
/// becomes a `batch.kubernetes.io/job-name` label value, which Kubernetes caps
/// at 63 bytes.
///
/// `<h8>` is 8 hex of FNV-1a over the claiming PVC's `metadata.uid` and is
/// **never** clipped: it is what makes N claimants of one `Restore` produce N
/// distinct Jobs, and what makes a DELETED-and-re-created claim (a new uid) get a
/// fresh Job instead of adopting the dead claim's. The pre-#443 name was the
/// bare `<restore>-populate`, which is why only one claimant could ever be
/// populated; that legacy name is still driven for an adopted in-flight claim
/// (see the controller's `JobNameReuse`), never minted anew.
///
/// ```
/// # use kopiur_api::expand::populate_job_name;
/// let n = populate_job_name("restore-pg", "9f1c-uid");
/// assert!(n.starts_with("restore-pg-populate-"));
/// assert!(n.len() <= 63);
/// // Distinct claimants never collide.
/// assert_ne!(n, populate_job_name("restore-pg", "other-uid"));
/// ```
pub fn populate_job_name(restore: &str, consumer_uid: &str) -> String {
    const MARKER: &str = "-populate-";
    let tag = fnv8(consumer_uid);
    let keep = MAX_CHILD_NAME.saturating_sub(MARKER.len() + tag.len());
    format!("{}{MARKER}{tag}", clip(restore, keep).trim_end_matches('-'))
        .trim_matches('-')
        .to_string()
}

/// Render a `LabelSelector` as the API server's selector string.
///
/// Lives here so the controller and the CLI build byte-identical queries: a
/// divergence would make `kubectl kopiur snapshot now` and the schedule expand
/// to DIFFERENT PVC sets for the same recipe, which is the kind of discrepancy
/// nobody notices until a restore is missing a volume.
pub fn label_selector_string(
    sel: &k8s_openapi::apimachinery::pkg::apis::meta::v1::LabelSelector,
) -> String {
    use k8s_openapi::apimachinery::pkg::apis::meta::v1::LabelSelectorRequirement;
    let mut terms: Vec<String> = Vec::new();
    if let Some(labels) = &sel.match_labels {
        for (k, v) in labels {
            terms.push(format!("{k}={v}"));
        }
    }
    if let Some(exprs) = &sel.match_expressions {
        for LabelSelectorRequirement {
            key,
            operator,
            values,
        } in exprs
        {
            let vals = values.clone().unwrap_or_default().join(",");
            match operator.as_str() {
                "In" => terms.push(format!("{key} in ({vals})")),
                "NotIn" => terms.push(format!("{key} notin ({vals})")),
                "Exists" => terms.push(key.clone()),
                "DoesNotExist" => terms.push(format!("!{key}")),
                // Unknown operator: skip (the webhook/schema constrain the set).
                _ => {}
            }
        }
    }
    terms.join(",")
}

/// Whether an object with these `labels` matches `selector` — the client-side
/// counterpart to [`label_selector_string`].
///
/// Implements `matchLabels` (every key present with the required value) plus the
/// `matchExpressions` operators `In`/`NotIn`/`Exists`/`DoesNotExist`. An empty
/// selector matches everything, as Kubernetes defines it.
///
/// Lives here rather than in a consumer because three of them need the same
/// answer and a partial re-implementation is worse than none: the schedule
/// reconciler decides which policies to fire, its watch mapper decides which
/// schedules a policy edit wakes, and the web UI shows a policy which schedules
/// fire it. A copy that handled only `matchLabels` would silently claim a
/// `matchExpressions` selector matched every policy in the namespace.
///
/// An unknown operator is treated as *no constraint* rather than as no match:
/// the CRD schema constrains the set, so an unrecognized one can only come from
/// a newer API than this build, and refusing to match would silently drop a
/// policy the operator itself is scheduling.
///
/// ```
/// use std::collections::BTreeMap;
/// use k8s_openapi::apimachinery::pkg::apis::meta::v1::LabelSelector;
/// use kopiur_api::expand::labels_match_selector;
///
/// let labels = BTreeMap::from([("tier".to_string(), "gold".to_string())]);
/// // An empty selector selects everything.
/// assert!(labels_match_selector(&labels, &LabelSelector::default()));
///
/// let gold: LabelSelector = serde_json::from_value(serde_json::json!({
///     "matchLabels": { "tier": "gold" }
/// }))
/// .unwrap();
/// assert!(labels_match_selector(&labels, &gold));
/// ```
pub fn labels_match_selector(
    labels: &std::collections::BTreeMap<String, String>,
    selector: &k8s_openapi::apimachinery::pkg::apis::meta::v1::LabelSelector,
) -> bool {
    use k8s_openapi::apimachinery::pkg::apis::meta::v1::LabelSelectorRequirement;
    if let Some(ml) = &selector.match_labels {
        for (k, v) in ml {
            if labels.get(k) != Some(v) {
                return false;
            }
        }
    }
    if let Some(exprs) = &selector.match_expressions {
        for LabelSelectorRequirement {
            key,
            operator,
            values,
        } in exprs
        {
            let vals = values.clone().unwrap_or_default();
            let present = labels.get(key);
            let ok = match operator.as_str() {
                "In" => present.is_some_and(|v| vals.iter().any(|x| x == v)),
                "NotIn" => present.is_none_or(|v| !vals.iter().any(|x| x == v)),
                "Exists" => present.is_some(),
                "DoesNotExist" => present.is_none(),
                // Unknown operator: the schema constrains the set; treat as no constraint.
                _ => true,
            };
            if !ok {
                return false;
            }
        }
    }
    true
}

// --- naming -----------------------------------------------------------------

/// The deterministic name of the fanned-out `Snapshot` for one PVC.
///
/// `<base>-pvc-<slug>-<h8>`, capped at [`MAX_CHILD_NAME`].
///
/// * `<slug>` is human-legible (`<name>`, or `<namespace>-<name>` when the PVC
///   is outside the policy's namespace) and may be clipped.
/// * `<h8>` is 8 hex of FNV-1a over the exact string `"<namespace>/<name>"` and
///   is **never** clipped — it is the injectivity guarantee.
///
/// Collision-free against the three existing schemes (`<schedule>-<slot>`,
/// `<schedule>-<policy>-<slot>`, `<policy>-manual-<slot>`): each ends in a
/// dash-free 14-digit slot stamp, while a fanned name's tail after `-pvc-`
/// always contains a `-`.
///
/// This does NOT delegate to `io::staging::staged_child_name`, whose
/// `MAX_NAME_LEN - tag.len() - suffix.len() - 2` is unchecked `usize`
/// subtraction: its callers pass 4-char suffixes (`snap`, `src`), and a PVC
/// name up to 253 chars would underflow it.
///
/// ```
/// # use kopiur_api::expand::fanout_child_name;
/// let n = fanout_child_name("nightly-20260805020000", "db", "db", "pgdata");
/// assert!(n.starts_with("nightly-20260805020000-pvc-pgdata-"));
/// assert!(n.len() <= 63);
/// ```
pub fn fanout_child_name(base: &str, policy_ns: &str, pvc_ns: &str, pvc_name: &str) -> String {
    fanout_child_name_for(base, policy_ns, Some((pvc_ns, pvc_name)), None)
}

/// The marker segment naming the target repository in a multi-repo child name.
const REPO_MARKER: &str = "-repo-";

/// 8 hex chars of FNV-1a over `input` — the never-clipped injectivity tag
/// shared by every fan-out naming scheme in this module. One definition so
/// the hash function can never drift between the schemes.
fn fnv8(input: &str) -> String {
    let mut hash: u64 = 0xcbf2_9ce4_8422_2325;
    for b in input.as_bytes() {
        hash ^= *b as u64;
        hash = hash.wrapping_mul(0x100_0000_01b3);
    }
    format!("{:08x}", (hash & 0xffff_ffff) as u32)
}

/// The deterministic child `Snapshot` name for one (source-member, repository)
/// cell of a fan-out — the general form behind [`fanout_child_name`].
///
/// * `member`: `Some((pvc_ns, pvc_name))` for a selector-expanded member,
///   `None` for a policy whose single source needs no per-PVC expansion.
/// * `repo`: `Some(ref)` for a multi-repository fan-out child, `None` for the
///   classic single-repo shapes.
///
/// The four combinations produce:
///
/// | member | repo | name |
/// |--------|------|------|
/// | `None` | `None` | `<base>` — the legacy non-fanned name, byte-identical |
/// | `Some` | `None` | `<base>-pvc-<pslug>-<h8>` — byte-identical to the legacy [`fanout_child_name`] |
/// | `None` | `Some` | `<base>-repo-<rslug>-<h8>` |
/// | `Some` | `Some` | `<base>-pvc-<pslug>-repo-<rslug>-<h8>` |
///
/// `<h8>` is FNV-1a over the newline-framed FULL unclipped tuple — `base`,
/// then (when present) `"<pvc_ns>/<pvc_name>"`, then (when present) the
/// normalized [`repo_key`](crate::common::repo_key) — so clipping any legible
/// segment never merges two distinct cells, and a policy slug that happens to
/// contain `-repo-` (pslug `x-repo-y` vs pslug `x` + rslug `y`) still yields
/// distinct names because the hash inputs differ. Markers and the tag are
/// NEVER clipped; when over budget the clip order is base → pslug → rslug.
/// Always ≤ [`MAX_CHILD_NAME`] (63).
///
/// The legacy hash input for `repo: None` is exactly the pre-multi-repo
/// `"{base}\n{pvc_ns}/{pvc_name}"` — pinned by golden tests.
pub fn fanout_child_name_for(
    base: &str,
    policy_ns: &str,
    member: Option<(&str, &str)>,
    repo: Option<&crate::common::RepositoryRef>,
) -> String {
    // (normalized repo_key for the hash, legible slug for the name).
    let repo = repo.map(|r| (crate::common::repo_key(r, policy_ns), r.name.as_str()));
    match (member, repo) {
        // Legacy single child: the name IS the base (already ≤63 for every
        // caller-produced base; clipped defensively, identical when in budget).
        (None, None) => clip(base, MAX_CHILD_NAME).trim_matches('-').to_string(),
        (Some((pvc_ns, pvc_name)), repo) => {
            member_child_name(base, policy_ns, pvc_ns, pvc_name, repo.as_ref())
        }
        (None, Some((rkey, repo_name))) => {
            // <base>-repo-<rslug>-<h8>
            let tag = fnv8(&format!("{base}\n{rkey}"));
            let rslug = sanitize_dns1123(repo_name);
            let fixed = REPO_MARKER.len() + 1 + tag.len();
            let mut base_keep = base.len();
            let mut rslug_keep = rslug.len();
            if fixed + base_keep + rslug_keep > MAX_CHILD_NAME {
                let room = MAX_CHILD_NAME.saturating_sub(fixed);
                // Clip base first (every sibling shares it), rslug second.
                rslug_keep = rslug_keep.min(room / 3);
                base_keep = base_keep.min(room.saturating_sub(rslug_keep));
            }
            format!(
                "{}{REPO_MARKER}{}-{tag}",
                clip(base, base_keep),
                clip(&rslug, rslug_keep)
            )
            .trim_matches('-')
            .to_string()
        }
    }
}

/// The `member: Some(..)` arm of [`fanout_child_name_for`]: the legacy
/// `-pvc-` form (`rkey: None`, byte-identical to the pre-multi-repo
/// [`fanout_child_name`]) and the combined `-pvc-…-repo-…` form.
fn member_child_name(
    base: &str,
    policy_ns: &str,
    pvc_ns: &str,
    pvc_name: &str,
    repo: Option<&(String, &str)>,
) -> String {
    // The tag hashes the BASE as well as the PVC (and, for a multi-repo child,
    // the repo key). That is not belt-and-braces: `base` is
    // `<schedule>-<YYYYMMDDHHMMSS>` and it is the part that gets CLIPPED when
    // the name is too long, so a tag over the PVC alone leaves two different
    // slots of the same schedule+PVC with byte-identical names. The schedule
    // server-side-applies with force and only skips *terminating* twins, so
    // the second fire would re-apply onto the already-`Succeeded` first
    // Snapshot, `run_decision` would return `SucceededSteadyState`, and no
    // mover Job would ever launch — a whole backup slot vanishing with no
    // error anywhere.
    let tag = match repo {
        None => fnv8(&format!("{base}\n{pvc_ns}/{pvc_name}")),
        Some((rkey, _)) => fnv8(&format!("{base}\n{pvc_ns}/{pvc_name}\n{rkey}")),
    };

    let slug_full = if pvc_ns == policy_ns {
        pvc_name.to_string()
    } else {
        format!("{pvc_ns}-{pvc_name}")
    };
    let slug = sanitize_dns1123(&slug_full);

    match repo {
        None => {
            // Budget: base + "-pvc-" + slug + "-" + tag. Clip `base` first (it
            // is the most redundant part — every sibling shares it), then the
            // slug. Never the tag or the marker.
            let fixed = FANOUT_MARKER.len() + 1 + tag.len();
            let mut base_keep = base.len();
            let mut slug_keep = slug.len();
            if fixed + base_keep + slug_keep > MAX_CHILD_NAME {
                let room = MAX_CHILD_NAME.saturating_sub(fixed);
                // Give the slug up to a third of the room, the base the rest.
                slug_keep = slug_keep.min(room / 3);
                base_keep = base_keep.min(room.saturating_sub(slug_keep));
            }
            let name = format!(
                "{}{FANOUT_MARKER}{}-{tag}",
                clip(base, base_keep),
                clip(&slug, slug_keep)
            );
            // A clip can leave a trailing '-', which is not a legal DNS-1123 name.
            name.trim_matches('-').to_string()
        }
        Some((_, repo_name)) => {
            // Combined form: base + "-pvc-" + pslug + "-repo-" + rslug + "-" +
            // tag. Overhead 5+6+1+8 = 20, room 43. Clip base → pslug → rslug.
            let rslug = sanitize_dns1123(repo_name);
            let fixed = FANOUT_MARKER.len() + REPO_MARKER.len() + 1 + tag.len();
            let mut base_keep = base.len();
            let mut pslug_keep = slug.len();
            let mut rslug_keep = rslug.len();
            if fixed + base_keep + pslug_keep + rslug_keep > MAX_CHILD_NAME {
                let room = MAX_CHILD_NAME.saturating_sub(fixed);
                rslug_keep = rslug_keep.min(room / 3);
                pslug_keep = pslug_keep.min(room.saturating_sub(rslug_keep) / 2);
                base_keep = base_keep.min(room.saturating_sub(rslug_keep + pslug_keep));
            }
            format!(
                "{}{FANOUT_MARKER}{}{REPO_MARKER}{}-{tag}",
                clip(base, base_keep),
                clip(&slug, pslug_keep),
                clip(&rslug, rslug_keep)
            )
            .trim_matches('-')
            .to_string()
        }
    }
}

/// The kopia-cache PVC name for one (policy, repository) cell.
///
/// A kopia client cache is REPOSITORY-SPECIFIC state (indexes, metadata, owned
/// blobs of ONE repository), and the mover mounts it at a fixed
/// `KOPIA_CACHE_DIRECTORY` — so a multi-repo policy's children must not share
/// one PVC or they poison each other's cache. Hence:
///
/// * `repo: None` — the classic single-repo cache: `kopiur-cache-<policy>`,
///   byte-identical to the pre-multi-repo name (including its historical lack
///   of a length cap — an existing PVC must keep matching by name).
/// * `repo: Some(_)` — a pinned child of a multi-repo policy:
///   `kopiur-cache-<policy>-<rslug>-<h6>`, where `<h6>` is 6 hex of FNV-1a
///   over the normalized [`repo_key`](crate::common::repo_key) (never
///   clipped — the injectivity guarantee, same family as
///   [`fanout_child_name_for`]) and the whole name is capped at 63 (RFC 1123
///   label, the same Job-label bound child names honor). Clip order:
///   policy slug first, then rslug; marker dashes and the tag never.
///
/// Ownership stays the policy in both shapes (the caller's concern).
pub fn cache_pvc_name(
    policy_name: &str,
    policy_ns: &str,
    repo: Option<&crate::common::RepositoryRef>,
) -> String {
    const PREFIX: &str = "kopiur-cache-";
    let Some(repo) = repo else {
        return format!("{PREFIX}{policy_name}");
    };
    let rkey = crate::common::repo_key(repo, policy_ns);
    // h6: the fnv8 family hash, truncated to 6 hex — enough to disambiguate a
    // policy's ≤8 repositories while keeping the legible slugs roomy.
    let tag: String = fnv8(&rkey).chars().take(6).collect();
    let pslug = sanitize_dns1123(policy_name);
    let rslug = sanitize_dns1123(&repo.name);
    let fixed = PREFIX.len() + 2 + tag.len(); // two joining dashes + tag
    let mut pslug_keep = pslug.len();
    let mut rslug_keep = rslug.len();
    if fixed + pslug_keep + rslug_keep > MAX_CHILD_NAME {
        let room = MAX_CHILD_NAME.saturating_sub(fixed);
        rslug_keep = rslug_keep.min(room / 3);
        pslug_keep = pslug_keep.min(room.saturating_sub(rslug_keep));
    }
    // Trim any clip-produced trailing '-' per segment (a doubled dash is not a
    // legal-name problem, but keeping segments clean avoids `--` cosmetics; the
    // never-clipped tag carries injectivity regardless).
    format!(
        "{PREFIX}{}-{}-{tag}",
        clip(&pslug, pslug_keep).trim_end_matches('-'),
        clip(&rslug, rslug_keep).trim_end_matches('-')
    )
    .trim_matches('-')
    .to_string()
}

/// Truncate on a char boundary.
fn clip(s: &str, keep: usize) -> &str {
    if keep >= s.len() {
        return s;
    }
    let mut end = keep;
    while end > 0 && !s.is_char_boundary(end) {
        end -= 1;
    }
    &s[..end]
}

/// Lowercase, replacing anything outside `[a-z0-9-]` with `-`.
fn sanitize_dns1123(s: &str) -> String {
    s.chars()
        .map(|c| {
            if c.is_ascii_alphanumeric() {
                c.to_ascii_lowercase()
            } else {
                '-'
            }
        })
        .collect()
}

// --- expansion ---------------------------------------------------------------

/// One member of an expanded selector: the child's name and the `spec.source`
/// to stamp on it.
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct ExpandedMember {
    /// Deterministic child `Snapshot` name.
    pub name: String,
    /// The pin recording which PVC this child covers.
    pub source: SnapshotSourceRef,
}

/// The shared `VolumeGroupSnapshot` name for one expansion in one namespace.
///
/// Derived from the per-invocation base name, NOT from any one member: every
/// member must compute the identical name without talking to its siblings,
/// which is what makes their racing server-side-applies converge on one object
/// instead of N.
///
/// Bounded to 63 chars for the same reason child names are — see
/// [`fanout_child_name`].
/// `repo_key` adds the repository dimension for multi-repo fan-out — one
/// VolumeGroupSnapshot per (repository, slot), because each repo's members are
/// an independent capture wave (independent-captures semantics: the N groups
/// are N separate point-in-time CSI snapshots, at N× the CSI quota/load).
/// Existing callers pass `None` and get the byte-identical legacy name.
pub fn group_name(
    base: &str,
    namespace: &str,
    source_index: usize,
    repo_key: Option<&str>,
) -> String {
    let suffix = "-grp";
    // The SOURCE INDEX is part of the key, not just the namespace. Two selector
    // sources in one policy have DIFFERENT label selectors, and
    // `resolve_group_stage` builds the VolumeGroupSnapshot's `source.selector`
    // from `sources[sourceIndex]`. Sharing a name across them would have the two
    // members force-SSA conflicting selectors onto one object, so the loser's
    // PVC is never captured and it fails with `GroupMemberMissing`.
    let tag = match repo_key {
        None => fnv8(&format!("{namespace}#{source_index}")),
        Some(k) => fnv8(&format!("{namespace}#{source_index}#{k}")),
    };
    let room = MAX_CHILD_NAME - suffix.len() - tag.len() - 1;
    format!("{}-{tag}{suffix}", clip(base, room))
        .trim_matches('-')
        .to_string()
}

/// One `Snapshot` to mint for a slot/invocation: its deterministic name, the
/// source pin (for a `pvcSelector` member) and the repository pin (for a
/// multi-repo fan-out child, NORMALIZED via
/// [`normalized_repository_ref`](crate::common::normalized_repository_ref)).
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct MintCell {
    /// Deterministic child `Snapshot` name.
    pub name: String,
    /// The `spec.source` pin recording which PVC this child covers.
    pub source: Option<SnapshotSourceRef>,
    /// The `spec.repository` pin recording which repository this child targets
    /// (multi-repo fan-out only; `None` keeps the legacy unpinned wire).
    pub repository: Option<crate::common::RepositoryRef>,
}

/// **Pure.** Cross one policy's expanded source members with its repository
/// dimension (#368) into the exact set of `Snapshot`s to mint for one slot or
/// one `snapshot now` invocation. Lives here so the `SnapshotSchedule`
/// reconciler and `kubectl kopiur snapshot now` mint byte-identical sets — a
/// divergence would give the same recipe different children depending on who
/// fired it.
///
/// * Single-repo policy: byte-identical to the pre-multi-repo behavior — the
///   bare `base_name` child (no pins) for a selector-less recipe, or one
///   unpinned per-member child.
/// * Multi-repo policy: one child per (member × repository), named via
///   [`fanout_child_name_for`] with the repo dimension, `spec.repository`
///   stamped NORMALIZED; a grouped member's shared `VolumeGroupSnapshot` name
///   is re-derived PER REPOSITORY ([`group_name`] with the repo key) — each
///   repo's members are an independent capture wave, so N repos = N groups.
///
/// `members: Some(vec![])` (a selector that matched nothing) yields no cells —
/// the caller warns, as before.
pub fn mint_cells(
    policy: &SnapshotPolicy,
    base_name: &str,
    members: Option<Vec<ExpandedMember>>,
) -> Vec<MintCell> {
    use crate::common::{normalized_repository_ref, repo_key};
    let policy_ns = policy.namespace().unwrap_or_default();
    let repos: Option<Vec<&crate::common::RepositoryRef>> =
        crate::snapshot_policy::is_multi_repo(&policy.spec)
            .then(|| policy.spec.repositories.iter().collect());
    match (members, repos) {
        // Single-repo, no selector: the legacy bare child.
        (None, None) => vec![MintCell {
            name: base_name.to_string(),
            source: None,
            repository: None,
        }],
        // Single-repo selector fan-out: unpinned members, byte-identical.
        (Some(members), None) => members
            .into_iter()
            .map(|m| MintCell {
                name: m.name,
                source: Some(m.source),
                repository: None,
            })
            .collect(),
        // Multi-repo, no selector: one child per repository.
        (None, Some(repos)) => repos
            .into_iter()
            .map(|r| MintCell {
                name: fanout_child_name_for(base_name, &policy_ns, None, Some(r)),
                source: None,
                repository: Some(normalized_repository_ref(r, &policy_ns)),
            })
            .collect(),
        // Multi-repo selector fan-out: members × repositories.
        (Some(members), Some(repos)) => members
            .iter()
            .flat_map(|m| {
                let policy_ns = policy_ns.clone();
                let target = match &m.source.target {
                    SnapshotSourceTarget::Pvc(t) => t.clone(),
                };
                repos.iter().map(move |r| {
                    let rkey = repo_key(r, &policy_ns);
                    let mut source = m.source.clone();
                    if let Some(group) = source.group.as_mut() {
                        group.volume_group_snapshot_name = group_name(
                            base_name,
                            &group.namespace,
                            source.source_index as usize,
                            Some(&rkey),
                        );
                    }
                    MintCell {
                        name: fanout_child_name_for(
                            base_name,
                            &policy_ns,
                            Some((&target.namespace, &target.name)),
                            Some(r),
                        ),
                        source: Some(source),
                        repository: Some(normalized_repository_ref(r, &policy_ns)),
                    }
                })
            })
            .collect(),
    }
}

/// **Pure.** Expand one policy's sources against an already-matched PVC set.
///
/// `matched` is `(namespace, name)` per source index — the caller does the
/// cluster IO ([`match_pvcs`]), this decides the names and pins.
///
/// Returns `Ok(None)` when the policy has no selector source at all, meaning
/// "mint exactly one child with no `spec.source`", i.e. today's behavior. That
/// distinction matters: an empty `Vec` would mean "a selector matched nothing",
/// which is a different and much louder situation.
///
/// # The collision guard
///
/// `sourcePathStrategy` defaults to `PvcName` → `/pvc/<name>`. A selector with
/// a cross-namespace `namespaceSelector` matching a PVC called `data` in two
/// namespaces therefore yields **two kopia sources at the identical
/// `user@host:/pvc/data`**, silently merging two volumes' histories into one
/// stream — under a single `KOPIA_KEEP_MAX` retention pin, so they also prune
/// each other. `detect_identity_collision` cannot catch this: it compares
/// across policies and skips self. So it is caught here, before anything is
/// created.
pub fn expand_sources(
    policy: &SnapshotPolicy,
    base_name: &str,
    matched: &BTreeMap<usize, Vec<PvcTargetRef>>,
) -> Result<Option<Vec<ExpandedMember>>, ValidationError> {
    let policy_ns = policy.namespace().unwrap_or_default();
    if !policy.spec.sources.iter().any(|s| s.pvc_selector.is_some()) {
        return Ok(None);
    }
    let mut members: Vec<ExpandedMember> = Vec::new();
    // path -> (target, source index) that first produced it.
    let mut paths: BTreeMap<String, (PvcTargetRef, usize)> = BTreeMap::new();
    let grouped =
        policy.spec.group_by == Some(crate::snapshot_policy::GroupBy::VolumeGroupSnapshot);
    // How many members land in each namespace, computed up front: a
    // VolumeGroupSnapshot is namespaced and its `source.selector` is
    // namespace-local, so a selector spanning namespaces yields ONE GROUP PER
    // NAMESPACE — the consistency guarantee is per-namespace, not global.
    // Keyed by SOURCE too, not just namespace: one VolumeGroupSnapshot is built
    // from ONE source's label selector, so two selector sources in the same
    // namespace are two separate captures, and counting them together would
    // wrongly promote a pair of one-PVC sources into a "group".
    let mut per_group: BTreeMap<(&str, usize), usize> = BTreeMap::new();
    if grouped {
        for (index, source) in policy.spec.sources.iter().enumerate() {
            if source.pvc_selector.is_none() {
                continue;
            }
            for t in matched.get(&index).into_iter().flatten() {
                *per_group.entry((t.namespace.as_str(), index)).or_default() += 1;
            }
        }
    }

    for (index, source) in policy.spec.sources.iter().enumerate() {
        if source.pvc_selector.is_none() {
            continue;
        }
        let strategy = strategy_for(source);
        for target in matched.get(&index).into_iter().flatten() {
            // Collision check against every path this expansion has produced.
            let eff = EffectiveSource {
                index,
                pvc: Some(target.clone()),
                nfs_path: None,
                // A selector source is always PVC-shaped; a stream source never
                // expands (it is one artifact per Snapshot, and admission requires
                // it to be its policy's only source).
                stream: None,
                source_path_override: source.source_path_override.clone(),
                read_only: snapshot_policy::source_read_only(source),
            };
            let path = eff
                .kopia_source_path(strategy)
                .unwrap_or_else(|| "/data".to_string());
            // ANY repeated path is refused, not just one produced by two
            // DIFFERENT PVCs. Two selector sources that both match the same PVC
            // land on one path AND one child name, so the second would
            // force-server-side-apply over the first and one backup would
            // vanish with no error — the same silent-overwrite class as a
            // clipped slot stamp.
            if let Some((prev, prev_index)) = paths.insert(path.clone(), (target.clone(), index)) {
                let same_pvc = prev.namespace == target.namespace && prev.name == target.name;
                // The same PVC listed twice by the SAME source is a listing
                // artifact (`match_pvcs` already dedupes; this keeps
                // `expand_sources` idempotent for any caller). Skip it rather
                // than reporting a configuration error that isn't one.
                if same_pvc && prev_index == index {
                    continue;
                }
                let reason = if same_pvc {
                    format!(
                        "SnapshotPolicy `{}` has two `pvcSelector` sources that both match \
                         `{}/{}` (path `{path}`). Fix: narrow the selectors so each PVC matches \
                         only one source",
                        policy.name_any(),
                        target.namespace,
                        target.name,
                    )
                } else if let Some(i) = [index, prev_index].into_iter().find(|i| {
                    policy
                        .spec
                        .sources
                        .get(*i)
                        .and_then(|s| s.source_path_override.as_deref())
                        == Some(path.as_str())
                }) {
                    // A `sourcePathOverride` PRODUCED this path: it wins at
                    // `kopia_source_path`'s first branch, so the strategy is
                    // never consulted and "set sourcePathStrategy" would be a
                    // remedy that does nothing. Name the real cause instead.
                    format!(
                        "SnapshotPolicy `{}` pvcSelector matches both `{}/{}` and `{}/{}`, but \
                         `spec.sources[{i}].sourcePathOverride` gives them the same path \
                         `{path}`, so their backups would overwrite each other. Fix: remove \
                         that `sourcePathOverride` and use `sourcePathStrategy` (`PvcName` or \
                         `PvcNamespacedName`), or move the override onto its own `pvc:` \
                         source",
                        policy.name_any(),
                        prev.namespace,
                        prev.name,
                        target.namespace,
                        target.name,
                    )
                } else {
                    format!(
                        "SnapshotPolicy `{}` pvcSelector matches both `{}/{}` and `{}/{}`, \
                         which get the same path `{path}` under `sourcePathStrategy: PvcName`, \
                         so their backups would overwrite each other. Fix: set \
                         `sourcePathStrategy: PvcNamespacedName` on that source",
                        policy.name_any(),
                        prev.namespace,
                        prev.name,
                        target.namespace,
                        target.name,
                    )
                };
                return Err(ValidationError::InvalidFieldValue {
                    field: "spec.sources[].pvcSelector".to_string(),
                    reason,
                });
            }
            members.push(ExpandedMember {
                name: fanout_child_name(base_name, &policy_ns, &target.namespace, &target.name),
                source: SnapshotSourceRef {
                    source_index: index as u32,
                    target: SnapshotSourceTarget::Pvc(target.clone()),
                    // A one-member "group" buys nothing and costs a
                    // VolumeGroupSnapshotClass requirement (and a Beta API
                    // group many clusters do not serve), so it degrades to the
                    // ordinary per-PVC VolumeSnapshot path.
                    group: (grouped
                        && per_group
                            .get(&(target.namespace.as_str(), index))
                            .copied()
                            .unwrap_or(0)
                            > 1)
                    .then(|| SnapshotSourceGroup {
                        namespace: target.namespace.clone(),
                        // Repo dimension: none HERE — `expand_sources` emits
                        // repo-agnostic members, and `mint_cells` re-derives
                        // this name with `Some(repo_key)` per repository for a
                        // multi-repo policy. Passing None keeps single-repo
                        // names byte-identical to the legacy form.
                        volume_group_snapshot_name: group_name(
                            base_name,
                            &target.namespace,
                            index,
                            None,
                        ),
                    }),
                },
            });
        }
    }
    Ok(Some(members))
}

#[cfg(test)]
mod tests;