limnifs-core 0.3.52

//! Live tree walker — visit every inode reachable from a root.
//!
//! Three callers used to each reimplement this walk:
//!
//! - `limni::extract_dir_collect` + `extract_file` (CLI extraction)
//! - `limnifs_write::RwImage::write_live_dir` (RW commit/turnover)
//! - `limnifs_write::compaction::find_referenced_drops` (drop GC)
//!
//! They had drifted: the CLI's slice-map extraction ignored
//! `drop_byte_start` / `drop_byte_len`, which works only because the
//! writer today never emits a slice that doesn't span the whole drop.
//! The compaction version skipped extraction entirely.
//!
//! This module provides one walker parameterised by a [`LiveTreeSink`].
//! Adding a new consumer = one more `LiveTreeSink` impl. Existing
//! callers and the walker never change (OCP).
//!
//! ## Bug fix
//!
//! [`FilesystemSink`] respects `SliceRef::drop_byte_start` and
//! `SliceRef::drop_byte_len`. A future writer that emits sub-drop
//! slices now produces correctly-extracted plaintext.

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

use crate::inode::ContentHandle;
use crate::slab_source::SlabSource;
use crate::{CoreError, Inode, MetadataBlob};

/// Visitor callback for [`walk_live_tree`]. Each method receives the
/// absolute path (relative to the walk root) and the relevant data.
pub trait LiveTreeSink {
    /// A directory inode was encountered. Create it on disk, or
    /// record it, depending on the sink's purpose.
    ///
    /// Called once per directory in pre-order (parent before
    /// children). The root directory is included.
    fn on_directory(&mut self, abs_path: &Path) -> Result<(), CoreError>;

    /// Same event as [`on_directory`], carrying the inode so sinks
    /// that need mode/mtime fidelity (e.g. archive writers) can read
    /// it. Sinks that don't care keep the default delegation.
    fn on_directory_inode(&mut self, abs_path: &Path, inode: &Inode) -> Result<(), CoreError> {
        let _ = inode;
        self.on_directory(abs_path)
    }

    /// A regular-file inode (inline or slice-backed). The sink
    /// decides whether to extract now or defer.
    fn on_regular_file(&mut self, abs_path: &Path, inode: &Inode) -> Result<(), CoreError>;

    /// A symbolic link.
    fn on_symlink(&mut self, abs_path: &Path, target: &str) -> Result<(), CoreError>;

    /// Same event as [`on_symlink`], carrying the inode so sinks
    /// that need mode/mtime fidelity (e.g. archive writers) can
    /// read it. Sinks that don't care keep the default delegation.
    fn on_symlink_inode(
        &mut self,
        abs_path: &Path,
        inode: &Inode,
        target: &str,
    ) -> Result<(), CoreError> {
        let _ = inode;
        self.on_symlink(abs_path, target)
    }

    /// Any other inode type (block/char device, FIFO, socket).
    /// Default impl is a no-op so sinks that don't care can ignore.
    fn on_other(&mut self, _abs_path: &Path, _inode: &Inode) -> Result<(), CoreError> {
        Ok(())
    }
}

/// Walk every inode reachable from `root_inode_number`, invoking
/// `sink` for each. Cycles are detected and broken (a directory
/// reachable via two parents is only visited once).
///
/// # Errors
///
/// Returns [`CoreError::Corrupt`] if the root inode is missing, a
/// referenced child inode is missing, or a directory's hash doesn't
/// resolve to a node in the blob.
pub fn walk_live_tree(
    blob: &MetadataBlob,
    root_inode_number: u64,
    sink: &mut dyn LiveTreeSink,
) -> Result<(), CoreError> {
    let root = blob
        .inode_by_number(root_inode_number)
        .ok_or_else(|| CoreError::Corrupt {
            reason: format!("walk_live_tree: root inode {root_inode_number} missing"),
        })?;
    let mut visited: Vec<u64> = Vec::new();
    walk_dir(blob, root, Path::new(""), sink, &mut visited)
}

/// A tree entry name is one filesystem component. Once a name
/// leaves an image file it is attacker-controlled: joined
/// unvalidated, `..` or an absolute name escapes the extract root,
/// and on Windows `\` replaces the whole path (UNC/drive escape).
/// Unix legacy images may legally contain `\` as a literal
/// filename byte, where joins are safe — it is refused only where
/// it is a separator.
///
/// # Errors
///
/// Returns the reason the name must never reach a `Path::join`.
pub fn validate_tree_name(name: &str) -> Result<(), &'static str> {
    if name.is_empty() {
        return Err("is empty");
    }
    if name == "." {
        return Err("is '.'");
    }
    if name == ".." {
        return Err("is '..'");
    }
    if name.starts_with('/') {
        return Err("is absolute");
    }
    if name.contains('/') {
        return Err("contains a '/' separator");
    }
    if name.contains('\0') {
        return Err("contains a NUL byte");
    }
    if cfg!(windows) && name.contains('\\') {
        return Err("contains a '\\' separator");
    }
    Ok(())
}

