armdb 0.11.0

//! The set of index rows every declared source says should exist.
//!
//! [`validate`](super::validate) and [`repair`](super::repair) must agree on
//! this to the byte: if they don't, CI fails on an index the repair tool
//! considers correct, or repair "fixes" one the validator was happy with. They
//! genuinely differ only in what they do with a problem — report it and keep
//! going, or stop before touching anything — so that is the one thing injected.

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

use crate::counter::{CounterSink, ErasedCounterTarget};
use crate::error::{DbError, DbResult};
use crate::unique::ClaimTarget;

use super::registry::{BuiltRegistry, Decl, IndexTargetDecl};

/// `(source collection, hex source key)` — the row that projected an entry.
pub(super) type Owner = (String, String);

/// A problem found while building the expected set.
pub(super) enum BuildIssue {
    /// Two source rows own the same `(index, key)`, so its value is a function
    /// of scan order.
    Collision {
        index: &'static str,
        key: Vec<u8>,
        owners: [Owner; 2],
    },
    /// A declaration's projection closure failed on a source row.
    Projection {
        source: &'static str,
        error: DbError,
    },
}

/// A source whose scan skipped rows it could not read or decode.
///
/// Their projections are missing from the expected set, so anything derived
/// from it is short — see [`super::repair`], which refuses to delete orphans on
/// the targets they feed.
pub(super) struct SkippedSource {
    pub name: &'static str,
    pub rows: u64,
    pub targets: Vec<&'static str>,
}

/// One target's expected rows: index key -> (encoded value, owner).
pub(super) type TargetRows = HashMap<Vec<u8>, (Vec<u8>, Owner)>;

/// Expected index rows, keyed by target and then by index key.
///
/// Keyed by *key*, not by emitted row: a projection may emit the same index key
/// more than once and the last value wins. Consuming rows as they are emitted
/// instead makes a correct index look mismatched, and writes it twice.
#[derive(Default)]
pub(super) struct ExpectedRows {
    by_target: HashMap<&'static str, TargetRows>,
}

impl ExpectedRows {
    pub(super) fn rows(&self, target: &str) -> Option<&TargetRows> {
        self.by_target.get(target)
    }

    pub(super) fn contains(&self, target: &str, key: &[u8]) -> bool {
        self.by_target
            .get(target)
            .is_some_and(|m| m.contains_key(key))
    }
}

/// Every target named by an index declaration, whether or not it has rows —
/// an empty target still needs its orphan sweep.
pub(super) fn declared_targets(reg: &BuiltRegistry) -> HashSet<&'static str> {
    reg.reg
        .decls
        .iter()
        .filter_map(|d| match d {
            Decl::Index { targets, .. } => Some(targets.as_slice()),
            _ => None,
        })
        .flat_map(|ts| ts.iter().map(|t| t.name))
        .collect()
}

/// First declaration of `name`. Several sources may declare the same target;
/// each declaration holds an `Arc` of the same index and equivalent codecs, so
/// any one of them drives the target's scan, compare and write.
pub(super) fn target_decl<'r>(reg: &'r BuiltRegistry, name: &str) -> Option<&'r IndexTargetDecl> {
    reg.reg
        .decls
        .iter()
        .filter_map(|d| match d {
            Decl::Index { targets, .. } => Some(targets),
            _ => None,
        })
        .flat_map(|ts| ts.iter())
        .find(|t| t.name == name)
}

