haematite 0.6.2

//! Metadata-universe consultation (§3): manifest entries first, then the
//! canonical in-dir locations, then explicitly supplied paths — each source
//! consulted exactly once, every consultation recorded in the report, and
//! every condition a sweep would refuse on captured as a blocker.

use std::collections::HashSet;
use std::path::{Path, PathBuf};

use crate::branch::{BranchRefRecord, SnapshotEntry};

use super::error::VacuumError;
use super::manifest::{ManifestSourceKind, MetadataManifest};
use super::metadata::{read_refs_dir, read_snapshot_registry};
use super::report::{
    ManifestEntryReport, ManifestReport, MetadataOrigin, MetadataSourceKind, MetadataSourceReport,
    SweepBlocker,
};
use super::{CANONICAL_REFS_DIR, CANONICAL_SNAPSHOT_REGISTRY, VacuumOptions};

/// The consulted metadata universe (§3).
pub struct MetadataPass {
    pub manifest_report: Option<ManifestReport>,
    pub sources: Vec<MetadataSourceReport>,
    pub refs_scans: Vec<RefsSource>,
    pub registries: Vec<RegistrySource>,
    pub supplied_missing: Vec<PathBuf>,
    pub blockers: Vec<SweepBlocker>,
    /// Manifest-listed resolved paths, for the top-level accounting.
    pub listed_paths: Vec<PathBuf>,
}

pub struct RefsSource {
    pub records: Vec<BranchRefRecord>,
}

pub struct RegistrySource {
    pub path: PathBuf,
    pub entries: Vec<SnapshotEntry>,
    /// Manifest-declared store association (§2 M3): `Some` binds the walk to
    /// exactly these shards; `None` is the unbound resolve-in-≥1 fallback.
    pub declared_shards: Option<Vec<usize>>,
}

pub fn consult_metadata(
    data_dir: &Path,
    options: &VacuumOptions,
    manifest: Option<&MetadataManifest>,
) -> Result<MetadataPass, VacuumError> {
    let mut pass = MetadataPass {
        manifest_report: None,
        sources: Vec::new(),
        refs_scans: Vec::new(),
        registries: Vec::new(),
        supplied_missing: Vec::new(),
        blockers: Vec::new(),
        listed_paths: Vec::new(),
    };
    let mut consulted: HashSet<PathBuf> = HashSet::new();

    if let Some(manifest) = manifest {
        consult_manifest_entries(data_dir, manifest, &mut pass, &mut consulted)?;
    } else if !options.attest_metadata_complete {
        // §3: no manifest and no attestation — sweep would refuse; stats
        // reports the fact and proceeds on canonical + supplied paths.
        pass.blockers.push(SweepBlocker::MetadataUnattested);
    }

    consult_canonical(data_dir, manifest.is_some(), &mut pass, &mut consulted)?;
    consult_supplied(options, &mut pass, &mut consulted)?;
    Ok(pass)
}

fn consult_manifest_entries(
    data_dir: &Path,
    manifest: &MetadataManifest,
    pass: &mut MetadataPass,
    consulted: &mut HashSet<PathBuf>,
) -> Result<(), VacuumError> {
    let mut entry_reports = Vec::new();
    for entry in &manifest.entries {
        let path = MetadataManifest::resolved_path(data_dir, entry);
        pass.listed_paths.push(path.clone());
        if !entry.state.is_settled() {
            pass.blockers.push(SweepBlocker::ManifestEntryUnsettled {
                path: path.clone(),
                state: entry.state.display_name(),
            });
        }
        // `source_readable` means the source was actually SCANNED, not that
        // a path lookup succeeded (Sol r1 fold): absent ⇒ missing blocker;
        // inaccessible LOOKUP (execute-denied/unmounted parent, Sol r2) or
        // unreadable contents ⇒ unreadable blocker, stats completes (§3:
        // stats reports each entry's state and whatever records are
        // visible); only codec corruption fails the run.
        let mut readable = false;
        match probe_path(&path) {
            PathPresence::Present => {
                // §3 ⟨r7⟩: stats reports whatever records are visible,
                // settled or not — over-marking from an unsettled source is
                // leak-safe.
                readable = attempt_consult(
                    entry.kind,
                    &path,
                    MetadataOrigin::Manifest,
                    entry.shards.clone(),
                    pass,
                )?;
                consulted.insert(path.clone());
            }
            PathPresence::Absent => {
                // ⟨r5, B5⟩ any missing listed source is a sweep refusal,
                // always; stats reports it with its entry state.
                pass.blockers
                    .push(SweepBlocker::ManifestSourceMissing { path: path.clone() });
            }
            PathPresence::Inaccessible(detail) => {
                pass.blockers.push(SweepBlocker::SourceUnreadable {
                    path: path.clone(),
                    detail,
                });
            }
        }
        entry_reports.push(ManifestEntryReport {
            path,
            kind: map_kind(entry.kind),
            state: entry.state.display_name(),
            source_readable: readable,
        });
    }
    pass.manifest_report = Some(ManifestReport {
        path: data_dir.join(super::manifest::MANIFEST_FILE),
        generation: manifest.generation,
        entries: entry_reports,
    });
    Ok(())
}