fn walk_dir(
    blob: &MetadataBlob,
    dir_inode: &Inode,
    dir_path: &Path,
    sink: &mut dyn LiveTreeSink,
    visited: &mut Vec<u64>,
) -> Result<(), CoreError> {
    let hash = match &dir_inode.content_handle {
        ContentHandle::Directory(h) => *h,
        _ => return Ok(()),
    };
    if visited.contains(&dir_inode.number) {
        return Ok(());
    }
    visited.push(dir_inode.number);
    sink.on_directory_inode(dir_path, dir_inode)?;

    let node = blob
        .dir_node_by_hash(&hash)
        .ok_or_else(|| CoreError::Corrupt {
            reason: format!(
                "walk_live_tree: directory node for hash {} missing",
                hex_prefix(&hash)
            ),
        })?;
    for entry in &node.entries {
        if let Err(reason) = validate_tree_name(&entry.name) {
            return Err(CoreError::Corrupt {
                reason: format!(
                    "walk_live_tree: directory node {} entry name {:?} {reason}",
                    hex_prefix(&hash),
                    entry.name
                ),
            });
        }
        let child_path = if dir_path.as_os_str().is_empty() {
            PathBuf::from(&entry.name)
        } else {
            dir_path.join(&entry.name)
        };
        let child = blob
            .inode_by_number(entry.inode_number)
            .ok_or_else(|| CoreError::Corrupt {
                reason: format!("walk_live_tree: inode {} missing", entry.inode_number),
            })?;
        match &child.content_handle {
            ContentHandle::Directory(_) => {
                walk_dir(blob, child, &child_path, sink, visited)?;
            }
            ContentHandle::InlineData(_) | ContentHandle::SliceMap(_) => {
                sink.on_regular_file(&child_path, child)?;
            }
            ContentHandle::Symlink(target) => {
                sink.on_symlink_inode(&child_path, child, target)?;
            }
            _ => {
                sink.on_other(&child_path, child)?;
            }
        }
    }
    Ok(())
}

fn hex_prefix(bytes: &[u8]) -> String {
    bytes
        .iter()
        .take(4)
        .map(|b| format!("{b:02x}"))
        .collect::<Vec<_>>()
        .join("")
}

/// Reconstruct a file's plaintext from its inode + (optional) slab store.
///
/// For inline files: returns the inline data directly. For slice-backed
/// files: fetches each referenced drop and concatenates the
/// sub-drop byte ranges (`SliceRef::drop_byte_start` ..
/// `drop_byte_start + drop_byte_len`). The previous callers ignored
/// the sub-drop range; this function is the canonical place that
/// honours it.
///
/// # Errors
///
/// Returns [`CoreError::Corrupt`] if a slice references a drop the
/// slab store doesn't have, or decompression fails.
/// Decode one slice's byte range from its drop. Shared by the
/// sequential and batched-parallel assembly paths so both honor
/// `SliceRef::drop_byte_start`/`drop_byte_len` identically.
fn decode_slice<S: SlabSource + ?Sized>(
    store: &S,
    slice: &crate::inode::SliceRef,
) -> Result<Vec<u8>, CoreError> {
    let plaintext = store
        .plaintext_for(slice.drop_id.as_bytes())
        .ok_or_else(|| CoreError::Corrupt {
            reason: format!(
                "file_plaintext: drop {:02x?} not in any slab",
                &slice.drop_id.as_bytes()[..4]
            ),
        })?
        .map_err(|e| CoreError::Corrupt {
            reason: format!("file_plaintext: decompress: {e}"),
        })?;
    let start = usize::try_from(slice.drop_byte_start).unwrap_or(0);
    let len = usize::try_from(slice.drop_byte_len).unwrap_or(plaintext.len());
    let end = start.saturating_add(len).min(plaintext.len());
    if start > plaintext.len() || end > plaintext.len() {
        return Err(CoreError::Corrupt {
            reason: format!(
                "file_plaintext: slice range {start}..{end} outside drop of len {}",
                plaintext.len()
            ),
        });
    }
    Ok(plaintext[start..end].to_vec())
}

