haematite 0.6.1

//! Strict store enumeration and the top-level data-dir inventory (§4).
//!
//! No stored-node listing API exists; this is the vacuum's crate-internal
//! `DiskStore` inventory. It accepts exactly the canonical layout
//! (`<store>/<2 lowercase hex>/<62 lowercase hex>`, regular files only),
//! recognises `.node-*.tmp` install debris, and reports everything else as
//! malformed — an unexpected layout means our model of the directory is
//! wrong, and deleting by complement under a wrong model deletes evidence.

use std::collections::HashMap;
use std::fs;
use std::io;
use std::path::{Path, PathBuf};

use crate::tree::Hash;
use crate::tree::node::HASH_SIZE;

use super::error::VacuumError;
use super::report::{EntryKind, MalformedEntry, UninventoriedEntry};

/// Node-file temp-install debris naming, pinned by `store/disk.rs`
/// (`tempfile::Builder::prefix(".node-").suffix(".tmp")`).
const NODE_TEMP_PREFIX: &str = ".node-";
const NODE_TEMP_SUFFIX: &str = ".tmp";

const PREFIX_HEX_LEN: usize = 2;
const NAME_HEX_LEN: usize = HASH_SIZE * 2 - PREFIX_HEX_LEN;

/// One enumerated node file: where it lives and its compressed size (the
/// operator-visible byte number, §4).
#[derive(Debug)]
pub struct NodeFile {
    pub path: PathBuf,
    pub compressed_len: u64,
}

/// The strict inventory of one shard store.
///
/// LIFETIME DISCIPLINE (lens Q2): at most ONE inventory is live at a time —
/// the stats run enumerates, walks, and reports each store before moving to
/// the next, so memory stays bounded by (mark roots + one store's node set),
/// never all stores at once. The test-only residency gauge pins this.
#[derive(Debug, Default)]
pub struct StoreInventory {
    /// Every canonical node file, keyed by the hash reconstructed (and
    /// validated) from its path components.
    pub nodes: HashMap<Hash, NodeFile>,
    pub temp_debris_files: usize,
    pub temp_debris_bytes: u64,
    pub malformed: Vec<MalformedEntry>,
    /// Whether this inventory came from [`enumerate_store`] (gauge-counted),
    /// as opposed to the empty placeholder for lazy shards.
    #[cfg(test)]
    counted: bool,
}

#[cfg(test)]
impl Drop for StoreInventory {
    fn drop(&mut self) {
        if self.counted {
            residency::on_drop();
        }
    }
}

/// Test-only gauge for the Q2 single-store residency pin: counts live
/// enumerated inventories per thread (a stats run is single-threaded).
#[cfg(test)]
pub mod residency {
    use std::cell::Cell;

    thread_local! {
        static LIVE: Cell<usize> = const { Cell::new(0) };
        static PEAK: Cell<usize> = const { Cell::new(0) };
    }

    pub(super) fn on_enumerate() {
        LIVE.with(|live| {
            let now = live.get() + 1;
            live.set(now);
            PEAK.with(|peak| peak.set(peak.get().max(now)));
        });
    }

    pub(super) fn on_drop() {
        LIVE.with(|live| live.set(live.get().saturating_sub(1)));
    }

    pub fn reset() {
        LIVE.with(|live| live.set(0));
        PEAK.with(|peak| peak.set(0));
    }

    pub fn peak() -> usize {
        PEAK.with(Cell::get)
    }
}

/// Enumerate `store_dir` against the canonical layout. Read-only.
pub fn enumerate_store(store_dir: &Path) -> Result<StoreInventory, VacuumError> {
    let mut inventory = StoreInventory::default();
    #[cfg(test)]
    {
        inventory.counted = true;
        residency::on_enumerate();
    }
    for entry in read_dir(store_dir)? {
        let entry = entry.map_err(|error| io_at(store_dir, error))?;
        let path = entry.path();
        let kind = entry_kind(&path)?;
        let Some(name) = file_name_str(&path) else {
            inventory.malformed.push(malformed(&path, "non-UTF8 name"));
            continue;
        };
        if kind != EntryKind::Directory {
            inventory
                .malformed
                .push(malformed(&path, "top-level store entry is not a directory"));
            continue;
        }
        if !is_lowercase_hex(name) || name.len() != PREFIX_HEX_LEN {
            inventory.malformed.push(malformed(
                &path,
                "directory name is not 2 lowercase hex characters",
            ));
            continue;
        }
        enumerate_prefix_dir(&path, name, &mut inventory)?;
    }
    Ok(inventory)
}

