//! Standalone file-based resource source (`resources_file` in
//! config.yaml).
//!
//! Loads every dynamic resource — provider keys, models, API keys,
//! guardrails and their attachments, MCP servers, A2A agents, cache
//! policies, observability exporters, rate-limit policies — from one YAML
//! file instead of etcd, so a single container can run fully
//! declaratively.
//!
//! Pipeline (identical for boot, SIGHUP reload, and `aisix validate`):
//!
//! ```text
//! read file → YAML parse → ${VAR} interpolation (string scalars)
//!   → per-entry JSON documents → file sugar desugared
//!   → canonical JSON-Schema validation (same validators as the etcd path)
//!   → typed serde models → cross-reference checks → AisixSnapshot
//! ```
//!
//! Errors are collected across the whole file — every load problem is
//! reported together with kind / entry / field context, instead of
//! failing on the first one. (Within a single entry, interpolation /
//! shape problems report the first offending field; aggregation is
//! per-entry and across entries.)
//!
//! File format v1 (`_format_version: "1"`, mandatory):
//! - fourteen top-level collection keys, each a sequence of maps, named
//!   by the plural resource kind; unknown top-level keys are load errors.
//! - after desugaring, every entry must be exactly a canonical resource
//!   document (`schemas/resources/*.schema.json`) — the file source
//!   never relaxes the canonical schemas.
//! - entries carry no `id`: ids are derived deterministically
//!   (UUIDv5 of `"<kind>/<identity>"`, see
//!   [`desugar::FILE_RESOURCE_NAMESPACE`]) and identities must be
//!   unique per kind.
//! - cross-collection references are written as the identity the target
//!   collection is keyed by, and desugaring rewrites them to derived ids
//!   — including a guardrail attachment's `guardrail_id` and `scope_id`.
//!   A guardrail with no attachment is not an error: its scope target may
//!   simply be gone, and a guardrail's scope is its attachments and
//!   nothing else, so it governs nothing until something attaches it.

mod desugar;
mod status;
mod yaml;

pub use desugar::{derive_id, FILE_RESOURCE_NAMESPACE};
pub use status::load_resources_file_tracked;

use serde_json::Value;
use std::collections::BTreeMap;
use std::path::Path;
use yaml_rust2::{Yaml, YamlLoader};

use crate::models::{
    validate_a2a_agent, validate_apikey, validate_cache_policy, validate_claim_mapping,
    validate_guardrail, validate_guardrail_attachment, validate_mcp_auth_settings,
    validate_mcp_server, validate_model, validate_observability_exporter, validate_oidc_provider,
    validate_passthrough_route, validate_provider_key, validate_rate_limit_policy, A2aAgent,
    ApiKey, CachePolicy, ClaimMapping, Guardrail, GuardrailAttachment, McpAuthSettings, McpServer,
    Model, ObservabilityExporter, OidcProvider, PassthroughRoute, ProviderKey, RateLimitPolicy,
    SchemaError,
};
use crate::resource::ResourceEntry;
use crate::AisixSnapshot;

use desugar::{IdentityField, IdentityMaps};
use yaml::EnvLookup;

/// One load problem, with enough context to fix it: the entry scope
/// (`models[2] ("gpt-4o")`, or `(file)` for file-level problems) and a
/// message that names the offending field where applicable.
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct LoadError {
    pub scope: String,
    pub message: String,
}

impl std::fmt::Display for LoadError {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        write!(f, "{}: {}", self.scope, self.message)
    }
}

/// Aggregated load failure: every error found across the whole file.
#[derive(Debug)]
pub struct FileSourceErrors {
    /// The file the errors refer to, as given by the caller.
    pub file: String,
    pub errors: Vec<LoadError>,
}

impl std::fmt::Display for FileSourceErrors {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        writeln!(
            f,
            "resources file {}: {} error(s):",
            self.file,
            self.errors.len()
        )?;
        for e in &self.errors {
            writeln!(f, "  - {e}")?;
        }
        Ok(())
    }
}

impl std::error::Error for FileSourceErrors {}

/// The supported format version. A missing or unrecognized
/// `_format_version` is a load error, so future format revisions can
/// change semantics without silently misreading old gateways' files.
const SUPPORTED_FORMAT_VERSION: &str = "1";

/// True when a URL embeds a credential-bearing query parameter, which must
/// not sit in a public JWKS endpoint. String-scanned rather than
/// URL-parsed to avoid pulling a URL crate into `aisix-core`.
pub(crate) fn url_has_credentials(url: &str) -> bool {
    if let Some((_, query)) = url.split_once('?') {
        let query = query.split('#').next().unwrap_or(query);
        for pair in query.split('&') {
            let key = pair.split('=').next().unwrap_or(pair).to_ascii_lowercase();
            if matches!(
                key.as_str(),
                "access_token" | "token" | "client_secret" | "password" | "api_key" | "apikey"
            ) {
                return true;
            }
        }
    }
    false
}