/// Slices per parallel decode batch. Bounded so peak extra memory
/// stays ~one batch (64 × 256 KiB ≈ 16 MiB at default chunking)
/// instead of materializing the whole file's decoded parts at once.
const DECODE_BATCH_SLICES: usize = 64;
/// Below this slice count the sequential loop beats task overhead.
const DECODE_PARALLEL_THRESHOLD: usize = 8;

/// Streaming chunk iterator over a file's plaintext: one item per
/// slice (drop-windowed), or the single inline blob. Memory is
/// bounded by one chunk (~max chunk size); callers that need the
/// whole file use [`file_plaintext`], which shares this decode
/// path. Iteration yields `Result`s so a mid-file decode failure
/// surfaces with position context from the caller.
pub struct FileChunks<'a, S: SlabSource + ?Sized> {
    inode: &'a Inode,
    store: Option<&'a S>,
    /// Index into the slice map; `usize::MAX` = not started,
    /// exhausted when idx >= len.
    idx: usize,
    /// Buffered inline chunk, yielded once.
    inline: Option<Vec<u8>>,
    done: bool,
}

impl<'a, S: SlabSource + ?Sized> FileChunks<'a, S> {
    #[must_use]
    pub fn new(inode: &'a Inode, store: Option<&'a S>) -> Self {
        match &inode.content_handle {
            ContentHandle::InlineData(data) => Self {
                inode,
                store,
                idx: 0,
                inline: Some(data.clone()),
                done: false,
            },
            ContentHandle::SliceMap(_) => Self {
                inode,
                store,
                idx: 0,
                inline: None,
                done: false,
            },
            _ => Self {
                inode,
                store,
                idx: 0,
                inline: None,
                done: true,
            },
        }
    }
}

impl<S: SlabSource + ?Sized> Iterator for FileChunks<'_, S> {
    type Item = Result<Vec<u8>, CoreError>;

    fn next(&mut self) -> Option<Self::Item> {
        if self.done {
            return None;
        }
        if let Some(data) = self.inline.take() {
            self.done = true;
            return Some(Ok(data));
        }
        let ContentHandle::SliceMap(slices) = &self.inode.content_handle else {
            self.done = true;
            return None;
        };
        if self.idx >= slices.len() {
            self.done = true;
            return None;
        }
        let store = match self.store {
            Some(s) => s,
            None => {
                self.done = true;
                return Some(Err(CoreError::Corrupt {
                    reason: "file_chunks: slice-backed file but no slab store provided".into(),
                }));
            }
        };
        let slice = &slices[self.idx];
        self.idx += 1;
        Some(decode_slice(store, slice))
    }
}

/// Streaming form of [`file_plaintext`]: bounded memory, one chunk
/// per item. `file_plaintext` remains for callers that want a Vec.
pub fn file_chunks<'a, S: SlabSource + ?Sized>(
    inode: &'a Inode,
    store: Option<&'a S>,
) -> FileChunks<'a, S> {
    FileChunks::new(inode, store)
}

pub fn file_plaintext<S: SlabSource + ?Sized>(
    inode: &Inode,
    slab_store: Option<&S>,
) -> Result<Vec<u8>, CoreError> {
    match &inode.content_handle {
        ContentHandle::InlineData(data) => Ok(data.clone()),
        ContentHandle::SliceMap(slices) => {
            let store = slab_store.ok_or_else(|| CoreError::Corrupt {
                reason: "file_plaintext: slice-backed file but no slab store provided".into(),
            })?;
            let mut out = Vec::new();
            if slices.len() >= DECODE_PARALLEL_THRESHOLD {
                // Batched parallel decode: a single large file's
                // slices would otherwise decode on one core while
                // the caller's cross-file parallelism idles — the
                // read-side mirror of the write path's Phase-1
                // finding. The cached slab store drops its lock
                // during decompression, so concurrent decodes scale.
                // Batches are appended in order; output is
                // byte-identical to the sequential loop.
                use rayon::prelude::*;
                for batch in slices.chunks(DECODE_BATCH_SLICES) {
                    let parts: Result<Vec<Vec<u8>>, CoreError> =
                        batch.par_iter().map(|s| decode_slice(store, s)).collect();
                    for part in parts? {
                        out.extend_from_slice(&part);
                    }
                }
                return Ok(out);
            }
            for slice in slices {
                out.extend_from_slice(&decode_slice(store, slice)?);
            }
            Ok(out)
        }
        _ => Ok(Vec::new()),
    }
}