/// Enumerate one `<2 hex>` fanout directory into the inventory.
fn enumerate_prefix_dir(
    prefix_dir: &Path,
    prefix: &str,
    inventory: &mut StoreInventory,
) -> Result<(), VacuumError> {
    for entry in read_dir(prefix_dir)? {
        let entry = entry.map_err(|error| io_at(prefix_dir, error))?;
        let path = entry.path();
        let kind = entry_kind(&path)?;
        let Some(name) = file_name_str(&path) else {
            inventory.malformed.push(malformed(&path, "non-UTF8 name"));
            continue;
        };
        if kind == EntryKind::File
            && name.starts_with(NODE_TEMP_PREFIX)
            && name.ends_with(NODE_TEMP_SUFFIX)
        {
            // Install debris: reported, never treated as a node, never
            // deleted by sweep v1 (§4). The kind check comes FIRST: a
            // symlink or directory wearing the debris name is malformed,
            // not debris.
            inventory.temp_debris_files += 1;
            inventory.temp_debris_bytes += file_len(&path)?;
            continue;
        }
        if kind != EntryKind::File {
            inventory
                .malformed
                .push(malformed(&path, "node entry is not a regular file"));
            continue;
        }
        if name.len() != NAME_HEX_LEN || !is_lowercase_hex(name) {
            inventory.malformed.push(malformed(
                &path,
                "file name is not 62 lowercase hex characters",
            ));
            continue;
        }
        // Reconstruct and validate the 64-hex hash from path components.
        let Some(hash) = parse_hash(prefix, name) else {
            inventory
                .malformed
                .push(malformed(&path, "path components do not form a valid hash"));
            continue;
        };
        let compressed_len = file_len(&path)?;
        inventory.nodes.insert(
            hash,
            NodeFile {
                path,
                compressed_len,
            },
        );
    }
    Ok(())
}

/// ⟨r15, C1⟩ the top-level inventory of `data_dir` itself.
#[derive(Debug, Default)]
pub struct TopLevelInventory {
    pub uninventoried: Vec<UninventoriedEntry>,
    /// `shard-{id}` directories with `id >= shard_count` — typed sweep
    /// refusal; stats completes and reports them.
    pub shards_beyond_count: Vec<usize>,
}

/// Inventory the top level of `data_dir`: every entry not accounted for by
/// the model is enumerated, never assumed. `expected_names` carries the
/// model's fixed names (config, lock, manifest, canonical metadata);
/// `listed_paths` carries manifest-listed source paths (accounted when they
/// live directly under `data_dir`).
pub fn inventory_top_level(
    data_dir: &Path,
    shard_count: usize,
    expected_names: &[&str],
    listed_paths: &[PathBuf],
) -> Result<TopLevelInventory, VacuumError> {
    let mut top = TopLevelInventory::default();
    for entry in read_dir(data_dir)? {
        let entry = entry.map_err(|error| io_at(data_dir, error))?;
        let path = entry.path();
        let kind = entry_kind(&path)?;
        let Some(name) = file_name_str(&path) else {
            top.uninventoried.push(UninventoriedEntry {
                name: path.file_name().map_or_else(
                    || String::from("<unnamed>"),
                    |os| os.to_string_lossy().into_owned(),
                ),
                kind,
                size_bytes: entry_size(&path, kind),
            });
            continue;
        };
        if expected_names.contains(&name) {
            continue;
        }
        if listed_paths.iter().any(|listed| listed == &path) {
            continue;
        }
        if let Some(id) = parse_shard_dir_name(name) {
            if id < shard_count {
                // In-range shard names belong to the shard pass entirely —
                // it validates the entry's real type and refuses non-dirs.
                continue;
            }
            // Out-of-model: ALWAYS enumerated with kind and size (§4 —
            // its bytes must not disappear from the inventory), and a real
            // directory additionally escalates to the ⟨r15, C1⟩ sweep
            // refusal. A file or symlink coincidentally named `shard-{id}`
            // is an ordinary uninventoried entry, not a shard.
            if kind == EntryKind::Directory {
                top.shards_beyond_count.push(id);
            }
        }
        top.uninventoried.push(UninventoriedEntry {
            name: name.to_owned(),
            kind,
            size_bytes: entry_size(&path, kind),
        });
    }
    top.shards_beyond_count.sort_unstable();
    Ok(top)
}

