kopiur_api/validate/

mod.rs

//! Cross-field validation the type system can't express (ADR §2.2 principle 8).
//!
//! These are the rules a single struct's types can't enforce: "field X is
//! forbidden only when sibling Y has a particular variant," "this string must
//! parse as a cron," "a discovered backup may only Retain." They live here as pure
//! functions so the **webhook calls them at admission and the controller calls them
//! defensively** — one validator, two callers (SKILL hard-rule 4). No `kube::Client`,
//! no `tokio`.
//!
//! ## Fail-fast vs. accumulate (see [`crate::error`])
//!
//! Single-rule helpers return [`ValidationResult`] (fail-fast — first problem).
//! The per-CRD aggregate validators (`validate_backup_config`, …) return
//! `Vec<ValidationError>` so a user sees every independent problem in one apply.
//! An empty vec means valid.

use crate::backend::NfsVolume;
use crate::common::{FailurePolicy, MoverSpec, PvcAccessMode, RepositoryMode};
use crate::error::{ValidationError, ValidationResult};
use crate::server::{ServerAuth, ServerSpec};
use crate::snapshot_policy::{Source, SourceShape, StreamExec, StreamSource};
use k8s_openapi::api::core::v1::ResourceRequirements;
use kube_quantity::ParsedQuantity;

mod admission;
mod backend;
mod identity;
mod repository;
mod restore;
mod snapshot;
mod snapshot_replication_overlap;

pub use admission::*;
pub use backend::*;
pub use identity::*;
pub use repository::*;
pub use restore::*;
pub use snapshot::*;
pub use snapshot_replication_overlap::*;

/// Validate a `pvcSelector`'s shape.
///
/// Two refusals, both structural rather than stylistic:
///
/// * **`namespaceSelector` is rejected outright.** A Pod can only mount
///   PersistentVolumeClaims in its OWN namespace — that is a Kubernetes
///   invariant, not a kopiur limitation. The mover Job runs in the `Snapshot`'s
///   namespace, which is the policy's, so a PVC matched elsewhere could never be
///   mounted. Accepting the field would mean either failing at reconcile with
///   "source PVC not found" or, far worse, silently snapshotting a
///   SAME-NAMED PVC in the policy's own namespace under the matched one's
///   identity. Use one `SnapshotPolicy` per namespace.
/// * **An unusable `matchExpressions` entry is rejected** rather than dropped.
///   The generated schema does not enum-constrain `operator`, so a typo (`in`
///   for `In`) would otherwise be silently discarded — WIDENING the selector, so
///   PVCs the user meant to exclude get backed up. An `In`/`NotIn` with no
///   values renders as `key in ()`, which the API server rejects with a 400 that
///   would abort the whole schedule fire.
fn validate_pvc_selector(selector: &crate::snapshot_policy::PvcSelector) -> ValidationResult {
    if selector
        .namespace_selector
        .as_ref()
        .is_some_and(|n| !n.match_names.is_empty())
    {
        return Err(ValidationError::InvalidFieldValue {
            field: "spec.sources[].pvcSelector.namespaceSelector".to_string(),
            reason: "a backup can only read PVCs in its own namespace. Fix: use one \
                     SnapshotPolicy per namespace (they can share a repository)"
                .to_string(),
        });
    }
    if let Some(ls) = selector.label_selector.as_ref()
        && let Some(exprs) = ls.match_expressions.as_ref()
    {
        for e in exprs {
            match e.operator.as_str() {
                "In" | "NotIn" => {
                    if e.values.as_ref().is_none_or(|v| v.is_empty()) {
                        return Err(ValidationError::InvalidFieldValue {
                            field: format!(
                                "spec.sources[].pvcSelector.labelSelector.matchExpressions[{}]",
                                e.key
                            ),
                            reason: format!(
                                "operator `{}` requires at least one value; an empty list \
                                 renders as `{} {} ()`, which the API server rejects",
                                e.operator,
                                e.key,
                                e.operator.to_lowercase()
                            ),
                        });
                    }
                }
                "Exists" | "DoesNotExist" => {}
                other => {
                    return Err(ValidationError::InvalidFieldValue {
                        field: format!(
                            "spec.sources[].pvcSelector.labelSelector.matchExpressions[{}].operator",
                            e.key
                        ),
                        reason: format!(
                            "`{other}` is not a label-selector operator. Fix: use In, NotIn, \
                             Exists or DoesNotExist"
                        ),
                    });
                }
            }
        }
    }
    Ok(())
}