/// Stamp a written path with the inode's permission bits and mtime
/// (unix). Mtime first — chmod-before-stamp would block the write on
/// a soon-to-be read-only file; the read-open fallback covers
/// directories, which cannot be opened for writing. Ownership stays
/// untouched (needs privileges). Single source for extract, RW
/// staging, and any future materializer.
#[cfg(unix)]
pub fn apply_unix_identity(path: &Path, mode: u32, mtime_ns: u64) {
    use std::os::unix::fs::PermissionsExt as _;
    let secs = mtime_ns / 1_000_000_000;
    let nanos = u32::try_from(mtime_ns % 1_000_000_000).unwrap_or(0);
    if let Some(t) =
        std::time::SystemTime::UNIX_EPOCH.checked_add(std::time::Duration::new(secs, nanos))
    {
        let times = std::fs::FileTimes::new().set_modified(t);
        let file = std::fs::File::options()
            .read(true)
            .write(true)
            .open(path)
            .or_else(|_| std::fs::File::open(path));
        if let Ok(file) = file {
            let _ = file.set_times(times);
        }
    }
    let _ = std::fs::set_permissions(path, std::fs::Permissions::from_mode(mode & 0o7777));
}

#[cfg(not(unix))]
pub fn apply_unix_identity(path: &Path, _mode: u32, mtime_ns: u64) {
    // Mtime is portable (`File::set_modified`, std 1.75); mode and
    // ownership are unix concepts with no non-unix equivalent.
    // Directories need FILE_FLAG_BACKUP_SEMANTICS to open for
    // write on Windows (no std surface) and keep their now().
    let secs = mtime_ns / 1_000_000_000;
    let nanos = u32::try_from(mtime_ns % 1_000_000_000).unwrap_or(0);
    let Some(t) =
        std::time::SystemTime::UNIX_EPOCH.checked_add(std::time::Duration::new(secs, nanos))
    else {
        return;
    };
    if let Ok(f) = std::fs::File::options().write(true).read(true).open(path) {
        let _ = f.set_modified(t);
    }
}

/// Sink that writes the live tree to a filesystem directory.
/// Directories are created pre-order; regular files are written
/// inline (no parallelism — wrap with rayon outside if you need
/// parallel extraction).
pub struct FilesystemSink<'a> {
    root: &'a Path,
    slab_store: Option<&'a dyn SlabSource>,
    /// Directory identities for the `finish()` pass — a directory
    /// stamped read-only before its children are written would block
    /// them; its xattrs ride along for callers that can apply them
    /// (limnifs-core carries no xattr dependency by design).
    dirs: Vec<DirIdentity>,
    /// inode number -> first-written path (hardlink targets).
    seen_inodes: std::collections::HashMap<u64, PathBuf>,
}

impl<'a> FilesystemSink<'a> {
    /// Construct a sink that writes the tree under `root`. If
    /// `slab_store` is `None`, slice-backed files fail extraction.
    #[must_use]
    pub fn new(root: &'a Path, slab_store: Option<&'a dyn SlabSource>) -> Self {
        Self {
            root,
            slab_store,
            dirs: Vec::new(),
            seen_inodes: std::collections::HashMap::new(),
        }
    }

    /// Apply directory identities (mode, mtime) after every file
    /// and hardlink is written, and expose the recorded identities —
    /// including xattrs — for callers able to apply the latter.
    /// Must be called once the walk completes.
    pub fn finish(&mut self) -> &[DirIdentity] {
        for dir in &self.dirs {
            apply_unix_identity(&dir.path, dir.mode, dir.mtime_ns);
        }
        &self.dirs
    }
}

impl<'a> LiveTreeSink for FilesystemSink<'a> {
    fn on_directory(&mut self, abs_path: &Path) -> Result<(), CoreError> {
        let path = if abs_path.as_os_str().is_empty() {
            self.root.to_path_buf()
        } else {
            self.root.join(abs_path)
        };
        std::fs::create_dir_all(&path).map_err(io_to_core)
    }

    fn on_directory_inode(&mut self, abs_path: &Path, inode: &Inode) -> Result<(), CoreError> {
        self.on_directory(abs_path)?;
        let path = if abs_path.as_os_str().is_empty() {
            self.root.to_path_buf()
        } else {
            self.root.join(abs_path)
        };
        self.dirs.push(DirIdentity {
            path,
            mode: inode.mode,
            mtime_ns: inode.mtime_ns,
            xattrs: inode.xattrs.clone(),
        });
        Ok(())
    }

