armdb 0.11.0

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

use crate::armour::db::Db;
use crate::error::{DbError, DbResult};

use super::registry::{BuiltRegistry, Decl, scan_target_actual};
use super::unique_repair;
use super::validate::hex;

pub use super::unique_repair::{NamespaceConflict, NamespaceRepairStats};

/// Options for [`repair`].
#[derive(Debug, Clone, Default)]
pub struct RepairOptions {
    /// Delete orphan index rows on targets marked [`super::SchemaRegistry::complete_index`].
    pub delete_orphans: bool,
}

/// Per-index repair counters.
#[derive(Debug, Clone, Default)]
pub struct IndexRepairStats {
    pub name: &'static str,
    pub puts: u64,
    pub rewrites: u64,
    pub deletes: u64,
    pub declined_orphans: u64,
}

/// Per-counter repair counters — one row per declared `(collection, field)`
/// counter target actually compared. A target declined because its source
/// had undecodable rows (see [`RepairReport::undecodable_sources`])
/// contributes no row here — nothing was compared for it.
#[derive(Debug, Clone, Default)]
pub struct CounterRepairStats {
    pub collection: &'static str,
    pub field: &'static str,
    pub fixed: u64,
}

/// A source collection whose scan skipped rows it could not read or decode.
#[derive(Debug, Clone)]
pub struct UndecodableSource {
    pub name: &'static str,
    pub rows: u64,
    /// Targets whose orphan deletion — or, for a counter source, whose
    /// rewrite — was suppressed because of it.
    pub protected_targets: Vec<&'static str>,
}

/// Aggregate result of [`repair`].
#[derive(Debug, Clone, Default)]
pub struct RepairReport {
    pub per_index: Vec<IndexRepairStats>,
    pub per_counter: Vec<CounterRepairStats>,
    /// Per-namespace unique-claim repair counters (T10 Task 10) — the
    /// namespace twin of `per_index`/`per_counter`. See
    /// `schema::unique_repair`'s module doc for what drives each field and
    /// why a namespace is repaired by its own pass rather than the generic
    /// carrier reconcile.
    pub per_namespace: Vec<NamespaceRepairStats>,
    /// Sources that could not be scanned in full. Their index rows — and, for
    /// a counter source, every declared counter target it feeds, and for a
    /// unique-claim source, every namespace it feeds — are missing from the
    /// expected set, so orphan deletion / counter rewriting / namespace
    /// orphan deletion is held back for everything they feed — see
    /// [`repair`].
    pub undecodable_sources: Vec<UndecodableSource>,
}

