armdb 0.12.0

use std::collections::HashSet;
use std::fmt;

use armour_core::{OnMissing, PathSeg, RowPart};

use super::plan::{gather_ids, ids_for_home_shape};
use super::registry::BuiltRegistry;
use crate::armour::db::Db;

#[allow(unused_imports)]
pub use super::plan::{AmbiguityReason, AmbiguousRef};

#[derive(Debug, Clone, PartialEq, Eq)]
pub enum RefDesc {
    /// A branded reference the validator assembled. `shape` is in the target
    /// home's **field order** (not sorted — see `AmbiguousRef::shape`) and
    /// `ids` is parallel to it.
    Assembled { shape: Vec<String>, ids: Vec<u64> },
    /// A reference built by a `reg.fk` closure: the target key may be unbranded
    /// and have no `u64` components at all, so only the label is meaningful.
    Explicit { label: String },
}

#[derive(Debug, Clone, PartialEq)]
#[non_exhaustive]
pub enum Finding {
    /// A format migration changed a claim-bearing collection. Namespace rows
    /// may happen to look correct, but only a complete registry-aware repair
    /// can retire this durable conservative marker.
    UniqueDirty {
        collection: String,
    },
    DanglingFk {
        collection: String,
        key: String,
        part: RowPart,
        path: Vec<PathSeg>,
        reference: RefDesc,
        homes: Vec<String>,
        on_missing: OnMissing,
    },
    AmbiguousRef(AmbiguousRef),
    IndexMissing {
        source: String,
        index: String,
        key: String,
    },
    IndexOrphan {
        index: String,
        key: String,
        /// Target marked complete via [`SchemaRegistry::complete_index`].
        complete: bool,
    },
    IndexValueMismatch {
        index: String,
        key: String,
    },
    IndexKeyCollision {
        index: String,
        key: String,
        owners: Vec<(String, String)>,
    },
    /// A source row claims a key with no matching namespace row. Scoped to
    /// the *expected* set (the union of every declared source's claims for
    /// this namespace) — see `schema::expected::build_unique`'s doc.
    UniqueMissing {
        namespace: String,
        source: String,
        key: String,
    },
    /// Two or more source rows claim the same namespace key — unreachable
    /// under live enforcement (`resolve_claims` refuses the second one at
    /// commit time), so this can only come from data written before
    /// adoption, a source written outside the registry, or a bug.
    UniqueConflict {
        namespace: String,
        key: String,
        owners: Vec<(String, String)>,
    },
    /// A permanent namespace row (`expires_at == 0`) that no declared
    /// claimant's owner matches — either nobody claims that key at all, or
    /// someone else's claim on it does (see
    /// `schema::registry::run_decls`'s unique-namespace pass, which routes
    /// the latter through [`crate::unique::resolve::decide`] and reports it
    /// as [`Finding::UniqueConflict`] instead — this variant is only the
    /// "nobody claims it" case). A reservation with no source row is its
    /// normal state and is never reported (see the T10 spec's decisions
    /// section).
    ///
    /// Caveat: a claiming source with [`Finding::UndecodableRows`] has an
    /// incomplete expected set (its unreadable rows' claims are simply
    /// absent), so one of *its* live, correctly-claimed rows can show up here
    /// as a false orphan. The spec only requires this to gate a destructive
    /// *repair*, not to suppress the report-only finding here — see
    /// `unique_validate_reports_a_false_orphan_when_the_source_has_undecodable_rows`
    /// for the current (deliberately unchanged) behavior.
    UniqueOrphan {
        namespace: String,
        key: String,
        owner: String,
        /// Namespace marked via [`super::SchemaRegistry::complete_namespace`]
        /// — the unique-namespace twin of [`Finding::IndexOrphan`]'s own
        /// `complete` field (same gate semantics, same reason: under partial
        /// adoption the other, not-yet-declared sources' live claims are
        /// indistinguishable from orphans, so an orphan here is only a hard
        /// error once every producer of this namespace is known to the
        /// registry).
        complete: bool,
    },
    /// A namespace row that fails [`crate::UniqueRow::is_well_formed`] —
    /// reported without decoding it further (`key_len` cannot be trusted to
    /// address its own key buffer).
    UniqueRowCorrupt {
        namespace: String,
        key: String,
    },
    CounterMismatch {
        collection: String,
        key: String,
        field: String,
        stored: u64,
        expected: u64,
    },
    DenormMissing {
        source: String,
        denorm: String,
        key: String,
    },
    DenormMismatch {
        source: String,
        denorm: String,
        key: String,
    },
    UnclassifiedCollection {
        name: String,
    },
    /// Brand in this collection never participates in any checkable assembled,
    /// declared, or single-brand edge (no home, or shape never assembles).
    UnvalidatedEdge {
        from: String,
        brand: String,
    },
    RegistryError {
        message: String,
    },
    UndecodableRows {
        collection: String,
        count: u64,
    },
}