    fn on_regular_file(&mut self, abs_path: &Path, inode: &Inode) -> Result<(), CoreError> {
        let path = self.root.join(abs_path);
        if let Some(parent) = path.parent() {
            std::fs::create_dir_all(parent).map_err(io_to_core)?;
        }
        // Hardlink: serial pre-order means the first name is already
        // on disk — link instead of writing a second copy.
        if let Some(first) = self.seen_inodes.get(&inode.number) {
            if std::fs::hard_link(first, &path).is_ok() {
                return Ok(());
            }
            // Fall through to a copy so content always lands.
        } else {
            self.seen_inodes.insert(inode.number, path.clone());
        }
        // Chunk-streamed (bounded memory): staging a multi-GiB
        // tree must not become multi-GiB RSS.
        use std::io::Write as _;
        let mut out = std::io::BufWriter::new(std::fs::File::create(&path).map_err(io_to_core)?);
        for chunk in file_chunks(inode, self.slab_store) {
            let chunk = chunk?;
            out.write_all(&chunk).map_err(io_to_core)?;
        }
        out.flush().map_err(io_to_core)?;
        drop(out);
        apply_unix_identity(&path, inode.mode, inode.mtime_ns);
        Ok(())
    }

    fn on_symlink(&mut self, abs_path: &Path, target: &str) -> Result<(), CoreError> {
        let path = self.root.join(abs_path);
        #[cfg(unix)]
        {
            std::os::unix::fs::symlink(target, &path).map_err(io_to_core)
        }
        #[cfg(windows)]
        {
            create_symlink_best_effort(&path, target)
        }
        #[cfg(not(any(unix, windows)))]
        {
            // No symlink surface on this platform; the link is
            // skipped rather than materialized as a bogus file.
            let _ = path;
            let _ = target;
            Ok(())
        }
    }
}

/// Sink that collects every `DropId` referenced by the live tree.
/// Used by compaction to determine which drops to keep.
pub struct DropIdCollectorSink {
    pub drop_ids: std::collections::HashSet<[u8; 32]>,
}

impl Default for DropIdCollectorSink {
    fn default() -> Self {
        Self {
            drop_ids: std::collections::HashSet::new(),
        }
    }
}

impl LiveTreeSink for DropIdCollectorSink {
    fn on_directory(&mut self, _abs_path: &Path) -> Result<(), CoreError> {
        Ok(())
    }

    fn on_regular_file(&mut self, _abs_path: &Path, inode: &Inode) -> Result<(), CoreError> {
        if let ContentHandle::SliceMap(slices) = &inode.content_handle {
            for slice in slices {
                self.drop_ids.insert(*slice.drop_id.as_bytes());
            }
        }
        Ok(())
    }

    fn on_symlink(&mut self, _abs_path: &Path, _target: &str) -> Result<(), CoreError> {
        Ok(())
    }
}

/// Sink for parallel extraction: creates directories inline (the
/// sequential walk guarantees ordering — parents before children,
/// no races) and collects regular-file inodes as owned clones for
/// a later rayon fan-out phase.
///
/// Used by `limni::extract`'s two-phase pipeline:
/// 1. `walk_live_tree` + `ParallelExtractSink` (this struct) —
///    sequential, creates dirs, collects file tasks.
/// 2. rayon across `self.tasks` — parallel file writes.
/// Directory identity captured during the walk for the post-file
/// application pass (mode, mtime, xattrs).
pub struct DirIdentity {
    pub path: PathBuf,
    pub mode: u32,
    pub mtime_ns: u64,
    pub xattrs: Vec<crate::inode::XAttr>,
}

pub struct ParallelExtractSink<'a> {
    root: &'a Path,
    pub tasks: Vec<(PathBuf, Inode)>,
    /// Directories in creation order with their inode identity.
    /// Extract applies these AFTER file writes so a read-only
    /// directory cannot block its own children.
    pub dirs: Vec<DirIdentity>,
    /// Hardlinks: (new_path, first_path) pairs for inodes already
    /// written under an earlier name. Created serially after the
    /// parallel file phase (the target must exist).
    pub links: Vec<(PathBuf, PathBuf)>,
    /// inode number -> path of its first occurrence.
    seen_inodes: std::collections::HashMap<u64, PathBuf>,
    pub dir_count: usize,
}

impl<'a> ParallelExtractSink<'a> {
    /// Construct a sink that creates directories under `root` and
    /// collects file tasks for later parallel processing.
    #[must_use]
    pub fn new(root: &'a Path) -> Self {
        Self {
            root,
            tasks: Vec::new(),
            dirs: Vec::new(),
            links: Vec::new(),
            seen_inodes: std::collections::HashMap::new(),
            dir_count: 0,
        }
    }
}