/// A single backup `Source` is well-formed: **exactly one** of `pvc`,
/// `pvcSelector`, `nfs`, or `stream` is set (ADR §3.3 — modeled as sibling Options
/// because the forms share `sourcePath*` keys, so it's a webhook check, not an
/// enum), and that form's own content is valid.
///
/// The dispatch is an EXHAUSTIVE match over [`SourceShape`] rather than a chain of
/// `if let`s: a fifth source form then cannot compile until it is given content
/// validation here, instead of silently falling through with none.
pub fn validate_source(source: &Source) -> ValidationResult {
    match crate::snapshot_policy::source_shape(source)? {
        SourceShape::Pvc(_) => Ok(()),
        // `pvcSelector` + `sourcePathOverride` is deliberately NOT refused here. It
        // is a footgun, but a pure validator cannot see how many PVCs the selector
        // currently matches, and the two cases differ:
        //
        //  * ONE matched PVC — a working single-volume policy with a custom path.
        //    The override wins at `kopia_source_path`'s first branch, there is no
        //    path COLLISION for `expand_sources` to refuse, the backup is minted
        //    and real snapshots exist there. Refusing it would park the WHOLE
        //    policy on upgrade (`reconcile_inner` validates first and returns on
        //    the first error), stopping retention pruning, adoption, status and the
        //    repository summary along with verification.
        //  * TWO OR MORE — `expand_sources` refuses the RUN via its path-collision
        //    check (N members on one path would merge their histories into one
        //    stream and prune each other), so no further snapshot is minted for
        //    that source.
        //
        // So the broken case IS caught at run time, at the moment it becomes
        // broken, and the working one keeps working. The loud signal there is the
        // refused backup — the failed schedule fire — NOT verification: on a policy
        // that was working, the earlier snapshots still sit at that path, so
        // verification keeps passing. (It fails only where the path never received
        // a backup, which the mover now treats as terminal rather than a false
        // pass, #456.) The `sourcePathOverride` field documentation says so too.
        SourceShape::PvcSelector(selector) => validate_pvc_selector(selector),
        SourceShape::Nfs(nfs) => {
            // `readOnly: false` exists for exactly one purpose — letting the kubelet
            // apply `fsGroup` to the source — and the kubelet does not apply `fsGroup`
            // to in-tree NFS volumes at all. So on NFS it buys nothing and only makes
            // the export writable to the mover. Reject it rather than ship a knob that
            // silently does the opposite of what its user wants.
            if !crate::snapshot_policy::source_read_only(source) {
                return Err(ValidationError::InvalidFieldValue {
                    field: "spec.sources[].readOnly".to_string(),
                    reason: "readOnly: false is not supported on an nfs source (fsGroup does \
                             not apply to NFS). Fix: remove readOnly, and grant access with \
                             mover.podSecurityContext supplementalGroups or \
                             mover.securityContext runAsUser matching the export's owner"
                        .to_string(),
                });
            }
            validate_nfs_volume(nfs, "snapshot source")
        }
        SourceShape::Stream(stream) => validate_stream_source(source, stream),
    }
}

/// A `stream` source is well-formed: a safe single-segment `fileName`, a runnable
/// `command`, a non-empty `podSelector`, a parseable `timeout`, and none of the
/// PVC-only knobs that cannot mean anything without a mounted volume.
fn validate_stream_source(source: &Source, stream: &StreamSource) -> ValidationResult {
    validate_stream_file_name("spec.sources[].stream.fileName", &stream.file_name)?;
    validate_stream_exec("spec.sources[].stream.workloadExec", &stream.workload_exec)?;

    // PVC-only knobs. A stream source mounts nothing, so `readOnly` has no volume to
    // apply to, `acknowledgeLiveMutation` acknowledges a mutation that cannot happen,
    // and `sourcePathStrategy` derives a path from a PVC that does not exist.
    //
    // CRITICALLY, two of these carry a SCHEMA DEFAULT (`readOnly: true`,
    // `sourcePathStrategy: PvcName`), which the API server materializes onto every
    // source before admission ever sees it. So `is_some()` is NOT a signal that the
    // user wrote the field — on a stream source it is always true, and checking it
    // rejects every valid policy. Reject only the values that would actually MEAN
    // something; a materialized default is inert and indistinguishable from absence.
    if source.read_only == Some(false) {
        return Err(ValidationError::InvalidFieldValue {
            field: "spec.sources[].readOnly".to_string(),
            reason: "readOnly: false does not apply to a stream source, which mounts no \
                     volume. Fix: remove readOnly"
                .to_string(),
        });
    }
    if source.acknowledge_live_mutation.is_some() {
        return Err(ValidationError::InvalidFieldValue {
            field: "spec.sources[].acknowledgeLiveMutation".to_string(),
            reason: "acknowledgeLiveMutation does not apply to a stream source, which mounts \
                     no volume. Fix: remove acknowledgeLiveMutation"
                .to_string(),
        });
    }
    // Same materialized-default reasoning: only a NON-default strategy is a real
    // request. `PvcName` is what the API server stamps on everything.
    if source
        .source_path_strategy
        .is_some_and(|s| s != crate::snapshot_policy::SourcePathStrategy::PvcName)
    {
        return Err(ValidationError::InvalidFieldValue {
            field: "spec.sources[].sourcePathStrategy".to_string(),
            reason: "sourcePathStrategy only applies to a pvcSelector source. Fix: remove it; \
                     to change a stream source's path (/stream/<fileName>), set \
                     sourcePathOverride"
                .to_string(),
        });
    }
    Ok(())
}

/// `fileName` is ONE safe path segment.
///
/// Security-relevant, not cosmetic: kopia stores the `--stdin-file` value verbatim as
/// the snapshot entry's name and does not sanitize it, so an entry named `../x` makes
/// a later `kopia restore <id> <dir>` write OUTSIDE `<dir>` (verified against kopia
/// 0.23.1). An empty value is worse still — kopia then ignores stdin entirely and
/// tries to snapshot the source path from the mover's own filesystem.
pub fn validate_stream_file_name(field: &str, name: &str) -> ValidationResult {
    let bad = name.is_empty()
        || name.contains('/')
        || name == "."
        || name == ".."
        || name.contains('\0')
        || name.chars().any(|c| c.is_control());
    if bad {
        return Err(ValidationError::InvalidFieldValue {
            field: field.to_string(),
            reason: format!(
                "`{name}` is not a usable file name. Fix: give a single file name (e.g. \
                 `postgres.sql`): not empty, no `/`, not `.` or `..`, no control characters"
            ),
        });
    }
    Ok(())
}

