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;
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's mover Pod can only mount PersistentVolumeClaims in its own \
                     namespace, so a selector cannot reach PVCs in another one. Use one \
                     SnapshotPolicy per namespace (each may point at the same 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 (expected In, NotIn, \
                             Exists or DoesNotExist). An unrecognized operator would be dropped, \
                             WIDENING the selector so PVCs you meant to exclude get backed up"
                        ),
                    });
                }
            }
        }
    }
    Ok(())
}

/// A single backup `Source` is well-formed: **exactly one** of `pvc`,
/// `pvcSelector`, or `nfs` is set (ADR §3.3 — modeled as sibling Options because
/// the forms share `sourcePath*` keys, so it's a webhook check, not an enum). When
/// the source is `nfs`, its server/path are also validated.
pub fn validate_source(source: &Source) -> ValidationResult {
    let set: Vec<&str> = [
        ("pvc", source.pvc.is_some()),
        ("pvcSelector", source.pvc_selector.is_some()),
        ("nfs", source.nfs.is_some()),
    ]
    .into_iter()
    .filter_map(|(name, present)| present.then_some(name))
    .collect();

    match set.as_slice() {
        [] => Err(ValidationError::MissingRequiredField {
            field: "source.pvc, source.pvcSelector, or source.nfs".to_string(),
        }),
        [first, second, ..] => Err(ValidationError::MutuallyExclusive {
            a: (*first).to_string(),
            b: (*second).to_string(),
            context: "snapshot source".to_string(),
        }),
        [_only] => match &source.nfs {
            Some(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: the kubelet \
                                 does not apply fsGroup to in-tree NFS volumes, so a read-write \
                                 mount grants the mover no additional readability and only \
                                 exposes the export to writes. Remove readOnly (NFS is read \
                                 directly), and grant access with mover.podSecurityContext \
                                 supplementalGroups / mover.securityContext runAsUser matching \
                                 the export's ownership, or with a server-side ID remap"
                            .to_string(),
                    });
                }
                validate_nfs_volume(nfs, "snapshot source")
            }
            None => match source.pvc_selector.as_ref() {
                Some(selector) => validate_pvc_selector(selector),
                None => 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 (valid: {}). No PVC can be \
                     provisioned from it — if this value was stored before kopiur enforced \
                     the schema, it was already broken then; edit the resource to one of the \
                     valid modes.",
                    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 may not be combined with other access modes — the \
                     apiserver rejects such a PVC at create time, so the run would wedge in \
                     a retry loop instead of failing here. 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 => {
                "kopia treats an ABSENT field, not 0, as \"no limit\"; 0 would stall the transfer, \
                 so omit the field to run uncapped"
            }
            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}`; the API server rejects a pod whose \
                 requests exceed its limits, so the mover Job would never create a pod (it hangs \
                 instead of failing). 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 bounds how long a non-starting mover pod is tolerated 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: each fire creates one Snapshot CR per source, and \
         they accumulate up to the SnapshotPolicy retention window (CR count ≈ fires × \
         retained snapshots). A sub-hourly schedule with a wide or absent retention can \
         produce thousands of Snapshot CRs, each re-reconciled for its whole retention \
         window. If unintended, use a coarser schedule, bound SnapshotPolicy.spec.retention, \
         or set the Snapshot deletionPolicy to Retain/Orphan. See docs/backups.md \
         ('How many Snapshot CRs will I have?')."
    ))
}

/// 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. 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 — jitter is a per-slot spread \
                 within a cron period, not a schedule offset; use a window smaller than the \
                 cron period (e.g. 10m), or move the offset into the cron expression"
            ),
        });
    }
    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 — the operator stamps this key on every \
                         mover pod and its own value wins the merge, so the value here would be \
                         silently dropped. 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; remove \
                     server.readOnly (the ReadOnly mode forces the UI read-only) or set the \
                     repository's 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 to derive a \
                 workload from. {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;