impl<'a> LiveTreeSink for ParallelExtractSink<'a> {
    fn on_directory(&mut self, abs_path: &Path) -> Result<(), CoreError> {
        let path = if abs_path.as_os_str().is_empty() {
            self.root.to_path_buf()
        } else {
            self.root.join(abs_path)
        };
        std::fs::create_dir_all(&path).map_err(io_to_core)?;
        self.dir_count += 1;
        Ok(())
    }

    fn on_directory_inode(&mut self, abs_path: &Path, inode: &Inode) -> Result<(), CoreError> {
        self.on_directory(abs_path)?;
        let path = if abs_path.as_os_str().is_empty() {
            self.root.to_path_buf()
        } else {
            self.root.join(abs_path)
        };
        self.dirs.push(DirIdentity {
            path,
            mode: inode.mode,
            mtime_ns: inode.mtime_ns,
            xattrs: inode.xattrs.clone(),
        });
        Ok(())
    }

    fn on_regular_file(&mut self, abs_path: &Path, inode: &Inode) -> Result<(), CoreError> {
        // Hardlink: a later name for an inode already extracted —
        // link to the first occurrence instead of writing a copy.
        if let Some(first) = self.seen_inodes.get(&inode.number) {
            self.links.push((self.root.join(abs_path), first.clone()));
            return Ok(());
        }
        self.seen_inodes
            .insert(inode.number, self.root.join(abs_path));
        // Clone the inode so the rayon phase can outlive the borrow
        // of `blob` held open by `walk_live_tree`. Path is absolute
        // under `root`; the caller joins when popping tasks.
        self.tasks.push((self.root.join(abs_path), inode.clone()));
        Ok(())
    }

    fn on_symlink(&mut self, abs_path: &Path, target: &str) -> Result<(), CoreError> {
        let path = self.root.join(abs_path);
        #[cfg(unix)]
        {
            std::os::unix::fs::symlink(target, &path).map_err(io_to_core)?;
        }
        #[cfg(windows)]
        {
            create_symlink_best_effort(&path, target)?;
        }
        #[cfg(not(any(unix, windows)))]
        {
            let _ = (path, target);
        }
        Ok(())
    }
}

/// Best-effort symlink off unix: Windows can create them with
/// Developer Mode or SeCreateSymbolicLinkPrivilege. The flavor
/// must match the target's kind for resolution; without knowing
/// it, try file then dir, and skip only when the OS refuses —
/// the same best-effort law as xattr application (v0.3.50).
#[cfg(windows)]
fn create_symlink_best_effort(path: &Path, target: &str) -> Result<(), CoreError> {
    use std::os::windows::fs as win_fs;
    if win_fs::symlink_file(target, path)
        .or_else(|_| win_fs::symlink_dir(target, path))
        .is_ok()
    {
        return Ok(());
    }
    Ok(())
}

fn io_to_core(e: std::io::Error) -> CoreError {
    CoreError::Corrupt {
        reason: format!("live_tree: I/O: {e}"),
    }
}

#[cfg(test)]
mod tests {
    use super::*;
    use crate::merkle::hash_section;

    fn build_blob_with_one_file() -> (MetadataBlob, u64, [u8; 32], Vec<u8>) {
        // Build a metadata blob containing one inline-data file under
        // the root directory. Returns (blob, root_inode, drop_id_for_assertion, plaintext).
        let plaintext = b"hello live tree".to_vec();
        let file_inode = make_regular_inline_inode(2, 0o100_644, &plaintext);
        let entries = vec![("hello.txt".to_string(), 2u64, 0x01u8)];
        let dir_node_bytes = make_dir_node_bytes(&entries);
        let dir_hash = hash_section(&dir_node_bytes);
        let root_inode = make_directory_inode(1, dir_hash);

        let inode_count: u32 = 2;
        let dir_node_count: u32 = 1;
        let mut bytes = Vec::new();
        bytes.extend_from_slice(&inode_count.to_le_bytes());
        bytes.extend_from_slice(&root_inode);
        bytes.extend_from_slice(&file_inode);
        bytes.extend_from_slice(&dir_node_count.to_le_bytes());
        bytes.extend_from_slice(&dir_node_bytes);

        let mut cursor = crate::cursor::ManifestCursor::new(&bytes);
        let blob = crate::metadata::parse_metadata_blob(&mut cursor).expect("parse");
        (blob, 1, [0u8; 32], plaintext)
    }

