armdb 0.9.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 crate::error::{DbError, DbResult};

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