/// Scan every index declaration and collapse its projection into one expected
/// row per `(target, key)`.
///
/// `on_issue` decides the policy: returning `Ok(())` records the problem and
/// keeps scanning, returning `Err` aborts the build — which is how `repair`
/// refuses a registry before it has mutated anything.
pub(super) fn build(
    reg: &BuiltRegistry,
    on_issue: &mut dyn FnMut(BuildIssue) -> DbResult<()>,
) -> DbResult<(ExpectedRows, Vec<SkippedSource>)> {
    let mut expected = ExpectedRows::default();
    let mut skipped: Vec<SkippedSource> = Vec::new();

    for d in &reg.reg.decls {
        let Decl::Index {
            source,
            targets,
            project,
        } = d
        else {
            continue;
        };
        let Some(src) = reg.reg.entries.iter().find(|e| e.name == *source) else {
            continue;
        };

        // The scan callback cannot return, so an abort is stashed and checked
        // once the scan is over.
        let mut abort: Option<DbError> = None;
        let mut issue = |i: BuildIssue, abort: &mut Option<DbError>| {
            if let Err(e) = on_issue(i) {
                *abort = Some(e);
            }
        };
        let limits = src.projection_limits;
        (src.scan_any)(&mut |src_kb, k, v| {
            if abort.is_some() {
                return;
            }
            let mut rows = Vec::new();
            if let Err(e) = project(k, v, limits, &mut rows) {
                issue(BuildIssue::Projection { source, error: e }, &mut abort);
                return;
            }
            let owner = (source.to_string(), super::validate::hex(src_kb));
            for (target_idx, key, value) in rows {
                let Some(t) = targets.get(target_idx) else {
                    issue(
                        BuildIssue::Projection {
                            source,
                            error: DbError::Internal("index decl: target index out of range"),
                        },
                        &mut abort,
                    );
                    continue;
                };
                let rows = expected.by_target.entry(t.name).or_default();
                if let Some((_, existing)) = rows.get(&key)
                    && existing != &owner
                {
                    let mut owners = [existing.clone(), owner.clone()];
                    owners.sort();
                    issue(
                        BuildIssue::Collision {
                            index: t.name,
                            key: key.clone(),
                            owners,
                        },
                        &mut abort,
                    );
                    continue;
                }
                rows.insert(key, (value, owner.clone()));
            }
        });
        if let Some(e) = abort {
            return Err(e);
        }

        // The scan drops rows it cannot read or decode, so this source's half of
        // the expected set is short by their projections.
        let rows = (src.undecodable_rows)();
        if rows > 0 {
            let names: Vec<&'static str> = targets.iter().map(|t| t.name).collect();
            tracing::warn!(
                source = source,
                rows,
                targets = ?names,
                "schema: source rows could not be read; the expected index set is incomplete"
            );
            skipped.push(SkippedSource {
                name: source,
                rows,
                targets: names,
            });
        }
    }

    Ok((expected, skipped))
}

/// `(target collection, target field, key bytes)` — the field discriminator
/// is required: [`SchemaRegistry::counted`](super::registry::SchemaRegistry::counted)'s
/// own contract allows several *different* `(target, field)` pairs to be fed
/// by their own sources onto the very same target collection (only one
/// declared source per pair, enforced by
/// [`super::registry::RegistryBuildError::DuplicateCounterSource`]). Keying by
/// collection alone would let two such declarations collide on the same map
/// slot — see the regression fixed in the T9 Task 11 review.
pub(super) type CounterKey = (&'static str, &'static str, Vec<u8>);

