mkit_core/ops/

restore.rs

//! Restore.
//!
//! Materialises a stored [`Tree`](crate::object::Tree) into a target
//! directory: writes blobs as files, recurses into subtrees as
//! directories, creates symlinks for symlink entries, and (when
//! `clean=true`) deletes anything in the target dir that is not in the
//! tree (preserving `.mkit/` and `.git/`).
//!
//! ### Symlink invariant
//!
//! Like `worktree::build_tree`, this module **never follows external
//! symlinks**. Symlinks created here always have a relative,
//! `..`-free target — checked by [`worktree::validate_symlink_target`]
//! before the symlink is materialised.
//!
//! ### Sparse checkout
//!
//! When `RestoreOptions.sparse_patterns` is set, only files whose
//! computed full path (relative to the root) matches the pattern set
//! are restored. Pattern grammar:
//!
//! - Lines beginning with `#` are comments. Empty lines are skipped.
//! - Leading `!` negates the pattern.
//! - Trailing `/` makes the pattern dir-only.
//! - Patterns are evaluated in order; **last match wins**. No match =
//!   excluded.
//! - Bare names without a `/` also match against the basename of
//!   nested files.

use std::fs;
use std::io::{self, Write};
use std::path::{Path, PathBuf};
use std::process;
use std::sync::atomic::{AtomicU64, Ordering};

use crate::hash::Hash;
use crate::ignore::{self, IgnoreList};
use crate::layout::RepoLayout;
use crate::object::{self, EntryMode, Object, TreeEntry};
use crate::store::{MAX_TREE_DEPTH, ObjectStore};
use crate::worktree;

const MAX_SPARSE_BYTES: u64 = 1024 * 1024;

/// Chunk-count bound per `read_chunks` batch when materialising a
/// [`Object::ChunkedBlob`] — mirrors `worktree::STREAM_HASH_BATCH` (16
/// MiB at the current 256 KiB max chunk size) so the read side of a
/// large-file checkout bounds in-flight memory the same way the write
/// side already does. Public so a caller-supplied `read_chunks`'s own
/// fan-out threshold (e.g. `mkit-cli`'s `restore_fanout`) can be capped
/// against the same number: a threshold above this value could never be
/// reached by any batch this module ever hands `read_chunks`, silently
/// disabling that caller's parallel path on a high-core-count host.
pub const RESTORE_CHUNK_BATCH: usize = 64;

static TMP_COUNTER: AtomicU64 = AtomicU64::new(0);