/// Parse `shard-{id}` exactly: no leading zeros beyond the number itself, no
/// prefix/suffix noise. `shard-00` or `shard-1x` are NOT shard directories —
/// they fall through to the uninventoried list.
fn parse_shard_dir_name(name: &str) -> Option<usize> {
    let id_text = name.strip_prefix("shard-")?;
    let id: usize = id_text.parse().ok()?;
    // Round-trip to reject non-canonical spellings ("007", "+7", etc.).
    (id.to_string() == id_text).then_some(id)
}

fn read_dir(path: &Path) -> Result<fs::ReadDir, VacuumError> {
    fs::read_dir(path).map_err(|error| io_at(path, error))
}

fn io_at(path: &Path, error: io::Error) -> VacuumError {
    VacuumError::Io {
        path: path.to_path_buf(),
        error,
    }
}

/// Classify without following symlinks (§4 ⟨r2⟩: only regular files count;
/// symlinks and other file types are malformed entries).
fn entry_kind(path: &Path) -> Result<EntryKind, VacuumError> {
    let metadata = fs::symlink_metadata(path).map_err(|error| io_at(path, error))?;
    Ok(kind_of(metadata.file_type()))
}

/// Classify a path without following symlinks; `Ok(None)` when absent.
/// The model-boundary primitive: every shard/store/WAL presence decision
/// goes through this so a symlink can never smuggle external data into the
/// inventory (or hide as a lazy shard).
pub fn lstat_kind(path: &Path) -> Result<Option<EntryKind>, VacuumError> {
    match fs::symlink_metadata(path) {
        Ok(metadata) => Ok(Some(kind_of(metadata.file_type()))),
        Err(error) if error.kind() == io::ErrorKind::NotFound => Ok(None),
        Err(error) => Err(io_at(path, error)),
    }
}

fn kind_of(file_type: fs::FileType) -> EntryKind {
    if file_type.is_symlink() {
        EntryKind::Symlink
    } else if file_type.is_dir() {
        EntryKind::Directory
    } else if file_type.is_file() {
        EntryKind::File
    } else {
        EntryKind::Other
    }
}

fn file_len(path: &Path) -> Result<u64, VacuumError> {
    fs::symlink_metadata(path)
        .map(|metadata| metadata.len())
        .map_err(|error| io_at(path, error))
}

fn file_name_str(path: &Path) -> Option<&str> {
    path.file_name().and_then(|name| name.to_str())
}

fn is_lowercase_hex(text: &str) -> bool {
    !text.is_empty()
        && text
            .bytes()
            .all(|byte| byte.is_ascii_digit() || (b'a'..=b'f').contains(&byte))
}

/// Reconstruct the content hash from `<prefix><name>` (64 lowercase hex).
fn parse_hash(prefix: &str, name: &str) -> Option<Hash> {
    let mut hex = String::with_capacity(HASH_SIZE * 2);
    hex.push_str(prefix);
    hex.push_str(name);
    let mut bytes = [0_u8; HASH_SIZE];
    for (index, chunk) in hex.as_bytes().chunks_exact(2).enumerate() {
        let high = hex_value(chunk[0])?;
        let low = hex_value(chunk[1])?;
        bytes[index] = (high << 4) | low;
    }
    Some(Hash::from_bytes(bytes))
}

const fn hex_value(byte: u8) -> Option<u8> {
    match byte {
        b'0'..=b'9' => Some(byte - b'0'),
        b'a'..=b'f' => Some(byte - b'a' + 10),
        _ => None,
    }
}

/// Best-effort recursive size for an uninventoried entry: `None` when any
/// part is unreadable — the entry is still enumerated, never hidden.
fn entry_size(path: &Path, kind: EntryKind) -> Option<u64> {
    match kind {
        EntryKind::File | EntryKind::Symlink | EntryKind::Other => fs::symlink_metadata(path)
            .map(|metadata| metadata.len())
            .ok(),
        EntryKind::Directory => directory_size(path),
    }
}

fn directory_size(path: &Path) -> Option<u64> {
    let mut total = 0_u64;
    for entry in fs::read_dir(path).ok()? {
        let entry = entry.ok()?;
        let child = entry.path();
        let metadata = fs::symlink_metadata(&child).ok()?;
        if metadata.file_type().is_dir() {
            total = total.checked_add(directory_size(&child)?)?;
        } else {
            total = total.checked_add(metadata.len())?;
        }
    }
    Some(total)
}

fn malformed(path: &Path, reason: &str) -> MalformedEntry {
    MalformedEntry {
        path: path.to_path_buf(),
        reason: reason.to_owned(),
    }
}