/// A [`StreamExec`] is runnable: a non-empty argv, a selector that identifies
/// something, and a parseable timeout. Shared by the backup source and the restore
/// target so both reject the same mistakes with the same words.
pub fn validate_stream_exec(field: &str, exec: &StreamExec) -> ValidationResult {
    if exec.command.is_empty() {
        return Err(ValidationError::InvalidFieldValue {
            field: format!("{field}.command"),
            reason: "is empty. Fix: give the command to run, e.g. [\"sh\", \"-ec\", \
                     \"pg_dumpall -U postgres\"] (it runs without a shell)"
                .to_string(),
        });
    }
    if exec.command.iter().all(|a| a.trim().is_empty()) {
        return Err(ValidationError::InvalidFieldValue {
            field: format!("{field}.command"),
            reason: "the command is entirely blank; element 0 must name the program to run"
                .to_string(),
        });
    }
    let selector_empty = exec
        .pod_selector
        .match_labels
        .as_ref()
        .is_none_or(|m| m.is_empty())
        && exec
            .pod_selector
            .match_expressions
            .as_ref()
            .is_none_or(|e| e.is_empty());
    if selector_empty {
        return Err(ValidationError::InvalidFieldValue {
            field: format!("{field}.podSelector"),
            reason: "is empty, so it matches every pod in the namespace. Fix: set \
                     matchLabels/matchExpressions for the one pod to run the command in"
                .to_string(),
        });
    }
    if let Some(t) = &exec.timeout
        && crate::duration::parse_go_duration(t).is_none()
    {
        return Err(ValidationError::InvalidFieldValue {
            field: format!("{field}.timeout"),
            reason: format!("`{t}` is not a Go duration. Use forms like `30m`, `2h`, `90s`"),
        });
    }
    Ok(())
}

/// An inline [`NfsVolume`] is well-formed: a non-empty server and an absolute
/// export path. The structural schema can't express either, so the webhook does.
/// `context` names where it appears (e.g. `"snapshot source"`, `"filesystem repo"`)
/// for an actionable message.
pub fn validate_nfs_volume(nfs: &NfsVolume, context: &str) -> ValidationResult {
    if nfs.server.trim().is_empty() {
        return Err(ValidationError::MissingRequiredField {
            field: format!("{context} nfs.server"),
        });
    }
    if !nfs.path.starts_with('/') {
        return Err(ValidationError::InvalidFieldValue {
            field: format!("{context} nfs.path"),
            reason: format!(
                "must be an absolute export path beginning with '/' (got {:?})",
                nfs.path
            ),
        });
    }
    Ok(())
}

/// Validate a PVC access-modes list wherever one appears (`spec.staging.accessModes`,
/// `restore.target.pvc.accessModes`). Three rules, one place, both callers (webhook
/// at admission, controller defensively):
///
///   * every entry must be canonical — an [`PvcAccessMode::Unknown`] value is either
///     a legacy stored string from before schema enforcement or a typo, and no PVC
///     could ever be provisioned from it;
///   * no duplicates;
///   * `ReadWriteOncePod` must be the **sole** mode — the apiserver rejects the
///     combination at PVC-create time, so catching it here fails at admission with
///     the reason instead of wedging the first run in a create-retry loop.
///
/// `field` names the exact path for the message. Accumulates so every bad entry is
/// reported in one apply.
pub fn validate_access_modes(field: &str, modes: &[PvcAccessMode]) -> Vec<ValidationError> {
    let mut errs = Vec::new();
    let mut seen = std::collections::BTreeSet::new();
    for (i, mode) in modes.iter().enumerate() {
        if let PvcAccessMode::Unknown(value) = mode {
            errs.push(ValidationError::InvalidFieldValue {
                field: format!("{field}[{i}]"),
                reason: format!(
                    "{value:?} is not a Kubernetes access mode. Fix: use one of: {}",
                    PvcAccessMode::CANONICAL.join(", ")
                ),
            });
        }
        if !seen.insert(mode.mode_str().to_string()) {
            errs.push(ValidationError::InvalidFieldValue {
                field: format!("{field}[{i}]"),
                reason: format!(
                    "duplicate access mode {:?}; list each mode at most once",
                    mode.mode_str()
                ),
            });
        }
    }
    if modes.len() > 1
        && modes
            .iter()
            .any(|m| matches!(m, PvcAccessMode::ReadWriteOncePod))
    {
        errs.push(ValidationError::InvalidFieldValue {
            field: field.to_string(),
            reason: "ReadWriteOncePod cannot be combined with other access modes. Fix: use \
                     ReadWriteOncePod alone, or drop it"
                .to_string(),
        });
    }
    errs
}

