kopiur_api/

identity.rs

//! Kopia identity resolution (ADR §4.2).
//!
//! Kopia records every snapshot under `username@hostname:sourcePath`. Kopiur makes
//! that identity an explicit, overridable part of the API rather than an accident
//! of `metadata.name`/`metadata.namespace` (ADR §2.2 principle 9). This module is
//! the single place the defaulting + templating rules live. The webhook calls it
//! at admission (so a bad expression/component is rejected on `kubectl apply`) and
//! the controller calls it again on every reconcile, resolving from the LIVE
//! `SnapshotPolicy.spec.identity` and the LIVE referenced repository's
//! `identityDefaults` — **not** a value pinned once and frozen. `status.resolved.identity`
//! mirrors the most recent resolution for observability; it is not the source of
//! truth a later run reads back. What actually keeps an already-snapshotted policy
//! stable is the fork guard (`ValidationError::IdentityWouldFork`/
//! `RepositoryIdentityWouldFork`): an edit that would change the resolved identity
//! on a policy (or repository) with existing history is rejected at admission
//! unless acknowledged with the `allow-identity-change` annotation.
//!
//! ## Defaults (ADR §4.2)
//! - `username` ← `SnapshotPolicy.metadata.name`
//! - `hostname` ← namespace
//! - `sourcePath` ← `/pvc/<pvcName>`
//!
//! ## Repository/ClusterRepository identity expressions (CEL)
//!
//! A [`crate::common::IdentityDefaults`] (`Repository.spec.identityDefaults` or
//! `ClusterRepository.spec.identityDefaults`) supplies `hostnameExpr`/
//! `usernameExpr`, **CEL** expressions ([`cel`]) validated (compiled + trial-evaluated
//! via [`validate_identity_expr`]) at admission, then actually rendered against the
//! LIVE consumer + LIVE repository on every reconcile (ADR-0004 §5) — so editing
//! `identityDefaults` re-renders every consumer that resolves through it on its
//! next backup (guarded by [`ValidationError::RepositoryIdentityWouldFork`], see
//! the module intro above). A consumer's explicit [`Identity`] override **always
//! wins** over the expression.
//!
//! ### CEL environment
//!
//! Each expression returns a **string** and is evaluated against:
//! `namespace` (the consumer's namespace), `policyName` (the `SnapshotPolicy`'s
//! name), `labels` and `annotations` (its metadata maps), and `cluster`
//! ([`IdentityDefaults::cluster`], or `""` when unset — see [`identity_context`]).
//! Examples: `hostnameExpr: "namespace"`,
//! `usernameExpr: "namespace + '-' + policyName"`,
//! `"'team' in labels ? labels['team'] : namespace"`,
//! `"namespace + '.' + cluster"`. CEL is sandboxed (no I/O, no arbitrary code); a
//! syntax error or out-of-scope variable is rejected at `kubectl apply` via
//! [`validate_identity_expr`], and a non-string result is a typed error.
//! Expressions are length-capped ([`MAX_EXPR_LEN`]) as the cost-budget surrogate.
//!
//! ### Multi-cluster hostname default
//!
//! When a repository's [`IdentityDefaults::cluster`] is set, the default
//! (no override, no `hostnameExpr`) kopia identity hostname becomes
//! `<namespace>.<cluster>` instead of bare `<namespace>`, so N clusters sharing one
//! repository never collide on a same-named namespace. [`classify_hostname`]
//! recovers the namespace/cluster split on the read path (retention, discovered
//! `Snapshot` placement).

use std::collections::BTreeMap;

use cel::{Context, Program, Value};

use crate::common::{Identity, IdentityDefaults, ResolvedIdentity};
use crate::error::{ValidationError, ValidationResult};

/// Maximum CEL expression length accepted at admission (the cost-budget surrogate;
/// `cel` 0.13 has no built-in cost API). 1 KiB is far beyond any real identity
/// expression and bounds parse/eval work on adversarial input.
pub const MAX_EXPR_LEN: usize = 1024;

/// Inputs to identity resolution. Grouped into a struct so call sites are readable
/// and future inputs slot in without churning the signature.
#[derive(Debug, Clone)]
pub struct IdentityInputs<'a> {
    /// The consumer object's `metadata.name` (default `username`; the `policyName`
    /// CEL variable).
    pub object_name: &'a str,
    /// The consumer object's namespace (default `hostname`; the `namespace` CEL
    /// variable).
    pub namespace: &'a str,
    /// Explicit overrides from `SnapshotPolicy.spec.identity`, if any.
    pub overrides: Option<&'a Identity>,
    /// The referenced repository's `spec.identityDefaults` — `Repository` or
    /// `ClusterRepository`, whichever the consumer targets. Carries the CEL
    /// `*Expr` pair (`hostnameExpr`/`usernameExpr`) AND `cluster`, the
    /// multi-cluster hostname-default suffix (see [`IdentityDefaults::cluster`]
    /// and the module's "Multi-cluster hostname default" section).
    pub defaults: Option<&'a IdentityDefaults>,
    /// The consumer's `metadata.labels`, exposed to CEL as `labels`.
    pub labels: Option<&'a BTreeMap<String, String>>,
    /// The consumer's `metadata.annotations`, exposed to CEL as `annotations`.
    pub annotations: Option<&'a BTreeMap<String, String>>,
    /// The kopia `sourcePath` to record, fully derived by the caller.
    ///
    /// Deliberately ONE field rather than the `pvc_name` / `default_source_path`
    /// / `source_path_override` triple it replaces. Those three invited every
    /// identity site to re-derive the path inline from a source's `pvc`/`nfs`,
    /// and five of them therefore resolved a pathless identity for a `stream`
    /// source while the backup side recorded `/stream/<fileName>` (#451). With
    /// only the answer accepted here, a caller MUST go through
    /// [`crate::expand::identity_source_path`] — the single derivation — so a new
    /// source kind cannot silently resolve the wrong path at some sites and the
    /// right one at others.
    ///
    /// `None` leaves `sourcePath` unset: kopia's identity-only
    /// `username@hostname` form, which matches any path. Used by surfaces that
    /// genuinely have no single source path (a maintenance identity, a selector
    /// policy with no fan-out pin, a zero-source legacy policy).
    pub source_path: Option<&'a str>,
}