/// Every counter source's declared contributions, summed per source row and
/// resolved to the concrete target row they belong to:
/// [`CounterKey`] `-> expected absolute value`.
///
/// Shared by [`super::validate`] and [`super::repair`] for the same reason
/// [`build`] is (see the module doc): if the two computed this independently
/// they could disagree about what a counter should hold.
///
/// Unlike an index projection, a counter's contributions are meant to be
/// *summed*, not compared for a same-owner collision, so there is no
/// `BuildIssue`-style "record and keep going" callback here — the only real
/// failures are hard ones: the declaration's projection closure erroring on a
/// row, or an accumulated sum overflowing `u64`. Either aborts the whole
/// build — mirrors `repair`'s index pass aborting on a key collision: there
/// is nothing safe to fall back to, so both callers treat a failure here as
/// fatal to the counter section rather than comparing against a map that
/// silently under-reports (see [`super::run_decls`] and [`super::repair`]).
///
/// A contribution resolving to no live target is **not** treated as fatal:
/// [`crate::counter::CounterTargets::resolve_group`]'s `Ok(None)` is skipped
/// (reserved, not reachable today) and its `CounterTargetMissing` — a
/// dangling contribution, e.g. a vacated `one_of` group — is warned and
/// skipped, exactly mirroring [`crate::counter::recount`]'s pass 2. That
/// ruling exists so a single dangling contribution cannot make a database
/// unopenable (see `counter.rs`); propagating it as an error here would
/// re-break that guarantee the moment `validate`/`repair` run on a database
/// an open-time recount already warned about. `AmbiguousCounterTarget` (more
/// than one current holder) has no safe skip target and stays a hard error,
/// same as `recount`.
///
/// A source whose scan skipped undecodable rows contributes a short sum here
/// (the skipped rows' contributions are simply absent) — the caller must not
/// treat that as ground truth for a destructive repair. `repair` checks the
/// same source's `undecodable_rows` separately and declines to rewrite its
/// targets; `validate` reports the row skip as its own
/// [`super::Finding::UndecodableRows`] (via the per-collection check every
/// registered entry already gets) alongside whatever `CounterMismatch`
/// findings the short sum produces.
///
/// A declaration whose source is not a registered entry is skipped here —
/// exactly like [`build`]'s index half — but that is **not** license to read
/// its targets as "expected zero": both callers check
/// `reg.reg.entries.iter().any(|e| e.name == *source)` themselves before
/// comparing or repairing a declaration's targets, so a merely-unregistered
/// source protects its targets the same way an undecodable one does, instead
/// of the missing data being mistaken for a confirmed count.
pub(super) fn build_counters(reg: &BuiltRegistry) -> DbResult<HashMap<CounterKey, u64>> {
    let mut out: HashMap<CounterKey, u64> = HashMap::new();

    for d in &reg.reg.decls {
        let Decl::Counter {
            source,
            targets,
            project,
        } = d
        else {
            continue;
        };
        let Some(src) = reg.reg.entries.iter().find(|e| e.name == *source) else {
            continue;
        };
        let (max_rows, max_bytes) = src.projection_limits.unwrap_or((
            crate::projection::DEFAULT_MAX_PROJECTED_ROWS,
            crate::projection::DEFAULT_MAX_PROJECTED_BYTES,
        ));

        // Pass 1 (scoped to this one declaration's `CounterTargets`): sum
        // every source row's contributions into (group index, key) -> u128 —
        // the same shape `counter::recount`'s pass 1 builds.
        let mut sums: HashMap<(usize, Vec<u8>), u128> = HashMap::new();
        let mut abort: Option<DbError> = None;
        (src.scan_any)(&mut |_src_kb, k, v| {
            if abort.is_some() {
                return;
            }
            let mut sink = CounterSink::new(targets, max_rows, max_bytes);
            if let Err(e) = project(k, v, &mut sink) {
                abort = Some(e);
                return;
            }
            for (slot, w) in sink.into_acc() {
                let entry = sums.entry(slot).or_insert(0u128);
                match entry.checked_add(w) {
                    Some(sum) => *entry = sum,
                    None => abort = Some(DbError::Internal("counter expected-sum overflow")),
                }
            }
        });
        if let Some(e) = abort {
            return Err(e);
        }

        // Resolve each (group, key) to the concrete target member holding it
        // — exactly as the write path resolves a live delta
        // (`counter::into_tx_counter_deltas`) and `counter::recount`'s pass 2
        // do, including the same dangling-contribution leniency (see the
        // rustdoc above).
        for ((gi, key), sum) in sums {
            let member: &std::sync::Arc<dyn ErasedCounterTarget> = match targets
                .resolve_group(gi, &key)
            {
                Ok(Some(m)) => m,
                // Reserved by `resolve_group`, not reachable today — see its
                // own rustdoc. Kept a no-op rather than an `unwrap`.
                Ok(None) => continue,
                Err(DbError::CounterTargetMissing { targets, key }) => {
                    tracing::warn!(
                        group = gi,
                        targets = %targets,
                        key = %key,
                        "schema: dangling contribution — no live target row holds this key, skipping (see counter::recount)"
                    );
                    continue;
                }
                Err(e) => return Err(e),
            };
            // `OnMissingTarget::Skip` on an absent row means the write path
            // **dropped** this contribution — deliberately, by declaration.
            // Recording it as expected would report a `CounterMismatch` on a
            // database that is exactly right, forever, and send `repair` at a
            // row it can only no-op on. (`Error` and `Create` both stay: the
            // first would have failed the write, the second would have
            // materialised the row, so in either case an absent row is real
            // drift. A `one_of` group never reaches this branch — its own
            // `contains` probe above already resolved the holder.)
            if member.on_missing() == crate::counter::OnMissingTarget::Skip
                && !member.contains(&key)?
            {
                continue;
            }
            let want = u64::try_from(sum).map_err(|_| DbError::CounterOverflow {
                collection: member.collection_name(),
                field: member.field(),
                key: super::validate::hex(&key),
            })?;
            out.insert((member.collection_name(), member.field(), key), want);
        }
    }

    Ok(out)
}

