limnifs-core 0.3.21

//! 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>;

    /// 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>;

    /// 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)
}

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(dir_path)?;

    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 {
        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(&child_path, 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;

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()),
    }
}

/// 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>,
}

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 }
    }
}

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_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)?;
        }
        let plaintext = file_plaintext(inode, self.slab_store)?;
        std::fs::write(&path, &plaintext).map_err(io_to_core)
    }

    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(not(unix))]
        {
            // No symlink support on this platform; write the target
            // as a regular file so the tree shape is preserved.
            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.
pub struct ParallelExtractSink<'a> {
    root: &'a Path,
    pub tasks: Vec<(PathBuf, Inode)>,
    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(),
            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_regular_file(&mut self, abs_path: &Path, inode: &Inode) -> Result<(), CoreError> {
        // 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(not(unix))]
        {
            let _ = (path, target);
        }
        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"
        );
    }
}