/// Compile a CEL identity expression, enforcing the [`MAX_EXPR_LEN`] budget first.
/// Maps a parse failure to [`ValidationError::IdentityExprCompile`].
fn compile_expr(expr: &str) -> ValidationResult<Program> {
    if expr.len() > MAX_EXPR_LEN {
        return Err(ValidationError::IdentityExprCompile {
            expr: expr.to_string(),
            reason: format!(
                "expression is {} bytes; the maximum is {MAX_EXPR_LEN}",
                expr.len()
            ),
        });
    }
    let aliased = alias_namespace_ident(expr);
    Program::compile(&aliased).map_err(|e| ValidationError::IdentityExprCompile {
        expr: expr.to_string(),
        reason: restore_source(&e.to_string(), expr, &aliased),
    })
}

/// The name the `namespace` variable is actually bound under.
///
/// `namespace` is a CEL reserved word: the spec reserves it, and from `cel`
/// 0.14.5 the parser enforces that (as cel-go always has) — `namespace` alone is
/// a parse error. But it is also this API's documented, longest-standing
/// variable, in every example and in live `identityDefaults`; refusing it would
/// break each of those expressions on upgrade, and with them every backup that
/// resolves its identity through one. So it keeps working: before compiling,
/// [`alias_namespace_ident`] rewrites the bare identifier to this name, which is
/// what [`identity_context`] binds.
///
/// Exactly as long as `namespace`, so the rewritten source lines up with the
/// user's column for column: a `cel` error's `line:col` and `^` caret stay
/// true, and [`restore_source`] can map its quoted source back exactly.
const NAMESPACE_ALIAS: &str = "__nsvar__";
const _: () = assert!(NAMESPACE_ALIAS.len() == "namespace".len());

/// Rewrite every bare `namespace` identifier in `expr` to [`NAMESPACE_ALIAS`].
///
/// Token-aware, so only a variable reference changes: string and bytes literals
/// (every quoting and prefix form), `//` comments, and field selectors
/// (`labels.namespace` — the spec permits reserved words there) are copied
/// through untouched. Anything this scan does not understand is copied through
/// too, leaving the parser to report it against the user's own text.
fn alias_namespace_ident(expr: &str) -> String {
    let b = expr.as_bytes();
    let mut out = String::with_capacity(expr.len());
    let mut i = 0;
    // Whether the last significant token was `.`, i.e. an identifier here is a
    // field selector rather than a variable.
    let mut after_dot = false;
    while i < b.len() {
        let c = b[i];
        if c == b'/' && b.get(i + 1) == Some(&b'/') {
            let end = expr[i..].find('\n').map_or(b.len(), |n| i + n);
            out.push_str(&expr[i..end]);
            i = end;
        } else if c == b'"' || c == b'\'' {
            let end = string_end(b, i, false);
            out.push_str(&expr[i..end]);
            i = end;
            after_dot = false;
        } else if c.is_ascii_alphabetic() || c == b'_' {
            let start = i;
            while i < b.len() && (b[i].is_ascii_alphanumeric() || b[i] == b'_') {
                i += 1;
            }
            let word = &expr[start..i];
            let is_prefix = word.len() <= 2 && word.bytes().all(|p| b"rRbB".contains(&p));
            if is_prefix && matches!(b.get(i), Some(b'"' | b'\'')) {
                // A string prefix: `r'…'`, `b"…"`, `rb'''…'''`. Raw strings take
                // no escapes.
                let raw = word.bytes().any(|p| p == b'r' || p == b'R');
                let end = string_end(b, i, raw);
                out.push_str(&expr[start..end]);
                i = end;
            } else if word == "namespace" && !after_dot {
                out.push_str(NAMESPACE_ALIAS);
            } else {
                out.push_str(word);
            }
            after_dot = false;
        } else if c.is_ascii_digit() {
            // A number (`42`, `0xFF`, `3u`, `1e9`): its letters are not identifiers.
            let start = i;
            while i < b.len() && (b[i].is_ascii_alphanumeric() || b[i] == b'_') {
                i += 1;
            }
            out.push_str(&expr[start..i]);
            after_dot = false;
        } else {
            let ch = expr[i..].chars().next().unwrap_or_default();
            if !ch.is_whitespace() {
                after_dot = ch == '.';
            }
            out.push(ch);
            i += ch.len_utf8();
        }
    }
    out
}

/// Index just past the string literal whose opening quote is at `open`, handling
/// `'`/`"` and their tripled forms. An unterminated literal runs to the end.
fn string_end(b: &[u8], open: usize, raw: bool) -> usize {
    let q = b[open];
    let triple = b.get(open + 1) == Some(&q) && b.get(open + 2) == Some(&q);
    let mut i = open + if triple { 3 } else { 1 };
    while i < b.len() {
        if b[i] == b'\\' && !raw {
            i += 2;
        } else if triple {
            if b[i] == q && b.get(i + 1) == Some(&q) && b.get(i + 2) == Some(&q) {
                return i + 3;
            }
            i += 1;
        } else if b[i] == q {
            return i + 1;
        } else {
            i += 1;
        }
    }
    b.len()
}