    fn make_regular_inline_inode(number: u64, mode: u32, data: &[u8]) -> Vec<u8> {
        let mut bytes = Vec::new();
        bytes.extend_from_slice(&number.to_le_bytes());
        bytes.extend_from_slice(&mode.to_le_bytes());
        bytes.extend_from_slice(&0u32.to_le_bytes());
        bytes.extend_from_slice(&0u32.to_le_bytes());
        bytes.extend_from_slice(&0u64.to_le_bytes());
        bytes.extend_from_slice(&0u64.to_le_bytes());
        bytes.extend_from_slice(&1u32.to_le_bytes());
        bytes.push(crate::inode::INODE_FLAG_INLINE_DATA);
        let len = u32::try_from(data.len()).unwrap();
        bytes.extend_from_slice(&len.to_le_bytes());
        bytes.extend_from_slice(data);
        bytes
    }

    fn make_directory_inode(number: u64, hash: [u8; 32]) -> Vec<u8> {
        let mode = crate::inode::S_IFDIR | 0o755;
        let mut bytes = Vec::new();
        bytes.extend_from_slice(&number.to_le_bytes());
        bytes.extend_from_slice(&mode.to_le_bytes());
        bytes.extend_from_slice(&0u32.to_le_bytes());
        bytes.extend_from_slice(&0u32.to_le_bytes());
        bytes.extend_from_slice(&0u64.to_le_bytes());
        bytes.extend_from_slice(&0u64.to_le_bytes());
        bytes.extend_from_slice(&2u32.to_le_bytes());
        bytes.push(0);
        bytes.extend_from_slice(&hash);
        bytes
    }

    fn make_dir_node_bytes(entries: &[(String, u64, u8)]) -> Vec<u8> {
        let mut bytes = Vec::new();
        bytes.push(1u8); // version
        let count = u32::try_from(entries.len()).unwrap();
        bytes.extend_from_slice(&count.to_le_bytes());
        for (name, ino, etype) in entries {
            let name_bytes = name.as_bytes();
            let name_len = u32::try_from(name_bytes.len()).unwrap();
            bytes.extend_from_slice(&name_len.to_le_bytes());
            bytes.extend_from_slice(name_bytes);
            bytes.extend_from_slice(&ino.to_le_bytes());
            bytes.push(*etype);
        }
        bytes
    }

    #[test]
    fn walk_visits_root_and_children() {
        let (blob, root, _drop_id, plaintext) = build_blob_with_one_file();
        let temp = std::env::temp_dir().join(format!(
            "limnifs-live-tree-test-{}-{}",
            std::process::id(),
            std::time::SystemTime::now()
                .duration_since(std::time::UNIX_EPOCH)
                .map(|d| d.as_nanos() as u64)
                .unwrap_or(0),
        ));

        let mut sink = FilesystemSink::new(&temp, None);
        walk_live_tree(&blob, root, &mut sink).expect("walk");

        // The root dir + hello.txt should both exist.
        assert!(temp.is_dir());
        let hello_path = temp.join("hello.txt");
        assert!(hello_path.is_file());
        let recovered = std::fs::read(&hello_path).expect("read");
        assert_eq!(recovered, plaintext);

        std::fs::remove_dir_all(&temp).ok();
    }

    #[test]
    fn drop_id_collector_gathers_slice_refs() {
        // Inline-only tree → collector finds nothing. (Slice-backed
        // collection is exercised via the CachedSlabStore test in
        // slab_cache.rs which builds a real slab.)
        let (blob, root, _drop_id, _plaintext) = build_blob_with_one_file();
        let mut sink = DropIdCollectorSink::default();
        walk_live_tree(&blob, root, &mut sink).expect("walk");
        assert!(
            sink.drop_ids.is_empty(),
            "inline-only tree references no drops"
        );
    }

    fn build_blob_with_named_entry(name: &str) -> MetadataBlob {
        let plaintext = b"hostile".to_vec();
        let file_inode = make_regular_inline_inode(2, 0o100_644, &plaintext);
        let entries = vec![(name.to_string(), 2u64, 0x01u8)];
        let dir_node_bytes = make_dir_node_bytes(&entries);
        let dir_hash = hash_section(&dir_node_bytes);
        let root_inode = make_directory_inode(1, dir_hash);

        let mut bytes = Vec::new();
        bytes.extend_from_slice(&2u32.to_le_bytes());
        bytes.extend_from_slice(&root_inode);
        bytes.extend_from_slice(&file_inode);
        bytes.extend_from_slice(&1u32.to_le_bytes());
        bytes.extend_from_slice(&dir_node_bytes);

        let mut cursor = crate::cursor::ManifestCursor::new(&bytes);
        crate::metadata::parse_metadata_blob(&mut cursor).expect("parse")
    }

