haematite 0.6.1

//! `vacuum_stats` orchestration: preflight → lock → config → metadata →
//! per-store enumerate+mark+report → assemble. Report-only: the single
//! permitted write is the advisory lock anchor on a VALID legacy dir
//! (⟨r4, M5⟩ — validity established by a non-mutating config preflight
//! BEFORE the create-capable lock open), and the report names it.
//!
//! MEMORY DISCIPLINE (lens Q2): stores are processed strictly one at a
//! time — enumerate, walk every mark source that touches this store, emit
//! the shard's report numbers, DROP the inventory. Peak residency is the
//! global mark-root lists plus ONE store's node set, never all stores.

use std::path::Path;
use std::time::Instant;

use crate::db::lock::{DataDirLock, LOCK_FILE, LockError};
use crate::tree::Hash;

#[path = "shard_scan.rs"]
mod shard_scan;
use shard_scan::{ShardScan, scan_shard};

use super::consult::{MetadataPass, consult_metadata};
use super::error::{MarkSourceId, VacuumError};
use super::inventory::{StoreInventory, inventory_top_level};
use super::manifest::read_manifest;
use super::mark::{ProbeOutcome, StoreMarks, commit_marks, probe_store, walk_declared};
use super::report::{
    MarkTallies, MetadataReport, NodeTotals, ShardPresence, ShardReport, SweepBlocker, VacuumMode,
    VacuumReport,
};
use super::{CANONICAL_REFS_DIR, CANONICAL_SNAPSHOT_REGISTRY, VacuumOptions};

/// §6: the trust boundary named in every report — what the lock cannot see.
const TRUST_BOUNDARY: &str = "This run held the A4 data-dir writer lock, which excludes every \
     writer routed through a live Database. It does NOT exclude direct \
     DiskStore/tree/WAL/branch-layer/sync access opened on paths under this data_dir by other \
     code: directories under a database data_dir are owned by that database, and direct access \
     forfeits vacuum safety (STORAGE-VACUUM.md §6). If any such tool touches this directory, \
     stop before sweeping.";

