armdb 0.9.0

use std::any::Any;
use std::collections::{HashMap, HashSet};
use std::sync::Arc;

use armour_core::{
    CollectionKind, CollectionNode, CompositeRef, FieldRef, GetRefs, GetType, KeyShape, OnMissing,
    PathSeg, RefVisitor, RelationEdge, RelationKind, RowPart, SchemaGraph, StorageClass, Typ,
};

use super::access::SchemaCollection;
use super::plan::{IdFlush, IdMap, MatchPlan, VisitRefsFn, build_match_plan, scope_of_path};
use super::validate::RefDesc;
use crate::CollectionMeta;
use crate::error::DbResult;
use crate::hook::NoHook;
use crate::key::Key;
use crate::projection::{IndexTargets, Indexed, ProjectableIndex, ProjectionSink};

type ScanAnyFn = Box<dyn Fn(&mut dyn FnMut(&[u8], &dyn Any, &dyn Any)) + Send + Sync>;
type UndecodableRowsFn = Box<dyn Fn() -> u64 + Send + Sync>;
type ContainsAnyFn = Box<dyn Fn(&dyn Any) -> Option<bool> + Send + Sync>;
type WithValueAnyFn = Box<dyn Fn(&dyn Any, &mut dyn FnMut(&dyn Any)) -> Option<bool> + Send + Sync>;
type ValueEqBytesFn = Box<dyn Fn(&[u8], &[u8]) -> DbResult<bool> + Send + Sync>;
/// `(src K any, src V any, source projection budget, out rows per target index)`.
///
/// The budget is the source collection's already-resolved
/// `min(hook budget, Config::max_projected_*)`, read off the open handle at
/// registration — so a declaration cannot evaluate a contract writes never used.
/// `None` (a non-projecting family) falls back to the crate defaults.
type IndexProjectFn = Box<
    dyn Fn(
            &dyn Any,
            &dyn Any,
            Option<(usize, usize)>,
            &mut Vec<(usize, Vec<u8>, Vec<u8>)>,
        ) -> DbResult<()>
        + Send
        + Sync,
>;
pub(super) type ValueToBytesFn = Arc<dyn Fn(&dyn Any) -> Option<Vec<u8>> + Send + Sync>;

pub(super) struct IndexTargetDecl {
    pub name: &'static str,
    pub index: Arc<dyn ProjectableIndex>,
    pub value_eq: ValueEqBytesFn,
    pub value_to_bytes: ValueToBytesFn,
}
type DenormProjectFn =
    Box<dyn Fn(&dyn Any, &dyn Any) -> Option<(Vec<u8>, Box<dyn Any>)> + Send + Sync>;
type DenormConsistentFn = Box<dyn Fn(&dyn Any, &dyn Any) -> Option<bool> + Send + Sync>;
type CounterReadFn = Box<dyn Fn(&dyn Any) -> Option<u64> + Send + Sync>;
type CounterPrefixFn = Box<dyn Fn(&dyn Any) -> Option<Vec<u8>> + Send + Sync>;
type ContainsFn = Box<dyn Fn(&[u64]) -> Option<bool> + Send + Sync>;

pub(in crate::schema) struct Home {
    pub(super) name: &'static str,
    pub(super) shape: &'static [&'static str],
    pub(super) contains: ContainsFn,
}

pub(super) struct HomePending {
    shape: &'static [&'static str],
    contains: ContainsFn,
}