/// Apply the same scan as `validate` and fix index drift: put missing rows,
/// rewrite mismatches, optionally delete orphans on complete targets. Also
/// fixes counter drift (`Decl::Counter` targets): resolves each declared
/// counter to its expected absolute value via
/// [`super::expected::build_counters`] — the same shared pass `validate`
/// reads — and applies the difference through
/// [`crate::counter::ErasedCounterTarget::apply_delta`].
///
/// Aborts with [`DbError::IndexKeyCollision`] if two source owners project the
/// same `(target, key)` — collisions are unrepairable and must not leave an
/// arbitrary scan-order winner on disk.
///
/// # Undecodable source rows
///
/// A source scan silently skips rows whose value cannot be read or decoded (Var
/// and LSM families; see `Finding::UndecodableRows`). Those rows contribute
/// nothing to the expected set, so their live index entries would look like
/// orphans, and their counter contributions are simply missing — a rewrite
/// from either would create drift, not fix it. Deleting an orphan under this
/// condition would destroy the only remaining record of a row whose
/// value is already unreadable, so **orphan deletion is suppressed** for every
/// index target fed by such a source, and **counter rewriting is suppressed**
/// for every counter target it feeds, regardless of
/// [`RepairOptions::delete_orphans`] and `complete_index`. Missing rows are
/// still put and mismatches still rewritten — only the destructive half is held
/// back. The affected sources are listed in
/// [`RepairReport::undecodable_sources`].
///
/// # Unique namespaces (T10)
///
/// Pass 5 repairs every declared unique namespace — see
/// `schema::unique_repair`'s module doc for the full rule (missing claims
/// written, conflicts refused and reported with the key and both owners,
/// orphans deleted only under `complete_namespace` + `delete_orphans`,
/// reservations and corrupt rows never touched). It runs *after* the index
/// and counter passes above, so a namespace scan failure (an I/O-level
/// error from `ClaimTarget::scan_rows`, propagated via `?`) aborts `repair`
/// with whatever the earlier passes already wrote already landed on disk —
/// the same partial-failure shape `expected::build_counters(reg)?` already
/// has relative to the index passes before it, not a new one this pass
/// introduces.
pub fn repair(db: &Db, reg: &BuiltRegistry, opts: RepairOptions) -> DbResult<RepairReport> {
    let _ = db;
    use super::expected::{self, BuildIssue};

    // Pass 1: the expected set — shared with `validate`, so the two cannot
    // disagree about what an index should hold. Any issue aborts here, before
    // anything below has written a byte.
    let (expected, skipped) = expected::build(reg, &mut |issue| match issue {
        BuildIssue::Collision { index, key, owners } => Err(DbError::IndexKeyCollision {
            index,
            key: hex(&key),
            owners: owners.to_vec(),
        }),
        BuildIssue::Projection { error, .. } => Err(error),
    })?;

    // Targets whose expected set is known to be short, so an "orphan" there may
    // be a live row of a source record that can no longer be read.
    let protected: HashSet<&'static str> = skipped
        .iter()
        .flat_map(|s| s.targets.iter().copied())
        .collect();
    let mut undecodable_sources: Vec<UndecodableSource> = skipped
        .into_iter()
        .map(|s| UndecodableSource {
            name: s.name,
            rows: s.rows,
            protected_targets: s.targets,
        })
        .collect();

    let mut actual_by_target: HashMap<&'static str, HashMap<Vec<u8>, Vec<u8>>> = HashMap::new();
    let mut stats_by_name: HashMap<&'static str, IndexRepairStats> = HashMap::new();
    let all_targets = expected::declared_targets(reg);

    // Pass 2: put what is missing, rewrite what differs. Keyed by target, not by
    // declaration — the expected set already unions every source, and `actual`
    // is a pre-mutation snapshot, so walking a shared target once per
    // declaration would re-put each row and double-count the report.
    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);
        for (key, (value, _owner)) in rows {
            let action = match actual.get(key) {
                None => Some(true),
                Some(actual_bytes) => match (t.value_eq)(value, actual_bytes) {
                    Ok(false) => Some(false),
                    Ok(true) => None,
                    Err(e) => return Err(e),
                },
            };
            if let Some(missing) = action {
                t.index.put_bytes(key, value)?;
                let stats = stats_by_name
                    .entry(t.name)
                    .or_insert_with(|| IndexRepairStats {
                        name: t.name,
                        ..Default::default()
                    });
                if missing {
                    stats.puts += 1;
                } else {
                    stats.rewrites += 1;
                }
            }
        }
    }

    // Pass 3: orphans — index rows no 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) && !protected.contains(target_name);
        let orphans: Vec<Vec<u8>> = actual
            .keys()
            .filter(|kb| !expected.contains(target_name, kb))
            .cloned()
            .collect();
        for kb in orphans {
            let stats = stats_by_name
                .entry(target_name)
                .or_insert_with(|| IndexRepairStats {
                    name: target_name,
                    ..Default::default()
                });
            if complete && opts.delete_orphans {
                tdecl.index.delete_bytes(&kb)?;
                stats.deletes += 1;
            } else {
                stats.declined_orphans += 1;
            }
        }
    }

    let mut per_index: Vec<_> = stats_by_name.into_values().collect();
    per_index.sort_by_key(|s| s.name);

    // Pass 4: counters — shares `expected::build_counters` with `validate`,
    // exactly as the index passes above share `expected::build`.
    //
    // A source whose scan skipped rows contributes an undercount, so its
    // declared targets are declined here — merged into the same
    // `undecodable_sources` list the index passes populate above, since a
    // source could in principle feed both an index and a counter — rather
    // than rewritten from a short sum. Rewriting from a short scan would
    // *create* drift, not fix it.
    let counter_expected = expected::build_counters(reg)?;
    let mut per_counter: Vec<CounterRepairStats> = Vec::new();
    for d in &reg.reg.decls {
        let Decl::Counter {
            source, targets, ..
        } = d
        else {
            continue;
        };
        // An unregistered source must never be read as "expected zero" — it
        // has no `CollectionEntry`, so `build_counters` silently skipped this
        // whole declaration and `counter_expected` holds nothing for its
        // targets. Skip repairing them entirely rather than "fixing" every
        // stored row down to zero; `validate`'s structural check is what
        // reports this (see `run_decls`).
        let Some(src_entry) = reg.reg.entries.iter().find(|e| e.name == *source) else {
            continue;
        };
        let skipped_rows = (src_entry.undecodable_rows)();
        if skipped_rows > 0 {
            let target_names: Vec<&'static str> =
                targets.all().map(|t| t.collection_name()).collect();
            match undecodable_sources.iter_mut().find(|s| s.name == *source) {
                Some(existing) => {
                    for name in target_names {
                        if !existing.protected_targets.contains(&name) {
                            existing.protected_targets.push(name);
                        }
                    }
                }
                None => undecodable_sources.push(UndecodableSource {
                    name: source,
                    rows: skipped_rows,
                    protected_targets: target_names,
                }),
            }
            continue;
        }
        for t in targets.all() {
            let drift = expected::counter_drift(t, &counter_expected)?;
            let mut fixed = 0u64;
            for d in drift {
                let delta = i128::from(d.want) - i128::from(d.stored);
                // Mirror the open-time recount's leniency (`counter::recount`)
                // — and what `docs/schema-metadata.md` already documents for
                // `validate`/`repair`: a **dangling contribution** (a live
                // source row whose target row is gone) degrades to a `warn!`
                // and is skipped. Propagating it here would abort the whole
                // repair part-way through, after earlier deltas already
                // landed, over a row `validate` reports anyway.
                // `AmbiguousCounterTarget` has no safe target to skip to and
                // still propagates.
                // `Ok(false)` — the resolved decision was `Upsert::Keep`, so
                // nothing was written. Counting it would report a repair that
                // did not happen, on a mismatch that is still there at the
                // next run.
                match t.apply_delta(&d.key, delta) {
                    Ok(true) => fixed += 1,
                    Ok(false) => {}
                    Err(DbError::CounterTargetMissing { targets, key }) => {
                        tracing::warn!(
                            collection = t.collection_name(),
                            field = t.field(),
                            targets = %targets,
                            key = %key,
                            "repair: dangling contribution — target row missing, skipping (see schema::validate)"
                        );
                    }
                    Err(e) => return Err(e),
                }
            }
            per_counter.push(CounterRepairStats {
                collection: t.collection_name(),
                field: t.field(),
                fixed,
            });
        }
    }
    per_counter.sort_by(|a, b| (a.collection, a.field).cmp(&(b.collection, b.field)));

    // Pass 5: unique namespaces — shares `expected::build_unique` with
    // `validate`, exactly as passes 1-4 share `expected::build`/
    // `build_counters`. A namespace is repaired by ITS OWN pass, never the
    // generic `reconcile_carrier_projections` above — see
    // `schema::unique_repair`'s module doc.
    let (per_namespace, unique_skipped) = unique_repair::repair_namespaces(reg, &opts)?;
    for s in unique_skipped {
        match undecodable_sources.iter_mut().find(|u| u.name == s.name) {
            Some(existing) => {
                for name in s.targets {
                    if !existing.protected_targets.contains(&name) {
                        existing.protected_targets.push(name);
                    }
                }
            }
            None => undecodable_sources.push(UndecodableSource {
                name: s.name,
                rows: s.rows,
                protected_targets: s.targets,
            }),
        }
    }

    Ok(RepairReport {
        per_index,
        per_counter,
        per_namespace,
        undecodable_sources,
    })
}