/// Show a `cel` error against the source the user wrote rather than the
/// [`alias_namespace_ident`] rewrite it parsed.
///
/// `cel` reports each error as `ERROR: <input>:LINE:COL: …` followed by
/// `| <source line LINE>` and a caret line. Only that quoted line is restored,
/// and only when it is exactly the rewritten line LINE: nothing is searched
/// for, so no other text in the message — a user's own `'__nsvar__'` literal,
/// another error's quote — can be touched. The rewrite never adds or removes a
/// line, and the alias is the length of `namespace`, so LINE:COL and the caret
/// need no adjusting. Anything not in that shape is passed through unchanged.
fn restore_source(message: &str, original: &str, aliased: &str) -> String {
    let original: Vec<&str> = original.lines().collect();
    let aliased: Vec<&str> = aliased.lines().collect();
    let mut line = None;
    message
        .split('\n')
        .map(|m| {
            if let Some(n) = error_line(m) {
                line = Some(n);
            } else if let (Some(quoted), Some(n)) = (m.strip_prefix("| "), line.take()) {
                let idx = n.wrapping_sub(1);
                if aliased.get(idx) == Some(&quoted)
                    && let Some(o) = original.get(idx)
                {
                    return format!("| {o}");
                }
            }
            m.to_string()
        })
        .collect::<Vec<_>>()
        .join("\n")
}

/// The 1-based LINE of a `cel` `ERROR: <input>:LINE:COL: …` header line.
fn error_line(m: &str) -> Option<usize> {
    let rest = m.strip_prefix("ERROR: <input>:")?;
    rest.split(':').next()?.parse().ok()
}

/// Build the CEL evaluation context for identity expressions: `namespace`,
/// `policyName`, `labels`, `annotations`, `cluster`. `cluster` is **always**
/// declared (so an expression referencing it never hits an undeclared-variable
/// error), taking `inputs.defaults.and_then(|d| d.cluster.as_deref())` or `""`
/// when the consumer's repository has no `identityDefaults.cluster` set (bare
/// single-cluster deployments never see a `cluster` variable value).
fn identity_context<'a>(inputs: &IdentityInputs<'_>) -> Context<'a> {
    let mut ctx = Context::default();
    // `add_variable` only errors if a value cannot serialize; these are
    // `&str`/`BTreeMap<String,String>`, which always serialize.
    let empty = BTreeMap::<String, String>::new();
    // Bound under its alias: `namespace` itself cannot be referenced in CEL.
    let _ = ctx.add_variable(NAMESPACE_ALIAS, inputs.namespace);
    let _ = ctx.add_variable("policyName", inputs.object_name);
    let _ = ctx.add_variable("labels", inputs.labels.unwrap_or(&empty));
    let _ = ctx.add_variable("annotations", inputs.annotations.unwrap_or(&empty));
    let cluster = inputs
        .defaults
        .and_then(|d| d.cluster.as_deref())
        .unwrap_or("");
    let _ = ctx.add_variable("cluster", cluster);
    ctx
}

/// Evaluate a compiled identity [`Program`] against `inputs`, requiring a string
/// result. Maps evaluation failure to [`ValidationError::IdentityExprEval`] and a
/// non-string result to [`ValidationError::IdentityExprType`].
fn eval_expr(
    expr: &str,
    program: &Program,
    inputs: &IdentityInputs<'_>,
) -> ValidationResult<String> {
    let ctx = identity_context(inputs);
    match program.execute(&ctx) {
        Ok(Value::String(s)) => Ok(s.to_string()),
        Ok(other) => Err(ValidationError::IdentityExprType {
            expr: expr.to_string(),
            got: other.type_of().to_string(),
        }),
        Err(e) => Err(ValidationError::IdentityExprEval {
            expr: expr.to_string(),
            reason: restore_source(&e.to_string(), expr, &alias_namespace_ident(expr)),
        }),
    }
}

/// Compile + evaluate an identity expression in one step.
fn render_expr(expr: &str, inputs: &IdentityInputs<'_>) -> ValidationResult<String> {
    let program = compile_expr(expr)?;
    eval_expr(expr, &program, inputs)
}

/// Validate a `Repository`/`ClusterRepository` `identityDefaults` CEL expression
/// at admission (ADR-0004 §5): it must compile, and — because CEL reports an
/// out-of-scope variable only at *evaluation* time — it must also evaluate
/// against a representative context without referencing an undeclared variable.
/// A non-string result is rejected. Missing *map keys* (e.g. `labels['env']`
/// when the trial data lacks `env`) are tolerated: they are data-dependent, not
/// a structural error.
pub fn validate_identity_expr(expr: &str) -> ValidationResult {
    let program = compile_expr(expr)?;
    // Representative trial context: non-empty maps so `'k' in labels`-style guards
    // behave, plus placeholder scalars. `cluster` gets a non-empty placeholder too
    // (rather than the "unset" empty string [`identity_context`] would otherwise
    // supply) so an expression referencing it — e.g. `namespace + '.' + cluster` —
    // trial-evaluates the same way regardless of whether *this particular*
    // repository happens to set `identityDefaults.cluster`.
    let labels = BTreeMap::from([("app".to_string(), "trial".to_string())]);
    let annotations = BTreeMap::from([("note".to_string(), "trial".to_string())]);
    let trial_defaults = IdentityDefaults {
        cluster: Some("trial-cluster".to_string()),
        hostname_expr: None,
        username_expr: None,
    };
    let inputs = IdentityInputs {
        object_name: "policy",
        namespace: "namespace",
        overrides: None,
        defaults: Some(&trial_defaults),
        labels: Some(&labels),
        annotations: Some(&annotations),
        source_path: None,
    };
    let ctx = identity_context(&inputs);
    match program.execute(&ctx) {
        Ok(Value::String(_)) => Ok(()),
        Ok(other) => Err(ValidationError::IdentityExprType {
            expr: expr.to_string(),
            got: other.type_of().to_string(),
        }),
        // An undeclared-variable reference (a typo / out-of-scope var) is a hard
        // rejection. Other runtime errors (e.g. NoSuchKey on a label the trial data
        // lacks) are data-dependent and tolerated — the real object may supply them.
        Err(cel::ExecutionError::UndeclaredReference(name)) => {
            Err(ValidationError::IdentityExprEval {
                expr: expr.to_string(),
                reason: format!("undeclared reference to '{name}'"),
            })
        }
        Err(_) => Ok(()),
    }
}