fn consult_canonical(
    data_dir: &Path,
    manifest_present: bool,
    pass: &mut MetadataPass,
    consulted: &mut HashSet<PathBuf>,
) -> Result<(), VacuumError> {
    let canonical = [
        (
            data_dir.join(CANONICAL_REFS_DIR),
            ManifestSourceKind::RefsDir,
        ),
        (
            data_dir.join(CANONICAL_SNAPSHOT_REGISTRY),
            ManifestSourceKind::SnapshotRegistry,
        ),
    ];
    for (path, kind) in canonical {
        if consulted.contains(&path) {
            continue;
        }
        match probe_path(&path) {
            // Only the CANONICAL locations may be validly absent (§3): a
            // database that never used branches has no branches/ dir.
            PathPresence::Absent => continue,
            PathPresence::Inaccessible(detail) => {
                pass.blockers
                    .push(SweepBlocker::SourceUnreadable { path, detail });
                continue;
            }
            PathPresence::Present => {}
        }
        if manifest_present {
            // ⟨r4⟩ stale-manifest detection: a canonical source the manifest
            // omits is proof the manifest lies.
            pass.blockers.push(SweepBlocker::StaleManifest {
                unlisted: path.clone(),
            });
        }
        attempt_consult(kind, &path, MetadataOrigin::Canonical, None, pass)?;
        consulted.insert(path);
    }
    Ok(())
}

fn consult_supplied(
    options: &VacuumOptions,
    pass: &mut MetadataPass,
    consulted: &mut HashSet<PathBuf>,
) -> Result<(), VacuumError> {
    let supplied = options
        .refs_dirs
        .iter()
        .map(|path| (path.clone(), ManifestSourceKind::RefsDir))
        .chain(
            options
                .snapshot_files
                .iter()
                .map(|path| (path.clone(), ManifestSourceKind::SnapshotRegistry)),
        );
    for (path, kind) in supplied {
        if consulted.contains(&path) {
            continue;
        }
        match probe_path(&path) {
            PathPresence::Absent => {
                // ⟨r4, B4⟩ attestation attests completeness, not existence:
                // a typo'd supplied path is a sweep refusal, never silently
                // skipped. Stats reports it.
                pass.supplied_missing.push(path.clone());
                pass.blockers
                    .push(SweepBlocker::SuppliedPathMissing { path });
                continue;
            }
            PathPresence::Inaccessible(detail) => {
                pass.blockers
                    .push(SweepBlocker::SourceUnreadable { path, detail });
                continue;
            }
            PathPresence::Present => {}
        }
        attempt_consult(kind, &path, MetadataOrigin::Supplied, None, pass)?;
        consulted.insert(path);
    }
    Ok(())
}

/// Tri-state metadata-path probe (Sol r2 fold): a lookup that fails for an
/// ACCESS reason (execute-denied or unmounted parent) is `Inaccessible` —
/// reported like any unreadable source, never a hard stats abort. Only
/// `NotFound` means absent.
enum PathPresence {
    Present,
    Absent,
    Inaccessible(String),
}

fn probe_path(path: &Path) -> PathPresence {
    match std::fs::symlink_metadata(path) {
        Ok(_metadata) => PathPresence::Present,
        Err(error) if error.kind() == std::io::ErrorKind::NotFound => PathPresence::Absent,
        Err(error) => PathPresence::Inaccessible(error.to_string()),
    }
}

/// Consult one source, converting ACCESS failures (open/read I/O) into the
/// `SourceUnreadable` sweep blocker so stats completes and the operator sees
/// the source it could not read (§3). Codec corruption still propagates —
/// bytes that were read but do not decode are a dropped pin, and the vacuum
/// fails loud on those. Returns whether the source was actually scanned.
fn attempt_consult(
    kind: ManifestSourceKind,
    path: &Path,
    origin: MetadataOrigin,
    declared_shards: Option<Vec<usize>>,
    pass: &mut MetadataPass,
) -> Result<bool, VacuumError> {
    match consult_source(kind, path, origin, declared_shards, pass) {
        Ok(()) => Ok(true),
        Err(VacuumError::Io { error, .. }) => {
            pass.blockers.push(SweepBlocker::SourceUnreadable {
                path: path.to_path_buf(),
                detail: error.to_string(),
            });
            Ok(false)
        }
        Err(other) => Err(other),
    }
}

fn consult_source(
    kind: ManifestSourceKind,
    path: &Path,
    origin: MetadataOrigin,
    declared_shards: Option<Vec<usize>>,
    pass: &mut MetadataPass,
) -> Result<(), VacuumError> {
    match kind {
        ManifestSourceKind::RefsDir => {
            let scan = read_refs_dir(path)?;
            pass.sources.push(MetadataSourceReport {
                path: path.to_path_buf(),
                kind: MetadataSourceKind::RefsDir,
                origin,
                records: scan.records.len(),
                temp_debris_files: scan.temp_debris_files,
                foreign_entries: scan.foreign_entries,
            });
            pass.refs_scans.push(RefsSource {
                records: scan.records,
            });
        }
        ManifestSourceKind::SnapshotRegistry => {
            let entries = read_snapshot_registry(path)?;
            pass.sources.push(MetadataSourceReport {
                path: path.to_path_buf(),
                kind: MetadataSourceKind::SnapshotRegistry,
                origin,
                records: entries.len(),
                temp_debris_files: 0,
                foreign_entries: 0,
            });
            pass.registries.push(RegistrySource {
                path: path.to_path_buf(),
                entries,
                declared_shards,
            });
        }
    }
    Ok(())
}

const fn map_kind(kind: ManifestSourceKind) -> MetadataSourceKind {
    match kind {
        ManifestSourceKind::RefsDir => MetadataSourceKind::RefsDir,
        ManifestSourceKind::SnapshotRegistry => MetadataSourceKind::SnapshotRegistry,
    }
}