/// One row where a counter target's stored value differs from what
/// [`build_counters`] expects.
pub(super) struct CounterDrift {
    pub key: Vec<u8>,
    pub want: u64,
    pub stored: u64,
}

/// Union `target`'s currently-stored rows with the keys `expected` has for
/// it, and return every key where the two differ — the counter twin of the
/// index compare loop above, shared verbatim by `validate` and `repair` so
/// the two cannot compute drift differently one level below
/// [`build_counters`] itself (that was exactly how the shape of this bug
/// hid: the map was already shared, but each caller re-derived "what differs"
/// from it independently).
///
/// Not only keys the source contributed to: a target row nobody contributes
/// to any more has no entry in `expected`, and the union with its physically
/// stored rows is what catches it — comparing it against the implied zero,
/// mirroring `counter::recount`'s pass 2 unioning `expected` with each
/// target's currently-stored rows.
pub(super) fn counter_drift(
    target: &std::sync::Arc<dyn ErasedCounterTarget>,
    expected: &HashMap<CounterKey, u64>,
) -> DbResult<Vec<CounterDrift>> {
    let mut existing: HashMap<Vec<u8>, u64> = HashMap::new();
    target.for_each_counter_row(&mut |key, stored| {
        existing.insert(key.to_vec(), stored);
        Ok(())
    })?;

    let name = target.collection_name();
    let field = target.field();
    let mut keys: HashSet<Vec<u8>> = existing.keys().cloned().collect();
    keys.extend(
        expected
            .keys()
            .filter(|(n, f, _)| *n == name && *f == field)
            .map(|(_, _, key)| key.clone()),
    );

    let mut out = Vec::new();
    for key in keys {
        let want = expected
            .get(&(name, field, key.clone()))
            .copied()
            .unwrap_or(0);
        let stored = existing.get(&key).copied().unwrap_or(0);
        if want != stored {
            out.push(CounterDrift { key, want, stored });
        }
    }
    Ok(out)
}

// -- T10 Task 9: unique-namespace expected claim set -------------------------
//
// Shared by `validate` (this task) and `repair` (T10 Task 10) for the same
// reason `build`/`build_counters` above are: if the two computed "what a
// namespace should hold" independently, they could disagree about it.

/// One source row's claim on a unique-namespace key — the unique-namespace
/// twin of [`Owner`], keeping the decoded [`crate::UniqueOwner`] alongside so
/// a namespace write does not need to reconstruct it. `validate` only reads
/// `source`/`source_key` for its findings; `schema::unique_repair::repair_namespaces`
/// (T10 Task 10) is what reads `owner` — it is the claimant `owner` passed to
/// [`crate::unique::resolve::decide`] whenever a `UniqueMissing`-shaped key
/// needs a `Write` decision turned into an actual row.
pub(super) struct UniqueClaim {
    pub source: &'static str,
    pub source_key: String,
    pub owner: crate::UniqueOwner,
}

/// `(namespace, key bytes) -> every source row that claims it`.
///
/// Unlike [`TargetRows`] (the index half), a second claimant is **not**
/// first-wins-plus-collision-report during the build: this module's `build`
/// aborts recording a second owner and reports a [`BuildIssue::Collision`]
/// because an index target genuinely can only hold one value per key at a
/// time. A unique namespace has the same one-row-per-key constraint, but
/// getting *two claimants in the expected set* here is not a build-time
/// anomaly — it is the exact condition [`super::Finding::UniqueConflict`]
/// exists to report (see this namespace's own module doc: unreachable under
/// live enforcement, so it only shows up on data written before adoption, a
/// source written outside the registry, or a bug). So every claimant is kept.
pub(super) type UniqueExpected = HashMap<(&'static str, Vec<u8>), Vec<UniqueClaim>>;

/// Every namespace named by any `Decl::Unique` declaration, mapped to its
/// live [`ClaimTarget`] handle — deduplicated, since several sources may
/// declare into the same namespace. Unlike [`target_decl`] (the index half),
/// this does **not** go through `reg.reg.entries`: a namespace registers with
/// [`super::registry::SchemaRegistry::register_external`] (no handle —
/// `UniqueIndex` is not a [`super::access::SchemaCollection`], so it cannot
/// go through `register`/`register_entity`), so the declaration itself
/// (`Decl::Unique.targets`, a [`crate::UniqueTargets`]) is the only place a
/// live handle to scan it lives.
pub(super) fn declared_unique_handles(
    reg: &BuiltRegistry,
) -> HashMap<&'static str, Arc<dyn ClaimTarget>> {
    let mut out: HashMap<&'static str, Arc<dyn ClaimTarget>> = HashMap::new();
    for d in &reg.reg.decls {
        if let Decl::Unique { targets, .. } = d {
            for t in targets.all() {
                out.entry(t.collection_name())
                    .or_insert_with(|| Arc::clone(t));
            }
        }
    }
    out
}