/// Resolve a [`ResolvedIdentity`] from defaults, an optional `Repository`/
/// `ClusterRepository` identity expression set, and explicit consumer overrides
/// (ADR §4.2 / ADR-0004 §5).
///
/// Precedence per component: **explicit override, then expression, then default**.
/// For `hostname` specifically, the default step itself has two tiers: with
/// [`IdentityDefaults::cluster`] set, the default is `<namespace>.<cluster>`
/// (M1, multi-cluster shared repositories); otherwise it's the bare `namespace`
/// (ADR §4.2). So the full hostname chain, highest precedence first, is:
/// explicit override, `hostnameExpr`, `<namespace>.<cluster>` (cluster set),
/// `namespace`. `username` is unaffected — `cluster` is not part of its default.
/// Returns a
/// [`ValidationError::IdentityExprCompile`]/`IdentityExprEval`/`IdentityExprType` if a
/// supplied expression fails (so the webhook rejects it at admission rather than
/// pinning garbage).
///
/// ```
/// use kopiur_api::{IdentityInputs, resolve_identity, identity_string};
///
/// // Bare defaults: username <- object name, hostname <- namespace,
/// // sourcePath <- /pvc/<pvcName> (ADR §4.2).
/// let inputs = IdentityInputs {
///     object_name: "postgres-data",
///     namespace: "billing",
///     overrides: None,
///     defaults: None,
///     labels: None,
///     annotations: None,
///     source_path: Some("/pvc/postgres-data"),
/// };
/// let id = resolve_identity(&inputs).unwrap();
/// assert_eq!(id.username, "postgres-data");
/// assert_eq!(id.hostname, "billing");
/// assert_eq!(id.source_path.as_deref(), Some("/pvc/postgres-data"));
/// assert_eq!(identity_string(&id), "postgres-data@billing:/pvc/postgres-data");
/// ```
pub fn resolve_identity(inputs: &IdentityInputs<'_>) -> ValidationResult<ResolvedIdentity> {
    let override_username = inputs.overrides.and_then(|o| o.username.as_deref());
    let override_hostname = inputs.overrides.and_then(|o| o.hostname.as_deref());

    let username = match override_username {
        Some(u) => u.to_string(),
        None => match inputs.defaults.and_then(|t| t.username_expr.as_deref()) {
            Some(expr) => render_expr(expr, inputs)?,
            None => inputs.object_name.to_string(),
        },
    };

    let hostname = match override_hostname {
        Some(h) => h.to_string(),
        None => match inputs.defaults.and_then(|t| t.hostname_expr.as_deref()) {
            Some(expr) => render_expr(expr, inputs)?,
            // No explicit expression: fall back to `<namespace>.<cluster>` when this
            // repository has a cluster identity configured (M1 — distinguishes
            // same-named namespaces across clusters sharing one repository), else the
            // plain namespace (ADR §4.2).
            None => match inputs.defaults.and_then(|d| d.cluster.as_deref()) {
                Some(cluster) => format!("{}.{cluster}", inputs.namespace),
                None => inputs.namespace.to_string(),
            },
        },
    };

    let source_path = inputs.source_path.map(String::from);

    // Shape-check the fully-resolved identity, regardless of where each component came
    // from (explicit override, CEL expression result, or the name/namespace/PVC
    // default). kopia parses `username@hostname:path` on the first `@`/`:` with no
    // escaping, so a component carrying a delimiter/whitespace/control char would
    // silently misparse or become un-findable. Rejecting here means the controller
    // never PINS a bad identity into status, and the webhook surfaces it at admission.
    crate::validate::validate_identity_component("resolved username", &username)?;
    crate::validate::validate_identity_component("resolved hostname", &hostname)?;
    if let Some(p) = &source_path {
        crate::validate::validate_source_path("resolved sourcePath", p)?;
    }

    Ok(ResolvedIdentity {
        username,
        hostname,
        source_path,
    })
}

/// Format a kopia identity string. With a source path: `username@hostname:path`;
/// without one: `username@hostname` (kopia's identity-only form, used for catalog
/// queries that aren't pinned to a path).
///
/// ```
/// use kopiur_api::{IdentityInputs, resolve_identity, identity_string};
///
/// // No PVC => no sourcePath => kopia's identity-only `username@hostname` form.
/// let inputs = IdentityInputs {
///     object_name: "cfg",
///     namespace: "ns",
///     overrides: None,
///     defaults: None,
///     labels: None,
///     annotations: None,
///     source_path: None,
/// };
/// let id = resolve_identity(&inputs).unwrap();
/// assert_eq!(id.source_path, None);
/// assert_eq!(identity_string(&id), "cfg@ns");
/// ```
pub fn identity_string(id: &ResolvedIdentity) -> String {
    match &id.source_path {
        Some(p) => format!("{}@{}:{}", id.username, id.hostname, p),
        None => format!("{}@{}", id.username, id.hostname),
    }
}

/// Classification of a snapshot identity hostname relative to THIS cluster.
/// Kubernetes namespace names cannot contain `.`, so the FIRST `.` unambiguously
/// ends the namespace part of a `<namespace>.<cluster>` hostname. Total: with no
/// cluster identity every hostname is `Bare`; an empty namespace part (".east")
/// or empty suffix ("ns.") also classifies `Bare` (never a panic, never an
/// invalid empty namespace). Suffix comparison is case-sensitive.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum HostClass<'a> {
    /// No dot (or cluster mode off): the legacy `hostname == namespace` form.
    Bare {
        /// The hostname verbatim (there is no separate suffix to strip).
        namespace: &'a str,
    },
    /// `<namespace>.<suffix>` where suffix == this cluster: written by us.
    OwnCluster {
        /// The part of the hostname before the first `.`.
        namespace: &'a str,
    },
    /// `<namespace>.<suffix>` with a different suffix: another cluster's.
    ForeignCluster {
        /// Everything after the first `.` (may itself contain further dots).
        suffix: &'a str,
    },
}