/// Fixed processing order for the fourteen resource collections.
const KINDS: [(&str, IdentityField); 14] = [
    ("provider_keys", IdentityField::DisplayName),
    ("models", IdentityField::DisplayName),
    ("api_keys", IdentityField::DisplayName),
    ("guardrails", IdentityField::Name),
    ("guardrail_attachments", IdentityField::AttachmentTriple),
    ("mcp_servers", IdentityField::NameOrDisplayName),
    ("a2a_agents", IdentityField::NameOrDisplayName),
    ("cache_policies", IdentityField::Name),
    ("observability_exporters", IdentityField::Name),
    ("rate_limit_policies", IdentityField::Name),
    ("oidc_providers", IdentityField::Name),
    ("claim_mappings", IdentityField::Name),
    ("passthrough_routes", IdentityField::NameOrDisplayName),
    // Singleton: the fixed identity makes a second entry a duplicate
    // at pass 1, enforcing at-most-one row per file.
    (
        "mcp_auth_settings",
        IdentityField::Fixed("mcp_auth_settings"),
    ),
];

/// One id-form model reference a document can carry: the field, the
/// name-form field that replaces it, and how an operator should spell that
/// name form.
///
/// `hint` is not decoration. It is the same as `name_field` for every
/// reference whose name form takes a bare model name, and differs for the
/// one that does not: a cache policy's model scope is written into the
/// free-form `applies_to`, where a bare name parses as no discriminator at
/// all and the policy silently WIDENS to every request instead of failing.
/// An error message that told an operator to "use `applies_to`" would be
/// walking them into that.
pub struct ModelRefIdField {
    pub field: &'static str,
    pub name_field: &'static str,
    pub hint: &'static str,
}

const fn pair(field: &'static str, name_field: &'static str) -> ModelRefIdField {
    ModelRefIdField {
        field,
        name_field,
        hint: name_field,
    }
}

/// The id-form model references a document of `kind` can carry.
///
/// A projected document may point at a Model by resource id instead of by
/// display name, which is what makes a reference survive a rename of the
/// model. The list is public because two consumers must agree on it: the
/// resources file refuses the id form (see [`load_from_str`]) and `aisix
/// export` rewrites it back to the name form, and a field one of them
/// knows about and the other does not is a silent round-trip loss.
pub fn model_ref_id_fields(kind: &str) -> &'static [ModelRefIdField] {
    const API_KEYS: [ModelRefIdField; 1] = [pair("allowed_model_ids", "allowed_models")];
    const MODELS: [ModelRefIdField; 4] = [
        pair("model_id", "model"),
        pair("target_id", "target"),
        pair("embedding_model_id", "embedding_model"),
        pair("default_id", "default"),
    ];
    const CACHE_POLICIES: [ModelRefIdField; 2] = [
        ModelRefIdField {
            field: "applies_to_model_id",
            name_field: "applies_to",
            hint: "applies_to: \"model:<name>\"",
        },
        pair("embedding_model_id", "embedding_model"),
    ];
    const GUARDRAILS: [ModelRefIdField; 1] = [pair("embedding_model_id", "embedding_model")];
    match kind {
        "api_keys" => &API_KEYS,
        "models" => &MODELS,
        "cache_policies" => &CACHE_POLICIES,
        "guardrails" => &GUARDRAILS,
        _ => &[],
    }
}

/// One id-form MCP server reference a document can carry: where it sits,
/// the field, and the name-form field that replaces it.
///
/// The name form spells the server as the `<server>` half of a namespaced
/// `<server>__<tool>` pattern (an ACL side), as a map key (the per-server
/// rate limits), or as a bare entry in a list (the anonymous ceiling);
/// either way the file names the server, and the id form is what a file
/// cannot express.
pub struct McpRefIdField {
    /// Object path from the document root to the object carrying `field`.
    pub path: &'static [&'static str],
    pub field: &'static str,
    pub name_field: &'static str,
}

impl McpRefIdField {
    /// Dotted path an operator sees in the error, e.g. `mcp_access.allow_ids`.
    fn display_path(&self) -> String {
        let mut out = String::new();
        for segment in self.path {
            out.push_str(segment);
            out.push('.');
        }
        out.push_str(self.field);
        out
    }
}

/// The id-form MCP server references a document of `kind` can carry.
///
/// Same contract as [`model_ref_id_fields`], for the other reference that
/// has an id spelling: the resources file refuses it (see [`load_from_str`])
/// and `aisix export` rewrites it back to the name form, and a field one of
/// them knows about and the other does not is a silent round-trip loss.
///
/// `mcp_policies` is deliberately absent — the file source carries no such
/// collection, so there is no document of that kind for a file to reject.
pub fn mcp_ref_id_fields(kind: &str) -> &'static [McpRefIdField] {
    const MCP_AUTH_SETTINGS: [McpRefIdField; 1] = [McpRefIdField {
        path: &["anonymous"],
        field: "server_ids",
        name_field: "servers",
    }];
    const API_KEYS: [McpRefIdField; 3] = [
        McpRefIdField {
            path: &[],
            field: "mcp_rate_limits_by_id",
            name_field: "mcp_rate_limits",
        },
        McpRefIdField {
            path: &["mcp_access"],
            field: "allow_ids",
            name_field: "allow",
        },
        McpRefIdField {
            path: &["mcp_access"],
            field: "deny_ids",
            name_field: "deny",
        },
    ];
    match kind {
        "api_keys" => &API_KEYS,
        "mcp_auth_settings" => &MCP_AUTH_SETTINGS,
        _ => &[],
    }
}