/// Union every declared source's claims into one expected set per namespace —
/// a namespace fed by three sources has one expected set, not three (see the
/// module doc). Mirrors [`build_counters`]'s shape (a hard `DbResult` failure
/// on a projection error, no per-issue callback) rather than [`build`]'s: see
/// [`UniqueExpected`]'s doc for why a second claimant is not itself a
/// build-time problem here. A declaration whose source is not a registered
/// entry is skipped — exactly like `build`'s and `build_counters`'s own guard
/// against reading an absent scan as "confirmed empty".
pub(super) fn build_unique(reg: &BuiltRegistry) -> DbResult<(UniqueExpected, Vec<SkippedSource>)> {
    use crate::unique::{ClaimSink, UniqueOwner, collection_tag};

    let mut expected: UniqueExpected = HashMap::new();
    let mut skipped: Vec<SkippedSource> = Vec::new();

    for d in &reg.reg.decls {
        let Decl::Unique {
            source,
            targets,
            project,
        } = d
        else {
            continue;
        };
        let Some(src) = reg.reg.entries.iter().find(|e| e.name == *source) else {
            continue;
        };
        let (max_rows, max_bytes) = src.projection_limits.unwrap_or((
            crate::projection::DEFAULT_MAX_PROJECTED_ROWS,
            crate::projection::DEFAULT_MAX_PROJECTED_BYTES,
        ));

        let mut abort: Option<DbError> = None;
        (src.scan_any)(&mut |src_kb, k, v| {
            if abort.is_some() {
                return;
            }
            let mut sink = ClaimSink::new(targets, max_rows, max_bytes);
            if let Err(e) = project(k, v, &mut sink) {
                abort = Some(e);
                return;
            }
            let Some(owner) = UniqueOwner::new(collection_tag(source), src_kb, 0) else {
                abort = Some(DbError::Internal(
                    "unique decl: source key exceeds the owner capacity",
                ));
                return;
            };
            // Dedup this one row's own claims before recording them: a row
            // whose projection calls `sink.claim` twice for the same
            // `(namespace, key)` (e.g. two aliases that normalise to the same
            // claim, or a duplicate list entry) makes exactly one claim at
            // commit time — `unique::diff::claim_set` collapses into a
            // `BTreeSet` for the same reason. Recording it twice here would
            // manufacture a false `UniqueConflict` against itself.
            let mut this_row: std::collections::BTreeSet<(&'static str, Vec<u8>)> =
                std::collections::BTreeSet::new();
            for intent in sink.claims() {
                this_row.insert((intent.namespace, intent.key.clone()));
            }
            for (namespace, key) in this_row {
                expected
                    .entry((namespace, key))
                    .or_default()
                    .push(UniqueClaim {
                        source,
                        source_key: super::validate::hex(src_kb),
                        owner,
                    });
            }
        });
        if let Some(e) = abort {
            return Err(e);
        }

        // The scan drops rows it cannot read or decode, so this source's half
        // of the expected set is short by their claims.
        let rows = (src.undecodable_rows)();
        if rows > 0 {
            let names: Vec<&'static str> = targets.all().map(|t| t.collection_name()).collect();
            tracing::warn!(
                source = source,
                rows,
                targets = ?names,
                "schema: source rows could not be read; the expected unique-claim set is incomplete"
            );
            skipped.push(SkippedSource {
                name: source,
                rows,
                targets: names,
            });
        }
    }

    Ok((expected, skipped))
}