pub fn run(options: &VacuumOptions) -> Result<VacuumReport, VacuumError> {
    let started = Instant::now();
    let data_dir = options.data_dir.as_path();

    // NON-MUTATING PREFLIGHT: `DataDirLock::acquire` opens the anchor with
    // create(true), and the ⟨r4, M5⟩ exception licenses that write only on
    // a VALID legacy data dir — so config must read and parse BEFORE the
    // create-capable open; a refusal on a wrong directory changes nothing.
    read_config_checked(data_dir)?;

    // ⟨r4, M5⟩ acquiring the lock on a valid legacy dir creates exactly one
    // empty lockfile, the design's single named report-only write exception.
    // Creation is CAUSAL (Sol r3 fold): `acquire` reports it from its own
    // atomic `create_new`, never from before/after probes, and a symlinked
    // anchor refuses before any create-capable open.
    let lock = DataDirLock::acquire(data_dir).map_err(map_lock_error)?;
    let anchor_created = lock.created_anchor();
    let _lock = lock;

    // Authoritative re-read UNDER the lock — the preflight is never trusted
    // against a live writer that raced us to the directory.
    let config = read_config_checked(data_dir)?;
    let shard_count = config.shard_count;

    let manifest = read_manifest(data_dir)?;
    let metadata = consult_metadata(data_dir, options, manifest.as_ref())?;
    let top = inventory_top_level(
        data_dir,
        shard_count,
        &[
            crate::db::CONFIG_FILE,
            LOCK_FILE,
            super::manifest::MANIFEST_FILE,
            CANONICAL_REFS_DIR,
            CANONICAL_SNAPSHOT_REGISTRY,
        ],
        &metadata.listed_paths,
    )?;

    let mut tallies = MarkTallies::default();
    let branch_roots = group_branch_roots(&metadata, shard_count, &mut tallies)?;
    let mut snapshots = group_snapshot_roots(&metadata, shard_count, &mut tallies)?;

    // One store at a time (module doc): an inventory lives one iteration.
    // Reports grow incrementally — the persisted count never drives an
    // eager allocation (Sol r3 fold).
    let mut verified = 0_usize;
    let mut totals = NodeTotals::default();
    let mut shard_reports = Vec::new();
    let mut never_materialised = 0_usize;
    let mut blockers = metadata.blockers;
    for shard_id in 0..shard_count {
        let scan = scan_shard(data_dir, shard_id)?;
        let shard_branch_roots: &[(Hash, MarkSourceId)] =
            branch_roots.get(&shard_id).map_or(&[], Vec::as_slice);
        if scan.presence == ShardPresence::NeverMaterialised && shard_branch_roots.is_empty() {
            // Lazy and unreferenced: nothing can mark or resolve in an
            // empty store, so aggregate instead of materialising a report
            // entry — the report must stay O(materialised), not
            // O(shard_count). (A lazy shard a branch record DOES reference
            // still processes below and refuses UnresolvableRoot.)
            never_materialised += 1;
            continue;
        }
        let report = process_shard(
            scan,
            shard_branch_roots,
            &mut snapshots,
            &mut tallies,
            &mut verified,
        )?;
        totals.accumulate(report.nodes);
        if !report.malformed.is_empty() {
            blockers.push(SweepBlocker::MalformedStoreEntries {
                shard_id,
                count: report.malformed.len(),
            });
        }
        shard_reports.push(report);
    }

    // §2 M3: a snapshot root that resolved in NO candidate store — nowhere
    // (unbound) or nowhere DECLARED (bound) — is a typed refusal.
    for snapshot in &snapshots {
        if !snapshot.resolved {
            return Err(VacuumError::UnresolvableRoot {
                source: snapshot.source.clone(),
                root: snapshot.root,
                missing: snapshot.last_missing,
            });
        }
    }

    for id in top.shards_beyond_count {
        blockers.push(SweepBlocker::ShardBeyondCount { id, shard_count });
    }
    #[cfg(not(unix))]
    blockers.push(SweepBlocker::UnsupportedDurability);

    Ok(VacuumReport {
        mode: VacuumMode::Stats,
        data_dir: options.data_dir.clone(),
        shard_count,
        duration_ms: u64::try_from(started.elapsed().as_millis()).unwrap_or(u64::MAX),
        lock_anchor_created: anchor_created,
        never_materialised_shards: never_materialised,
        shards: shard_reports,
        totals,
        metadata: MetadataReport {
            manifest: metadata.manifest_report,
            attestation: options.attest_metadata_complete,
            sources: metadata.sources,
            supplied_missing: metadata.supplied_missing,
        },
        marks: tallies,
        verified_nodes: verified,
        uninventoried: top.uninventoried,
        sweep_blockers: blockers,
        trust_boundary: TRUST_BOUNDARY.to_owned(),
    })
}

/// Map lock acquisition failures to vacuum refusals. A symlinked anchor is
/// refused with its remedy named; Io failures carry the CAUSAL
/// anchor-created disclosure (⟨r4, M5⟩ transparency on the error path too —
/// there is no report to carry it, so the refusal does).
fn map_lock_error(error: LockError) -> VacuumError {
    match error {
        LockError::AlreadyLocked {
            lock_path,
            created_anchor,
        } => VacuumError::Locked {
            lock_path,
            anchor_created: created_anchor,
        },
        LockError::AnchorNotARegularFile { lock_path } => VacuumError::LockIo {
            error: std::io::Error::other(
                "writer.lock is not a regular file — a symlinked anchor would make the \
                 create-capable open mint an inode OUTSIDE the data dir; remove or replace \
                 it with a plain empty file, then re-run",
            ),
            lock_path,
            anchor_created: false,
        },
        LockError::Io {
            lock_path,
            error,
            created_anchor,
        } => VacuumError::LockIo {
            lock_path,
            error,
            anchor_created: created_anchor,
        },
    }
}