    #[test]
    fn walk_refuses_unsafe_entry_names() {
        // The dir-node parser already refuses empty/'/'/NUL names,
        // so the walk-side guard's uncovered vectors are the
        // path-semantic ones: '.' and '..' escape the walk root
        // through a legal parse.
        for name in ["..", "."] {
            let blob = build_blob_with_named_entry(name);
            let mut sink = DropIdCollectorSink::default();
            let err = walk_live_tree(&blob, 1, &mut sink)
                .expect_err("unsafe entry name must refuse the walk");
            assert!(
                matches!(err, CoreError::Corrupt { ref reason } if reason.contains("entry name")),
                "name {name:?}: wrong error {err:?}"
            );
        }
    }

    /// v0.3.51: extraction carries file mtimes off unix and keeps
    /// symlink names (creating real links when the runner has the
    /// privilege). Runs on the Windows CI legs only.
    #[cfg(windows)]
    #[test]
    fn windows_extract_stamps_mtime_and_keeps_link_names() {
        let dir = std::env::temp_dir().join(format!(
            "limnifs-win-fidelity-{}-{}",
            std::process::id(),
            std::time::SystemTime::now()
                .duration_since(std::time::UNIX_EPOCH)
                .map(|d| d.subsec_nanos() as u64)
                .unwrap_or(0),
        ));
        let src = dir.join("src");
        let out = dir.join("out");
        std::fs::create_dir_all(&src).expect("mkdir src");
        std::fs::create_dir_all(&out).expect("mkdir out");

        const MTIME_NS: u64 = 1_700_000_123_000_000_000;
        let config: &'static limnifs_write::WriteConfig =
            Box::leak(Box::new(limnifs_write::WriteConfig::default_v0_1()));
        let mut w = limnifs_write::stream::StreamWriter::new(config).expect("writer");
        w.add_file(
            "f.txt",
            limnifs_write::stream::EntryMeta::new(MTIME_NS, 0o644),
            &[],
            &mut b"body\n".as_slice(),
        )
        .expect("file");
        w.add_symlink(
            "lnk",
            "f.txt",
            limnifs_write::stream::EntryMeta::new(1, 0o777),
        )
        .expect("symlink");
        let artifact = w.finish().expect("finish");

        let mut cursor = crate::cursor::ManifestCursor::new(&artifact.bytes);
        crate::parse_manifest_header(&mut cursor).expect("header");
        crate::parse_feature_flags_section(&mut cursor).expect("flags");
        let meta_ref = crate::parse_metadata_reference(&mut cursor).expect("meta");
        let inline = meta_ref.inline_metadata.as_ref().expect("inline");
        let mut blob_cursor = crate::cursor::ManifestCursor::new(inline);
        let blob = crate::metadata::parse_metadata_blob(&mut blob_cursor).expect("blob");
        let root_ino = blob.root_inode_number().expect("root");

        let mut sink = FilesystemSink::new(&out, None);
        walk_live_tree(&blob, root_ino, &mut sink).expect("walk");

        let got = std::fs::metadata(out.join("f.txt"))
            .expect("file extracted")
            .modified()
            .expect("mtime");
        let got_ns = got
            .duration_since(std::time::UNIX_EPOCH)
            .expect("post-epoch")
            .as_nanos() as i128;
        let want_ns = i128::from(MTIME_NS);
        assert!(
            (got_ns - want_ns).abs() < 2_000_000_000,
            "mtime drifted {got_ns} vs {want_ns}"
        );

        let link = std::fs::symlink_metadata(out.join("lnk")).expect("link name preserved");
        if link.file_type().is_symlink() {
            assert_eq!(
                std::fs::read_link(out.join("lnk"))
                    .expect("read link")
                    .to_str()
                    .unwrap_or_default(),
                "f.txt"
            );
        }

        let _ = std::fs::remove_dir_all(&dir);
    }

    #[test]
    fn validate_tree_name_accepts_plain_components() {
        assert!(validate_tree_name("hello.txt").is_ok());
        assert!(validate_tree_name("a b").is_ok());
        assert!(validate_tree_name("..hidden").is_ok());
        assert!(validate_tree_name("...").is_ok());
        assert!(validate_tree_name(r"a\b").is_ok() || cfg!(windows));
    }

    #[test]
    fn validate_tree_name_refuses_encoding_and_traversal() {
        for name in ["", ".", "..", "/etc/passwd", "a/b", "nul\0"] {
            assert!(validate_tree_name(name).is_err(), "name {name:?}");
        }
        if cfg!(windows) {
            assert!(validate_tree_name(r"\\evil\share").is_err());
        }
    }
}