fn sorted_shape(shape: &[&'static str]) -> Box<[&'static str]> {
    let mut v: Vec<_> = shape.to_vec();
    v.sort_unstable();
    v.into_boxed_slice()
}

/// Registry consistency error reported by [`SchemaRegistry::finish`].
#[derive(Debug, Clone, Copy, PartialEq, Eq, thiserror::Error)]
pub enum RegistryBuildError {
    #[error("collection `{0}` registered twice")]
    DuplicateCollection(&'static str),
}

/// Display label for an FK edge in the graph/report (not extraction — closures do that).
#[derive(Debug, Clone)]
pub struct FkLabel(pub String);

impl FkLabel {
    pub fn field(name: &str) -> Self {
        Self(name.to_string())
    }
}

pub(super) struct CollectionEntry {
    pub name: &'static str,
    pub kind: CollectionKind,
    pub typ: Typ,
    pub key_typ: Typ,
    pub self_shape: Option<&'static [&'static str]>,
    pub storage: StorageClass,
    pub auto_refs: Vec<(RowPart, FieldRef)>,
    pub composites: Vec<(RowPart, CompositeRef)>,
    pub match_plan: MatchPlan,
    /// Scan rows; callback receives key bytes and id maps for key/value halves.
    pub visit_refs: VisitRefsFn,
    /// Scan with erased types; callback: (key_bytes, &K as Any, &V as Any).
    pub scan_any: ScanAnyFn,
    /// Rows a `scan` silently skips (undecodable values) — reported as a finding
    /// so a corrupt collection cannot pass validation as clean.
    pub undecodable_rows: UndecodableRowsFn,
    /// The budget this collection's writes project under, read off the open
    /// handle. A declaration sourced here evaluates the same contract.
    pub projection_limits: Option<(usize, usize)>,
    /// None — wrong key type (declaration error), Some(b) — presence.
    pub contains_any: ContainsAnyFn,
    /// Some(false) — no record, Some(true) — record existed and `f` was called.
    pub with_value_any: WithValueAnyFn,
    /// Taken in `finish` to populate the homes map.
    pub entity_home: Option<HomePending>,
}

/// Result of an explicit FK key-builder.
pub(super) enum FkBuild {
    /// No reference (e.g. `Option::None`) — valid.
    NoRef,
    Key(Box<dyn Any>),
    TypeMismatch,
}

type FkBuildFn = Box<dyn Fn(&dyn Any, &dyn Any) -> FkBuild + Send + Sync>;

pub(super) enum Decl {
    Index {
        source: &'static str,
        targets: Vec<IndexTargetDecl>,
        project: IndexProjectFn,
    },
    Fk {
        from: &'static str,
        to: &'static str,
        label: String,
        on_missing: OnMissing,
        /// (src K any, src V any) -> target key; inner `None` = no reference (valid).
        build: FkBuildFn,
    },
    Denorm {
        source: &'static str,
        denorm: &'static str,
        project: DenormProjectFn,
        consistent: DenormConsistentFn,
    },
    Counter {
        collection: &'static str,
        field: &'static str,
        source: &'static str,
        read: CounterReadFn,
        prefix: CounterPrefixFn,
    },
}

#[derive(Default)]
pub struct SchemaRegistry {
    pub(super) entries: Vec<CollectionEntry>,
    pub(super) decls: Vec<Decl>,
    pub(super) external: Vec<(&'static str, CollectionKind)>,
    projection_sources: HashSet<&'static str>,
    projection_targets: HashSet<&'static str>,
    complete_indices: HashSet<&'static str>,
}

/// Registry after a successful consistency check.
pub struct BuiltRegistry {
    pub(super) reg: SchemaRegistry,
    /// Sorted brand multiset -> entity homes for that key shape.
    pub(super) homes: HashMap<Box<[&'static str]>, Vec<Home>>,
    pub(super) complete_indices: HashSet<&'static str>,
}

impl SchemaRegistry {
    pub fn new() -> Self {
        Self::default()
    }

    /// Register a collection with an open handle.
    pub fn register<S: SchemaCollection>(&mut self, handle: &Arc<S>, kind: CollectionKind) {
        self.register_inner(handle, kind, None);
    }

    /// Entity whose key shape becomes an auto-FK home.
    pub fn register_entity<S: SchemaCollection>(&mut self, handle: &Arc<S>)
    where
        S::K: KeyShape,
    {
        let h = handle.clone();
        let shape = S::K::SHAPE;
        if shape.iter().any(|b| b.is_empty()) {
            self.register_inner(handle, CollectionKind::Entity, None);
            return;
        }
        let contains = Box::new(move |ids: &[u64]| -> Option<bool> {
            let key = S::K::from_ids(ids)?;
            Some(h.contains_key(&key))
        });
        self.register_inner(
            handle,
            CollectionKind::Entity,
            Some(HomePending { shape, contains }),
        );
    }

    /// Graph node without a handle (no validation; clears UnclassifiedCollection).
    pub fn register_external(&mut self, name: &'static str, kind: CollectionKind) {
        self.external.push((name, kind));
    }

    fn register_inner<S: SchemaCollection>(
        &mut self,
        handle: &Arc<S>,
        kind: CollectionKind,
        home: Option<HomePending>,
    ) {
        let mut auto_refs: Vec<(RowPart, FieldRef)> = S::K::refs()
            .into_iter()
            .map(|r| (RowPart::Key, r))
            .collect();
        auto_refs.extend(S::V::refs().into_iter().map(|r| (RowPart::Value, r)));

        let mut composites: Vec<(RowPart, CompositeRef)> = S::K::composites()
            .into_iter()
            .map(|c| (RowPart::Key, c))
            .collect();
        composites.extend(S::V::composites().into_iter().map(|c| (RowPart::Value, c)));

        let h = handle.clone();
        #[allow(clippy::type_complexity)]
        let visit_refs = Box::new(move |f: &mut dyn FnMut(&[u8], &IdMap, &IdMap, IdFlush)| {
            type ElemFlush<'a> = dyn FnMut(&IdMap, IdFlush) + 'a;
            struct Visitor<'a> {
                stack: Vec<PathSeg>,
                ids: &'a mut IdMap,
                /// When set, leaving `Elem` flushes planned edges for that instance.
                flush: Option<&'a mut ElemFlush<'a>>,
            }
            impl RefVisitor for Visitor<'_> {
                fn enter(&mut self, seg: PathSeg) {
                    self.stack.push(seg);
                }
                fn leave(&mut self) {
                    if self.stack.last() == Some(&PathSeg::Elem)
                        && let Some(flush) = self.flush.as_mut()
                    {
                        let scope = self.stack.clone();
                        flush(self.ids, IdFlush::Scope(scope.clone()));
                        self.ids.retain(|(_, _, full), _| {
                            full.len() < scope.len() || full[..scope.len()] != scope[..]
                        });
                    }
                    self.stack.pop();
                }
                fn id(&mut self, seg: Option<PathSeg>, brand: &'static str, id: u64) {
                    let mut full_path = self.stack.clone();
                    // `None` — the visited value *is* the id (`Fuid`/`Id64` at a
                    // root), whose static `refs()` path is the stack alone.
                    if let Some(seg) = seg {
                        full_path.push(seg);
                    }
                    let scope = scope_of_path(&full_path);
                    self.ids.insert((scope, brand, full_path), id);
                }
            }
            h.scan(&mut |k, v| {
                let kb = k.as_bytes();
                let mut key_ids = IdMap::new();
                k.visit_refs(&mut Visitor {
                    stack: Vec::new(),
                    ids: &mut key_ids,
                    flush: None,
                });
                let mut value_ids = IdMap::new();
                {
                    let mut on_flush = |ids: &IdMap, flush: IdFlush| {
                        f(kb, &key_ids, ids, flush);
                    };
                    v.visit_refs(&mut Visitor {
                        stack: Vec::new(),
                        ids: &mut value_ids,
                        flush: Some(&mut on_flush),
                    });
                }
                f(kb, &key_ids, &value_ids, IdFlush::Final);
            });
        }) as VisitRefsFn;

        let h = handle.clone();
        let scan_any: ScanAnyFn = Box::new(move |f| {
            h.scan(&mut |k, v| f(k.as_bytes(), k as &dyn Any, v as &dyn Any));
        });

        let h = handle.clone();
        let undecodable_rows: UndecodableRowsFn = Box::new(move || h.undecodable_rows());

        let h = handle.clone();
        let contains_any =
            Box::new(move |key: &dyn Any| key.downcast_ref::<S::K>().map(|k| h.contains_key(k)))
                as ContainsAnyFn;

        let h = handle.clone();
        let with_value_any = Box::new(move |key: &dyn Any, f: &mut dyn FnMut(&dyn Any)| {
            key.downcast_ref::<S::K>()
                .map(|k| h.with_value(k, &mut |v| f(v as &dyn Any)))
        }) as WithValueAnyFn;

        let self_shape = home.as_ref().map(|h| h.shape);

        self.entries.push(CollectionEntry {
            name: S::V::NAME,
            kind,
            typ: <S::V as GetType>::TYPE,
            key_typ: <S::K as GetType>::TYPE,
            self_shape,
            storage: S::STORAGE,
            auto_refs,
            composites,
            match_plan: MatchPlan {
                ambiguities: Vec::new(),
                edges: Vec::new(),
                unvalidated_brands: HashSet::new(),
            },
            visit_refs,
            scan_any,
            undecodable_rows,
            projection_limits: handle.effective_projection_limits(),
            contains_any,
            with_value_any,
            entity_home: home,
        });
    }

    /// Consistency check: unique collection names; assemble homes map.
    /// Declarations are validated as they are added in later tasks.
    pub fn finish(mut self) -> Result<BuiltRegistry, RegistryBuildError> {
        let mut names: HashSet<&'static str> = HashSet::new();
        for e in &self.entries {
            if !names.insert(e.name) {
                return Err(RegistryBuildError::DuplicateCollection(e.name));
            }
        }
        let mut homes: HashMap<Box<[&'static str]>, Vec<Home>> = HashMap::new();
        for e in &mut self.entries {
            if let Some(pending) = e.entity_home.take() {
                let key = sorted_shape(pending.shape);
                homes.entry(key).or_default().push(Home {
                    name: e.name,
                    shape: pending.shape,
                    contains: pending.contains,
                });
            }
        }
        for e in &mut self.entries {
            e.match_plan =
                build_match_plan(e.name, e.self_shape, &e.auto_refs, &e.composites, &homes);
        }
        let complete_indices = std::mem::take(&mut self.complete_indices);
        Ok(BuiltRegistry {
            reg: self,
            homes,
            complete_indices,
        })
    }

    /// Every producer of this index is declared in this registry.
    pub fn complete_index<I: SchemaCollection>(&mut self, _index: &Arc<I>) {
        self.complete_indices.insert(I::V::NAME);
    }

    fn reject_index_decl(&self, source: &'static str, targets: &IndexTargets) {
        assert!(
            !targets.is_empty(),
            "secondary_index: IndexTargets must declare at least one target"
        );
        for t in targets.erased_targets() {
            assert_ne!(
                t.name, source,
                "secondary_index: target `{0}` must not equal source `{source}`",
                t.name
            );
            assert!(
                !self.projection_sources.contains(t.name),
                "secondary_index: target `{0}` is already a projecting source (cascade forbidden)",
                t.name
            );
        }
        assert!(
            !self.projection_targets.contains(source),
            "secondary_index: source `{source}` is already a projection target (cascade forbidden)"
        );
    }

    /// Declare a secondary index: returns an [`Indexed`] carrier for opening the source.
    ///
    /// Several sources may feed one or more index collections (one declaration per
    /// source): the validator's orphan check runs against the union of expected keys
    /// from all declared sources of that index.
    pub fn secondary_index<SV, P>(
        &mut self,
        targets: IndexTargets,
        project: P,
    ) -> Indexed<SV::SelfId, SV, NoHook>
    where
        SV: CollectionMeta + Send + Sync + 'static,
        SV::SelfId: Key + 'static,
        P: Fn(&SV::SelfId, &SV, &mut ProjectionSink<'_>) -> DbResult<()> + Send + Sync + 'static,
    {
        self.reject_index_decl(SV::NAME, &targets);

        let target_decls: Vec<IndexTargetDecl> = targets
            .erased_targets()
            .iter()
            .map(|t| {
                let index = Arc::clone(&t.inner);
                IndexTargetDecl {
                    name: t.name,
                    index: Arc::clone(&t.inner),
                    value_eq: Box::new(move |a, b| index.values_equal(a, b)),
                    value_to_bytes: Arc::clone(&t.value_to_bytes),
                }
            })
            .collect();

        self.projection_sources.insert(SV::NAME);
        for t in &target_decls {
            self.projection_targets.insert(t.name);
        }

        let project_fn = Arc::new(project);
        let project_erased = Arc::clone(&project_fn);
        let sink_targets = targets.clone();

        self.decls.push(Decl::Index {
            source: SV::NAME,
            targets: target_decls,
            project: Box::new(move |k, v, limits, out| {
                let k = k.downcast_ref::<SV::SelfId>().ok_or_else(|| {
                    crate::error::DbError::Internal("index decl: key type mismatch")
                })?;
                let v = v.downcast_ref::<SV>().ok_or_else(|| {
                    crate::error::DbError::Internal("index decl: value type mismatch")
                })?;
                // Resolved off the open source collection, so this is the exact
                // budget its writes project under — not a reconstruction of it.
                let (max_rows, max_bytes) = limits.unwrap_or((
                    crate::projection::DEFAULT_MAX_PROJECTED_ROWS,
                    crate::projection::DEFAULT_MAX_PROJECTED_BYTES,
                ));
                let mut sink = ProjectionSink::new(&sink_targets, max_rows, max_bytes);
                project_erased(k, v, &mut sink)?;
                for row in sink.rows() {
                    let Some(idx) = sink_targets
                        .erased_targets()
                        .iter()
                        .position(|t| t.name == row.target)
                    else {
                        return Err(crate::error::DbError::UndeclaredIndexTarget {
                            index: row.target,
                        });
                    };
                    out.push((idx, row.key.clone(), row.value.clone()));
                }
                Ok(())
            }),
        });

        Indexed::new(targets, move |k, v, sink| project_fn(k, v, sink))
    }

    /// Explicit FK: typed key-builder (covers unbranded fields and composite targets).
    pub fn fk<SV, TV, F>(&mut self, on_missing: OnMissing, label: FkLabel, build: F)
    where
        SV: CollectionMeta + 'static,
        SV::SelfId: Key + 'static,
        TV: CollectionMeta + 'static,
        TV::SelfId: Key + 'static,
        F: Fn(&SV::SelfId, &SV) -> Option<TV::SelfId> + Send + Sync + 'static,
    {
        self.decls.push(Decl::Fk {
            from: SV::NAME,
            to: TV::NAME,
            label: label.0,
            on_missing,
            build: Box::new(move |k, v| {
                let (Some(k), Some(v)) = (k.downcast_ref::<SV::SelfId>(), v.downcast_ref::<SV>())
                else {
                    return FkBuild::TypeMismatch;
                };
                match build(k, v) {
                    Some(tk) => FkBuild::Key(Box::new(tk)),
                    None => FkBuild::NoRef,
                }
            }),
        });
    }

    /// Denormalization: key projection + consistency predicate.
    pub fn denormalized<SV, DV, PF, EQ>(&mut self, project: PF, consistent: EQ)
    where
        SV: CollectionMeta + 'static,
        SV::SelfId: Key + 'static,
        DV: CollectionMeta + 'static,
        DV::SelfId: Key + 'static,
        PF: Fn(&SV::SelfId, &SV) -> DV::SelfId + Send + Sync + 'static,
        EQ: Fn(&SV, &DV) -> bool + Send + Sync + 'static,
    {
        self.decls.push(Decl::Denorm {
            source: SV::NAME,
            denorm: DV::NAME,
            project: Box::new(move |k, v| {
                let k = k.downcast_ref::<SV::SelfId>()?;
                let v = v.downcast_ref::<SV>()?;
                let dk = project(k, v);
                Some((dk.as_bytes().to_vec(), Box::new(dk) as Box<dyn Any>))
            }),
            consistent: Box::new(move |s, d| {
                Some(consistent(s.downcast_ref::<SV>()?, d.downcast_ref::<DV>()?))
            }),
        });
    }

    /// Counter: `field` in CV is recomputed as count of `source` records whose key
    /// starts with `prefix(cv key)`.
    pub fn counter<CV, RF, PF>(
        &mut self,
        field: &'static str,
        read: RF,
        source: &'static str,
        prefix: PF,
    ) where
        CV: CollectionMeta + 'static,
        CV::SelfId: Key + 'static,
        RF: Fn(&CV) -> u64 + Send + Sync + 'static,
        PF: Fn(&CV::SelfId) -> Vec<u8> + Send + Sync + 'static,
    {
        self.decls.push(Decl::Counter {
            collection: CV::NAME,
            field,
            source,
            read: Box::new(move |v| Some(read(v.downcast_ref::<CV>()?))),
            prefix: Box::new(move |k| Some(prefix(k.downcast_ref::<CV::SelfId>()?))),
        });
    }
}

impl BuiltRegistry {
    pub(super) fn homes_for_shape(&self, shape: &[&'static str]) -> Option<&[Home]> {
        self.homes.get(&sorted_shape(shape)).map(Vec::as_slice)
    }

    pub fn graph(&self) -> SchemaGraph {
        let mut g = SchemaGraph::default();
        for e in &self.reg.entries {
            g.collections.push(CollectionNode {
                name: e.name.to_string(),
                kind: e.kind,
                ty: e.typ,
                key_ty: Some(e.key_typ),
                self_shape: e
                    .self_shape
                    .map(|s| s.iter().map(|b| (*b).to_string()).collect()),
                storage: Some(e.storage),
            });
        }
        for (name, kind) in &self.reg.external {
            g.collections.push(CollectionNode {
                name: name.to_string(),
                kind: *kind,
                ty: Typ::Custom("external", &[]),
                key_ty: None,
                self_shape: None,
                storage: None,
            });
        }
        for e in &self.reg.entries {
            for edge in &e.match_plan.edges {
                for (home_name, home_shape) in edge.home_names.iter().zip(edge.home_shapes.iter()) {
                    let home_shape_refs: Vec<&str> = home_shape.to_vec();
                    g.relations.push(RelationEdge {
                        from: e.name.to_string(),
                        to: home_name.to_string(),
                        kind: RelationKind::Fk {
                            field_path: super::plan::reference_path_for_graph(
                                &home_shape_refs,
                                &edge.slots,
                            ),
                            brand: home_shape.join("+"),
                            on_missing: OnMissing::Error,
                            validated: true,
                        },
                    });
                }
            }
        }
        for d in &self.reg.decls {
            match d {
                Decl::Index {
                    source, targets, ..
                } => {
                    for t in targets {
                        g.relations.push(RelationEdge {
                            from: source.to_string(),
                            to: t.name.to_string(),
                            kind: RelationKind::Index,
                        });
                    }
                }
                Decl::Fk {
                    from,
                    to,
                    label,
                    on_missing,
                    ..
                } => g.relations.push(RelationEdge {
                    from: from.to_string(),
                    to: to.to_string(),
                    kind: RelationKind::Fk {
                        field_path: vec![],
                        brand: label.clone(),
                        on_missing: *on_missing,
                        validated: true,
                    },
                }),
                Decl::Denorm { source, denorm, .. } => g.relations.push(RelationEdge {
                    from: source.to_string(),
                    to: denorm.to_string(),
                    kind: RelationKind::Denorm,
                }),
                Decl::Counter {
                    collection,
                    field,
                    source,
                    ..
                } => g.relations.push(RelationEdge {
                    from: collection.to_string(),
                    to: source.to_string(),
                    kind: RelationKind::Counter {
                        field: field.to_string(),
                    },
                }),
            }
        }
        g
    }
}

fn find<'a>(entries: &'a [CollectionEntry], name: &str) -> Option<&'a CollectionEntry> {
    entries.iter().find(|e| e.name == name)
}

pub(crate) fn scan_target_actual<'a>(
    reg: &'a BuiltRegistry,
    cache: &'a mut HashMap<&'static str, HashMap<Vec<u8>, Vec<u8>>>,
    name: &'static str,
    encode: &ValueToBytesFn,
) -> &'a HashMap<Vec<u8>, Vec<u8>> {
    cache.entry(name).or_insert_with(|| {
        let Some(entry) = reg.reg.entries.iter().find(|e| e.name == name) else {
            return HashMap::new();
        };
        let mut map = HashMap::new();
        (entry.scan_any)(&mut |kb, _k, v| {
            if let Some(vb) = encode(v) {
                map.insert(kb.to_vec(), vb);
            }
        });
        map
    })
}