/// Read `config.json` and validate ONLY what stats consumes (Sol r4 fold):
/// parse/format validity and a nonzero `shard_count`. Sweep-interval and
/// distributed-topology validation deliberately do NOT run here — offline
/// stats must stay available when scheduler/topology config is damaged
/// (that is when an inventory matters most), and `FullMesh` validation
/// materialises O(nodes²) pairs, the count-derived allocation class again.
/// Every failure is the ⟨r15, C1⟩ `ConfigUnreadable` refusal: there is no
/// shard iteration without a trusted `shard_count`, and a zero count would
/// silently skip the exhaustive M1 source. No upper ceiling: the engine
/// accepts any nonzero count at create/open, and a vacuum-only bound would
/// invent a compatibility rule the certified design does not license (an
/// engine-wide maximum is routed to the pair; the vacuum adopts whatever
/// they sign, where they sign it).
fn read_config_checked(data_dir: &Path) -> Result<crate::db::DatabaseConfig, VacuumError> {
    let config_unreadable = |reason: String| VacuumError::ConfigUnreadable {
        path: data_dir.join(crate::db::CONFIG_FILE),
        reason,
    };
    let config = crate::db::config::read_config(data_dir)
        .map_err(|error| config_unreadable(error.to_string()))?;
    if config.shard_count == 0 {
        return Err(config_unreadable(
            "shard_count is 0; the engine requires at least 1 shard, so this config \
             cannot be the engine's own work"
                .to_owned(),
        ));
    }
    Ok(config)
}

/// M2 roots grouped per shard (§2: each anchor/head walks in exactly
/// `shard-{record.shard_id}/store`), ids validated against `shard_count`.
/// SPARSE by shard id (Sol r3 fold): storage is O(records), never
/// `O(shard_count)` — the persisted count must not drive eager allocation.
fn group_branch_roots(
    metadata: &MetadataPass,
    shard_count: usize,
    tallies: &mut MarkTallies,
) -> Result<std::collections::HashMap<usize, Vec<(Hash, MarkSourceId)>>, VacuumError> {
    let mut per_shard: std::collections::HashMap<usize, Vec<(Hash, MarkSourceId)>> =
        std::collections::HashMap::new();
    for scan in &metadata.refs_scans {
        for record in &scan.records {
            for shard_ref in &record.shards {
                if shard_ref.shard_id >= shard_count {
                    return Err(VacuumError::ShardBeyondCount {
                        id: shard_ref.shard_id,
                        shard_count,
                    });
                }
                let roots = per_shard.entry(shard_ref.shard_id).or_default();
                roots.push((
                    shard_ref.fork_anchor,
                    MarkSourceId::BranchAnchor {
                        branch: record.name.clone(),
                        shard_id: shard_ref.shard_id,
                    },
                ));
                roots.push((
                    shard_ref.head,
                    MarkSourceId::BranchHead {
                        branch: record.name.clone(),
                        shard_id: shard_ref.shard_id,
                    },
                ));
                tallies.branch_anchors += 1;
                tallies.branch_heads += 1;
            }
        }
    }
    Ok(per_shard)
}

/// The candidate scope for one M3 root (§2 M3): the manifest-DECLARED
/// shards for a bound registry, or every shard for the unbound
/// resolve-in-at-least-one fallback. Candidate storage is O(1) per root
/// (Sol r2 fold): unbound roots carry NO list at all, and every root of one
/// bound registry SHARES that registry's single `Arc` of declared ids — the
/// lens-Q2 bound must not be violated by the work-list either.
#[derive(Clone)]
enum SnapshotCandidates {
    AllShards,
    Declared(std::sync::Arc<[usize]>),
}

impl SnapshotCandidates {
    fn applies_to(&self, shard_id: usize) -> bool {
        match self {
            Self::AllShards => true,
            Self::Declared(ids) => ids.contains(&shard_id),
        }
    }
}

/// One M3 root's cross-store resolution state. Resolution is tracked across
/// the sequential store passes; marking happens in EVERY candidate store
/// where the root fully resolves (over-mark is leak-safe).
struct SnapshotWork {
    root: Hash,
    source: MarkSourceId,
    candidates: SnapshotCandidates,
    resolved: bool,
    last_missing: Hash,
}