impl Finding {
    pub fn is_error(&self) -> bool {
        match self {
            Finding::DanglingFk { on_missing, .. } => *on_missing == OnMissing::Error,
            Finding::AmbiguousRef(_) => true,
            Finding::UnclassifiedCollection { .. } | Finding::UnvalidatedEdge { .. } => false,
            Finding::IndexOrphan { complete, .. } => *complete,
            Finding::UniqueOrphan { complete, .. } => *complete,
            Finding::UndecodableRows { .. } => true,
            _ => true,
        }
    }
}

#[derive(Debug, Default)]
pub struct ValidationReport {
    pub findings: Vec<Finding>,
}

impl ValidationReport {
    pub fn errors(&self) -> impl Iterator<Item = &Finding> {
        self.findings.iter().filter(|f| f.is_error())
    }
    pub fn is_clean(&self) -> bool {
        self.findings.is_empty()
    }
}

impl fmt::Display for ValidationReport {
    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
        if self.findings.is_empty() {
            return writeln!(f, "schema validation: clean");
        }
        for finding in &self.findings {
            writeln!(
                f,
                "{}: {finding:?}",
                if finding.is_error() { "ERROR" } else { "WARN" }
            )?;
        }
        Ok(())
    }
}

pub(super) use crate::hex::hex;

pub fn validate(db: &Db, reg: &BuiltRegistry) -> ValidationReport {
    let mut report = ValidationReport::default();

    let known: HashSet<&str> = reg
        .reg
        .entries
        .iter()
        .map(|e| e.name)
        .chain(reg.reg.external.iter().map(|(n, _)| *n))
        .collect();
    for name in db.collection_names() {
        if !known.contains(name.as_str()) {
            report
                .findings
                .push(Finding::UnclassifiedCollection { name });
        }
    }

    for entry in &reg.reg.entries {
        if db.unique_dirty(entry.name) {
            report.findings.push(Finding::UniqueDirty {
                collection: entry.name.to_owned(),
            });
        }
        let plan = &entry.match_plan;

        for amb in &plan.ambiguities {
            report.findings.push(Finding::AmbiguousRef(amb.clone()));
        }

        // Sorted: the set is a `HashSet`, and every other finding in the report
        // is order-stable — a golden report must not depend on hash order.
        let mut unvalidated: Vec<&str> = plan.unvalidated_brands.iter().copied().collect();
        unvalidated.sort_unstable();
        for brand in unvalidated {
            report.findings.push(Finding::UnvalidatedEdge {
                from: entry.name.to_string(),
                brand: brand.to_string(),
            });
        }

        let skipped = (entry.undecodable_rows)();
        if skipped > 0 {
            report.findings.push(Finding::UndecodableRows {
                collection: entry.name.to_string(),
                count: skipped,
            });
        }

        // Resolved once per edge, not once per (row × flush × edge): the lookup
        // sorts the shape into a fresh allocation every call.
        let edge_homes: Vec<Option<&[super::registry::Home]>> = plan
            .edges
            .iter()
            .map(|e| reg.homes_for_shape(&e.home_shape))
            .collect();

        (entry.visit_refs)(&mut |key_bytes, key_ids, value_ids, flush| {
            for (edge, home_list) in plan.edges.iter().zip(&edge_homes) {
                if !flush.matches_edge_scope(&edge.scope) {
                    continue;
                }
                let Some(ids_vec) = gather_ids(&edge.slots, key_ids, value_ids) else {
                    continue;
                };

                let Some(home_list) = home_list else {
                    continue;
                };

                let mut satisfied = false;
                let mut registry_err = false;
                for home in *home_list {
                    let Some(home_ids) =
                        ids_for_home_shape(&edge.slots, &ids_vec, home.shape, &edge.home_shape)
                    else {
                        continue;
                    };
                    match (home.contains)(&home_ids) {
                        Some(true) => {
                            satisfied = true;
                            break;
                        }
                        Some(false) => {}
                        None => {
                            report.findings.push(Finding::RegistryError {
                                message: format!(
                                    "collection `{}`: home `{}` key construction failed",
                                    entry.name, home.name
                                ),
                            });
                            registry_err = true;
                            break;
                        }
                    }
                }
                if registry_err || satisfied {
                    continue;
                }

                report.findings.push(Finding::DanglingFk {
                    collection: entry.name.to_string(),
                    key: hex(key_bytes),
                    part: edge.part,
                    path: edge.reference_path.clone(),
                    reference: RefDesc::Assembled {
                        shape: edge.home_shape.iter().map(|b| (*b).to_string()).collect(),
                        ids: ids_vec,
                    },
                    homes: edge.home_names.iter().map(|n| (*n).to_string()).collect(),
                    on_missing: OnMissing::Error,
                });
            }
        });
    }

    super::registry::run_decls(reg, &mut report);

    report
}