/// The semantic class of a numeric lower-bound, so [`require_min`] can attach the
/// right minimum AND a one-line *why* without every caller re-deriving it — the
/// difference between the old terse `"must be >= 1 (got 0)"` and a message that
/// says why 0 is refused and what to do instead. Exhaustive `min`/`because`, so a
/// new bound cannot be added without deciding both (mirrors the type-safety
/// thesis: a knob's meaning is not free-text).
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum NumericBound {
    /// A worker/parallelism count (`parallel`, `fileParallelism`, …). 0 does no work.
    Count,
    /// A per-second cap (bytes/s or ops/s). kopia reads an ABSENT field as "no limit".
    RatePerSecond,
    /// A megabyte size cap (e.g. `upload.limitMb`).
    Megabytes,
    /// A number of seconds — a Job/pod deadline.
    Seconds,
    /// A retry backoff count, which legitimately allows 0.
    BackoffCount,
}

impl NumericBound {
    /// The inclusive minimum this bound enforces.
    pub fn min(self) -> i64 {
        match self {
            NumericBound::BackoffCount => 0,
            NumericBound::Count
            | NumericBound::RatePerSecond
            | NumericBound::Megabytes
            | NumericBound::Seconds => 1,
        }
    }

    /// Why the minimum exists — the `because` clause of the rejection message.
    pub fn because(self) -> &'static str {
        match self {
            NumericBound::Count => "0 would do no work; set a positive count",
            NumericBound::RatePerSecond => {
                "0 would stall the transfer; omit the field for no limit"
            }
            NumericBound::Megabytes => {
                "0 would cap the upload at nothing; omit the field to leave it uncapped"
            }
            NumericBound::Seconds => {
                "a non-positive deadline is rejected by the kubelet (or fails the pod immediately)"
            }
            NumericBound::BackoffCount => "a negative retry budget is meaningless",
        }
    }
}

/// A numeric knob must meet its [`NumericBound`]'s minimum — the shared one-liner
/// behind every `Option<u32>` count / `Option<i64>` rate field (e.g.
/// `RepositoryReplication.spec.sync.parallel`), so the rule, its minimum, AND its
/// rationale are written once instead of re-derived per field. `field` names the
/// exact path for the message (e.g. `"RepositoryReplication spec.sync.parallel"`).
/// Callers only invoke this for a `Some` value — an absent knob is always valid
/// and never reaches this helper.
pub fn require_min(field: &str, value: i64, bound: NumericBound) -> Option<ValidationError> {
    let min = bound.min();
    (value < min).then(|| ValidationError::InvalidFieldValue {
        field: field.to_string(),
        reason: crate::message::Diagnostic::new(format!("must be at least {min} (got {value})"))
            .because(bound.because())
            .to_string(),
    })
}

/// Validate every SET knob of a [`Throttle`](crate::common::Throttle): a cap is a
/// positive rate, so each present field must be >= 1. `field` names the block's
/// path for the message (e.g. `"SnapshotReplication spec.migrate.throttle.source"`),
/// and each knob's own camelCase name is appended.
///
/// A `0` is refused rather than read as "unlimited": kopia's own "no limit" is the
/// ABSENT field, so accepting `0` would give one wire value two plausible meanings
/// (uncapped vs. stalled) — and the one it actually has in kopia is a hard stop.
/// Exhaustively destructured, so a fifth knob cannot be added without deciding its
/// bound here.
pub fn validate_throttle(field: &str, throttle: &crate::common::Throttle) -> Vec<ValidationError> {
    let crate::common::Throttle {
        upload_bytes_per_second,
        download_bytes_per_second,
        read_ops_per_second,
        write_ops_per_second,
    } = throttle;
    [
        ("uploadBytesPerSecond", upload_bytes_per_second),
        ("downloadBytesPerSecond", download_bytes_per_second),
        ("readOpsPerSecond", read_ops_per_second),
        ("writeOpsPerSecond", write_ops_per_second),
    ]
    .into_iter()
    .filter_map(|(knob, value)| {
        value.and_then(|v| require_min(&format!("{field}.{knob}"), v, NumericBound::RatePerSecond))
    })
    .collect()
}

/// Validate a `MoverSpec`. `context` names the owning resource for the message (e.g.
/// `"Restore mover"`).
///
/// `inheritSecurityContextFrom` and the explicit `securityContext`/`podSecurityContext` are
/// **compatible**, not mutually exclusive: they are adjacent layers of the merge ladder
/// (`hardened ⊂ moverDefaults ⊂ inherited ⊂ explicit`), so the explicit context overrides the
/// inherited one field-wise and fills whatever the workload does not pin — and stands in alone
/// when inheritance cannot resolve a pod.
///
/// This pair used to be rejected here, on the rationale that "the mover's effective contexts
/// must have a single, unambiguous source so the privileged-mover gate runs on exactly one".
/// That rationale was never true: the gate has always evaluated the *merged* product of
/// hardened + `moverDefaults` + recipe (see the callers of
/// [`crate::common::requires_privilege_resolved`]), which
/// [`crate::invariants::enforce_security_context_invariants`] normalizes first — INV-1 exists
/// precisely to reconcile an inherited `runAsUser: 0` against the hardened `runAsNonRoot:
/// true`. Merging one more layer in cannot smuggle an elevated mover past it.
pub fn validate_mover(mover: &MoverSpec, context: &str) -> ValidationResult {
    if let Some(resources) = &mover.resources {
        validate_resources(resources, context)?;
    }
    Ok(())
}