fn group_snapshot_roots(
    metadata: &MetadataPass,
    shard_count: usize,
    tallies: &mut MarkTallies,
) -> Result<Vec<SnapshotWork>, VacuumError> {
    let mut work = Vec::new();
    for registry in &metadata.registries {
        let candidates = match &registry.declared_shards {
            Some(declared) => {
                for &shard_id in declared {
                    if shard_id >= shard_count {
                        return Err(VacuumError::ShardBeyondCount {
                            id: shard_id,
                            shard_count,
                        });
                    }
                }
                SnapshotCandidates::Declared(std::sync::Arc::from(declared.as_slice()))
            }
            None => SnapshotCandidates::AllShards,
        };
        for entry in &registry.entries {
            work.push(SnapshotWork {
                root: entry.root_hash,
                source: MarkSourceId::Snapshot {
                    name: entry.name.clone(),
                    registry: registry.path.clone(),
                },
                candidates: candidates.clone(),
                resolved: false,
                last_missing: entry.root_hash,
            });
            tallies.snapshot_roots += 1;
        }
    }
    Ok(work)
}

/// Walk every mark source that touches this store, then reduce the scan to
/// its report numbers. The inventory is consumed and dropped HERE — this is
/// what keeps peak residency at one store (module doc).
fn process_shard(
    scan: ShardScan,
    branch_roots: &[(Hash, MarkSourceId)],
    snapshots: &mut [SnapshotWork],
    tallies: &mut MarkTallies,
    verified: &mut usize,
) -> Result<ShardReport, VacuumError> {
    let ShardScan {
        shard_id,
        presence,
        committed_root,
        store_dir,
        mut inventory,
        mut shard_malformed,
    } = scan;
    let mut marks = StoreMarks::new();

    // M1: the WAL committed root marks within its own store.
    if let Some(root) = committed_root {
        walk_declared(
            &inventory,
            &store_dir,
            root,
            &MarkSourceId::WalRoot { shard_id },
            &mut marks,
            verified,
        )?;
        tallies.wal_roots += 1;
    }
    // M2: anchors/heads whose records declare THIS shard.
    for (root, source) in branch_roots {
        walk_declared(&inventory, &store_dir, *root, source, &mut marks, verified)?;
    }
    // M3: probe each candidate snapshot root; mark where it fully resolves.
    for snapshot in snapshots.iter_mut() {
        if !snapshot.candidates.applies_to(shard_id) {
            continue;
        }
        match probe_store(&inventory, &store_dir, snapshot.root, &marks, verified)? {
            ProbeOutcome::Resolved(resolved) => {
                commit_marks(&inventory, &resolved, &mut marks);
                snapshot.resolved = true;
            }
            ProbeOutcome::Missing(missing) => snapshot.last_missing = missing,
        }
    }

    let nodes = shard_totals(&inventory, &marks);
    shard_malformed.append(&mut inventory.malformed);
    Ok(ShardReport {
        shard_id,
        presence,
        wal_committed_root: committed_root.is_some(),
        nodes,
        temp_debris_files: inventory.temp_debris_files,
        temp_debris_bytes: inventory.temp_debris_bytes,
        malformed: shard_malformed,
    })
}

fn shard_totals(inventory: &StoreInventory, marks: &StoreMarks) -> NodeTotals {
    let enumerated_nodes = inventory.nodes.len();
    let enumerated_bytes: u64 = inventory
        .nodes
        .values()
        .map(|file| file.compressed_len)
        .sum();
    let marked_nodes = marks.len();
    let marked_bytes: u64 = marks.values().sum();
    NodeTotals {
        enumerated_nodes,
        enumerated_bytes,
        marked_nodes,
        marked_bytes,
        unmarked_nodes: enumerated_nodes.saturating_sub(marked_nodes),
        unmarked_bytes: enumerated_bytes.saturating_sub(marked_bytes),
    }
}

#[cfg(test)]
#[path = "stats_candidate_tests.rs"]
mod candidate_shape_tests;