/// Errors raised by this module.
#[derive(Debug, thiserror::Error)]
pub enum RestoreError {
    #[error("requested object is not a tree")]
    NotATree,
    #[error("requested object is not a blob or chunked-blob")]
    NotABlob,
    #[error("symlink target '{0}' is invalid (absolute or contains '..')")]
    InvalidSymlinkTarget(String),
    #[error("path '{0}' is occupied by something other than a directory")]
    NotADirectory(PathBuf),
    #[error("path component is not valid UTF-8")]
    InvalidUtf8,
    #[error("tree nesting exceeds {} levels", MAX_TREE_DEPTH)]
    TreeTooDeep,
    #[error(transparent)]
    Object(#[from] object::MkitError),
    #[error(transparent)]
    Store(#[from] crate::store::StoreError),
    #[error(transparent)]
    Io(#[from] io::Error),
    /// A `read_chunks` callback passed to
    /// [`restore_tree_to_worktree_with`] returned a different number of
    /// chunk buffers than the batch it was given — a contract violation
    /// by the caller (mirrors [`crate::worktree::WorktreeError::ChunkBatchLengthMismatch`]
    /// on the ingest side), never a normal runtime condition.
    #[error("read_chunks callback returned {actual} buffers for a {expected}-chunk batch")]
    ChunkBatchLengthMismatch { expected: usize, actual: usize },
}

/// Result alias.
pub type RestoreResult<T> = Result<T, RestoreError>;

/// One sparse-checkout pattern.
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct SparsePattern {
    pub pattern: String,
    pub negated: bool,
    pub dir_only: bool,
}

/// Options for [`restore_tree`].
#[derive(Debug, Clone)]
pub struct RestoreOptions {
    /// If `true`, delete anything in the target dir that is not in the
    /// tree (preserving `.mkit/` and `.git/`). Default `true`.
    pub clean: bool,
    /// If `Some`, only restore entries whose path matches the patterns.
    pub sparse_patterns: Option<Vec<SparsePattern>>,
}

impl Default for RestoreOptions {
    fn default() -> Self {
        Self {
            clean: true,
            sparse_patterns: None,
        }
    }
}

/// Parse the contents of a `.mkit/sparse-checkout` file into patterns.
#[must_use]
pub fn parse_sparse_patterns(content: &str) -> Vec<SparsePattern> {
    let mut out = Vec::new();
    for raw in content.split('\n') {
        let line = raw.trim_end_matches(['\r', ' ']);
        if line.is_empty() || line.starts_with('#') {
            continue;
        }
        let (negated, rest) = if let Some(stripped) = line.strip_prefix('!') {
            (true, stripped)
        } else {
            (false, line)
        };
        let (dir_only, pat) = if let Some(stripped) = rest.strip_suffix('/') {
            (true, stripped)
        } else {
            (false, rest)
        };
        if pat.is_empty() {
            continue;
        }
        out.push(SparsePattern {
            pattern: pat.to_string(),
            negated,
            dir_only,
        });
    }
    out
}

/// True iff `path` matches at least one (non-negated, last-match-wins)
/// pattern in `patterns`.
#[must_use]
pub fn matches_sparse(patterns: &[SparsePattern], path: &str, is_dir: bool) -> bool {
    let mut matched = false;
    for pat in patterns {
        if path_matches_pattern(&pat.pattern, path) {
            if pat.dir_only && !is_dir {
                let pat_stripped = pat.pattern.strip_suffix('/').unwrap_or(&pat.pattern);
                if pat_stripped == path {
                    continue;
                }
            }
            matched = !pat.negated;
        }
    }
    matched
}

/// True iff a directory at `dir_prefix` could contain matched
/// descendants. Used to short-circuit recursion under sparse mode.
#[must_use]
pub fn could_match_descendant(patterns: &[SparsePattern], dir_prefix: &str) -> bool {
    for pat in patterns {
        if pat.negated {
            continue;
        }
        if pat.pattern.starts_with(dir_prefix) {
            return true;
        }
        if dir_prefix.starts_with(&pat.pattern) {
            return true;
        }
        if !dir_prefix.is_empty()
            && !dir_prefix.ends_with('/')
            && pat.pattern.len() > dir_prefix.len()
            && pat.pattern.starts_with(dir_prefix)
            && pat.pattern.as_bytes()[dir_prefix.len()] == b'/'
        {
            return true;
        }
    }
    false
}

/// Load patterns from `<repo_root>/.mkit/sparse-checkout`. Returns
/// `Ok(None)` if the file does not exist or has zero patterns.
///
/// # Errors
/// - [`RestoreError::Io`] for filesystem failures other than "not found".
pub fn load_sparse_checkout(layout: &RepoLayout) -> RestoreResult<Option<Vec<SparsePattern>>> {
    let path = layout.sparse_checkout_file();
    let meta = match fs::metadata(&path) {
        Ok(m) => m,
        Err(e) if e.kind() == io::ErrorKind::NotFound => return Ok(None),
        Err(e) => return Err(RestoreError::Io(e)),
    };
    if meta.len() > MAX_SPARSE_BYTES {
        return Err(RestoreError::Io(io::Error::other(
            "sparse-checkout too large",
        )));
    }
    let raw = fs::read_to_string(&path)?;
    let patterns = parse_sparse_patterns(&raw);
    if patterns.is_empty() {
        Ok(None)
    } else {
        Ok(Some(patterns))
    }
}

/// Write a sparse-checkout file (one pattern per line). Atomic on the
/// destination.
///
/// # Errors
/// - [`RestoreError::Io`] for filesystem failures.
pub fn write_sparse_checkout(layout: &RepoLayout, lines: &[&str]) -> RestoreResult<()> {
    let mkit_dir = layout.worktree_state_dir().to_path_buf();
    fs::create_dir_all(&mkit_dir)?;
    let mut buf = String::new();
    for l in lines {
        buf.push_str(l);
        buf.push('\n');
    }
    let path = layout.sparse_checkout_file();
    crate::atomic::write_atomic(&path, buf.as_bytes(), true)?;
    Ok(())
}

fn path_matches_pattern(pattern: &str, path: &str) -> bool {
    let pat = pattern.strip_suffix('/').unwrap_or(pattern);
    if pat == path {
        return true;
    }
    if path.len() > pat.len() && path.starts_with(pat) && path.as_bytes()[pat.len()] == b'/' {
        return true;
    }
    if !pat.contains('/')
        && let Some(last) = path.rfind('/')
    {
        let basename = &path[last + 1..];
        if basename == pat {
            return true;
        }
    }
    false
}

/// Summary of a [`restore_tree_to_worktree`] call. The counts are no
/// longer printed by any caller (`mkit checkout` emits a git-shaped
/// switch confirmation instead); they are retained for programmatic
/// callers and asserted by this module's own unit tests.
#[derive(Debug, Default, Clone, PartialEq, Eq)]
pub struct RestoreReport {
    /// Number of regular / executable files materialised.
    pub files_written: u32,
    /// Number of symlinks materialised.
    pub symlinks_written: u32,
    /// Number of directories created (or that already existed as dirs).
    pub directories_created: u32,
}

/// Materialise `tree_hash` into `root` as a working tree.
///
/// Thin wrapper around [`restore_tree`] that additionally:
/// 1. Loads `<root>/.gitignore` + `<root>/.mkitignore`. Ignore rules do NOT
///    gate which tree entries are materialised — tracked content is always
///    written (git parity) — they only protect *untracked* worktree files
///    (editor swapfiles, local-only build artefacts, …) from the
///    `clean=true` sweep.
/// 2. Returns a [`RestoreReport`] with counts of what was materialised
///    (consumed by programmatic callers and this module's tests; not
///    printed by the CLI).
///
/// Symlink safety is inherited from [`restore_tree`] /
/// [`worktree::validate_symlink_target`]: targets are validated BEFORE
/// the symlink is created, and any target that is absolute or contains
/// `..` is rejected with [`RestoreError::InvalidSymlinkTarget`]. The
/// net effect is that no symlink produced by this function can point
/// outside `root`.
///
/// # Errors
///
/// Same variants as [`restore_tree`]. [`RestoreError::InvalidSymlinkTarget`]
/// on a rejected target is the load-bearing "outside-of-root" check.
pub fn restore_tree_to_worktree(
    store: &ObjectStore,
    tree: &Hash,
    root: &Path,
    opts: &RestoreOptions,
) -> RestoreResult<RestoreReport> {
    // Batch size 1, not `RESTORE_CHUNK_BATCH`: `sequential_read_chunks`
    // never fans anything out, so batching it would only raise peak
    // memory from one chunk to a full batch for zero benefit — the same
    // regression `restore_blob` (the `restore_tree`/stash path) avoids
    // by not batching at all. `restore_tree_to_worktree_with` below is
    // the path real (parallel-capable) `read_chunks` callers want.
    restore_tree_to_worktree_impl(store, tree, root, opts, 1, &sequential_read_chunks)
}

/// [`restore_tree_to_worktree`], parameterised over how a
/// [`Object::ChunkedBlob`]'s chunk objects are read back during
/// materialisation.
///
/// `read_chunks` receives up to [`RESTORE_CHUNK_BATCH`] chunk hashes at a
/// time, in file order, and MUST return exactly one chunk's raw bytes per
/// input hash, **in the same order** — this function only verifies the
/// returned count against the input batch's length (surfacing a
/// mismatch as [`RestoreError::ChunkBatchLengthMismatch`]); a same-length
/// but reordered result is undetectable here (each chunk's *content* is
/// separately BLAKE3-verified against its own hash inside
/// [`ObjectStore::read_object`], but nothing at this layer ties a
/// returned buffer back to *which* input hash it answered, so a
/// transposed pair of same-length chunks would silently produce a
/// byte-swapped file instead of an error). This is the exact same trust
/// boundary [`worktree::store_large_file_streaming_with`]'s `hash_chunks`
/// callback already documents on the write side (also length-checked
/// only, via its own `ChunkBatchLengthMismatch`) — order is a documented
/// caller contract on both sides, not a runtime-verified one, because
/// verifying it here would mean re-hashing every chunk a second time
/// after `read_chunks` already did the real, expensive read; that would
/// undo most of the point of fanning the read out in the first place.
/// Both of this crate's own `read_chunks` implementations satisfy the
/// contract by construction: `sequential_read_chunks`'s plain
/// `iter().map().collect()` and `mkit-cli`'s `rayon::par_iter().map().collect()`
/// (the latter relies on rayon's `IndexedParallelIterator` guarantee that
/// collecting into a `Vec` preserves input order regardless of which
/// worker finishes a given index first) are both order-preserving, so
/// this gap has no live exploit through any code path this crate or
/// `mkit-cli` ships — it is a contract a *hand-written, buggy or hostile*
/// third-party `read_chunks` could violate, same as an equally
/// third-party `hash_chunks` already could on the write side.
/// `mkit-core` has no thread-pool dependency of its own — it stays
/// usable from wasm targets, which have no OS threads — so the fan-out
/// decision lives with the caller instead of living in this function.
/// Reading each chunk (open + integrity-verify + decode) is independent
/// of every other chunk in a batch, so `mkit-cli`'s native
/// checkout/clone/reset/restore paths pass a rayon-backed `read_chunks`,
/// the read-side counterpart of `add`'s ingest-side chunk-hashing
/// fan-out.
///
/// # Errors
/// Same variants as [`restore_tree_to_worktree`], plus
/// [`RestoreError::ChunkBatchLengthMismatch`] if `read_chunks` returns a
/// different number of buffers than the batch it was given (see above
/// for what this check does and does not catch).
pub fn restore_tree_to_worktree_with<F>(
    store: &ObjectStore,
    tree: &Hash,
    root: &Path,
    opts: &RestoreOptions,
    read_chunks: &F,
) -> RestoreResult<RestoreReport>
where
    F: Fn(&ObjectStore, &[Hash]) -> RestoreResult<Vec<Vec<u8>>> + Sync,
{
    restore_tree_to_worktree_impl(store, tree, root, opts, RESTORE_CHUNK_BATCH, read_chunks)
}

/// Shared implementation behind [`restore_tree_to_worktree`] and
/// [`restore_tree_to_worktree_with`]. `batch_size` is not part of either
/// public signature — it is 1 for the sequential default (see
/// [`restore_tree_to_worktree`]'s doc) and [`RESTORE_CHUNK_BATCH`] for a
/// caller-supplied `read_chunks`.
fn restore_tree_to_worktree_impl<F>(
    store: &ObjectStore,
    tree: &Hash,
    root: &Path,
    opts: &RestoreOptions,
    batch_size: usize,
    read_chunks: &F,
) -> RestoreResult<RestoreReport>
where
    F: Fn(&ObjectStore, &[Hash]) -> RestoreResult<Vec<Vec<u8>>> + Sync,
{
    // Load the root-level ignore list. Missing = empty list.
    let ignore_list = match ignore::load(root) {
        Ok(il) => il,
        Err(_) => IgnoreList::new(),
    };
    fs::create_dir_all(root)?;
    let mut report = RestoreReport::default();
    restore_tree_to_worktree_inner(
        store,
        *tree,
        root,
        opts,
        "",
        &ignore_list,
        &mut report,
        0,
        batch_size,
        read_chunks,
    )?;
    Ok(report)
}

#[allow(clippy::too_many_arguments)]
fn restore_tree_to_worktree_inner<F>(
    store: &ObjectStore,
    tree_hash: Hash,
    target_dir: &Path,
    options: &RestoreOptions,
    path_prefix: &str,
    ignore: &IgnoreList,
    report: &mut RestoreReport,
    depth: usize,
    batch_size: usize,
    read_chunks: &F,
) -> RestoreResult<()>
where
    F: Fn(&ObjectStore, &[Hash]) -> RestoreResult<Vec<Vec<u8>>> + Sync,
{
    if depth > MAX_TREE_DEPTH {
        return Err(RestoreError::TreeTooDeep);
    }
    let obj = store.read_object(&tree_hash)?;
    let Object::Tree(tree) = obj else {
        return Err(RestoreError::NotATree);
    };

    if options.clean {
        clean_directory(
            target_dir,
            &tree.entries,
            options.sparse_patterns.as_deref(),
            path_prefix,
            Some(ignore),
        )?;
    }

    for entry in &tree.entries {
        if !crate::object::TreeEntry::validate_name(&entry.name) {
            continue;
        }
        let name = std::str::from_utf8(&entry.name).map_err(|_| RestoreError::InvalidUtf8)?;
        let full_path = if path_prefix.is_empty() {
            name.to_string()
        } else {
            format!("{path_prefix}/{name}")
        };
        // NOTE: ignore rules do NOT gate materialization. Tree entries are
        // tracked content and must always be written (git parity — skipping
        // them would desync the index from the worktree). Ignore rules only
        // protect *untracked* worktree files during the ignore-aware
        // `clean_directory` sweep below.
        match entry.mode {
            EntryMode::Blob | EntryMode::Executable => {
                if let Some(patterns) = options.sparse_patterns.as_deref()
                    && !matches_sparse(patterns, &full_path, false)
                {
                    continue;
                }
                restore_blob_with(
                    store,
                    target_dir,
                    name,
                    entry.object_hash,
                    entry.mode == EntryMode::Executable,
                    batch_size,
                    read_chunks,
                )?;
                report.files_written += 1;
            }
            EntryMode::Tree => {
                if let Some(patterns) = options.sparse_patterns.as_deref()
                    && !could_match_descendant(patterns, &full_path)
                {
                    continue;
                }
                ensure_directory(target_dir, name)?;
                report.directories_created += 1;
                let dir_path = target_dir.join(name);
                let dir_meta = fs::symlink_metadata(&dir_path)?;
                if !dir_meta.is_dir() {
                    return Err(RestoreError::NotADirectory(dir_path));
                }
                restore_tree_to_worktree_inner(
                    store,
                    entry.object_hash,
                    &dir_path,
                    options,
                    &full_path,
                    ignore,
                    report,
                    depth + 1,
                    batch_size,
                    read_chunks,
                )?;
            }
            EntryMode::Symlink => {
                if let Some(patterns) = options.sparse_patterns.as_deref()
                    && !matches_sparse(patterns, &full_path, false)
                {
                    continue;
                }
                restore_symlink(store, target_dir, name, entry.object_hash)?;
                report.symlinks_written += 1;
            }
        }
    }
    Ok(())
}

/// Materialise `tree_hash` into `target_dir`. See module docs for the
/// invariants.
///
/// Unlike [`restore_tree_to_worktree`]/[`restore_tree_to_worktree_with`],
/// this has no `_with` counterpart — `stash` (this function's only
/// caller) restores a snapshot on `push`/`pop`/`apply`, not the
/// large-file checkout/clone/reset/restore path the perf work behind
/// `restore_tree_to_worktree_with` targeted, so its `ChunkedBlob` chunks
/// are always read sequentially via the plain `restore_blob`. Giving
/// `stash` the same rayon fan-out would mean threading a `read_chunks`
/// callback through `ops::stash`'s own public API (`save`/`pop`/`apply`)
/// and `mkit-cli`'s stash command — a separate, larger change, not a
/// side effect of this one.
///
/// # Errors
/// - [`RestoreError::NotATree`] if `tree_hash` is not a tree object.
/// - [`RestoreError::InvalidSymlinkTarget`] if a symlink entry's target
///   is absolute or contains `..`.
pub fn restore_tree(
    store: &ObjectStore,
    tree_hash: Hash,
    target_dir: &Path,
    options: &RestoreOptions,
) -> RestoreResult<()> {
    fs::create_dir_all(target_dir)?;
    restore_tree_inner(store, tree_hash, target_dir, options, "")
}

fn restore_tree_inner(
    store: &ObjectStore,
    tree_hash: Hash,
    target_dir: &Path,
    options: &RestoreOptions,
    path_prefix: &str,
) -> RestoreResult<()> {
    let obj = store.read_object(&tree_hash)?;
    let Object::Tree(tree) = obj else {
        return Err(RestoreError::NotATree);
    };

    if options.clean {
        clean_directory(
            target_dir,
            &tree.entries,
            options.sparse_patterns.as_deref(),
            path_prefix,
            None,
        )?;
    }

    for entry in &tree.entries {
        if !crate::object::TreeEntry::validate_name(&entry.name) {
            continue;
        }
        let name = std::str::from_utf8(&entry.name).map_err(|_| RestoreError::InvalidUtf8)?;
        let full_path = if path_prefix.is_empty() {
            name.to_string()
        } else {
            format!("{path_prefix}/{name}")
        };
        match entry.mode {
            EntryMode::Blob | EntryMode::Executable => {
                if let Some(patterns) = options.sparse_patterns.as_deref()
                    && !matches_sparse(patterns, &full_path, false)
                {
                    continue;
                }
                restore_blob(
                    store,
                    target_dir,
                    name,
                    entry.object_hash,
                    entry.mode == EntryMode::Executable,
                )?;
            }
            EntryMode::Tree => {
                if let Some(patterns) = options.sparse_patterns.as_deref()
                    && !could_match_descendant(patterns, &full_path)
                {
                    continue;
                }
                ensure_directory(target_dir, name)?;
                let dir_path = target_dir.join(name);
                // Refuse to follow a symlink that took the place of the dir.
                let dir_meta = fs::symlink_metadata(&dir_path)?;
                if !dir_meta.is_dir() {
                    return Err(RestoreError::NotADirectory(dir_path));
                }
                restore_tree_inner(store, entry.object_hash, &dir_path, options, &full_path)?;
            }
            EntryMode::Symlink => {
                if let Some(patterns) = options.sparse_patterns.as_deref()
                    && !matches_sparse(patterns, &full_path, false)
                {
                    continue;
                }
                restore_symlink(store, target_dir, name, entry.object_hash)?;
            }
        }
    }
    Ok(())
}

/// Sequential blob restore: reads each chunk of a [`Object::ChunkedBlob`]
/// one at a time on the calling thread, straight into the open tmp file.
/// Used by [`restore_tree`] (the non-worktree, `stash`-facing path,
/// which has no parallel-capable `read_chunks` to offer). Kept as its
/// own true one-chunk-at-a-time loop rather than delegating to
/// [`restore_blob_with`] with [`sequential_read_chunks`]: batching
/// chunks before writing them (as `restore_blob_with` does, to give a
/// parallel `read_chunks` something to fan out) would raise this path's
/// peak memory from one chunk (≤256 KiB) to a full [`RESTORE_CHUNK_BATCH`]
/// batch (≤16 MiB) for no benefit, regressing issue #828's original
/// "reassemble without ever holding more than one chunk" guarantee on a
/// path that can never use the extra batch size for anything.
fn restore_blob(
    store: &ObjectStore,
    dir: &Path,
    name: &str,
    blob_hash: Hash,
    executable: bool,
) -> RestoreResult<()> {
    let obj = store.read_object(&blob_hash)?;
    match obj {
        Object::Blob(b) => write_file_atomic(dir, name, &b.data, executable)?,
        Object::ChunkedBlob(cb) => {
            let (tmp_path, final_path, mut tmp) = create_tmp_for_write(dir, name)?;
            let mut written: u64 = 0;
            for ch in &cb.chunks {
                let buf = read_chunk_bytes(store, ch)?;
                tmp.write_all(&buf)?;
                written += buf.len() as u64;
            }
            cb.check_reassembled_size(usize::try_from(written).unwrap_or(usize::MAX))?;
            #[cfg(not(target_arch = "wasm32"))] // wasm File has no destructor
            drop(tmp);
            finish_atomic_write(&tmp_path, &final_path, executable)?;
        }
        _ => return Err(RestoreError::NotABlob),
    }
    Ok(())
}

/// Materialise `blob_hash` (a [`Object::Blob`] or [`Object::ChunkedBlob`])
/// as `dir/name`. For a `ChunkedBlob`, `read_chunks` reads back each
/// batch of up to `batch_size` chunk hashes (at most [`RESTORE_CHUNK_BATCH`]
/// — see [`restore_tree_to_worktree_with`]'s doc for the callback
/// contract; `batch_size` itself is `restore_tree_to_worktree_impl`'s
/// internal knob, 1 for the sequential default or `RESTORE_CHUNK_BATCH`
/// for a caller-supplied `read_chunks`, never part of either public
/// entry point's own signature).
///
/// # Panics
/// If `batch_size` is 0 (`[T]::chunks` itself panics on that) — never
/// true for either internal caller.
fn restore_blob_with<F>(
    store: &ObjectStore,
    dir: &Path,
    name: &str,
    blob_hash: Hash,
    executable: bool,
    batch_size: usize,
    read_chunks: &F,
) -> RestoreResult<()>
where
    F: Fn(&ObjectStore, &[Hash]) -> RestoreResult<Vec<Vec<u8>>> + Sync,
{
    let obj = store.read_object(&blob_hash)?;
    match obj {
        Object::Blob(b) => write_file_atomic(dir, name, &b.data, executable)?,
        Object::ChunkedBlob(cb) => {
            // Stream each batch straight to the open tmp file instead of
            // concatenating the whole reassembled file into memory first
            // (issue #828): peak memory is one `batch_size`-chunk batch
            // (bounded, see this function's doc), not the file's total
            // size. Batching (rather than one chunk at a time) lets a
            // parallel-capable `read_chunks` fan a batch's independent
            // reads out across threads — see this function's and
            // `restore_tree_to_worktree_with`'s docs.
            let (tmp_path, final_path, mut tmp) = create_tmp_for_write(dir, name)?;
            let mut written: u64 = 0;
            for batch in cb.chunks.chunks(batch_size) {
                let bufs = read_chunks(store, batch)?;
                if bufs.len() != batch.len() {
                    return Err(RestoreError::ChunkBatchLengthMismatch {
                        expected: batch.len(),
                        actual: bufs.len(),
                    });
                }
                for buf in &bufs {
                    tmp.write_all(buf)?;
                    written += buf.len() as u64;
                }
            }
            cb.check_reassembled_size(usize::try_from(written).unwrap_or(usize::MAX))?;
            #[cfg(not(target_arch = "wasm32"))] // wasm File has no destructor
            drop(tmp);
            finish_atomic_write(&tmp_path, &final_path, executable)?;
        }
        _ => return Err(RestoreError::NotABlob),
    }
    Ok(())
}

/// Default `read_chunks`: reads and type-checks each chunk hash in
/// `hashes` one at a time via [`ObjectStore::read_object`] (which
/// integrity-verifies the bytes against the hash).
fn sequential_read_chunks(store: &ObjectStore, hashes: &[Hash]) -> RestoreResult<Vec<Vec<u8>>> {
    hashes.iter().map(|h| read_chunk_bytes(store, h)).collect()
}

fn read_chunk_bytes(store: &ObjectStore, h: &Hash) -> RestoreResult<Vec<u8>> {
    match store.read_object(h)? {
        Object::Blob(b) => Ok(b.data),
        _ => Err(RestoreError::NotABlob),
    }
}

fn restore_symlink(
    store: &ObjectStore,
    dir: &Path,
    name: &str,
    blob_hash: Hash,
) -> RestoreResult<()> {
    let obj = store.read_object(&blob_hash)?;
    let Object::Blob(b) = obj else {
        return Err(RestoreError::NotABlob);
    };
    let target = std::str::from_utf8(&b.data).map_err(|_| RestoreError::InvalidUtf8)?;
    if !worktree::validate_symlink_target(target) {
        return Err(RestoreError::InvalidSymlinkTarget(target.to_string()));
    }
    let tmp_name = make_tmp_sibling_name(name);
    let tmp_path = dir.join(&tmp_name);
    let final_path = dir.join(name);
    let _ = fs::remove_file(&tmp_path);
    create_symlink(target, &tmp_path)?;
    prepare_path_for_rename(&final_path)?;
    fs::rename(&tmp_path, &final_path)?;
    Ok(())
}

#[cfg(unix)]
fn create_symlink(target: &str, link: &Path) -> io::Result<()> {
    std::os::unix::fs::symlink(target, link)
}

#[cfg(windows)]
fn create_symlink(target: &str, link: &Path) -> io::Result<()> {
    // Symlinks on Windows require either Developer Mode or admin
    // privileges. We pick `symlink_file` because every blob is a file
    // when materialised through this code path.
    std::os::windows::fs::symlink_file(target, link)
}

#[cfg(not(any(unix, windows)))]
fn create_symlink(_target: &str, _link: &Path) -> io::Result<()> {
    // Targets without a filesystem (notably `wasm32-unknown-unknown`)
    // cannot materialise symlinks. The demo wasm crate does not exercise
    // the restore path; this stub exists so the crate still compiles.
    Err(io::Error::new(
        io::ErrorKind::Unsupported,
        "symlink creation is not supported on this target",
    ))
}

/// Write a restored worktree file via tmp + rename so concurrent
/// readers never observe a torn file.
///
/// Deliberately NOT flushed: worktree contents are not part of the
/// store's durability invariant (SPEC-OBJECTS §10.1) — the object
/// store is the source of truth and checkout is re-runnable after a
/// crash. Flushing every restored file made checkout O(files) full
/// flushes (`F_FULLFSYNC` each on macOS) for no recoverable state; git
/// likewise does not flush checked-out files.
fn write_file_atomic(dir: &Path, name: &str, data: &[u8], executable: bool) -> io::Result<()> {
    let (tmp_path, final_path, mut tmp) = create_tmp_for_write(dir, name)?;
    tmp.write_all(data)?;
    #[cfg(not(target_arch = "wasm32"))] // wasm File has no destructor
    drop(tmp);
    finish_atomic_write(&tmp_path, &final_path, executable)
}

/// Open a fresh tmp file beside `name`, removing any stale tmp file a
/// prior aborted attempt at the same name left behind. Returns the tmp
/// path, final path, and the open handle so a caller can write
/// incrementally (see [`restore_blob`]'s `ChunkedBlob` arm) instead of
/// buffering the whole file first — pair with [`finish_atomic_write`]
/// to apply the executable bit and atomically rename into place.
fn create_tmp_for_write(dir: &Path, name: &str) -> io::Result<(PathBuf, PathBuf, fs::File)> {
    let tmp_name = make_tmp_sibling_name(name);
    let tmp_path = dir.join(&tmp_name);
    let final_path = dir.join(name);
    let _ = fs::remove_file(&tmp_path);
    let tmp = fs::File::create(&tmp_path)?;
    Ok((tmp_path, final_path, tmp))
}

/// Apply the executable bit (if requested) and atomically rename the
/// tmp file from [`create_tmp_for_write`] into its final path. The tmp
/// file's handle must already be closed/dropped before calling this
/// (renaming an open handle is unreliable on some platforms).
fn finish_atomic_write(tmp_path: &Path, final_path: &Path, executable: bool) -> io::Result<()> {
    if executable {
        apply_executable_bit(tmp_path)?;
    }
    prepare_path_for_rename(final_path)?;
    fs::rename(tmp_path, final_path)?;
    Ok(())
}

#[cfg(unix)]
fn apply_executable_bit(path: &Path) -> io::Result<()> {
    use std::os::unix::fs::PermissionsExt;
    let mut perm = fs::metadata(path)?.permissions();
    perm.set_mode(0o755);
    fs::set_permissions(path, perm)
}

#[cfg(not(unix))]
#[allow(clippy::unnecessary_wraps)]
fn apply_executable_bit(_path: &Path) -> io::Result<()> {
    Ok(())
}

fn ensure_directory(parent: &Path, name: &str) -> io::Result<()> {
    let path = parent.join(name);
    match fs::symlink_metadata(&path) {
        Ok(meta) if meta.is_dir() => return Ok(()),
        Ok(_) => fs::remove_file(&path)?,
        Err(e) if e.kind() == io::ErrorKind::NotFound => {}
        Err(e) => return Err(e),
    }
    match fs::create_dir_all(&path) {
        Ok(()) => Ok(()),
        Err(e) if e.kind() == io::ErrorKind::AlreadyExists => {
            let meta = fs::symlink_metadata(&path)?;
            if meta.is_dir() {
                Ok(())
            } else {
                Err(io::Error::new(
                    io::ErrorKind::AlreadyExists,
                    "expected directory",
                ))
            }
        }
        Err(e) => Err(e),
    }
}

fn prepare_path_for_rename(final_path: &Path) -> io::Result<()> {
    match fs::symlink_metadata(final_path) {
        Ok(meta) if meta.is_dir() => fs::remove_dir_all(final_path),
        Ok(_) => Ok(()),
        Err(e) if e.kind() == io::ErrorKind::NotFound => Ok(()),
        Err(e) => Err(e),
    }
}

fn make_tmp_sibling_name(name: &str) -> String {
    let pid = process::id();
    let counter = TMP_COUNTER.fetch_add(1, Ordering::Relaxed);
    format!(".{name}.tmp.{pid}.{counter}")
}

/// Sweep `target_dir` of untracked entries before re-materialising a
/// tree. Entries present in `tree_entries`, the repo-metadata dirs
/// (`.mkit`/`.git`, ASCII case-insensitive so case-insensitive
/// filesystems can't smuggle a `.MKIT`/`.Git` entry past the sweep — Git
/// CVE-2021-21300 family), and `.mkitignore` are always preserved.
///
/// When `ignore` is `Some`, this is the worktree-checkout path: entries
/// matching the ignore list are additionally preserved (ancestor-aware,
/// so an untracked file *under* an ignored directory survives too), and
/// `.gitignore` is preserved as well.
fn clean_directory(
    target_dir: &Path,
    tree_entries: &[TreeEntry],
    sparse_patterns: Option<&[SparsePattern]>,
    path_prefix: &str,
    ignore: Option<&IgnoreList>,
) -> RestoreResult<()> {
    struct CleanItem {
        name: String,
        is_dir: bool,
    }
    let mut to_delete: Vec<CleanItem> = Vec::new();

    let read = match fs::read_dir(target_dir) {
        Ok(r) => r,
        Err(e) if e.kind() == io::ErrorKind::NotFound => return Ok(()),
        Err(e) => return Err(RestoreError::Io(e)),
    };
    for entry in read {
        let entry = entry?;
        let file_name = entry.file_name();
        let name_str = file_name
            .to_str()
            .ok_or(RestoreError::InvalidUtf8)?
            .to_string();
        if name_str.eq_ignore_ascii_case(".mkit") || name_str.eq_ignore_ascii_case(".git") {
            continue;
        }
        if name_str == ".mkitignore" {
            continue;
        }
        // The ignore-aware checkout path also preserves `.gitignore`.
        if ignore.is_some() && name_str == ".gitignore" {
            continue;
        }
        let mut found = false;
        for te in tree_entries {
            if te.name.as_slice() == name_str.as_bytes() {
                found = true;
                break;
            }
        }
        if found {
            continue;
        }
        let meta = entry.metadata()?;
        let is_dir = meta.is_dir();
        let full_path = if path_prefix.is_empty() {
            name_str.clone()
        } else {
            format!("{path_prefix}/{name_str}")
        };
        // Respect ignore rules — don't touch locally-ignored files. Use
        // ancestor-aware matching so an untracked file *under* an ignored
        // directory is preserved too (the safety gate exempts it, so the
        // sweep must not delete it).
        if let Some(ignore) = ignore
            && ignore.is_ignored_with_ancestors(&full_path, is_dir)
        {
            continue;
        }
        if let Some(patterns) = sparse_patterns {
            let allow = matches_sparse(patterns, &full_path, is_dir)
                || (is_dir && could_match_descendant(patterns, &full_path));
            if !allow {
                continue;
            }
        }
        to_delete.push(CleanItem {
            name: name_str,
            is_dir,
        });
    }

    for item in to_delete {
        let path = target_dir.join(&item.name);
        if item.is_dir {
            let _ = fs::remove_dir_all(&path);
        } else {
            let _ = fs::remove_file(&path);
        }
    }
    Ok(())
}

#[cfg(test)]
mod tests {
    use super::*;
    use crate::object::{Tree, TreeEntry};
    use crate::serialize;
    use tempfile::TempDir;

    fn fresh_store() -> (TempDir, ObjectStore) {
        let dir = TempDir::new().unwrap();
        let store = ObjectStore::init(&RepoLayout::single(dir.path())).unwrap();
        (dir, store)
    }

    fn put_blob(store: &ObjectStore, data: &[u8]) -> Hash {
        let bytes = serialize::serialize(&Object::Blob(crate::object::Blob {
            data: data.to_vec(),
        }))
        .unwrap();
        store.write(&bytes).unwrap()
    }

    fn put_tree_with(store: &ObjectStore, entries: Vec<TreeEntry>) -> Hash {
        let bytes = serialize::serialize(&Object::Tree(Tree { entries })).unwrap();
        store.write(&bytes).unwrap()
    }

    #[test]
    fn parse_sparse_basic() {
        let content = "# comment line\nsrc\n!tests\ndocs/\n\nREADME.md\n";
        let p = parse_sparse_patterns(content);
        assert_eq!(p.len(), 4);
        assert_eq!(p[0].pattern, "src");
        assert!(!p[0].negated);
        assert!(!p[0].dir_only);
        assert_eq!(p[1].pattern, "tests");
        assert!(p[1].negated);
        assert_eq!(p[2].pattern, "docs");
        assert!(p[2].dir_only);
        assert_eq!(p[3].pattern, "README.md");
    }

    #[test]
    fn matches_sparse_exact_and_prefix() {
        let p = vec![SparsePattern {
            pattern: "src".to_string(),
            negated: false,
            dir_only: false,
        }];
        assert!(matches_sparse(&p, "src/main.rs", false));
        assert!(matches_sparse(&p, "src/lib/util.rs", false));
        assert!(!matches_sparse(&p, "tests/foo", false));
    }

    #[test]
    fn matches_sparse_negation() {
        let p = vec![
            SparsePattern {
                pattern: "src".to_string(),
                negated: false,
                dir_only: false,
            },
            SparsePattern {
                pattern: "src/secret".to_string(),
                negated: true,
                dir_only: false,
            },
        ];
        assert!(matches_sparse(&p, "src/main.rs", false));
        assert!(!matches_sparse(&p, "src/secret/key.pem", false));
    }

    #[test]
    fn matches_sparse_dir_only() {
        let p = vec![SparsePattern {
            pattern: "build".to_string(),
            negated: false,
            dir_only: true,
        }];
        assert!(matches_sparse(&p, "build", true));
        assert!(!matches_sparse(&p, "build", false));
    }

    #[test]
    fn matches_sparse_last_match_wins() {
        let p = vec![
            SparsePattern {
                pattern: "src".to_string(),
                negated: false,
                dir_only: false,
            },
            SparsePattern {
                pattern: "src".to_string(),
                negated: true,
                dir_only: false,
            },
        ];
        assert!(!matches_sparse(&p, "src/main.rs", false));
    }

    #[test]
    fn matches_sparse_bare_basename() {
        let p = vec![SparsePattern {
            pattern: "Makefile".to_string(),
            negated: false,
            dir_only: false,
        }];
        assert!(matches_sparse(&p, "Makefile", false));
        assert!(matches_sparse(&p, "sub/Makefile", false));
        assert!(!matches_sparse(&p, "Makefile.bak", false));
    }

    #[test]
    fn could_match_descendant_basic() {
        let p = vec![SparsePattern {
            pattern: "src/lib".to_string(),
            negated: false,
            dir_only: false,
        }];
        assert!(could_match_descendant(&p, "src"));
        assert!(could_match_descendant(&p, "src/lib"));
        assert!(!could_match_descendant(&p, "tests"));
    }

    #[test]
    fn restore_empty_tree_creates_no_files() {
        let (_d, store) = fresh_store();
        let target = TempDir::new().unwrap();
        let tree_h = put_tree_with(&store, vec![]);
        restore_tree(&store, tree_h, target.path(), &RestoreOptions::default()).unwrap();
        let count = fs::read_dir(target.path()).unwrap().count();
        assert_eq!(count, 0);
    }

    #[test]
    fn restore_single_file() {
        let (_d, store) = fresh_store();
        let target = TempDir::new().unwrap();
        let blob = put_blob(&store, b"hello");
        let tree = put_tree_with(
            &store,
            vec![TreeEntry {
                name: b"file.txt".to_vec(),
                mode: EntryMode::Blob,
                object_hash: blob,
            }],
        );
        restore_tree(&store, tree, target.path(), &RestoreOptions::default()).unwrap();
        let content = fs::read(target.path().join("file.txt")).unwrap();
        assert_eq!(content, b"hello");
    }

    #[test]
    fn restore_nested_directories() {
        let (_d, store) = fresh_store();
        let target = TempDir::new().unwrap();
        let blob = put_blob(&store, b"const main = 0;");
        let inner = put_tree_with(
            &store,
            vec![TreeEntry {
                name: b"main.rs".to_vec(),
                mode: EntryMode::Blob,
                object_hash: blob,
            }],
        );
        let root = put_tree_with(
            &store,
            vec![TreeEntry {
                name: b"src".to_vec(),
                mode: EntryMode::Tree,
                object_hash: inner,
            }],
        );
        restore_tree(&store, root, target.path(), &RestoreOptions::default()).unwrap();
        let content = fs::read(target.path().join("src/main.rs")).unwrap();
        assert_eq!(content, b"const main = 0;");
    }

    #[test]
    fn restore_overwrites_existing_files() {
        let (_d, store) = fresh_store();
        let target = TempDir::new().unwrap();
        fs::write(target.path().join("file.txt"), b"old").unwrap();
        let blob = put_blob(&store, b"new");
        let tree = put_tree_with(
            &store,
            vec![TreeEntry {
                name: b"file.txt".to_vec(),
                mode: EntryMode::Blob,
                object_hash: blob,
            }],
        );
        restore_tree(&store, tree, target.path(), &RestoreOptions::default()).unwrap();
        assert_eq!(fs::read(target.path().join("file.txt")).unwrap(), b"new");
    }

    #[test]
    fn restore_removes_untracked_when_clean() {
        let (_d, store) = fresh_store();
        let target = TempDir::new().unwrap();
        fs::write(target.path().join("extra.txt"), b"gone").unwrap();
        let blob = put_blob(&store, b"keep");
        let tree = put_tree_with(
            &store,
            vec![TreeEntry {
                name: b"tracked.txt".to_vec(),
                mode: EntryMode::Blob,
                object_hash: blob,
            }],
        );
        restore_tree(&store, tree, target.path(), &RestoreOptions::default()).unwrap();
        assert!(!target.path().join("extra.txt").exists());
        assert_eq!(
            fs::read(target.path().join("tracked.txt")).unwrap(),
            b"keep"
        );
    }

    #[test]
    fn restore_clean_false_keeps_untracked() {
        let (_d, store) = fresh_store();
        let target = TempDir::new().unwrap();
        fs::write(target.path().join("extra.txt"), b"survive").unwrap();
        let blob = put_blob(&store, b"keep");
        let tree = put_tree_with(
            &store,
            vec![TreeEntry {
                name: b"tracked.txt".to_vec(),
                mode: EntryMode::Blob,
                object_hash: blob,
            }],
        );
        restore_tree(
            &store,
            tree,
            target.path(),
            &RestoreOptions {
                clean: false,
                sparse_patterns: None,
            },
        )
        .unwrap();
        assert_eq!(
            fs::read(target.path().join("extra.txt")).unwrap(),
            b"survive"
        );
    }

    #[test]
    fn restore_preserves_mkit_directory() {
        let (_d, store) = fresh_store();
        let target = TempDir::new().unwrap();
        fs::create_dir_all(target.path().join(".mkit")).unwrap();
        fs::write(target.path().join(".mkit/config"), b"important").unwrap();
        let tree = put_tree_with(&store, vec![]);
        restore_tree(&store, tree, target.path(), &RestoreOptions::default()).unwrap();
        assert_eq!(
            fs::read(target.path().join(".mkit/config")).unwrap(),
            b"important"
        );
    }

    #[test]
    fn clean_directory_preserves_case_variant_mkit_and_git() {
        // Regression for Git CVE-2021-21300 family: on case-insensitive
        // filesystems a worktree directory named `.MKIT` or `.Git` must
        // never be swept by `clean_directory`, otherwise a hostile tree
        // entry could trick restore into deleting repo metadata.
        let target = TempDir::new().unwrap();
        fs::create_dir_all(target.path().join(".MKIT")).unwrap();
        fs::write(target.path().join(".MKIT/config"), b"meta").unwrap();
        fs::create_dir_all(target.path().join(".Git")).unwrap();
        fs::write(target.path().join(".Git/HEAD"), b"ref").unwrap();
        // Empty tree — without the case-insensitive guard, everything
        // unknown is removed.
        clean_directory(target.path(), &[], None, "", None).unwrap();
        assert!(
            target.path().join(".MKIT/config").exists(),
            ".MKIT swept by clean_directory (case-fold bypass)"
        );
        assert!(
            target.path().join(".Git/HEAD").exists(),
            ".Git swept by clean_directory (case-fold bypass)"
        );
    }

    #[test]
    fn clean_directory_with_ignore_list_preserves_case_variant_mkit_and_git() {
        let target = TempDir::new().unwrap();
        fs::create_dir_all(target.path().join(".MKIT")).unwrap();
        fs::write(target.path().join(".MKIT/config"), b"meta").unwrap();
        fs::create_dir_all(target.path().join(".GIT")).unwrap();
        fs::write(target.path().join(".GIT/HEAD"), b"ref").unwrap();
        let ignore = crate::ignore::IgnoreList::new();
        clean_directory(target.path(), &[], None, "", Some(&ignore)).unwrap();
        assert!(
            target.path().join(".MKIT/config").exists(),
            ".MKIT swept by clean_directory (ignore-aware, case-fold bypass)"
        );
        assert!(
            target.path().join(".GIT/HEAD").exists(),
            ".GIT swept by clean_directory (ignore-aware, case-fold bypass)"
        );
    }

    #[test]
    fn restore_chunked_blob_reassembled() {
        let (_d, store) = fresh_store();
        let target = TempDir::new().unwrap();
        let c0 = put_blob(&store, b"Hello, ");
        let c1 = put_blob(&store, b"chunked ");
        let c2 = put_blob(&store, b"world!");
        let cb = Object::ChunkedBlob(crate::object::ChunkedBlob {
            total_size: 7 + 8 + 6,
            chunk_size: 64 * 1024,
            chunks: vec![c0, c1, c2],
        });
        let cb_h = store.write(&serialize::serialize(&cb).unwrap()).unwrap();
        let tree = put_tree_with(
            &store,
            vec![TreeEntry {
                name: b"out.txt".to_vec(),
                mode: EntryMode::Blob,
                object_hash: cb_h,
            }],
        );
        restore_tree(&store, tree, target.path(), &RestoreOptions::default()).unwrap();
        let content = fs::read(target.path().join("out.txt")).unwrap();
        assert_eq!(content, b"Hello, chunked world!");
    }

    /// Mirrors `worktree`'s
    /// `streaming_rejects_miscounted_hashes_in_full_and_partial_batches`
    /// on the read side: a `read_chunks` callback that returns fewer
    /// buffers than its input batch must surface as
    /// [`RestoreError::ChunkBatchLengthMismatch`], not a panic or a
    /// silently short/garbled restored file.
    #[test]
    fn restore_tree_to_worktree_with_rejects_miscounted_chunks() {
        let (_d, store) = fresh_store();
        let target = TempDir::new().unwrap();
        let c0 = put_blob(&store, b"Hello, ");
        let c1 = put_blob(&store, b"chunked ");
        let c2 = put_blob(&store, b"world!");
        let cb = Object::ChunkedBlob(crate::object::ChunkedBlob {
            total_size: 7 + 8 + 6,
            chunk_size: 0,
            chunks: vec![c0, c1, c2],
        });
        let cb_h = store.write(&serialize::serialize(&cb).unwrap()).unwrap();
        let tree = put_tree_with(
            &store,
            vec![TreeEntry {
                name: b"out.txt".to_vec(),
                mode: EntryMode::Blob,
                object_hash: cb_h,
            }],
        );
        let err = restore_tree_to_worktree_with(
            &store,
            &tree,
            target.path(),
            &RestoreOptions::default(),
            &|_store, batch| Ok(vec![Vec::new(); batch.len() - 1]),
        )
        .unwrap_err();
        assert!(
            matches!(
                err,
                RestoreError::ChunkBatchLengthMismatch {
                    expected: 3,
                    actual: 2
                }
            ),
            "unexpected error: {err:?}"
        );
    }

    /// Mirrors `worktree`'s
    /// `hash_file_with_metadata_with_batch_fanout_matches_sequential`: a
    /// `read_chunks` callback that processes a batch out of order
    /// internally (simulating a rayon fan-out) but returns results in
    /// input order — the documented contract — must still reassemble the
    /// exact same bytes as the sequential default. Pins the `_with`
    /// contract `mkit-cli`'s `restore_fanout::read_chunks_fanout` relies
    /// on.
    #[test]
    fn restore_tree_to_worktree_with_out_of_order_processing_matches_sequential() {
        let (_d, store) = fresh_store();
        let chunk_data: Vec<Vec<u8>> = (0..10)
            .map(|i| format!("chunk-{i:02}-").into_bytes())
            .collect();
        let total_size: u64 = chunk_data.iter().map(|c| c.len() as u64).sum();
        let chunks: Vec<Hash> = chunk_data.iter().map(|c| put_blob(&store, c)).collect();
        let cb = Object::ChunkedBlob(crate::object::ChunkedBlob {
            total_size,
            chunk_size: 0,
            chunks: chunks.clone(),
        });
        let cb_h = store.write(&serialize::serialize(&cb).unwrap()).unwrap();
        let tree = put_tree_with(
            &store,
            vec![TreeEntry {
                name: b"out.txt".to_vec(),
                mode: EntryMode::Blob,
                object_hash: cb_h,
            }],
        );

        let sequential_target = TempDir::new().unwrap();
        restore_tree_to_worktree(
            &store,
            &tree,
            sequential_target.path(),
            &RestoreOptions::default(),
        )
        .unwrap();
        let sequential_content = fs::read(sequential_target.path().join("out.txt")).unwrap();

        let fanout_target = TempDir::new().unwrap();
        restore_tree_to_worktree_with(
            &store,
            &tree,
            fanout_target.path(),
            &RestoreOptions::default(),
            &|store, batch| {
                // Read in reverse, then un-reverse before returning —
                // proves the contract cares about the *returned* order,
                // not the order chunks are actually processed in.
                let mut out: Vec<Vec<u8>> = batch
                    .iter()
                    .rev()
                    .map(|h| read_chunk_bytes(store, h))
                    .collect::<RestoreResult<_>>()?;
                out.reverse();
                Ok(out)
            },
        )
        .unwrap();
        let fanout_content = fs::read(fanout_target.path().join("out.txt")).unwrap();

        assert_eq!(
            sequential_content, fanout_content,
            "an out-of-order-processing read_chunks callback must still match \
             the sequential default when it returns results in input order"
        );
    }

    /// SPEC-OBJECTS §7: "The concatenated length MUST equal `total_size`."
    /// A manifest whose forged `total_size` disagrees with its (valid)
    /// chunks must fail restore instead of writing wrong-length content.
    #[test]
    fn restore_rejects_chunked_total_size_mismatch() {
        let (_d, store) = fresh_store();
        let target = TempDir::new().unwrap();
        let c0 = put_blob(&store, b"Hello, ");
        let c1 = put_blob(&store, b"world!");
        let cb = Object::ChunkedBlob(crate::object::ChunkedBlob {
            total_size: 7 + 6 + 1,
            chunk_size: 0,
            chunks: vec![c0, c1],
        });
        let cb_h = store.write(&serialize::serialize(&cb).unwrap()).unwrap();
        let tree = put_tree_with(
            &store,
            vec![TreeEntry {
                name: b"out.txt".to_vec(),
                mode: EntryMode::Blob,
                object_hash: cb_h,
            }],
        );
        let err =
            restore_tree(&store, tree, target.path(), &RestoreOptions::default()).unwrap_err();
        assert!(
            matches!(
                err,
                RestoreError::Object(object::MkitError::ChunkedBlobSizeMismatch {
                    expected: 14,
                    actual: 13,
                })
            ),
            "expected ChunkedBlobSizeMismatch, got {err:?}"
        );
        assert!(!target.path().join("out.txt").exists());
    }

    #[cfg(unix)]
    #[test]
    fn restore_with_symlink() {
        let (_d, store) = fresh_store();
        let target = TempDir::new().unwrap();
        let link_target = put_blob(&store, b"target.txt");
        let file = put_blob(&store, b"real");
        let tree = put_tree_with(
            &store,
            vec![
                TreeEntry {
                    name: b"link".to_vec(),
                    mode: EntryMode::Symlink,
                    object_hash: link_target,
                },
                TreeEntry {
                    name: b"target.txt".to_vec(),
                    mode: EntryMode::Blob,
                    object_hash: file,
                },
            ],
        );
        restore_tree(&store, tree, target.path(), &RestoreOptions::default()).unwrap();
        let read = fs::read_link(target.path().join("link")).unwrap();
        assert_eq!(read.to_str().unwrap(), "target.txt");
        assert_eq!(fs::read(target.path().join("target.txt")).unwrap(), b"real");
    }

    #[cfg(unix)]
    #[test]
    fn restore_rejects_invalid_symlink_targets() {
        let (_d, store) = fresh_store();
        let target = TempDir::new().unwrap();
        let bad = put_blob(&store, b"/etc/passwd");
        let tree = put_tree_with(
            &store,
            vec![TreeEntry {
                name: b"link".to_vec(),
                mode: EntryMode::Symlink,
                object_hash: bad,
            }],
        );
        let err =
            restore_tree(&store, tree, target.path(), &RestoreOptions::default()).unwrap_err();
        assert!(matches!(err, RestoreError::InvalidSymlinkTarget(_)));
    }

    #[test]
    fn sparse_restore_only_restores_matched() {
        let (_d, store) = fresh_store();
        let target = TempDir::new().unwrap();
        let main = put_blob(&store, b"pub fn main(){}");
        let test = put_blob(&store, b"test {}");
        let readme = put_blob(&store, b"# Project");
        let src = put_tree_with(
            &store,
            vec![TreeEntry {
                name: b"main.rs".to_vec(),
                mode: EntryMode::Blob,
                object_hash: main,
            }],
        );
        let tests = put_tree_with(
            &store,
            vec![TreeEntry {
                name: b"test.rs".to_vec(),
                mode: EntryMode::Blob,
                object_hash: test,
            }],
        );
        let root = put_tree_with(
            &store,
            vec![
                TreeEntry {
                    name: b"README.md".to_vec(),
                    mode: EntryMode::Blob,
                    object_hash: readme,
                },
                TreeEntry {
                    name: b"src".to_vec(),
                    mode: EntryMode::Tree,
                    object_hash: src,
                },
                TreeEntry {
                    name: b"tests".to_vec(),
                    mode: EntryMode::Tree,
                    object_hash: tests,
                },
            ],
        );
        let opts = RestoreOptions {
            clean: true,
            sparse_patterns: Some(vec![SparsePattern {
                pattern: "src".to_string(),
                negated: false,
                dir_only: false,
            }]),
        };
        restore_tree(&store, root, target.path(), &opts).unwrap();
        assert!(target.path().join("src/main.rs").exists());
        assert!(!target.path().join("tests/test.rs").exists());
        assert!(!target.path().join("README.md").exists());
    }

    #[test]
    fn sparse_restore_with_negation_excludes_subtree() {
        let (_d, store) = fresh_store();
        let target = TempDir::new().unwrap();
        let main = put_blob(&store, b"main");
        let key = put_blob(&store, b"secret");
        let secret_tree = put_tree_with(
            &store,
            vec![TreeEntry {
                name: b"key.pem".to_vec(),
                mode: EntryMode::Blob,
                object_hash: key,
            }],
        );
        let src = put_tree_with(
            &store,
            vec![
                TreeEntry {
                    name: b"main.rs".to_vec(),
                    mode: EntryMode::Blob,
                    object_hash: main,
                },
                TreeEntry {
                    name: b"secret".to_vec(),
                    mode: EntryMode::Tree,
                    object_hash: secret_tree,
                },
            ],
        );
        let root = put_tree_with(
            &store,
            vec![TreeEntry {
                name: b"src".to_vec(),
                mode: EntryMode::Tree,
                object_hash: src,
            }],
        );
        let opts = RestoreOptions {
            clean: true,
            sparse_patterns: Some(vec![
                SparsePattern {
                    pattern: "src".to_string(),
                    negated: false,
                    dir_only: false,
                },
                SparsePattern {
                    pattern: "src/secret".to_string(),
                    negated: true,
                    dir_only: false,
                },
            ]),
        };
        restore_tree(&store, root, target.path(), &opts).unwrap();
        assert!(target.path().join("src/main.rs").exists());
        assert!(!target.path().join("src/secret/key.pem").exists());
    }

    #[test]
    fn sparse_checkout_roundtrip() {
        let target = TempDir::new().unwrap();
        let layout = RepoLayout::single(target.path());
        write_sparse_checkout(&layout, &["src", "!src/secret", "docs/"]).unwrap();
        let p = load_sparse_checkout(&layout).unwrap().unwrap();
        assert_eq!(p.len(), 3);
        assert_eq!(p[0].pattern, "src");
        assert!(p[1].negated);
        assert!(p[2].dir_only);
    }

    #[test]
    fn load_sparse_checkout_returns_none_when_missing() {
        let target = TempDir::new().unwrap();
        fs::create_dir_all(target.path().join(".mkit")).unwrap();
        let p = load_sparse_checkout(&RepoLayout::single(target.path())).unwrap();
        assert!(p.is_none());
    }

    // =================================================================
    // restore_tree_to_worktree — checkout-facing wrapper.
    // =================================================================

    #[test]
    fn worktree_restore_counts_files_and_dirs() {
        let (_d, store) = fresh_store();
        let target = TempDir::new().unwrap();
        let blob_a = put_blob(&store, b"a");
        let blob_b = put_blob(&store, b"b");
        let sub = put_tree_with(
            &store,
            vec![TreeEntry {
                name: b"b.txt".to_vec(),
                mode: EntryMode::Blob,
                object_hash: blob_b,
            }],
        );
        let root = put_tree_with(
            &store,
            vec![
                TreeEntry {
                    name: b"a.txt".to_vec(),
                    mode: EntryMode::Blob,
                    object_hash: blob_a,
                },
                TreeEntry {
                    name: b"sub".to_vec(),
                    mode: EntryMode::Tree,
                    object_hash: sub,
                },
            ],
        );
        let report =
            restore_tree_to_worktree(&store, &root, target.path(), &RestoreOptions::default())
                .unwrap();
        assert_eq!(report.files_written, 2);
        assert_eq!(report.directories_created, 1);
        assert!(target.path().join("a.txt").exists());
        assert!(target.path().join("sub/b.txt").exists());
    }

    #[test]
    fn worktree_restore_writes_tracked_entries_and_keeps_untracked_ignored() {
        let (_d, store) = fresh_store();
        let target = TempDir::new().unwrap();
        // Pre-seed an ignore file, an UNTRACKED ignored file (must survive the
        // clean sweep), and a tracked path that happens to match the ignore
        // pattern but IS in the target tree (must be written — git parity).
        fs::write(target.path().join(".mkitignore"), "*.tmp\nsecret.txt\n").unwrap();
        fs::write(target.path().join("scratch.tmp"), b"local-only").unwrap();
        fs::write(target.path().join("secret.txt"), b"OLD-LOCAL").unwrap();
        let secret_blob = put_blob(&store, b"COMMITTED-SECRET");
        let ok_blob = put_blob(&store, b"ok");
        let root = put_tree_with(
            &store,
            vec![
                TreeEntry {
                    name: b"ok.txt".to_vec(),
                    mode: EntryMode::Blob,
                    object_hash: ok_blob,
                },
                TreeEntry {
                    name: b"secret.txt".to_vec(),
                    mode: EntryMode::Blob,
                    object_hash: secret_blob,
                },
            ],
        );
        let report =
            restore_tree_to_worktree(&store, &root, target.path(), &RestoreOptions::default())
                .unwrap();
        // Both tracked entries materialise — ignore rules never gate writes.
        assert_eq!(report.files_written, 2);
        assert_eq!(
            fs::read(target.path().join("secret.txt")).unwrap(),
            b"COMMITTED-SECRET",
            "a tracked tree entry is written even if it matches an ignore rule"
        );
        assert_eq!(fs::read(target.path().join("ok.txt")).unwrap(), b"ok");
        // The UNTRACKED ignored file is preserved by the clean sweep.
        assert_eq!(
            fs::read(target.path().join("scratch.tmp")).unwrap(),
            b"local-only",
            "an untracked ignored file must survive the clean sweep"
        );
    }

    #[test]
    fn worktree_restore_clean_keeps_untracked_under_ignored_dir() {
        // The clean sweep must not delete an untracked file that lives *under*
        // an ignored directory, even when that directory is part of the
        // target tree (so it gets recursed into).
        let (_d, store) = fresh_store();
        let target = TempDir::new().unwrap();
        fs::write(target.path().join(".mkitignore"), "dist/\n").unwrap();
        fs::create_dir(target.path().join("dist")).unwrap();
        fs::write(target.path().join("dist/local.tmp"), b"local").unwrap();
        // Target tree: dist/ (tree) holding a tracked app.js.
        let app_blob = put_blob(&store, b"APP");
        let dist_tree = put_tree_with(
            &store,
            vec![TreeEntry {
                name: b"app.js".to_vec(),
                mode: EntryMode::Blob,
                object_hash: app_blob,
            }],
        );
        let root = put_tree_with(
            &store,
            vec![TreeEntry {
                name: b"dist".to_vec(),
                mode: EntryMode::Tree,
                object_hash: dist_tree,
            }],
        );
        restore_tree_to_worktree(&store, &root, target.path(), &RestoreOptions::default()).unwrap();
        // Tracked content materialised...
        assert_eq!(fs::read(target.path().join("dist/app.js")).unwrap(), b"APP");
        // ...and the untracked file under the ignored dir is preserved.
        assert_eq!(
            fs::read(target.path().join("dist/local.tmp")).unwrap(),
            b"local",
            "an untracked file under an ignored dir must survive the clean sweep"
        );
    }

    #[cfg(unix)]
    #[test]
    fn worktree_restore_rejects_escaping_symlink() {
        let (_d, store) = fresh_store();
        let target = TempDir::new().unwrap();
        let bad = put_blob(&store, b"../outside");
        let root = put_tree_with(
            &store,
            vec![TreeEntry {
                name: b"link".to_vec(),
                mode: EntryMode::Symlink,
                object_hash: bad,
            }],
        );
        let err =
            restore_tree_to_worktree(&store, &root, target.path(), &RestoreOptions::default())
                .unwrap_err();
        assert!(matches!(err, RestoreError::InvalidSymlinkTarget(_)));
    }
}