/// The first id-form MCP reference `doc` carries, if any.
fn mcp_ref_id_field(kind: &str, doc: &Value) -> Option<&'static McpRefIdField> {
    mcp_ref_id_fields(kind).iter().find(|f| {
        let mut node = doc;
        for segment in f.path {
            match node.get(segment) {
                Some(next) => node = next,
                None => return false,
            }
        }
        node.get(f.field).is_some()
    })
}

/// Call `f` on every object in `doc` that may carry one of
/// [`model_ref_id_fields`]'s fields, for a document of `kind`.
///
/// Addressed by path rather than by walking the whole document for the
/// field names: `secrets` and `headers` on a guardrail are operator-keyed
/// maps, so a blind walk would treat a secret named `model_id` as a model
/// reference and rewrite it. Extend this when a new nesting site gains a
/// model reference.
pub fn for_each_model_ref_node(
    kind: &str,
    doc: &mut Value,
    f: &mut dyn FnMut(&mut serde_json::Map<String, Value>),
) {
    let Some(root) = doc.as_object_mut() else {
        return;
    };
    let objects_in =
        |node: &mut Value, key: &str, f: &mut dyn FnMut(&mut serde_json::Map<String, Value>)| {
            if let Some(Value::Array(items)) = node.get_mut(key) {
                for item in items {
                    if let Some(obj) = item.as_object_mut() {
                        f(obj);
                    }
                }
            }
        };
    match kind {
        // The key's grant list is a root field; a guardrail's kind config
        // is `#[serde(flatten)]`ed onto the root, so its embedder is too.
        "api_keys" | "guardrails" => f(root),
        "cache_policies" => {
            f(root);
            if let Some(Value::Object(semantic)) = root.get_mut("semantic") {
                f(semantic);
            }
        }
        "models" => {
            if let Some(routing) = root.get_mut("routing") {
                objects_in(routing, "targets", f);
            }
            if let Some(ensemble) = root.get_mut("ensemble") {
                objects_in(ensemble, "panel", f);
                if let Some(Value::Object(judge)) = ensemble.get_mut("judge") {
                    f(judge);
                }
            }
            if let Some(semantic) = root.get_mut("semantic") {
                objects_in(semantic, "routes", f);
                if let Some(Value::Object(on_failure)) = semantic.get_mut("on_embedding_failure") {
                    f(on_failure);
                }
                if let Some(Value::Object(on_failure)) = semantic
                    .get_mut("classifier")
                    .and_then(|c| c.get_mut("on_failure"))
                {
                    f(on_failure);
                }
                if let Some(semantic) = semantic.as_object_mut() {
                    f(semantic);
                }
            }
        }
        _ => {}
    }
}

/// The first id-form model reference `doc` carries.
fn model_ref_id_field(kind: &str, doc: &mut Value) -> Option<&'static ModelRefIdField> {
    let fields = model_ref_id_fields(kind);
    if fields.is_empty() {
        return None;
    }
    let mut found = None;
    for_each_model_ref_node(kind, doc, &mut |node| {
        if found.is_none() {
            found = fields.iter().find(|f| node.contains_key(f.field));
        }
    });
    found
}

/// Load `path` into a fresh [`AisixSnapshot`], resolving `${VAR}`
/// interpolation against the current process environment. `revision` is
/// stamped on every entry (the file source's generation counter: 1 at
/// boot, incremented per successful reload).
pub fn load_resources_file(path: &Path, revision: i64) -> Result<AisixSnapshot, FileSourceErrors> {
    let label = path.display().to_string();
    let contents = std::fs::read_to_string(path).map_err(|e| FileSourceErrors {
        file: label.clone(),
        errors: vec![LoadError {
            scope: "(file)".into(),
            message: format!("cannot read file: {e}"),
        }],
    })?;
    load_from_str(&contents, &label, revision, &|name| {
        std::env::var(name).ok()
    })
}