/// The first resource key whose `requests` value exceeds its `limits` value (both present
/// and parseable), as `(key, request, limit)`. Quantity comparison uses `kube_quantity`'s
/// `ParsedQuantity` (the same `k8s-openapi` `Quantity` type the cluster uses), so
/// `"1Gi" > "512Mi"` is compared correctly across binary/SI/milli suffixes. **Best-effort:**
/// a key whose quantity fails to parse is skipped, never a false rejection.
fn requests_exceeding_limits(resources: &ResourceRequirements) -> Option<(String, String, String)> {
    let (Some(requests), Some(limits)) = (resources.requests.as_ref(), resources.limits.as_ref())
    else {
        return None;
    };
    for (key, req) in requests {
        let Some(lim) = limits.get(key) else { continue };
        let (Ok(req_p), Ok(lim_p)) = (ParsedQuantity::try_from(req), ParsedQuantity::try_from(lim))
        else {
            continue;
        };
        if req_p > lim_p {
            return Some((key.clone(), req.0.clone(), lim.0.clone()));
        }
    }
    None
}

/// Validate that a `ResourceRequirements` has no `requests > limits` for any key. A pod with
/// `requests > limits` is **rejected by the API server**, so the mover Job never creates a
/// pod and the run hangs — the same silent-wedge class as an impossible securityContext.
/// `context` names the owner (e.g. `"SnapshotPolicy mover"`).
pub fn validate_resources(resources: &ResourceRequirements, context: &str) -> ValidationResult {
    if let Some((key, req, lim)) = requests_exceeding_limits(resources) {
        return Err(ValidationError::InvalidFieldValue {
            field: format!("{context} resources.requests.{key}"),
            reason: format!(
                "request `{req}` exceeds limit `{lim}`, so the mover pod can never start. Fix: \
                 lower the request or raise the limit"
            ),
        });
    }
    Ok(())
}

/// Validate a [`FailurePolicy`]'s numeric fields are sane: `activeDeadlineSeconds` and
/// `podStartupDeadlineSeconds` must be positive (the kubelet rejects a non-positive Job
/// deadline, and a non-positive grace would fail every pod on its first reconcile);
/// `backoffLimit` must be non-negative. `context` names the owner (e.g. `"Snapshot"`).
pub fn validate_failure_policy(fp: &FailurePolicy, context: &str) -> ValidationResult {
    if let Some(d) = fp.active_deadline_seconds
        && let Some(e) = require_min(
            &format!("{context} failurePolicy.activeDeadlineSeconds"),
            d,
            NumericBound::Seconds,
        )
    {
        return Err(e);
    }
    if let Some(g) = fp.pod_startup_deadline_seconds
        && g <= 0
    {
        // Bespoke `because`: this deadline has a meaning worth naming beyond the
        // generic Seconds rationale.
        return Err(ValidationError::InvalidFieldValue {
            field: format!("{context} failurePolicy.podStartupDeadlineSeconds"),
            reason: crate::message::Diagnostic::new(format!("must be at least 1 second (got {g})"))
                .because("it is how long a mover pod may take to start before the run fails")
                .to_string(),
        });
    }
    if let Some(b) = fp.backoff_limit
        && let Some(e) = require_min(
            &format!("{context} failurePolicy.backoffLimit"),
            b.into(),
            NumericBound::BackoffCount,
        )
    {
        return Err(e);
    }
    Ok(())
}

/// A cron expression parses with the same parser the controller uses at runtime, so
/// bad expressions are rejected at apply time, not at first reconcile (ADR §4.1).
///
/// `croner` 2.x does not implement Jenkins-style `H`. Since kopiur resolves `H`
/// deterministically in [`crate::jitter::substitute_h`] (not in the parser), we
/// substitute every `H` field with the fixed placeholder `0` purely to validate the
/// expression's *shape* here. The real `H` spread is produced at scheduling time.
///
/// ```
/// use kopiur_api::validate::validate_cron;
/// use kopiur_api::ValidationError;
///
/// // Valid 5-field crons pass — including Jenkins-style `H` (resolved later).
/// assert!(validate_cron("0 2 * * *").is_ok());
/// assert!(validate_cron("H 2 * * *").is_ok());
///
/// // Garbage is rejected at apply time, not at first reconcile (ADR §4.1).
/// assert!(matches!(
///     validate_cron("not a cron"),
///     Err(ValidationError::InvalidCron { .. }),
/// ));
/// ```
pub fn validate_cron(expr: &str) -> ValidationResult {
    let probe = expr
        .split_whitespace()
        .map(|f| if f == "H" { "0" } else { f })
        .collect::<Vec<_>>()
        .join(" ");
    match crate::jitter::cron_parser().parse(&probe) {
        Ok(_) => Ok(()),
        Err(e) => Err(ValidationError::InvalidCron {
            expr: expr.to_string(),
            reason: e.to_string(),
        }),
    }
}