/// Classify a kopia identity `hostname` against `my_cluster`
/// ([`IdentityDefaults::cluster`], resolved from the consuming repository), so a
/// reader (retention pruning, discovered-`Snapshot` placement) can tell whether a
/// hostname it sees was written by this cluster, another cluster sharing the same
/// repository, or predates cluster identity entirely.
///
/// Total and pure — never panics, never fabricates an empty namespace:
/// - `my_cluster` is `None` or `""` (no cluster identity configured): always
///   [`HostClass::Bare`].
/// - No `.` in `hostname`: [`HostClass::Bare`].
/// - The part before the first `.` is empty (e.g. `".east"`), or the part after
///   is empty (e.g. `"ns."`): [`HostClass::Bare`] with the **whole** hostname as
///   `namespace` — these aren't well-formed `<namespace>.<cluster>` hostnames, so
///   classification declines to guess which part is which.
/// - The suffix (everything after the first `.`) exactly equals `my_cluster`:
///   [`HostClass::OwnCluster`].
/// - Otherwise: [`HostClass::ForeignCluster`] — note `"ns.east.x"` under cluster
///   `"east"` is `ForeignCluster { suffix: "east.x" }`, not `OwnCluster`; only an
///   **exact** suffix match is ours.
///
/// ```
/// use kopiur_api::identity::{HostClass, classify_hostname};
///
/// assert_eq!(
///     classify_hostname("billing.east", Some("east")),
///     HostClass::OwnCluster { namespace: "billing" }
/// );
/// assert_eq!(
///     classify_hostname("billing.west", Some("east")),
///     HostClass::ForeignCluster { suffix: "west" }
/// );
/// // No cluster identity configured => every hostname reads as legacy/bare.
/// assert_eq!(
///     classify_hostname("billing.east", None),
///     HostClass::Bare { namespace: "billing.east" }
/// );
/// ```
pub fn classify_hostname<'a>(hostname: &'a str, my_cluster: Option<&str>) -> HostClass<'a> {
    let my_cluster = match my_cluster {
        Some(c) if !c.is_empty() => c,
        _ => {
            return HostClass::Bare {
                namespace: hostname,
            };
        }
    };
    match hostname.split_once('.') {
        None => HostClass::Bare {
            namespace: hostname,
        },
        Some((namespace, suffix)) if namespace.is_empty() || suffix.is_empty() => HostClass::Bare {
            namespace: hostname,
        },
        Some((namespace, suffix)) if suffix == my_cluster => HostClass::OwnCluster { namespace },
        Some((_, suffix)) => HostClass::ForeignCluster { suffix },
    }
}

#[cfg(test)]
mod tests {
    use super::*;