/// The full pipeline over in-memory contents. Separated from
/// [`load_resources_file`] so tests can inject file contents and a
/// closed environment map without touching process state.
pub fn load_from_str(
    contents: &str,
    file_label: &str,
    revision: i64,
    env: EnvLookup<'_>,
) -> Result<AisixSnapshot, FileSourceErrors> {
    let mut errors: Vec<LoadError> = Vec::new();
    let fail = |errors: Vec<LoadError>| FileSourceErrors {
        file: file_label.to_string(),
        errors,
    };
    let file_error = |message: String| LoadError {
        scope: "(file)".into(),
        message,
    };

    // ── YAML parse ────────────────────────────────────────────────────
    let docs = match YamlLoader::load_from_str(contents) {
        Ok(docs) => docs,
        Err(e) => return Err(fail(vec![file_error(format!("YAML parse error: {e}"))])),
    };
    let root = match docs.len() {
        0 => return Err(fail(vec![file_error("file is empty".into())])),
        1 => &docs[0],
        n => {
            return Err(fail(vec![file_error(format!(
                "expected a single YAML document, found {n}"
            ))]))
        }
    };
    let Yaml::Hash(root_map) = root else {
        return Err(fail(vec![file_error(
            "top level must be a mapping with `_format_version` and resource collections".into(),
        )]));
    };

    // ── Format-version gate + unknown-top-level-key gate ─────────────
    match root_map.get(&Yaml::String("_format_version".into())) {
        Some(Yaml::String(v)) if v == SUPPORTED_FORMAT_VERSION => {}
        Some(Yaml::String(v)) => errors.push(file_error(format!(
            "unrecognized _format_version {v:?} (supported: \"{SUPPORTED_FORMAT_VERSION}\")"
        ))),
        Some(_) => errors.push(file_error(format!(
            "_format_version must be the string \"{SUPPORTED_FORMAT_VERSION}\" (quote it)"
        ))),
        None => errors.push(file_error(format!(
            "missing mandatory _format_version (expected \"{SUPPORTED_FORMAT_VERSION}\")"
        ))),
    }
    for key in root_map.keys() {
        let Yaml::String(key) = key else {
            errors.push(file_error(format!(
                "top-level keys must be plain strings, found {key:?}"
            )));
            continue;
        };
        // Named ahead of the generic unknown-key error: `pricing` is a
        // real collection the gateway loads from etcd, so "unknown
        // top-level key" would read as a typo rather than as the answer
        // it is. Pricing documents are a control-plane projection —
        // shared across environments and keyed by control-plane id — and
        // a file that declared its own could not be the same document any
        // other environment reads. Set `cost` on the model instead.
        if key == "pricing" {
            errors.push(file_error(
                "the resources file does not accept a `pricing` collection — pricing \
                 documents are written by the control plane and shared across \
                 environments, which a file cannot express; set `cost` on each model \
                 instead"
                    .to_string(),
            ));
            continue;
        }
        if key != "_format_version" && !KINDS.iter().any(|(k, _)| k == key) {
            let known: Vec<&str> = KINDS.iter().map(|(k, _)| *k).collect();
            errors.push(file_error(format!(
                "unknown top-level key `{key}` (expected _format_version and resource \
                 collections: {})",
                known.join(", ")
            )));
        }
    }

    // ── Pass 1: interpolate + convert + identity / duplicate checks ──
    struct Prepared {
        kind: &'static str,
        scope: String,
        identity: String,
        doc: Value,
    }
    let mut prepared: Vec<Prepared> = Vec::new();
    let mut identity_maps = IdentityMaps::new();

    for (kind, identity_field) in KINDS {
        let mut seen_at: BTreeMap<String, usize> = BTreeMap::new();
        let entries = match root_map.get(&Yaml::String(kind.into())) {
            None | Some(Yaml::Null) => continue,
            Some(Yaml::Array(items)) => items,
            Some(_) => {
                errors.push(file_error(format!("`{kind}` must be a sequence of maps")));
                continue;
            }
        };
        for (i, item) in entries.iter().enumerate() {
            let index_scope = format!("{kind}[{i}]");
            if !matches!(item, Yaml::Hash(_)) {
                errors.push(LoadError {
                    scope: index_scope,
                    message: "entry must be a mapping".into(),
                });
                continue;
            }
            let doc = match yaml::yaml_to_json(item, "", env) {
                Ok(doc) => doc,
                Err((path, message)) => {
                    errors.push(LoadError {
                        scope: index_scope,
                        message: format!("field `{path}`: {message}"),
                    });
                    continue;
                }
            };
            let Some(identity) = identity_field.extract(&doc) else {
                errors.push(LoadError {
                    scope: index_scope,
                    message: format!(
                        "{} is required and must be a non-empty string \
                         (it is the entry's identity)",
                        identity_field.describe()
                    ),
                });
                continue;
            };
            let scope = format!("{kind}[{i}] ({identity:?})");
            if let Some(first) = seen_at.get(&identity) {
                errors.push(LoadError {
                    scope,
                    message: format!(
                        "duplicate {kind} entry: {} {identity:?} is already \
                         defined at {kind}[{first}] — identities must be unique \
                         within a kind",
                        identity_field.describe()
                    ),
                });
                continue;
            }
            seen_at.insert(identity.clone(), i);
            identity_maps
                .entry(kind)
                .or_default()
                .insert(identity.clone(), derive_id(kind, &identity));
            prepared.push(Prepared {
                kind,
                scope,
                identity,
                doc,
            });
        }
    }

    // ── Pass 2: desugar → canonical validation → typed models ────────
    fn finish<T: serde::de::DeserializeOwned>(
        scope: &str,
        doc: &Value,
        validate: fn(&Value) -> Result<(), SchemaError>,
        errors: &mut Vec<LoadError>,
    ) -> Option<T> {
        if let Err(e) = validate(doc) {
            errors.push(LoadError {
                scope: scope.to_string(),
                message: e.to_string(),
            });
            return None;
        }
        match serde_json::from_value::<T>(doc.clone()) {
            Ok(t) => Some(t),
            Err(e) => {
                errors.push(LoadError {
                    scope: scope.to_string(),
                    message: format!("cannot decode canonical document: {e}"),
                });
                None
            }
        }
    }

    // Typed buckets carry `(derived_id, scope, value)` so cross-reference
    // errors can point back at the file entry.
    let mut models: Vec<(String, String, Model)> = Vec::new();
    let mut apikeys: Vec<(String, String, ApiKey)> = Vec::new();
    let mut provider_keys: Vec<(String, String, ProviderKey)> = Vec::new();
    let mut guardrails: Vec<(String, String, Guardrail)> = Vec::new();
    let mut guardrail_attachments: Vec<(String, String, GuardrailAttachment)> = Vec::new();
    let mut mcp_servers: Vec<(String, String, McpServer)> = Vec::new();
    let mut a2a_agents: Vec<(String, String, A2aAgent)> = Vec::new();
    let mut cache_policies: Vec<(String, String, CachePolicy)> = Vec::new();
    let mut observability_exporters: Vec<(String, String, ObservabilityExporter)> = Vec::new();
    let mut rate_limit_policies: Vec<(String, String, RateLimitPolicy)> = Vec::new();
    let mut oidc_providers: Vec<(String, String, OidcProvider)> = Vec::new();
    let mut claim_mappings: Vec<(String, String, ClaimMapping)> = Vec::new();
    let mut passthrough_routes: Vec<(String, String, PassthroughRoute)> = Vec::new();
    let mut mcp_auth_settings: Vec<(String, String, McpAuthSettings)> = Vec::new();

    for mut entry in prepared {
        let id = derive_id(entry.kind, &entry.identity);
        let scope = entry.scope;

        // The file accepts no `id` field on any kind: ids are always
        // derived. Strict schemas would reject it as an unknown field
        // anyway; the open ones (guardrail, cache_policy,
        // observability_exporter) would silently carry it, so the check
        // is made explicit and uniform here.
        if entry.doc.get("id").is_some() {
            errors.push(LoadError {
                scope,
                message: "the resources file does not accept `id` — ids are derived \
                          deterministically from the entry's name"
                    .into(),
            });
            continue;
        }

        // `pricing_key` is the same class of control-plane projection as
        // the model-reference id spellings below: it names a pricing
        // document a file has no way to declare, so a file that carried
        // it would leave the model with no price at all — silently, and
        // with `cost` the only thing that could have supplied one.
        if entry.kind == "models" && entry.doc.get("pricing_key").is_some() {
            errors.push(LoadError {
                scope,
                message: "the resources file does not accept `pricing_key` — it names a \
                          pricing document written by the control plane, which a file \
                          cannot declare; set the price inline with `cost`"
                    .into(),
            });
            continue;
        }

        // Every model reference has an id spelling that names the model by
        // the id the control plane assigned it. A file's ids are derived
        // from its entry names, so no id a file can carry ever resolves:
        // accepting one would make the reference point at nothing —
        // silently, and with the name spelling ignored on top. The file
        // rejects the id spelling instead, wherever it appears. (The etcd
        // path is the opposite: there an id that resolves to no model
        // degrades that one reference and must never fail the row, because
        // a rejected api_key stops authenticating entirely.)
        if let Some(reference) = model_ref_id_field(entry.kind, &mut entry.doc) {
            let (field, hint) = (reference.field, reference.hint);
            errors.push(LoadError {
                scope,
                message: format!(
                    "the resources file does not accept `{field}` — it names a model by \
                     control-plane id, which a file cannot resolve; name the model with \
                     `{hint}` instead"
                ),
            });
            continue;
        }

        // The same rule for the other reference with an id spelling: an
        // MCP server named by the id the control plane assigned it. A file
        // registers its servers by name and derives their ids from those
        // names, so an id a file carries resolves to nothing — the grant
        // would silently cover no tool, or the limit bind to no server,
        // with the name spelling ignored on top.
        if let Some(reference) = mcp_ref_id_field(entry.kind, &entry.doc) {
            let (path, name_field) = (reference.display_path(), reference.name_field);
            errors.push(LoadError {
                scope,
                message: format!(
                    "the resources file does not accept `{path}` — it names an MCP server by \
                     control-plane id, which a file cannot resolve; name the server with \
                     `{name_field}` instead"
                ),
            });
            continue;
        }

        let sugar_result = match entry.kind {
            "models" => desugar::desugar_model(&mut entry.doc, &identity_maps),
            "api_keys" => desugar::desugar_api_key(&mut entry.doc, env),
            "rate_limit_policies" => {
                desugar::desugar_rate_limit_policy(&mut entry.doc, &identity_maps)
            }
            "claim_mappings" => desugar::desugar_claim_mapping(&mut entry.doc, &identity_maps),
            "guardrail_attachments" => {
                desugar::desugar_guardrail_attachment(&mut entry.doc, &identity_maps)
            }
            "passthrough_routes" => {
                desugar::desugar_passthrough_route(&mut entry.doc, &identity_maps)
            }
            _ => Ok(()),
        };
        if let Err(message) = sugar_result {
            errors.push(LoadError { scope, message });
            continue;
        }

        match entry.kind {
            "provider_keys" => {
                if let Some(t) = finish(&scope, &entry.doc, validate_provider_key, &mut errors) {
                    provider_keys.push((id, scope, t));
                }
            }
            "models" => {
                if let Some(t) = finish(&scope, &entry.doc, validate_model, &mut errors) {
                    models.push((id, scope, t));
                }
            }
            "api_keys" => {
                if let Some(t) = finish(&scope, &entry.doc, validate_apikey, &mut errors) {
                    apikeys.push((id, scope, t));
                }
            }
            "guardrails" => {
                if let Some(t) = finish(&scope, &entry.doc, validate_guardrail, &mut errors) {
                    guardrails.push((id, scope, t));
                }
            }
            "guardrail_attachments" => {
                if let Some(t) = finish(
                    &scope,
                    &entry.doc,
                    validate_guardrail_attachment,
                    &mut errors,
                ) {
                    guardrail_attachments.push((id, scope, t));
                }
            }
            "mcp_servers" => {
                if let Some(t) =
                    finish::<McpServer>(&scope, &entry.doc, validate_mcp_server, &mut errors)
                {
                    // An OpenAPI document that yields no tool is beyond the
                    // schema — a load error like any schema failure.
                    if let Err(message) = t.validate_semantics() {
                        errors.push(LoadError { scope, message });
                    } else {
                        mcp_servers.push((id, scope, t));
                    }
                }
            }
            "a2a_agents" => {
                if let Some(t) = finish(&scope, &entry.doc, validate_a2a_agent, &mut errors) {
                    a2a_agents.push((id, scope, t));
                }
            }
            "passthrough_routes" => {
                if let Some(t) = finish(&scope, &entry.doc, validate_passthrough_route, &mut errors)
                {
                    passthrough_routes.push((id, scope, t));
                }
            }
            "cache_policies" => {
                if let Some(t) = finish(&scope, &entry.doc, validate_cache_policy, &mut errors) {
                    cache_policies.push((id, scope, t));
                }
            }
            "observability_exporters" => {
                if let Some(t) = finish(
                    &scope,
                    &entry.doc,
                    validate_observability_exporter,
                    &mut errors,
                ) {
                    observability_exporters.push((id, scope, t));
                }
            }
            "rate_limit_policies" => {
                if let Some(t) = finish::<RateLimitPolicy>(
                    &scope,
                    &entry.doc,
                    validate_rate_limit_policy,
                    &mut errors,
                ) {
                    // Semantic caps the schema can't express (condition-tree
                    // depth/leaf counts, operator×dimension admission, regex
                    // compilability) — a failing entry is a load error like
                    // any schema failure.
                    if let Err(message) = t.validate_semantics() {
                        errors.push(LoadError { scope, message });
                    } else {
                        rate_limit_policies.push((id, scope, t));
                    }
                }
            }
            "oidc_providers" => {
                if let Some(t) =
                    finish::<OidcProvider>(&scope, &entry.doc, validate_oidc_provider, &mut errors)
                {
                    // Mode-dependent field coupling and the shared-secret
                    // length floor are beyond the schema — a failing
                    // entry is a load error like any schema failure.
                    if let Err(message) = t.validate_semantics() {
                        errors.push(LoadError { scope, message });
                    } else {
                        oidc_providers.push((id, scope, t));
                    }
                }
            }
            "claim_mappings" => {
                if let Some(t) = finish(&scope, &entry.doc, validate_claim_mapping, &mut errors) {
                    claim_mappings.push((id, scope, t));
                }
            }
            "mcp_auth_settings" => {
                if let Some(t) = finish(&scope, &entry.doc, validate_mcp_auth_settings, &mut errors)
                {
                    mcp_auth_settings.push((id, scope, t));
                }
            }
            other => unreachable!("kind {other} is not in KINDS"),
        }
    }

    // ── Pass 3: cross-reference checks ────────────────────────────────
    // Every model-name reference must resolve at load time so a typo can
    // never become a silent runtime failure. Entries containing `*` are
    // glob patterns and are exempt from the existence check.
    let empty = BTreeMap::new();
    let model_names = identity_maps.get("models").unwrap_or(&empty);
    let mut check_model_ref = |scope: &str, field: &str, reference: &str| {
        if reference.contains('*') || model_names.contains_key(reference) {
            return;
        }
        let mut known: Vec<&str> = model_names.keys().map(String::as_str).collect();
        known.sort_unstable();
        errors.push(LoadError {
            scope: scope.to_string(),
            message: format!(
                "{field} references unknown model {reference:?} (defined models: {})",
                if known.is_empty() {
                    "none".to_string()
                } else {
                    known.join(", ")
                }
            ),
        });
    };

    for (_, scope, key) in &apikeys {
        for entry in &key.allowed_models {
            check_model_ref(scope, "allowed_models entry", entry);
        }
    }
    for (_, scope, model) in &models {
        if let Some(routing) = &model.routing {
            for target in &routing.targets {
                check_model_ref(scope, "routing target", &target.model);
            }
        }
        if let Some(ensemble) = &model.ensemble {
            for member in &ensemble.panel {
                check_model_ref(scope, "ensemble panel member", &member.model);
            }
            check_model_ref(scope, "ensemble judge", &ensemble.judge.model);
        }
        if let Some(semantic) = &model.semantic {
            if semantic.classifier.is_none() {
                check_model_ref(scope, "semantic embedding_model", &semantic.embedding_model);
            }
            check_model_ref(scope, "semantic default", &semantic.default);
            for route in &semantic.routes {
                check_model_ref(
                    scope,
                    &format!("semantic route {:?} target", route.name),
                    &route.target,
                );
            }
            if let crate::models::OnEmbeddingFailure::Target { target, .. } =
                &semantic.on_embedding_failure
            {
                check_model_ref(scope, "semantic on_embedding_failure target", target);
            }
            if let Some(crate::models::OnEmbeddingFailure::Target { target, .. }) =
                semantic.classifier.as_ref().map(|c| &c.on_failure)
            {
                check_model_ref(scope, "semantic classifier on_failure target", target);
            }
        }
    }

    // Same load-time protection for route grants: an `allowed_routes`
    // entry naming no defined route (globs exempt) is a typo that would
    // otherwise surface as a silent runtime 403.
    let route_names = identity_maps.get("passthrough_routes").unwrap_or(&empty);
    for (_, scope, key) in &apikeys {
        for entry in key.allowed_routes.iter().flatten() {
            if entry.contains('*') || route_names.contains_key(entry) {
                continue;
            }
            let mut known: Vec<&str> = route_names.keys().map(String::as_str).collect();
            known.sort_unstable();
            errors.push(LoadError {
                scope: scope.to_string(),
                message: format!(
                    "allowed_routes entry references unknown passthrough route {entry:?} \
                     (defined routes: {})",
                    if known.is_empty() {
                        "none".to_string()
                    } else {
                        known.join(", ")
                    }
                ),
            });
        }
    }

    // The runtime credential index is keyed by key_hash, so two api_keys
    // entries with distinct display_names but the same plaintext would
    // silently last-wins at auth time (the Admin API rejects exactly
    // this on create). Enforce credential uniqueness like the identity
    // uniqueness above. The message names entries, never hashes.
    let mut seen_hashes: BTreeMap<&str, &str> = BTreeMap::new();
    for (_, scope, key) in &apikeys {
        if let Some(first) = seen_hashes.insert(key.key_hash.as_str(), scope.as_str()) {
            errors.push(LoadError {
                scope: scope.clone(),
                message: format!(
                    "duplicate api key credential: this entry's key resolves to the \
                     same key_hash as {first} — every api key must have a distinct \
                     plaintext"
                ),
            });
        }
    }

    // JWT authentication selects the key by (jwt_provider, jwt_subject),
    // so a subject is set only alongside the provider allowed to assert
    // it, and that pair must be unique — otherwise auth would silently
    // tie-break, or (without a provider) a second trusted IdP could
    // impersonate this identity. Reject both at load.
    let mut seen_subjects: BTreeMap<(&str, &str), &str> = BTreeMap::new();
    for (_, scope, key) in &apikeys {
        match (key.jwt_subject.as_deref(), key.jwt_provider.as_deref()) {
            (Some(subject), Some(provider)) => {
                if let Some(first) = seen_subjects.insert((provider, subject), scope.as_str()) {
                    errors.push(LoadError {
                        scope: scope.clone(),
                        message: format!(
                            "duplicate jwt binding {provider:?}/{subject:?}: already used by \
                             {first} — every (jwt_provider, jwt_subject) pair must be distinct"
                        ),
                    });
                }
            }
            (Some(subject), None) => {
                errors.push(LoadError {
                    scope: scope.clone(),
                    message: format!(
                        "jwt_subject {subject:?} is set without jwt_provider — a subject must \
                         name the OIDC provider allowed to assert it, or a second trusted \
                         provider could impersonate this identity"
                    ),
                });
            }
            _ => {}
        }
    }

    // Two enabled providers sharing one issuer are ambiguous: they carry
    // different audience/scope/claim policies, and JWT auth would have to
    // pick one. Reject the duplicate at load (the DP resolver also fails
    // closed at runtime, but the file can and should surface it).
    let mut seen_issuers: BTreeMap<&str, &str> = BTreeMap::new();
    for (_, scope, provider) in &oidc_providers {
        if !provider.enabled {
            continue;
        }
        // Providers that pin no issuer (only possible in shared-secret
        // mode) are not ambiguous with each other: a token reaches them
        // by trial in name order, not by issuer, which is a total order.
        let Some(issuer) = provider.issuer.as_deref() else {
            continue;
        };
        if let Some(first) = seen_issuers.insert(issuer, scope.as_str()) {
            errors.push(LoadError {
                scope: scope.clone(),
                message: format!(
                    "duplicate enabled OIDC issuer {issuer:?}: already used by {first} — every \
                     enabled provider must have a distinct issuer"
                ),
            });
        }
    }

    // A claim mapping only ever evaluates against tokens verified by the
    // provider it names, so a typo'd `jwt_provider` would make the rule
    // silently dead. Resolve the reference at load like any other
    // cross-reference. (API keys deliberately allow a dangling
    // `jwt_provider`: the binding goes inert but the key still
    // authenticates by plaintext. A mapping has no such fallback role.)
    let provider_names = identity_maps.get("oidc_providers").unwrap_or(&empty);
    let api_key_ids = identity_maps.get("api_keys").unwrap_or(&empty);
    for (_, scope, mapping) in &claim_mappings {
        if !provider_names.contains_key(&mapping.jwt_provider) {
            let mut known: Vec<&str> = provider_names.keys().map(String::as_str).collect();
            known.sort_unstable();
            errors.push(LoadError {
                scope: scope.clone(),
                message: format!(
                    "jwt_provider references unknown OIDC provider {:?} (defined providers: {})",
                    mapping.jwt_provider,
                    if known.is_empty() {
                        "none".to_string()
                    } else {
                        known.join(", ")
                    }
                ),
            });
        }
        // The `resolve.api_key` name sugar resolves (or errors) in
        // desugar; a canonical `resolve.api_key_id` written directly must
        // equally land on a key defined in this file, or the mapping
        // would silently resolve nothing at runtime.
        if !api_key_ids
            .values()
            .any(|derived| derived == &mapping.resolve.api_key_id)
        {
            errors.push(LoadError {
                scope: scope.clone(),
                message: format!(
                    "resolve.api_key_id {:?} does not match any api key defined in this file — \
                     reference the key by name via `resolve.api_key` instead",
                    mapping.resolve.api_key_id
                ),
            });
        }
    }

    // A JWKS/discovery URL must never carry a credential query: JWKS
    // material is public,
    // credentials there would only leak (e.g. through a snapshot export).
    for (_, scope, provider) in &oidc_providers {
        for (field, url) in [
            ("issuer", provider.issuer.as_ref()),
            ("jwks_uri", provider.jwks_uri.as_ref()),
        ] {
            if let Some(url) = url {
                if url_has_credentials(url) {
                    errors.push(LoadError {
                        scope: scope.clone(),
                        message: format!(
                            "OIDC provider {field} must not embed credentials (a token query \
                             parameter) — JWKS endpoints are public"
                        ),
                    });
                }
            }
        }
    }

    // Same rule for the MCP OAuth resource URL: it is published verbatim
    // on the unauthenticated protected-resource-metadata endpoint, so an
    // embedded credential would be world-readable.
    for (_, scope, settings) in &mcp_auth_settings {
        let Some(resource_url) = settings.resource_url.as_deref() else {
            continue;
        };
        if url_has_credentials(resource_url) {
            errors.push(LoadError {
                scope: scope.clone(),
                message: "mcp_auth_settings resource_url must not embed credentials (a token \
                          query parameter) — protected resource metadata is public"
                    .into(),
            });
        }
    }

    // An explicit `provider_key_id` must also resolve: in file mode every
    // provider-key id is derived from its name, so any other value is
    // guaranteed dangling and would only surface per-request.
    let pk_ids: std::collections::BTreeSet<&str> = identity_maps
        .get("provider_keys")
        .unwrap_or(&empty)
        .values()
        .map(String::as_str)
        .collect();
    for (_, scope, model) in &models {
        if let Some(pk_id) = model.provider_key_id.as_deref() {
            if !pk_ids.contains(pk_id) {
                errors.push(LoadError {
                    scope: scope.clone(),
                    message: format!(
                        "provider_key_id {pk_id:?} does not match any provider key \
                         defined in this file — reference it by name with \
                         `provider_key` instead"
                    ),
                });
            }
        }
        let classifier = model.semantic.as_ref().and_then(|s| s.classifier.as_ref());
        if let Some(pk_id) = classifier.map(|c| c.provider_key_id.as_str()) {
            if !pk_ids.contains(pk_id) {
                errors.push(LoadError {
                    scope: scope.clone(),
                    message: format!(
                        "semantic classifier provider_key_id {pk_id:?} does not match any \
                         provider key defined in this file — reference it by name with \
                         `provider_key` instead"
                    ),
                });
            }
        }
    }

    if !errors.is_empty() {
        return Err(fail(errors));
    }

    // ── Materialize the snapshot ──────────────────────────────────────
    let snapshot = AisixSnapshot::new();
    for (id, _, v) in provider_keys {
        snapshot
            .provider_keys
            .insert(ResourceEntry::new(id, v, revision));
    }
    for (id, _, v) in models {
        snapshot.models.insert(ResourceEntry::new(id, v, revision));
    }
    for (id, _, v) in apikeys {
        snapshot.apikeys.insert(ResourceEntry::new(id, v, revision));
    }
    for (id, _, v) in guardrails {
        snapshot
            .guardrails
            .insert(ResourceEntry::new(id, v, revision));
    }
    for (id, _, v) in guardrail_attachments {
        snapshot
            .guardrail_attachments
            .insert(ResourceEntry::new(id, v, revision));
    }
    for (id, _, v) in mcp_servers {
        snapshot
            .mcp_servers
            .insert(ResourceEntry::new(id, v, revision));
    }
    for (id, _, v) in a2a_agents {
        snapshot
            .a2a_agents
            .insert(ResourceEntry::new(id, v, revision));
    }
    for (id, _, v) in cache_policies {
        snapshot
            .cache_policies
            .insert(ResourceEntry::new(id, v, revision));
    }
    for (id, _, v) in observability_exporters {
        snapshot
            .observability_exporters
            .insert(ResourceEntry::new(id, v, revision));
    }
    for (id, _, v) in rate_limit_policies {
        snapshot
            .rate_limit_policies
            .insert(ResourceEntry::new(id, v, revision));
    }
    for (id, _, v) in oidc_providers {
        snapshot
            .oidc_providers
            .insert(ResourceEntry::new(id, v, revision));
    }
    for (id, _, v) in claim_mappings {
        snapshot
            .claim_mappings
            .insert(ResourceEntry::new(id, v, revision));
    }
    for (id, _, v) in passthrough_routes {
        snapshot
            .passthrough_routes
            .insert(ResourceEntry::new(id, v, revision));
    }
    for (id, _, v) in mcp_auth_settings {
        snapshot
            .mcp_auth_settings
            .insert(ResourceEntry::new(id, v, revision));
    }
    Ok(snapshot)
}

#[cfg(test)]
mod tests;