/// A non-blocking admission WARNING (never a rejection) when a schedule's cron
/// fires more often than hourly (issue #249). Every fire creates one per-run
/// `Snapshot` CR per source, and they accumulate up to the `SnapshotPolicy`
/// retention window — each terminal one is then re-reconciled for that whole
/// window — so a sub-hourly cadence with a wide (or absent) retention can produce
/// thousands of CRs. Sub-hourly is legitimate for some workloads, so this is a
/// footgun heads-up, not a block.
///
/// Pure and cron-only: the schedule webhook is client-less and can't read the
/// referenced policy's retention, so the message states the cadence and the
/// CR-count relationship rather than an exact number. `None` for an hourly-or-slower
/// cadence, or a cron that doesn't parse (that error is surfaced by [`validate_cron`]).
pub fn schedule_cr_growth_warning(cron: &str) -> Option<String> {
    let fires = schedule_fires_per_hour(cron)?;
    if fires <= 1 {
        return None;
    }
    Some(format!(
        "schedule fires ~{fires}×/hour and each fire creates a Snapshot CR per source, kept \
         for the SnapshotPolicy retention window — this can add up to thousands of CRs. If \
         unintended, use a coarser schedule or bound SnapshotPolicy.spec.retention. See \
         https://kopiur.home-operations.com/backups/#how-many-snapshot-crs-will-i-have"
    ))
}

/// A non-blocking admission WARNING (#547) when a CR's name is longer than
/// [`crate::consts::JOB_NAME_MAX`] (57) and its mover `Job` takes that exact name
/// (a `Snapshot`'s backup Job, a `Restore`'s Job). Kubernetes 1.37's
/// `WorkloadWithJob` derives a `<job>-pgt-0` name that must fit 63 characters, so
/// such a Job breaks there. The Job cannot simply be renamed (its name is the
/// lookup key tying it to the CR), so admission says so up front. `None` when the
/// name fits.
pub fn job_name_length_warning(kind: &str, name: &str) -> Option<String> {
    use crate::consts::JOB_NAME_MAX;
    (name.len() > JOB_NAME_MAX).then(|| {
        format!(
            "{kind} name `{name}` is {} characters. Its mover Job uses the same name, and \
             Kubernetes 1.37+ (WorkloadWithJob) needs Job names of at most {JOB_NAME_MAX} \
             characters. Fix: use a {kind} name of {JOB_NAME_MAX} characters or fewer",
            name.len()
        )
    })
}

/// The scheduled-run sibling of [`job_name_length_warning`] (#547): every fire of
/// a `SnapshotSchedule` creates a `Snapshot` named `<schedule>-<YYYYMMDDHHMMSS>`
/// (15 more characters), whose backup Job takes that name. `None` when those
/// names fit [`crate::consts::JOB_NAME_MAX`].
pub fn scheduled_job_name_length_warning(schedule_name: &str) -> Option<String> {
    use crate::consts::JOB_NAME_MAX;
    // `-` + the 14-digit `%Y%m%d%H%M%S` slot stamp.
    const SLOT_SUFFIX: usize = 15;
    let max = JOB_NAME_MAX - SLOT_SUFFIX;
    (schedule_name.len() > max).then(|| {
        format!(
            "SnapshotSchedule name `{schedule_name}` is {} characters, so each run's Snapshot \
             and backup Job get a {}-character name. Kubernetes 1.37+ (WorkloadWithJob) needs \
             Job names of at most {JOB_NAME_MAX} characters. Fix: use a schedule name of {max} \
             characters or fewer",
            schedule_name.len(),
            schedule_name.len() + SLOT_SUFFIX
        )
    })
}

/// Count how many times `cron` fires within one representative active hour. Pure and
/// clock-free — anchored on the cron's FIRST fire from a fixed instant (not a fixed
/// calendar hour), so a day-of-week / day-of-month constrained cron is measured
/// during an hour it is actually active. `H` tokens are substituted to a fixed value
/// first (they pick a minute within the window, not the cadence, so the count is
/// H-independent). `None` if the cron doesn't parse.
fn schedule_fires_per_hour(cron: &str) -> Option<u32> {
    use chrono::{Duration, TimeZone, Utc};
    let probe = cron
        .split_whitespace()
        .map(|f| if f == "H" { "0" } else { f })
        .collect::<Vec<_>>()
        .join(" ");
    let parsed = crate::jitter::cron_parser().parse(&probe).ok()?;
    let anchor = Utc.with_ymd_and_hms(2001, 1, 1, 0, 0, 0).single()?;
    let first = parsed.find_next_occurrence(&anchor, true).ok()?;
    let horizon = first + Duration::hours(1);
    let mut cursor = first;
    let mut count = 0u32;
    // The cron grammar has no seconds field, so at most 60 fires/hour; cap defensively.
    for _ in 0..61 {
        let next = parsed.find_next_occurrence(&cursor, true).ok()?;
        if next >= horizon {
            break;
        }
        count += 1;
        cursor = next + Duration::seconds(1);
    }
    Some(count)
}

/// Validate an optional Go-style `jitter` duration (`30m`, `1h`, …) against the SAME
/// parser the controller uses at scheduling time, so a typo or an out-of-range value
/// is rejected at apply time rather than silently degrading to *no jitter* at the
/// next reconcile (`parse_go_duration` returns `None`, which the schedule treats as a
/// zero offset). `None` (no jitter) is always valid. `field` names the path for the
/// error message (e.g. `spec.schedule.jitter`).
pub fn validate_jitter(field: &str, jitter: Option<&str>) -> ValidationResult {
    if let Some(j) = jitter
        && crate::duration::parse_go_duration(j).is_none()
    {
        return Err(ValidationError::InvalidFieldValue {
            field: field.to_string(),
            reason: format!(
                "{j:?} is not a valid duration. Fix: use a Go-style duration like 30s, 5m, or 1h"
            ),
        });
    }
    Ok(())
}