    fn inputs<'a>(
        name: &'a str,
        ns: &'a str,
        overrides: Option<&'a Identity>,
        defaults: Option<&'a IdentityDefaults>,
        source_path: Option<&'a str>,
    ) -> IdentityInputs<'a> {
        IdentityInputs {
            object_name: name,
            namespace: ns,
            overrides,
            defaults,
            labels: None,
            annotations: None,
            source_path,
        }
    }

    fn defaults(host: Option<&str>, user: Option<&str>) -> IdentityDefaults {
        IdentityDefaults {
            cluster: None,
            hostname_expr: host.map(String::from),
            username_expr: user.map(String::from),
        }
    }

    /// Like [`defaults`] but also sets `identityDefaults.cluster` (M1 precedence tests).
    fn defaults_with_cluster(
        host: Option<&str>,
        user: Option<&str>,
        cluster: Option<&str>,
    ) -> IdentityDefaults {
        IdentityDefaults {
            cluster: cluster.map(String::from),
            hostname_expr: host.map(String::from),
            username_expr: user.map(String::from),
        }
    }

    #[test]
    fn nfs_source_uses_default_source_path() {
        // No PVC, but an NFS export supplies the sourcePath (derived by
        // `expand::identity_source_path` from the export path).
        let mut i = inputs("media", "default", None, None, None);
        i.source_path = Some("/mnt/eros/Media");
        let r = resolve_identity(&i).unwrap();
        assert_eq!(r.source_path.as_deref(), Some("/mnt/eros/Media"));
        assert_eq!(identity_string(&r), "media@default:/mnt/eros/Media");
    }

    #[test]
    fn override_beats_default_source_path() {
        // `sourcePathOverride` precedence now lives in
        // `EffectiveSource::kopia_source_path`, which is what supplies this field —
        // so the kernel simply records what it is handed.
        let mut i = inputs("media", "default", None, None, None);
        i.source_path = Some("/data");
        let r = resolve_identity(&i).unwrap();
        assert_eq!(r.source_path.as_deref(), Some("/data"));
    }

    #[test]
    fn defaults_use_name_namespace_and_pvc_path() {
        let r = resolve_identity(&inputs(
            "postgres-data",
            "billing",
            None,
            None,
            Some("/pvc/postgres-data"),
        ))
        .unwrap();
        assert_eq!(r.username, "postgres-data");
        assert_eq!(r.hostname, "billing");
        assert_eq!(r.source_path.as_deref(), Some("/pvc/postgres-data"));
    }

    // --- M1: identityDefaults.cluster hostname precedence -------------------
    // explicit override > hostnameExpr > `<namespace>.<cluster>` (cluster set) > namespace.

    #[test]
    fn cluster_default_is_used_when_no_override_or_expr() {
        let d = defaults_with_cluster(None, None, Some("east"));
        let r = resolve_identity(&inputs(
            "postgres-data",
            "billing",
            None,
            Some(&d),
            Some("data"),
        ))
        .unwrap();
        assert_eq!(r.hostname, "billing.east");
        // username is unaffected — cluster is not part of its default.
        assert_eq!(r.username, "postgres-data");
    }

    #[test]
    fn no_cluster_no_expr_no_override_falls_back_to_bare_namespace() {
        // The `defaults_use_name_namespace_and_pvc_path` test above already covers
        // `defaults: None`; this covers `defaults: Some(..)` with `cluster: None`.
        let d = defaults_with_cluster(None, None, None);
        let r = resolve_identity(&inputs("cfg", "billing", None, Some(&d), Some("data"))).unwrap();
        assert_eq!(r.hostname, "billing");
    }

    #[test]
    fn explicit_hostname_override_beats_cluster_default() {
        let d = defaults_with_cluster(None, None, Some("east"));
        let ovr = Identity {
            username: None,
            hostname: Some("pinned-host".to_string()),
        };
        let r = resolve_identity(&inputs(
            "postgres-data",
            "billing",
            Some(&ovr),
            Some(&d),
            Some("data"),
        ))
        .unwrap();
        assert_eq!(r.hostname, "pinned-host");
    }

    #[test]
    fn hostname_expr_beats_cluster_default() {
        // Both a hostnameExpr AND a cluster default are set; the expr wins (the
        // cluster tier only applies when there is no expression at all).
        let d = defaults_with_cluster(Some("namespace"), None, Some("east"));
        let r = resolve_identity(&inputs(
            "postgres-data",
            "billing",
            None,
            Some(&d),
            Some("data"),
        ))
        .unwrap();
        assert_eq!(r.hostname, "billing"); // NOT "billing.east"
    }

    #[test]
    fn hostname_expr_can_reference_cluster_variable() {
        // hostnameExpr explicitly renders using `cluster` — the brief's example.
        let d = defaults_with_cluster(Some("namespace + '.' + cluster"), None, Some("east"));
        let r = resolve_identity(&inputs(
            "postgres-data",
            "billing",
            None,
            Some(&d),
            Some("data"),
        ))
        .unwrap();
        assert_eq!(r.hostname, "billing.east");
    }

    #[test]
    fn cluster_variable_is_empty_string_when_defaults_lack_cluster() {
        // Same expression, but `identityDefaults.cluster` isn't set: `cluster`
        // evaluates to "" (identity_context's documented "unset" value).
        let d = defaults_with_cluster(Some("namespace + '.' + cluster"), None, None);
        let r = resolve_identity(&inputs(
            "postgres-data",
            "billing",
            None,
            Some(&d),
            Some("data"),
        ))
        .unwrap();
        assert_eq!(r.hostname, "billing.");
    }

    #[test]
    fn validate_identity_expr_admits_cluster_reference_even_without_a_real_cluster_default() {
        // The *admission-time trial* always declares `cluster` (a placeholder),
        // independent of whether the real ClusterRepository sets
        // identityDefaults.cluster — so this must admit even though no `defaults`
        // is threaded through validate_identity_expr's caller in this test.
        assert!(validate_identity_expr("namespace + '.' + cluster").is_ok());
        assert!(validate_identity_expr("cluster").is_ok());
    }

    #[test]
    fn adr_cluster_repository_cel_example() {
        // ADR-0004 §5 example:
        //   hostnameExpr: "namespace"
        //   usernameExpr: "namespace + '-' + policyName"
        // For namespace `billing`, policy `postgres-data`, must evaluate to
        //   username = billing-postgres-data, hostname = billing.
        let d = defaults(Some("namespace"), Some("namespace + '-' + policyName"));
        let r = resolve_identity(&inputs(
            "postgres-data",
            "billing",
            None,
            Some(&d),
            Some("data"),
        ))
        .unwrap();
        assert_eq!(r.username, "billing-postgres-data");
        assert_eq!(r.hostname, "billing");
    }

    #[test]
    fn cel_conditional_on_labels_resolves() {
        // ADR-0004 §5 conditional: 'team' in labels ? labels['team'] : namespace.
        let d = defaults(
            Some("'team' in labels ? labels['team'] : namespace"),
            Some("namespace + '-' + policyName"),
        );
        let labels = BTreeMap::from([("team".to_string(), "payments".to_string())]);
        let mut i = inputs("atuin", "default", None, Some(&d), Some("data"));
        i.labels = Some(&labels);
        let r = resolve_identity(&i).unwrap();
        assert_eq!(r.hostname, "payments"); // label present → label value
        assert_eq!(r.username, "default-atuin");

        // No `team` label → falls back to namespace.
        let mut i2 = inputs("atuin", "default", None, Some(&d), Some("data"));
        i2.labels = None;
        assert_eq!(resolve_identity(&i2).unwrap().hostname, "default");
    }

    #[test]
    fn override_beats_expr() {
        let d = defaults(Some("namespace"), Some("namespace + '-' + policyName"));
        let ovr = Identity {
            username: Some("custom-user".to_string()),
            hostname: Some("custom-host".to_string()),
        };
        let r = resolve_identity(&inputs("cfg", "ns", Some(&ovr), Some(&d), Some("p"))).unwrap();
        assert_eq!(r.username, "custom-user");
        assert_eq!(r.hostname, "custom-host");
    }

    #[test]
    fn partial_override_falls_through_to_expr_for_the_other_field() {
        let d = defaults(Some("namespace"), Some("namespace + '-' + policyName"));
        // Only hostname overridden; username still comes from the expression.
        let ovr = Identity {
            username: None,
            hostname: Some("pinned-host".to_string()),
        };
        let r = resolve_identity(&inputs(
            "postgres-data",
            "billing",
            Some(&ovr),
            Some(&d),
            Some("d"),
        ))
        .unwrap();
        assert_eq!(r.hostname, "pinned-host");
        assert_eq!(r.username, "billing-postgres-data");
    }

    #[test]
    /// The kernel records `source_path` VERBATIM — it never re-derives, never
    /// prepends `/pvc/`, never substitutes a default. That is the property that
    /// lets `expand::identity_source_path` be the single derivation (#451): if the
    /// kernel massaged the value, a stream path could still come out wrong here.
    fn source_path_is_recorded_verbatim() {
        for path in [
            "/data",
            "/pvc/vol",
            "/stream/postgres.sql",
            "/mnt/eros/Media",
        ] {
            let mut i = inputs("cfg", "ns", None, None, None);
            i.source_path = Some(path);
            let r = resolve_identity(&i).unwrap();
            assert_eq!(r.source_path.as_deref(), Some(path));
        }
    }

    #[test]
    fn no_pvc_yields_no_source_path() {
        let r = resolve_identity(&inputs("cfg", "ns", None, None, None)).unwrap();
        assert_eq!(r.source_path, None);
    }

    #[test]
    fn identity_string_with_and_without_path() {
        let with = ResolvedIdentity {
            username: "postgres-data".into(),
            hostname: "billing".into(),
            source_path: Some("/data".into()),
        };
        assert_eq!(identity_string(&with), "postgres-data@billing:/data");
        let without = ResolvedIdentity {
            source_path: None,
            ..with
        };
        assert_eq!(identity_string(&without), "postgres-data@billing");
    }

    #[test]
    fn resolve_rejects_override_with_kopia_delimiter() {
        // An explicit override carrying kopia's '@' delimiter would misparse — rejected
        // at resolution so the controller never pins it.
        let ovr = Identity {
            username: Some("bad@user".to_string()),
            hostname: None,
        };
        let err = resolve_identity(&inputs("c", "n", Some(&ovr), None, Some("p"))).unwrap_err();
        assert!(matches!(
            err,
            ValidationError::IdentityComponentInvalid { .. }
        ));
    }

    #[test]
    fn resolve_rejects_cel_result_with_delimiter() {
        // A CEL expression that resolves to a string with ':' is caught at resolution
        // (the static override validator can't see CEL results).
        let d = defaults(None, Some("'a:b'"));
        let err = resolve_identity(&inputs("c", "n", None, Some(&d), Some("p"))).unwrap_err();
        assert!(matches!(
            err,
            ValidationError::IdentityComponentInvalid { .. }
        ));
    }

    #[test]
    fn malformed_expr_is_rejected_at_resolve() {
        let d = defaults(Some("namespace +"), None); // syntax error
        let err = resolve_identity(&inputs("c", "n", None, Some(&d), Some("p"))).unwrap_err();
        assert!(matches!(err, ValidationError::IdentityExprCompile { .. }));
    }

    // --- validate_identity_expr (admission-time check, ADR-0004 §5) ---

    #[test]
    fn validate_accepts_valid_string_exprs() {
        assert!(validate_identity_expr("namespace").is_ok());
        assert!(validate_identity_expr("namespace + '-' + policyName").is_ok());
        assert!(validate_identity_expr("'team' in labels ? labels['team'] : namespace").is_ok());
        // Data-dependent map index is tolerated (the real object may supply the key).
        assert!(
            validate_identity_expr("namespace + (labels['env'] == 'prod' ? '-prod' : '')").is_ok()
        );
    }

    #[test]
    fn validate_rejects_syntax_error() {
        let err = validate_identity_expr("namespace +").unwrap_err();
        assert!(matches!(err, ValidationError::IdentityExprCompile { .. }));
    }

    #[test]
    fn validate_rejects_out_of_scope_variable() {
        // `namspace` is a typo — an undeclared reference, caught at trial-eval.
        let err = validate_identity_expr("namspace").unwrap_err();
        assert!(matches!(err, ValidationError::IdentityExprEval { .. }));
    }

    #[test]
    fn validate_rejects_non_string_result() {
        // A bool/int result is not a valid hostname/username.
        let err = validate_identity_expr("1 + 1").unwrap_err();
        assert!(matches!(err, ValidationError::IdentityExprType { .. }));
    }

    #[test]
    fn validate_rejects_over_length_expr() {
        let long = format!("'{}'", "a".repeat(MAX_EXPR_LEN));
        let err = validate_identity_expr(&long).unwrap_err();
        assert!(matches!(err, ValidationError::IdentityExprCompile { .. }));
    }

    // --- classify_hostname (M1) ---

    #[test]
    fn classify_hostname_table() {
        let cases: &[(&str, Option<&str>, HostClass<'_>)] = &[
            // No cluster identity configured at all: always Bare, dot or no dot.
            (
                "billing",
                None,
                HostClass::Bare {
                    namespace: "billing",
                },
            ),
            (
                "billing.east",
                None,
                HostClass::Bare {
                    namespace: "billing.east",
                },
            ),
            // Empty cluster string behaves like None (defensive; shouldn't occur post
            // admission-validation, but classify_hostname is total).
            (
                "billing",
                Some(""),
                HostClass::Bare {
                    namespace: "billing",
                },
            ),
            // No dot in the hostname at all.
            (
                "billing",
                Some("east"),
                HostClass::Bare {
                    namespace: "billing",
                },
            ),
            // Exact suffix match => ours.
            (
                "ns.east",
                Some("east"),
                HostClass::OwnCluster { namespace: "ns" },
            ),
            // Different suffix => another cluster's.
            (
                "ns.west",
                Some("east"),
                HostClass::ForeignCluster { suffix: "west" },
            ),
            // Only an EXACT suffix match is "own" — "east.x" != "east".
            (
                "ns.east.x",
                Some("east"),
                HostClass::ForeignCluster { suffix: "east.x" },
            ),
            // Empty namespace part or empty suffix: Bare, whole hostname verbatim.
            (
                ".east",
                Some("east"),
                HostClass::Bare { namespace: ".east" },
            ),
            ("ns.", Some("east"), HostClass::Bare { namespace: "ns." }),
            // Case-sensitive suffix comparison.
            (
                "ns.East",
                Some("east"),
                HostClass::ForeignCluster { suffix: "East" },
            ),
            // Hostname exactly equal to the cluster name, no dot => Bare, never
            // OwnCluster (there is no namespace/cluster split to make).
            ("east", Some("east"), HostClass::Bare { namespace: "east" }),
        ];
        for (hostname, cluster, expected) in cases {
            assert_eq!(
                classify_hostname(hostname, *cluster),
                *expected,
                "classify_hostname({hostname:?}, {cluster:?})"
            );
        }
    }

    #[test]
    fn classify_hostname_round_trips_the_rendered_default() {
        // For any valid namespace + cluster, classifying the hostname
        // resolve_identity's own default would render must come back OwnCluster
        // with the original namespace — the write path and the read path agree.
        for (ns, cluster) in [
            ("billing", "east"),
            ("a", "b"),
            ("kube-system", "prod-1"),
            ("default", "us-west-2"),
        ] {
            let rendered = format!("{ns}.{cluster}");
            assert_eq!(
                classify_hostname(&rendered, Some(cluster)),
                HostClass::OwnCluster { namespace: ns },
                "rendered={rendered:?} cluster={cluster:?}"
            );
        }
    }

    /// `namespace` is a CEL reserved word (enforced from `cel` 0.14.5); only its
    /// bare variable references may be aliased — never text the user quoted, a
    /// comment, or a field selector, all of which CEL already accepts as written.
    #[test]
    fn alias_rewrites_only_the_namespace_variable() {
        let cases = [
            ("namespace", "__nsvar__"),
            (
                "namespace + '-' + policyName",
                "__nsvar__ + '-' + policyName",
            ),
            ("namespace+namespace", "__nsvar__+__nsvar__"),
            // Not a variable: string/bytes literals in every quoting form.
            ("'namespace'", "'namespace'"),
            (r#""namespace""#, r#""namespace""#),
            ("'''a namespace'''", "'''a namespace'''"),
            (r#""""namespace""""#, r#""""namespace""""#),
            (r"r'namespace\'", r"r'namespace\'"),
            ("b'namespace'", "b'namespace'"),
            ("rb'namespace'", "rb'namespace'"),
            // An escaped quote does not end a literal early.
            (
                r"'it\'s namespace' + namespace",
                r"'it\'s namespace' + __nsvar__",
            ),
            // Field selectors may be reserved words, and stay so.
            ("labels.namespace", "labels.namespace"),
            ("labels . namespace", "labels . namespace"),
            // A comment.
            (
                "namespace // the namespace\n",
                "__nsvar__ // the namespace\n",
            ),
            // Longer identifiers that merely contain it.
            ("namespaces + my_namespace", "namespaces + my_namespace"),
            // Number suffixes are not identifiers.
            ("1e3 + 0xFF", "1e3 + 0xFF"),
            ("'ü' + namespace", "'ü' + __nsvar__"),
        ];
        for (input, want) in cases {
            assert_eq!(alias_namespace_ident(input), want, "input: {input:?}");
        }
    }

    #[test]
    fn namespace_variable_resolves_and_quoted_namespace_stays_text() {
        let labels = BTreeMap::from([("namespace".to_string(), "from-label".to_string())]);
        let d = defaults(
            Some("'namespace=' + namespace"),
            Some("labels.namespace + '/' + labels['namespace']"),
        );
        let mut i = inputs("p", "billing", None, Some(&d), None);
        i.labels = Some(&labels);
        let r = resolve_identity(&i).unwrap();
        assert_eq!(r.hostname, "namespace=billing");
        assert_eq!(r.username, "from-label/from-label");
    }

    #[test]
    fn compile_errors_quote_the_expression_the_user_wrote() {
        let err = validate_identity_expr("namespace +")
            .unwrap_err()
            .to_string();
        assert!(!err.contains(NAMESPACE_ALIAS), "alias leaked: {err}");
        assert!(err.contains("namespace +"), "{err}");
    }

    /// The error must be exactly what `cel` reports for an expression of the
    /// same shape with no reserved word in it — position and caret included.
    #[test]
    fn compile_error_positions_match_the_users_text() {
        let ours = validate_identity_expr("policyName + namespace + )")
            .unwrap_err()
            .to_string();
        let same_shape = Program::compile("policyName + xxxxxxxxx + )")
            .unwrap_err()
            .to_string()
            .replace("xxxxxxxxx", "namespace");
        assert!(
            ours.contains(&same_shape),
            "ours: {ours}\nwant: {same_shape}"
        );
    }

    /// A literal the user wrote is never rewritten in an error, even when it
    /// spells the alias.
    #[test]
    fn compile_errors_keep_a_literal_that_spells_the_alias() {
        let err = validate_identity_expr("'__nsvar__' + namespace +")
            .unwrap_err()
            .to_string();
        assert!(err.contains("'__nsvar__' + namespace +"), "{err}");
    }

    /// Each quoted line is restored by its own line number, so restoring one
    /// can never rewrite text in another — here, line 1's literal that spells
    /// the alias, while line 2 is nothing but the alias.
    #[test]
    fn multiline_errors_restore_each_line_independently() {
        let err = validate_identity_expr("'__nsvar__' + namespace +\nnamespace + )")
            .unwrap_err()
            .to_string();
        assert!(err.contains("<input>:2:13"), "{err}");
        assert!(err.contains("| namespace + )\n"), "{err}");

        // The case a plain search gets wrong: line 2 is the alias and nothing
        // else, which a search would also find inside line 1's literal.
        let expr = "'__nsvar__' + namespace +\nnamespace";
        let restored = restore_source(
            "ERROR: <input>:1:1: x\n| '__nsvar__' + __nsvar__ +\n| ^\n\
             ERROR: <input>:2:1: y\n| __nsvar__\n| ^",
            expr,
            &alias_namespace_ident(expr),
        );
        assert_eq!(
            restored,
            "ERROR: <input>:1:1: x\n| '__nsvar__' + namespace +\n| ^\n\
             ERROR: <input>:2:1: y\n| namespace\n| ^"
        );
    }

    /// A message that is not cel's quoted-source shape passes through as is.
    #[test]
    fn restore_source_leaves_other_messages_alone() {
        let msg = "No such key: __nsvar__";
        assert_eq!(restore_source(msg, "namespace", "__nsvar__"), msg);
    }

    #[test]
    fn other_reserved_words_are_still_rejected() {
        assert!(validate_identity_expr("package").is_err());
    }
}