armdb 0.9.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)]
pub enum Finding {
    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)>,
    },
    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::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) fn hex(bytes: &[u8]) -> String {
    bytes.iter().map(|b| format!("{b:02x}")).collect()
}

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 {
        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
}