/// The largest jitter window any cron may carry: 24 hours.
///
/// Jitter is a deterministic spread WITHIN a cron period, not an offset that moves
/// the schedule. Beyond a day the spread exceeds every cadence kopiur schedules
/// (the coarsest built-in default is a daily full maintenance), so a larger value
/// is always a misunderstanding — typically an attempt to say "run 30h after the
/// slot", which jitter cannot express.
pub const MAX_JITTER: std::time::Duration = std::time::Duration::from_secs(86_400);

/// Validate a **present** Go-style `jitter` duration: it must parse (same parser
/// [`validate_jitter`] and the scheduler use) AND must not exceed [`MAX_JITTER`].
///
/// The bounds half is a TIGHTENING of an existing rule, so it is admission-only —
/// the controller re-runs the shared aggregates as a hard stop at reconcile, and
/// adding this to one of them would brick a stored CR carrying an over-24h jitter
/// rather than merely refusing the next edit. Call it from the webhook's per-kind
/// `validate_*_admission_extras`, never from a shared aggregate.
///
/// Takes a bare `&str` (not `Option<&str>`) because every caller is already inside
/// an `if let Some(j)` over the field it is checking.
///
/// ```
/// use kopiur_api::validate::validate_jitter_bounds;
///
/// assert!(validate_jitter_bounds("spec.schedule.jitter", "10m").is_ok());
/// // Exactly the cap is fine; a second past it is not.
/// assert!(validate_jitter_bounds("spec.schedule.jitter", "24h").is_ok());
/// assert!(validate_jitter_bounds("spec.schedule.jitter", "86401s").is_err());
/// // Garbage is rejected by the parse half.
/// assert!(validate_jitter_bounds("spec.schedule.jitter", "soon").is_err());
/// ```
pub fn validate_jitter_bounds(field: &str, jitter: &str) -> ValidationResult {
    // Parse first, reusing `validate_jitter` so a typo reads identically whichever
    // validator catches it. Its `None` arm is the only unparseable case, so a bare
    // `let else` back into that same error keeps this panic-free rather than
    // leaning on an `expect` that "cannot" fire.
    let Some(parsed) = crate::duration::parse_go_duration(jitter) else {
        validate_jitter(field, Some(jitter))?;
        // Unreachable in practice (validate_jitter rejects exactly what fails to
        // parse) — but expressed as a value, not a panic: this is backup software.
        return Ok(());
    };
    if parsed > MAX_JITTER {
        return Err(ValidationError::InvalidFieldValue {
            field: field.to_string(),
            reason: format!(
                "jitter of {jitter} exceeds the 24h maximum. Fix: use a window smaller than \
                 the cron period (e.g. 10m); put a fixed offset in the cron expression instead"
            ),
        });
    }
    Ok(())
}

/// Keys a user may not set in `moverDefaults.podLabels`/`podAnnotations`: anything
/// under kopiur's own domain prefix, plus the exact `app.kubernetes.io/managed-by`
/// key.
///
/// Both are keys the controller stamps itself. Extra pod metadata merges UNDER
/// kopiur's, so a collision is silently DROPPED at render time — the manifest would
/// claim a label the pod never carries. Reject it instead of ignoring it (the same
/// reasoning as the replication mover's `inheritSecurityContextFrom` refusal).
const RESERVED_POD_METADATA_KEYS: &[&str] = &["app.kubernetes.io/managed-by"];

/// kopiur's own label/annotation domain prefix. Everything under it is
/// operator-owned wire contract (see [`crate::consts`]).
const KOPIUR_KEY_PREFIX: &str = "kopiur.home-operations.com/";

/// Validate `moverDefaults.podLabels`/`podAnnotations` key sets, accumulating every
/// reserved key so a user fixes them all in one apply.
///
/// New fields, so this is safe in the shared repository aggregates the controller
/// re-runs: no stored object can carry a key this rejects.
pub fn validate_pod_metadata(defaults: &crate::common::MoverDefaults) -> Vec<ValidationError> {
    let mut errs = Vec::new();
    for (field, map) in [
        ("moverDefaults.podLabels", defaults.pod_labels.as_ref()),
        (
            "moverDefaults.podAnnotations",
            defaults.pod_annotations.as_ref(),
        ),
    ] {
        let Some(map) = map else { continue };
        for key in map.keys() {
            let reserved = key.starts_with(KOPIUR_KEY_PREFIX)
                || RESERVED_POD_METADATA_KEYS.contains(&key.as_str());
            if reserved {
                errs.push(ValidationError::InvalidFieldValue {
                    field: format!("{field}[{key:?}]"),
                    reason: format!(
                        "{key:?} is reserved by kopiur and would be overwritten. Fix: use a key \
                         outside `{KOPIUR_KEY_PREFIX}` (and not `app.kubernetes.io/managed-by`)"
                    ),
                });
            }
        }
    }
    errs
}