pub(super) fn run_decls(reg: &BuiltRegistry, report: &mut super::validate::ValidationReport) {
    use super::expected::{self, BuildIssue};
    use super::validate::{Finding, hex};
    use std::collections::HashMap;

    let mut actual_by_target: HashMap<&'static str, HashMap<Vec<u8>, Vec<u8>>> = HashMap::new();

    // Structural checks first — they need no data, and a declaration naming an
    // unregistered collection is a usage bug, not index drift.
    for d in &reg.reg.decls {
        if let Decl::Index {
            source, targets, ..
        } = d
        {
            if reg.reg.entries.iter().all(|e| e.name != *source) {
                report.findings.push(Finding::RegistryError {
                    message: format!("index decl: source `{source}` not registered"),
                });
            }
            for t in targets {
                if reg.reg.entries.iter().all(|e| e.name != t.name) {
                    report.findings.push(Finding::RegistryError {
                        message: format!("index decl: target `{}` not registered", t.name),
                    });
                }
            }
        }
    }

    // The expected set, shared with `repair` so the two cannot disagree about
    // what an index should hold. The validator collects every issue rather than
    // stopping, so the callback always returns `Ok` and `build` cannot fail.
    let (expected, _skipped) = expected::build(reg, &mut |issue| {
        match issue {
            BuildIssue::Collision { index, key, owners } => {
                report.findings.push(Finding::IndexKeyCollision {
                    index: index.to_string(),
                    key: hex(&key),
                    owners: owners.to_vec(),
                });
            }
            BuildIssue::Projection { source, error } => {
                report.findings.push(Finding::RegistryError {
                    message: format!("index decl `{source}`: projection failed: {error}"),
                });
            }
        }
        Ok(())
    })
    .unwrap_or_default();

    let all_targets = expected::declared_targets(reg);

    // Compare per expected *key*, not per emitted row: a projection may emit one
    // key twice, and comparing the first value against a target that correctly
    // holds the second reports a mismatch on an index `repair` considers clean.
    for target_name in &all_targets {
        let Some(rows) = expected.rows(target_name) else {
            continue;
        };
        let Some(t) = expected::target_decl(reg, target_name) else {
            continue;
        };
        let actual = scan_target_actual(reg, &mut actual_by_target, t.name, &t.value_to_bytes);
        let mut found: Vec<(String, Finding)> = Vec::new();
        for (key, (value, owner)) in rows {
            let key_hex = hex(key);
            match actual.get(key) {
                None => found.push((
                    key_hex.clone(),
                    Finding::IndexMissing {
                        source: owner.0.clone(),
                        index: t.name.to_string(),
                        key: key_hex,
                    },
                )),
                Some(actual_bytes) => match (t.value_eq)(value, actual_bytes) {
                    Ok(false) => found.push((
                        key_hex.clone(),
                        Finding::IndexValueMismatch {
                            index: t.name.to_string(),
                            key: key_hex,
                        },
                    )),
                    Ok(true) => {}
                    Err(e) => found.push((
                        key_hex,
                        Finding::RegistryError {
                            message: format!("index decl `{}`: value compare failed: {e}", t.name),
                        },
                    )),
                },
            }
        }
        // The expected set is a hash map; sort so a report diffs cleanly.
        found.sort_by(|a, b| a.0.cmp(&b.0));
        report.findings.extend(found.into_iter().map(|(_, f)| f));
    }

    // Orphans: index rows no declared source claims.
    for target_name in &all_targets {
        let Some(tdecl) = expected::target_decl(reg, target_name) else {
            continue;
        };
        let actual = scan_target_actual(
            reg,
            &mut actual_by_target,
            target_name,
            &tdecl.value_to_bytes,
        );
        let complete = reg.complete_indices.contains(target_name);
        let mut orphans: Vec<String> = actual
            .keys()
            .filter(|kb| !expected.contains(target_name, kb))
            .map(|kb| hex(kb))
            .collect();
        orphans.sort();
        for key in orphans {
            report.findings.push(Finding::IndexOrphan {
                index: target_name.to_string(),
                key,
                complete,
            });
        }
    }

    for d in &reg.reg.decls {
        match d {
            Decl::Index { .. } => {}
            Decl::Fk {
                from,
                to,
                label,
                on_missing,
                build,
            } => {
                let (Some(src), Some(target)) =
                    (find(&reg.reg.entries, from), find(&reg.reg.entries, to))
                else {
                    report.findings.push(Finding::RegistryError {
                        message: format!("fk decl `{from}`->`{to}`: collection not registered"),
                    });
                    continue;
                };
                (src.scan_any)(&mut |kb, k, v| match build(k, v) {
                    FkBuild::NoRef => {}
                    FkBuild::TypeMismatch => report.findings.push(Finding::RegistryError {
                        message: format!("fk decl `{from}`->`{to}` ({label}): type mismatch"),
                    }),
                    FkBuild::Key(tk) => match (target.contains_any)(&*tk) {
                        Some(true) => {}
                        Some(false) => {
                            if *on_missing != OnMissing::AllowMissing {
                                report.findings.push(Finding::DanglingFk {
                                    collection: from.to_string(),
                                    key: hex(kb),
                                    part: RowPart::Value,
                                    path: Vec::new(),
                                    reference: RefDesc::Explicit {
                                        label: label.clone(),
                                    },
                                    homes: vec![to.to_string()],
                                    on_missing: *on_missing,
                                });
                            }
                        }
                        None => report.findings.push(Finding::RegistryError {
                            message: format!("fk decl `{to}`: key type mismatch"),
                        }),
                    },
                });
            }
            Decl::Denorm {
                source,
                denorm,
                project,
                consistent,
            } => {
                let (Some(src), Some(dn)) = (
                    find(&reg.reg.entries, source),
                    find(&reg.reg.entries, denorm),
                ) else {
                    report.findings.push(Finding::RegistryError {
                        message: format!(
                            "denorm decl `{source}`->`{denorm}`: collection not registered"
                        ),
                    });
                    continue;
                };
                (src.scan_any)(&mut |_kb, k, v| {
                    let Some((dk_bytes, dk)) = project(k, v) else {
                        report.findings.push(Finding::RegistryError {
                            message: format!(
                                "denorm decl `{source}`->`{denorm}`: type mismatch in projection"
                            ),
                        });
                        return;
                    };
                    match (dn.with_value_any)(&*dk, &mut |dv| {
                        if consistent(v, dv) == Some(false) {
                            report.findings.push(Finding::DenormMismatch {
                                source: source.to_string(),
                                denorm: denorm.to_string(),
                                key: hex(&dk_bytes),
                            });
                        }
                    }) {
                        Some(true) => {}
                        Some(false) => report.findings.push(Finding::DenormMissing {
                            source: source.to_string(),
                            denorm: denorm.to_string(),
                            key: hex(&dk_bytes),
                        }),
                        None => report.findings.push(Finding::RegistryError {
                            message: format!("denorm decl `{denorm}`: key type mismatch"),
                        }),
                    }
                });
            }
            Decl::Counter {
                collection,
                field,
                source,
                read,
                prefix,
            } => {
                let (Some(cv), Some(src)) = (
                    find(&reg.reg.entries, collection),
                    find(&reg.reg.entries, source),
                ) else {
                    report.findings.push(Finding::RegistryError {
                        message: format!(
                            "counter decl `{collection}`<-`{source}`: collection not registered"
                        ),
                    });
                    continue;
                };
                let mut counters: Vec<(Vec<u8>, u64, String)> = Vec::new();
                (cv.scan_any)(&mut |kb, k, v| {
                    if let (Some(p), Some(stored)) = (prefix(k), read(v)) {
                        counters.push((p, stored, hex(kb)));
                    }
                });
                let mut actual = vec![0u64; counters.len()];
                (src.scan_any)(&mut |kb, _k, _v| {
                    for (i, (p, _, _)) in counters.iter().enumerate() {
                        if kb.starts_with(p) {
                            actual[i] += 1;
                        }
                    }
                });
                for ((_, stored, key), got) in counters.into_iter().zip(actual) {
                    if stored != got {
                        report.findings.push(Finding::CounterMismatch {
                            collection: collection.to_string(),
                            key,
                            field: field.to_string(),
                            stored,
                            expected: got,
                        });
                    }
                }
            }
        }
    }
}