/// Validate an optional IANA timezone name against the same `chrono-tz` database the
/// controller uses at scheduling time, so a typo (e.g. `America/Chicgo`) is rejected at
/// apply time rather than silently resolving to UTC at the next reconcile. `None` (use
/// the controller default) is always valid.
///
/// ```
/// use kopiur_api::validate::validate_timezone;
///
/// assert!(validate_timezone(None).is_ok());
/// assert!(validate_timezone(Some("America/Chicago")).is_ok());
/// assert!(validate_timezone(Some("UTC")).is_ok());
/// assert!(validate_timezone(Some("America/Chicgo")).is_err());
/// ```
pub fn validate_timezone(name: Option<&str>) -> ValidationResult {
    match name {
        None => Ok(()),
        Some(tz) if tz.parse::<chrono_tz::Tz>().is_ok() => Ok(()),
        Some(tz) => Err(ValidationError::InvalidTimezone {
            name: tz.to_string(),
        }),
    }
}

/// The shared `spec.server` rules the type system can't express (server addendum):
///   * `auth.insecure` requires `acknowledgeInsecure: true` — a no-auth server exposes
///     full read/read of the repository, so it must be explicit.
///   * `service.port` must be non-zero.
///   * `readOnly: false` is contradictory on a `mode: ReadOnly` repository — a ReadOnly
///     repo can never serve a writable UI, so the explicit denial is rejected (omitting
///     the field is fine; the mode forces read-only).
///
/// `mode` is the parent repository's [`RepositoryMode`] (both callers have it). Accumulates
/// so a user sees every server problem at once. The PVC `ReadWriteMany` requirement for
/// filesystem-backend servers is **not** here — it needs a live PVC read and is enforced
/// at reconcile, not admission.
pub fn validate_server(server: &ServerSpec, mode: RepositoryMode) -> Vec<ValidationError> {
    let mut errs = Vec::new();
    if let Some(ServerAuth::Insecure(ack)) = &server.auth
        && !ack.acknowledge_insecure
    {
        errs.push(ValidationError::InsecureServerNotAcknowledged);
    }
    if let Some(service) = &server.service
        && service.port == Some(0)
    {
        errs.push(ValidationError::InvalidServerPort { port: 0 });
    }
    if server.read_only == Some(false) && !mode.allows_writes() {
        errs.push(ValidationError::InvalidFieldValue {
            field: "server.readOnly".to_string(),
            reason: "a Repository with spec.mode: ReadOnly cannot serve a read-write UI. Fix: \
                     remove server.readOnly, or set spec.mode: ReadWrite"
                .to_string(),
        });
    }
    if let Some(res) = &server.resources
        && let Err(e) = validate_resources(res, "server")
    {
        errs.push(e);
    }
    errs
}

/// `inheritSecurityContextFrom.pvcConsumer` derives the mover identity from a **backup
/// source** PVC's consumer; a kind that has no backup source (Restore, Maintenance) must
/// reject it at admission rather than fail at runtime. `field_prefix` names the owning kind
/// (e.g. `"restore"`/`"maintenance"`), `instead` is the kind-appropriate remedy.
pub(crate) fn forbid_pvc_consumer(
    mover: &MoverSpec,
    field_prefix: &str,
    instead: &str,
) -> ValidationResult {
    if matches!(
        mover.inherit_security_context_from,
        Some(crate::common::InheritSecurityContextFrom::PvcConsumer(_))
    ) {
        return Err(ValidationError::InvalidFieldValue {
            field: format!("{field_prefix}.mover.inheritSecurityContextFrom.pvcConsumer"),
            reason: format!(
                "is only valid for a backup source (there is no source PVC here). {instead}"
            ),
        });
    }
    Ok(())
}

/// `inheritSecurityContextFrom.snapshot` reproduces the identity RECORDED on a backup
/// (`Snapshot.status.recorded`), so it only makes sense where a snapshot is being consumed —
/// a `Restore`. A backup's identity comes from the live workload and maintenance touches no
/// snapshot at all, so those kinds reject the variant at admission rather than fail (or,
/// worse, silently no-op) at runtime. `field_prefix` names the owning kind (e.g.
/// `"snapshotPolicy"`/`"maintenance"`), `reason` is the kind-appropriate what/why/fix.
pub(crate) fn forbid_snapshot_inherit(
    mover: &MoverSpec,
    field_prefix: &str,
    reason: &str,
) -> ValidationResult {
    if matches!(
        mover.inherit_security_context_from,
        Some(crate::common::InheritSecurityContextFrom::Snapshot(_))
    ) {
        return Err(ValidationError::InvalidFieldValue {
            field: format!("{field_prefix}.mover.inheritSecurityContextFrom.snapshot"),
            reason: reason.to_string(),
        });
    }
    Ok(())
}

/// Reject `inheritSecurityContextFrom` **entirely**, for a kind whose reconciler never resolves
/// it. Stronger than [`forbid_pvc_consumer`]: that rejects one variant on a kind that *does*
/// honor the other, this rejects the whole field on a kind that honors none of it.
///
/// Accepting a field and then ignoring it is the failure mode this repo exists to design out —
/// the manifest says the mover runs as the workload, the mover runs as something else, and
/// nothing says otherwise. If a kind cannot honor it, admission must say so.
pub(crate) fn forbid_inherit(
    mover: &MoverSpec,
    field_prefix: &str,
    reason: &str,
) -> ValidationResult {
    if mover.inherit_security_context_from.is_some() {
        return Err(ValidationError::InvalidFieldValue {
            field: format!("{field_prefix}.mover.inheritSecurityContextFrom"),
            reason: reason.to_string(),
        });
    }
    Ok(())
}

#[cfg(test)]
mod tests;