zipatch_rs/apply/

mod.rs

//! Filesystem application of parsed `ZiPatch` chunks.
//!
//! # Parse / apply separation
//!
//! The crate is intentionally split into two independent layers:
//!
//! - **Parsing** (`src/chunk/`) — reads the binary wire format and produces
//!   [`Chunk`] values. Nothing in the parser allocates file handles, stats
//!   paths, or performs I/O against the install tree.
//! - **Applying** (this module) — takes a stream of [`Chunk`] values and
//!   writes the patch changes to disk (or to whatever backing the configured
//!   [`Vfs`](crate::apply::Vfs) provides).
//!
//! The only bridge between the two layers is [`Chunk::apply`], which
//! dispatches each parsed chunk to its filesystem-side logic. Callers that
//! only need to inspect patch contents can use the parser without ever
//! touching this module.
//!
//! # Two-type API: [`ApplyConfig`] and [`ApplySession`]
//!
//! The apply layer is split into two complementary types, each with one job:
//!
//! - [`ApplyConfig`] — the frozen *configuration* of an apply: install root,
//!   target platform, ignore flags, the [`Vfs`](crate::apply::Vfs) backing, observer, and
//!   checkpoint sink. Constructed via [`ApplyConfig::new`] and configured
//!   via `with_*` builder methods. Holds no runtime state and performs no
//!   I/O.
//! - [`ApplySession`] — the *runtime* state of an active apply: open
//!   file-handle cache, memoised directory and `SqPack` path caches, the
//!   reusable DEFLATE decompressor, and per-chunk progress bookkeeping.
//!   Created by consuming an [`ApplyConfig`] via
//!   [`ApplyConfig::into_session`], or implicitly by the high-level
//!   driver entry points.
//!
//! # Pluggable filesystem
//!
//! Every filesystem effect — open, write, truncate, fsync, mkdir, unlink,
//! readdir — is routed through the [`Vfs`](crate::apply::Vfs) trait. The default backing
//! ([`StdFs`](crate::apply::StdFs)) wraps `std::fs`. Swap in
//! [`InMemoryFs`](crate::apply::InMemoryFs) for tests, dry-run
//! previews, or sandboxed/embedded scenarios:
//!
//! ```no_run
//! use zipatch_rs::{ApplyConfig, ZiPatchReader};
//! use zipatch_rs::apply::InMemoryFs;
//!
//! let fs = InMemoryFs::new();
//! let cfg = ApplyConfig::new("/virtual/game").with_vfs(fs.clone());
//! // cfg.apply_patch(reader)?;
//! // fs.snapshot_files() — inspect the resulting in-memory layout.
//! # let _ = (cfg, fs);
//! ```
//!
//! ## Driver entry points
//!
//! The high-level drivers live on [`ApplyConfig`] as consuming methods so
//! the typical "build config, run patch, drop" flow stays a one-liner:
//!
//! ```no_run
//! use std::fs::File;
//! use zipatch_rs::{ApplyConfig, ZiPatchReader};
//!
//! let reader = ZiPatchReader::new(File::open("game.patch").unwrap()).unwrap();
//! ApplyConfig::new("/opt/ffxiv/game").apply_patch(reader).unwrap();
//! ```
//!
//! Callers that want to dispatch individual chunks (or apply several patches
//! back-to-back against the same install) materialise a session explicitly:
//!
//! ```no_run
//! use zipatch_rs::{ApplyConfig, Chunk};
//!
//! let mut session = ApplyConfig::new("/opt/ffxiv/game").into_session();
//! // chunk.apply(&mut session)?;
//! # let _ = &mut session;
//! ```
//!
//! # File-handle cache
//!
//! Every apply path that writes to a `SqPack` file calls an internal
//! `open_cached` method on [`ApplySession`] rather than opening the file
//! directly. The cache transparently returns an existing writable handle or
//! opens a new one via the configured [`Vfs`](crate::apply::Vfs).
//!
//! Cached handles are wrapped in a [`std::io::BufWriter`] with a per-handle
//! buffer (default 64 KiB, see [`DEFAULT_BUFFER_CAPACITY`](crate::apply::DEFAULT_BUFFER_CAPACITY))
//! to coalesce the many small writes the SQPK pipeline emits. Override via
//! [`ApplyConfig::with_buffer_capacity`].
//!
//! The cache is capped at [`DEFAULT_MAX_CACHED_FDS`](crate::apply::DEFAULT_MAX_CACHED_FDS)
//! (256) entries by default; override via [`ApplyConfig::with_max_cached_fds`]. When it is
//! full and a new, uncached path is requested, **all** cached handles are
//! flushed and closed at once before the new one is inserted.
//!
//! # Errors
//!
//! Every [`Chunk::apply`] call returns [`crate::ApplyResult`], which is
//! `Result<(), `[`crate::ApplyError`]`>`. On error, the apply operation
//! aborts at the failing chunk.

mod cancel;
pub mod checkpoint;
mod driver;
pub(crate) mod observer;
pub(crate) mod path;
pub(crate) mod sqpk;
pub mod vfs;

pub use cancel::CancelToken;
pub use checkpoint::{
    Checkpoint, CheckpointPolicy, CheckpointSink, InFlightAddFile, IndexedCheckpoint,
    NoopCheckpointSink, SequentialCheckpoint,
};
pub use observer::{ApplyObserver, ChunkEvent, NoopObserver};
pub use vfs::{InMemoryFs, StdFs, Vfs, VfsMetadata, VfsRead, VfsWrite};

use crate::ApplyResult as Result;
use crate::Platform;
use crate::chunk::Chunk;
use crate::chunk::adir::AddDirectory;
use crate::chunk::aply::{ApplyOption, ApplyOptionKind};
use crate::chunk::ddir::DeleteDirectory;
use std::collections::{HashMap, HashSet};
use std::io::{BufWriter, Seek, SeekFrom, Write};
use std::path::{Path, PathBuf};
use tracing::{trace, warn};

/// Buffered, seekable wrapper around a [`VfsWrite`] handle.
///
/// Stored in [`ApplySession`]'s file-handle cache. The `BufWriter` coalesces
/// the many small writes the SQPK pipeline emits into a smaller number of
/// underlying `write()` calls against the vfs handle.
pub(crate) struct CachedWriter {
    inner: BufWriter<Box<dyn VfsWrite>>,
}

impl std::fmt::Debug for CachedWriter {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        f.debug_struct("CachedWriter").finish_non_exhaustive()
    }
}

impl Write for CachedWriter {
    #[inline]
    fn write(&mut self, buf: &[u8]) -> std::io::Result<usize> {
        self.inner.write(buf)
    }
    #[inline]
    fn write_all(&mut self, buf: &[u8]) -> std::io::Result<()> {
        self.inner.write_all(buf)
    }
    #[inline]
    fn flush(&mut self) -> std::io::Result<()> {
        self.inner.flush()
    }
}

impl Seek for CachedWriter {
    #[inline]
    fn seek(&mut self, pos: SeekFrom) -> std::io::Result<u64> {
        self.inner.seek(pos)
    }
    #[inline]
    fn stream_position(&mut self) -> std::io::Result<u64> {
        self.inner.stream_position()
    }
}

impl CachedWriter {
    /// Truncate the underlying file to length zero.
    pub(crate) fn truncate_to_zero(&mut self) -> std::io::Result<()> {
        self.inner.flush()?;
        self.inner.get_mut().set_len(0)
    }

    /// Force buffered data through and then `sync_all` the underlying handle.
    fn sync_all_inner(&mut self) -> std::io::Result<()> {
        self.inner.flush()?;
        self.inner.get_mut().sync_all()
    }
}

/// Panics if `policy` is `FsyncEveryN(0)`. Called from both
/// [`ApplyConfig::with_checkpoint_sink`] and
/// [`crate::IndexApplier::with_checkpoint_sink`] so the two install points
/// surface the same diagnostic.
pub(crate) fn validate_checkpoint_policy(policy: CheckpointPolicy) {
    assert!(
        !matches!(policy, CheckpointPolicy::FsyncEveryN(0)),
        "CheckpointPolicy::FsyncEveryN(0) is invalid; use CheckpointPolicy::Fsync \
         for an every-record fsync cadence"
    );
}

/// Discriminator for the `path_cache` key: which `SqPack` file kind is being
/// resolved.
#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash)]
pub(crate) enum PathKind {
    Dat,
    Index,
}

/// Cache key for resolved `SqPack` `.dat`/`.index` paths.
#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash)]
pub(crate) struct PathCacheKey {
    pub(crate) main_id: u16,
    pub(crate) sub_id: u16,
    pub(crate) file_id: u32,
    pub(crate) kind: PathKind,
}

/// Default cap on the [`ApplySession`] open-file-handle cache.
///
/// Used by [`ApplyConfig::new`]; override with
/// [`ApplyConfig::with_max_cached_fds`].
pub const DEFAULT_MAX_CACHED_FDS: usize = 256;

/// Default per-handle write buffer capacity (64 KiB) wrapped around every
/// cached [`Vfs`] handle.
///
/// Chosen to comfortably absorb the largest single writes the SQPK pipeline
/// emits. Used by [`ApplyConfig::new`]; override with
/// [`ApplyConfig::with_buffer_capacity`].
pub const DEFAULT_BUFFER_CAPACITY: usize = 64 * 1024;

/// Frozen apply-time configuration: install root, target platform, ignore
/// flags, the [`Vfs`] backing, observer, and checkpoint sink.
///
/// `ApplyConfig` is the *configuration* half of the apply API. It owns
/// everything that should be settled before an apply starts; the *runtime*
/// half — open file handles, scratch buffers, per-chunk progress — lives
/// on [`ApplySession`].
///
/// # Construction
///
/// Build with [`ApplyConfig::new`], then chain the `with_*` builder methods
/// to override defaults:
///
/// ```
/// use zipatch_rs::{ApplyConfig, Platform};
///
/// let cfg = ApplyConfig::new("/opt/ffxiv/game")
///     .with_platform(Platform::Win32)
///     .with_ignore_missing(true);
///
/// assert_eq!(cfg.game_path().to_str().unwrap(), "/opt/ffxiv/game");
/// assert_eq!(cfg.platform(), Platform::Win32);
/// assert!(cfg.ignore_missing());
/// ```
///
/// # Pluggable filesystem
///
/// Defaults to [`StdFs`]. Override with [`ApplyConfig::with_vfs`] to swap in
/// [`InMemoryFs`] (for tests / previews) or any custom [`Vfs`] backing.
///
/// # Running a patch
///
/// The high-level drivers — [`ApplyConfig::apply_patch`] and
/// [`ApplyConfig::resume_apply_patch`] — consume the config, materialise
/// an [`ApplySession`] internally, run the patch end-to-end, and drop the
/// session on completion.
///
/// # Threading
///
/// [`ApplyObserver`], [`CheckpointSink`], and [`Vfs`] all carry `Send + Sync`
/// supertrait bounds, so an `ApplyConfig` can be constructed on one thread
/// and shipped to a worker for the actual apply.
pub struct ApplyConfig {
    game_path: PathBuf,
    platform: Platform,
    ignore_missing: bool,
    ignore_old_mismatch: bool,
    vfs: Box<dyn Vfs>,
    observer: Box<dyn ApplyObserver>,
    checkpoint_sink: Box<dyn CheckpointSink>,
    cancel_token: Option<CancelToken>,
    max_cached_fds: usize,
    buffer_capacity: usize,
}

impl std::fmt::Debug for ApplyConfig {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        f.debug_struct("ApplyConfig")
            .field("game_path", &self.game_path)
            .field("platform", &self.platform)
            .field("ignore_missing", &self.ignore_missing)
            .field("ignore_old_mismatch", &self.ignore_old_mismatch)
            .field("vfs", &"<dyn Vfs>")
            .field("observer", &"<dyn ApplyObserver>")
            .field("checkpoint_sink", &"<dyn CheckpointSink>")
            .field("cancel_token", &self.cancel_token)
            .field("max_cached_fds", &self.max_cached_fds)
            .field("buffer_capacity", &self.buffer_capacity)
            .finish()
    }
}

impl ApplyConfig {
    /// Create a configuration targeting the given game install directory.
    ///
    /// Defaults: platform is [`Platform::Win32`], both ignore-flags are off,
    /// vfs is [`StdFs`], observer is [`NoopObserver`], checkpoint sink is
    /// [`NoopCheckpointSink`].
    pub fn new(game_path: impl Into<PathBuf>) -> Self {
        Self {
            game_path: game_path.into(),
            platform: Platform::Win32,
            ignore_missing: false,
            ignore_old_mismatch: false,
            vfs: Box::new(StdFs::new()),
            observer: Box::new(NoopObserver),
            checkpoint_sink: Box::new(NoopCheckpointSink),
            cancel_token: None,
            max_cached_fds: DEFAULT_MAX_CACHED_FDS,
            buffer_capacity: DEFAULT_BUFFER_CAPACITY,
        }
    }

    /// Returns the configured cap on the open-file-handle cache.
    #[must_use]
    pub fn max_cached_fds(&self) -> usize {
        self.max_cached_fds
    }

    /// Returns the configured per-handle write buffer capacity, in bytes.
    #[must_use]
    pub fn buffer_capacity(&self) -> usize {
        self.buffer_capacity
    }

    /// Returns the game installation directory.
    #[must_use]
    pub fn game_path(&self) -> &Path {
        &self.game_path
    }

    /// Returns the configured target platform.
    #[must_use]
    pub fn platform(&self) -> Platform {
        self.platform
    }

    /// Returns the configured `ignore_missing` flag.
    #[must_use]
    pub fn ignore_missing(&self) -> bool {
        self.ignore_missing
    }

    /// Returns the configured `ignore_old_mismatch` flag.
    #[must_use]
    pub fn ignore_old_mismatch(&self) -> bool {
        self.ignore_old_mismatch
    }

    /// Sets the target platform. Defaults to [`Platform::Win32`].
    ///
    /// A [`crate::chunk::sqpk::SqpkTargetInfo`] chunk encountered during
    /// apply overrides this value on the active [`ApplySession`].
    #[must_use]
    pub fn with_platform(mut self, platform: Platform) -> Self {
        self.platform = platform;
        self
    }

    /// Silently ignore missing files instead of returning an error during apply.
    #[must_use]
    pub fn with_ignore_missing(mut self, v: bool) -> Self {
        self.ignore_missing = v;
        self
    }

    /// Silently ignore old-data mismatches instead of returning an error
    /// during apply.
    #[must_use]
    pub fn with_ignore_old_mismatch(mut self, v: bool) -> Self {
        self.ignore_old_mismatch = v;
        self
    }

    /// Install a [`Vfs`] implementation. Defaults to [`StdFs`].
    ///
    /// Swap in [`InMemoryFs`] for tests, dry-run previews, or sandboxed
    /// environments where the apply must not touch the host filesystem.
    #[must_use]
    pub fn with_vfs(mut self, vfs: impl Vfs + 'static) -> Self {
        self.vfs = Box::new(vfs);
        self
    }

    /// Install an [`ApplyObserver`] for progress reporting and cancellation.
    #[must_use]
    pub fn with_observer(mut self, observer: impl ApplyObserver + 'static) -> Self {
        self.observer = Box::new(observer);
        self
    }

    /// Install a [`CancelToken`] the apply driver polls between chunks (and
    /// inside long-running SQPK `AddFile` block loops) to abort cleanly.
    ///
    /// Hold a clone on whichever thread initiates cancellation (typically a
    /// UI handler) and pass the original here. The driver checks both the
    /// token and any installed [`ApplyObserver::should_cancel`] at every
    /// cancel point; either source aborts the apply with
    /// [`ApplyError::Cancelled`](crate::ApplyError::Cancelled).
    ///
    /// Consumers that only want cancellation do not need to implement
    /// [`ApplyObserver`] at all.
    #[must_use]
    pub fn with_cancel_token(mut self, token: CancelToken) -> Self {
        self.cancel_token = Some(token);
        self
    }

    /// Install a [`CheckpointSink`] to receive apply-time checkpoints.
    ///
    /// # Panics
    ///
    /// Panics if the sink reports [`CheckpointPolicy::FsyncEveryN`] with
    /// `n == 0`.
    #[must_use]
    pub fn with_checkpoint_sink(mut self, sink: impl CheckpointSink + 'static) -> Self {
        validate_checkpoint_policy(sink.policy());
        self.checkpoint_sink = Box::new(sink);
        self
    }

    /// Install a pre-boxed observer. Crate-internal escape hatch for
    /// callers (notably [`crate::IndexApplier`]) that already hold a
    /// `Box<dyn ApplyObserver>`.
    pub(crate) fn set_boxed_observer(&mut self, observer: Box<dyn ApplyObserver>) {
        self.observer = observer;
    }

    /// Install a pre-boxed checkpoint sink. Crate-internal escape hatch.
    pub(crate) fn set_boxed_checkpoint_sink(&mut self, sink: Box<dyn CheckpointSink>) {
        validate_checkpoint_policy(sink.policy());
        self.checkpoint_sink = sink;
    }

    /// Install a pre-boxed vfs. Crate-internal escape hatch.
    pub(crate) fn set_boxed_vfs(&mut self, vfs: Box<dyn Vfs>) {
        self.vfs = vfs;
    }

    /// Install a cancellation token. Crate-internal escape hatch mirroring
    /// the other `set_boxed_*` setters.
    pub(crate) fn set_cancel_token(&mut self, token: CancelToken) {
        self.cancel_token = Some(token);
    }

    /// Set the maximum number of writable file handles cached by the
    /// [`ApplySession`]. When the cache reaches this size and a new path is
    /// requested, every cached handle is flushed and dropped before the new
    /// one is inserted.
    ///
    /// Defaults to [`DEFAULT_MAX_CACHED_FDS`] (256). Lower this for hosts
    /// with tight FD limits; raise it for high-throughput appliers writing
    /// to many distinct `SqPack` files.
    ///
    /// # Panics
    ///
    /// Panics if `n` is zero — a zero-sized cache would force eviction on
    /// every open and is a programming error.
    #[must_use]
    pub fn with_max_cached_fds(mut self, n: usize) -> Self {
        assert!(n > 0, "with_max_cached_fds(0) is invalid");
        self.max_cached_fds = n;
        self
    }

    /// Set the per-handle [`std::io::BufWriter`] capacity, in bytes, wrapped
    /// around every cached [`Vfs`] handle.
    ///
    /// Defaults to [`DEFAULT_BUFFER_CAPACITY`] (64 KiB). Raise it for
    /// high-throughput batch appliers; lower it to reduce per-handle memory
    /// when many handles are cached concurrently.
    ///
    /// # Panics
    ///
    /// Panics if `bytes` is zero — a zero-sized buffer defeats the purpose
    /// of wrapping the handle and is a programming error.
    #[must_use]
    pub fn with_buffer_capacity(mut self, bytes: usize) -> Self {
        assert!(bytes > 0, "with_buffer_capacity(0) is invalid");
        self.buffer_capacity = bytes;
        self
    }

    /// Consume this config and materialise a fresh [`ApplySession`].
    ///
    /// Follows the `into_X` convention: same configuration data, handed off
    /// to the type that owns the per-apply runtime state.
    #[must_use]
    pub fn into_session(self) -> ApplySession {
        ApplySession::new(self)
    }
}

/// Active apply-time runtime state.
///
/// Holds the mutable scratch and bookkeeping a running apply needs: open
/// file-handle cache, memoised directory and path caches, the reusable
/// DEFLATE decompressor, and per-chunk progress counters.
pub struct ApplySession {
    config: ApplyConfig,
    pub(crate) file_cache: HashMap<PathBuf, CachedWriter>,
    pub(crate) dirs_created: HashSet<PathBuf>,
    pub(crate) path_cache: HashMap<PathCacheKey, PathBuf>,
    pub(crate) decompressor: flate2::Decompress,
    pub(crate) checkpoints_since_fsync: u32,
    #[cfg(any(test, feature = "test-utils"))]
    pub(crate) test_flush_count: usize,
    #[cfg(any(test, feature = "test-utils"))]
    pub(crate) test_sync_count: usize,
    pub(crate) current_chunk_index: u64,
    pub(crate) current_chunk_bytes_read: u64,
    pub(crate) patch_name: Option<String>,
    pub(crate) patch_size: Option<u64>,
}

impl std::fmt::Debug for ApplySession {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        let mut s = f.debug_struct("ApplySession");
        s.field("config", &self.config)
            .field("file_cache_len", &self.file_cache.len())
            .field("dirs_created_len", &self.dirs_created.len())
            .field("path_cache_len", &self.path_cache.len())
            .field("decompressor", &"<flate2::Decompress>")
            .field("checkpoints_since_fsync", &self.checkpoints_since_fsync)
            .field("current_chunk_index", &self.current_chunk_index)
            .field("current_chunk_bytes_read", &self.current_chunk_bytes_read)
            .field("patch_name", &self.patch_name)
            .field("patch_size", &self.patch_size);
        #[cfg(any(test, feature = "test-utils"))]
        s.field("test_flush_count", &self.test_flush_count)
            .field("test_sync_count", &self.test_sync_count);
        s.finish()
    }
}

impl ApplySession {
    fn new(config: ApplyConfig) -> Self {
        Self {
            config,
            file_cache: HashMap::new(),
            dirs_created: HashSet::new(),
            path_cache: HashMap::new(),
            decompressor: flate2::Decompress::new(false),
            checkpoints_since_fsync: 0,
            #[cfg(any(test, feature = "test-utils"))]
            test_flush_count: 0,
            #[cfg(any(test, feature = "test-utils"))]
            test_sync_count: 0,
            current_chunk_index: 0,
            current_chunk_bytes_read: 0,
            patch_name: None,
            patch_size: None,
        }
    }

    /// Returns the underlying [`ApplyConfig`].
    #[must_use]
    pub fn config(&self) -> &ApplyConfig {
        &self.config
    }

    /// Returns the game installation directory.
    #[must_use]
    pub fn game_path(&self) -> &Path {
        &self.config.game_path
    }

    /// Returns the current target platform.
    #[must_use]
    pub fn platform(&self) -> Platform {
        self.config.platform
    }

    /// Returns whether missing files are silently ignored.
    #[must_use]
    pub fn ignore_missing(&self) -> bool {
        self.config.ignore_missing
    }

    /// Returns whether old-data mismatches are silently ignored.
    #[must_use]
    pub fn ignore_old_mismatch(&self) -> bool {
        self.config.ignore_old_mismatch
    }

    /// Borrow the configured [`Vfs`] backing.
    pub(crate) fn vfs(&self) -> &dyn Vfs {
        &*self.config.vfs
    }

    /// Test-only: return `(flush_count, sync_count)` recorded since this
    /// session was constructed. Quarantined behind the `test-utils` feature
    /// and not part of the stable API.
    #[cfg(any(test, feature = "test-utils"))]
    #[doc(hidden)]
    #[must_use]
    pub fn test_counters(&self) -> (usize, usize) {
        (self.test_flush_count, self.test_sync_count)
    }

    pub(crate) fn set_platform(&mut self, platform: Platform) {
        self.config.platform = platform;
    }

    pub(crate) fn set_ignore_missing(&mut self, v: bool) {
        self.config.ignore_missing = v;
    }

    pub(crate) fn set_ignore_old_mismatch(&mut self, v: bool) {
        self.config.ignore_old_mismatch = v;
    }

    pub(crate) fn observer_mut(&mut self) -> &mut dyn ApplyObserver {
        &mut *self.config.observer
    }

    /// Returns `true` if cancellation is requested by either the installed
    /// [`CancelToken`] or the observer's
    /// [`ApplyObserver::should_cancel`]. The token is polled first; if it
    /// fires, the observer poll is skipped.
    pub(crate) fn cancel_requested(&mut self) -> bool {
        if let Some(token) = self.config.cancel_token.as_ref() {
            if token.is_cancelled() {
                return true;
            }
        }
        self.config.observer.should_cancel()
    }

    /// Flush every cached `BufWriter`, then `sync_all` the underlying handles.
    pub fn sync_all(&mut self) -> std::io::Result<()> {
        #[cfg(any(test, feature = "test-utils"))]
        {
            self.test_sync_count += 1;
        }
        let mut first_err: Option<std::io::Error> = None;
        for writer in self.file_cache.values_mut() {
            if let Err(e) = writer.sync_all_inner() {
                first_err.get_or_insert(e);
            }
        }
        match first_err {
            Some(e) => Err(e),
            None => Ok(()),
        }
    }

    /// Record a chunk-boundary `checkpoint` to the installed sink.
    pub(crate) fn record_checkpoint(&mut self, checkpoint: &Checkpoint) -> Result<()> {
        self.config.checkpoint_sink.record(checkpoint)?;
        match self.config.checkpoint_sink.policy() {
            CheckpointPolicy::Flush => {
                self.flush()?;
            }
            CheckpointPolicy::Fsync => {
                self.sync_all()?;
                self.checkpoints_since_fsync = 0;
            }
            CheckpointPolicy::FsyncEveryN(n) => {
                debug_assert!(n >= 1, "FsyncEveryN(0) must be rejected at install time");
                self.checkpoints_since_fsync = self.checkpoints_since_fsync.saturating_add(1);
                if self.checkpoints_since_fsync >= n {
                    self.sync_all()?;
                    self.checkpoints_since_fsync = 0;
                } else {
                    self.flush()?;
                }
            }
        }
        Ok(())
    }

    /// Record an in-flight mid-DEFLATE-block `checkpoint`. No flush, no fsync.
    pub(crate) fn record_checkpoint_mid_block(&mut self, checkpoint: &Checkpoint) -> Result<()> {
        self.config.checkpoint_sink.record(checkpoint)?;
        Ok(())
    }

    /// Return a writable handle to `path`, opening it via the configured
    /// [`Vfs`] if not already cached.
    pub(crate) fn open_cached(&mut self, path: &Path) -> std::io::Result<&mut CachedWriter> {
        if self.file_cache.contains_key(path) {
            return Ok(self
                .file_cache
                .get_mut(path)
                .expect("contains_key returned true above"));
        }
        if self.file_cache.len() >= self.config.max_cached_fds {
            self.drain_and_flush()?;
        }
        let handle = self.config.vfs.open_write(path)?;
        let writer = CachedWriter {
            inner: BufWriter::with_capacity(self.config.buffer_capacity, handle),
        };
        Ok(self.file_cache.entry(path.to_path_buf()).or_insert(writer))
    }

    /// Flush and remove the cached handle for `path`, if any.
    pub(crate) fn evict_cached(&mut self, path: &Path) -> std::io::Result<()> {
        if let Some(mut writer) = self.file_cache.remove(path) {
            writer.flush()?;
        }
        Ok(())
    }

    /// Flush and drop every cached file handle.
    pub(crate) fn clear_file_cache(&mut self) -> std::io::Result<()> {
        self.drain_and_flush()
    }

    /// Create `path` and every missing ancestor via the configured [`Vfs`],
    /// memoising the call.
    pub(crate) fn ensure_dir_all(&mut self, path: &Path) -> std::io::Result<()> {
        if self.dirs_created.contains(path) {
            return Ok(());
        }
        self.config.vfs.create_dir_all(path)?;
        self.dirs_created.insert(path.to_path_buf());
        Ok(())
    }

    /// Drop every memoised entry in the created-directories set.
    pub(crate) fn invalidate_dirs_created(&mut self) {
        self.dirs_created.clear();
    }

    /// Drop every memoised entry in the `SqPack` path cache.
    pub(crate) fn invalidate_path_cache(&mut self) {
        self.path_cache.clear();
    }

    fn drain_and_flush(&mut self) -> std::io::Result<()> {
        let mut first_err: Option<std::io::Error> = None;
        for (_, mut writer) in self.file_cache.drain() {
            if let Err(e) = writer.flush() {
                first_err.get_or_insert(e);
            }
        }
        match first_err {
            Some(e) => Err(e),
            None => Ok(()),
        }
    }

    /// Flush every buffered write through to the underlying [`Vfs`] handle.
    pub fn flush(&mut self) -> std::io::Result<()> {
        #[cfg(any(test, feature = "test-utils"))]
        {
            self.test_flush_count += 1;
        }
        let mut first_err: Option<std::io::Error> = None;
        for writer in self.file_cache.values_mut() {
            if let Err(e) = writer.flush() {
                first_err.get_or_insert(e);
            }
        }
        match first_err {
            Some(e) => Err(e),
            None => Ok(()),
        }
    }
}

impl Chunk {
    /// Apply this chunk to `session`.
    pub fn apply(&self, session: &mut ApplySession) -> Result<()> {
        match self {
            Chunk::FileHeader(_) | Chunk::ApplyFreeSpace(_) | Chunk::EndOfFile => Ok(()),
            Chunk::Sqpk(c) => c.apply(session),
            Chunk::ApplyOption(c) => apply_option(c, session),
            Chunk::AddDirectory(c) => apply_add_directory(c, session),
            Chunk::DeleteDirectory(c) => apply_delete_directory(c, session),
        }
    }
}

#[allow(clippy::unnecessary_wraps)]
pub(crate) fn apply_option(opt: &ApplyOption, session: &mut ApplySession) -> Result<()> {
    trace!(kind = ?opt.kind, value = opt.value, "apply option");
    match opt.kind {
        ApplyOptionKind::IgnoreMissing => session.set_ignore_missing(opt.value),
        ApplyOptionKind::IgnoreOldMismatch => session.set_ignore_old_mismatch(opt.value),
    }
    Ok(())
}

pub(crate) fn apply_add_directory(c: &AddDirectory, session: &mut ApplySession) -> Result<()> {
    trace!(name = %c.name, "create directory");
    let path = session.game_path().join(&c.name);
    session.ensure_dir_all(&path)?;
    Ok(())
}

pub(crate) fn apply_delete_directory(
    c: &DeleteDirectory,
    session: &mut ApplySession,
) -> Result<()> {
    let path = session.game_path().join(&c.name);
    match session.vfs().remove_dir(&path) {
        Ok(()) => {
            trace!(name = %c.name, "delete directory");
            session.invalidate_dirs_created();
            Ok(())
        }
        Err(e) if e.kind() == std::io::ErrorKind::NotFound && session.ignore_missing() => {
            warn!(name = %c.name, "delete directory: not found, ignored");
            Ok(())
        }
        Err(e) => Err(e.into()),
    }
}

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

    fn session(path: impl Into<PathBuf>) -> ApplySession {
        ApplyConfig::new(path).into_session()
    }

    // --- Cache semantics ---

    #[test]
    fn cache_eviction_clears_all_entries_when_at_capacity() {
        let tmp = tempfile::tempdir().unwrap();
        let mut s = session(tmp.path());

        for i in 0..DEFAULT_MAX_CACHED_FDS {
            s.open_cached(&tmp.path().join(format!("{i}.dat"))).unwrap();
        }
        assert_eq!(s.file_cache.len(), DEFAULT_MAX_CACHED_FDS);

        s.open_cached(&tmp.path().join("new.dat")).unwrap();
        assert_eq!(s.file_cache.len(), 1);
    }

    #[test]
    fn cache_hit_does_not_trigger_eviction_when_full() {
        let tmp = tempfile::tempdir().unwrap();
        let mut s = session(tmp.path());

        for i in 0..DEFAULT_MAX_CACHED_FDS {
            s.open_cached(&tmp.path().join(format!("{i}.dat"))).unwrap();
        }
        s.open_cached(&tmp.path().join("0.dat")).unwrap();
        assert_eq!(s.file_cache.len(), DEFAULT_MAX_CACHED_FDS);
    }

    #[test]
    fn evict_cached_removes_only_target_path() {
        let tmp = tempfile::tempdir().unwrap();
        let mut s = session(tmp.path());
        let a = tmp.path().join("a.dat");
        let b = tmp.path().join("b.dat");
        s.open_cached(&a).unwrap();
        s.open_cached(&b).unwrap();
        assert_eq!(s.file_cache.len(), 2);

        s.evict_cached(&a).unwrap();
        assert_eq!(s.file_cache.len(), 1);
        assert!(s.file_cache.contains_key(&b));
    }

    #[test]
    fn evict_cached_is_noop_for_absent_path() {
        let tmp = tempfile::tempdir().unwrap();
        let mut s = session(tmp.path());
        s.open_cached(&tmp.path().join("a.dat")).unwrap();
        s.evict_cached(&tmp.path().join("nonexistent.dat")).unwrap();
        assert_eq!(s.file_cache.len(), 1);
    }

    #[test]
    fn clear_file_cache_removes_all_handles() {
        let tmp = tempfile::tempdir().unwrap();
        let mut s = session(tmp.path());
        s.open_cached(&tmp.path().join("a.dat")).unwrap();
        s.open_cached(&tmp.path().join("b.dat")).unwrap();
        s.clear_file_cache().unwrap();
        assert_eq!(s.file_cache.len(), 0);
    }

    // --- Tuning-knob builders ---

    #[test]
    fn default_max_cached_fds_matches_constant() {
        let cfg = ApplyConfig::new("/irrelevant");
        assert_eq!(cfg.max_cached_fds(), DEFAULT_MAX_CACHED_FDS);
    }

    #[test]
    fn default_buffer_capacity_matches_constant() {
        let cfg = ApplyConfig::new("/irrelevant");
        assert_eq!(cfg.buffer_capacity(), DEFAULT_BUFFER_CAPACITY);
    }

    #[test]
    fn with_max_cached_fds_overrides_default() {
        let cfg = ApplyConfig::new("/irrelevant").with_max_cached_fds(16);
        assert_eq!(cfg.max_cached_fds(), 16);
    }

    #[test]
    fn with_buffer_capacity_overrides_default() {
        let cfg = ApplyConfig::new("/irrelevant").with_buffer_capacity(1 << 20);
        assert_eq!(cfg.buffer_capacity(), 1 << 20);
    }

    #[test]
    #[should_panic(expected = "with_max_cached_fds(0) is invalid")]
    fn with_max_cached_fds_zero_panics() {
        let _ = ApplyConfig::new("/irrelevant").with_max_cached_fds(0);
    }

    #[test]
    #[should_panic(expected = "with_buffer_capacity(0) is invalid")]
    fn with_buffer_capacity_zero_panics() {
        let _ = ApplyConfig::new("/irrelevant").with_buffer_capacity(0);
    }

    #[test]
    fn custom_max_cached_fds_changes_eviction_threshold() {
        let tmp = tempfile::tempdir().unwrap();
        let cfg = ApplyConfig::new(tmp.path()).with_max_cached_fds(4);
        let mut s = cfg.into_session();
        for i in 0..4 {
            s.open_cached(&tmp.path().join(format!("{i}.dat"))).unwrap();
        }
        assert_eq!(s.file_cache.len(), 4);
        s.open_cached(&tmp.path().join("new.dat")).unwrap();
        // Cache was drained on the 5th distinct open; only the new entry remains.
        assert_eq!(s.file_cache.len(), 1);
    }

    // --- Builder accessors ---

    #[test]
    fn game_path_returns_install_root_unchanged() {
        let tmp = tempfile::tempdir().unwrap();
        let cfg = ApplyConfig::new(tmp.path());
        assert_eq!(cfg.game_path(), tmp.path());
    }

    #[test]
    fn default_platform_is_win32() {
        let cfg = ApplyConfig::new("/irrelevant");
        assert_eq!(cfg.platform(), Platform::Win32);
    }

    #[test]
    fn with_platform_overrides_default() {
        let cfg = ApplyConfig::new("/irrelevant").with_platform(Platform::Ps4);
        assert_eq!(cfg.platform(), Platform::Ps4);
    }

    #[test]
    fn default_ignore_missing_is_false() {
        let cfg = ApplyConfig::new("/irrelevant");
        assert!(!cfg.ignore_missing());
    }

    #[test]
    fn with_ignore_missing_toggles_flag_both_ways() {
        let cfg = ApplyConfig::new("/irrelevant").with_ignore_missing(true);
        assert!(cfg.ignore_missing());
        let cfg = cfg.with_ignore_missing(false);
        assert!(!cfg.ignore_missing());
    }

    #[test]
    fn default_ignore_old_mismatch_is_false() {
        let cfg = ApplyConfig::new("/irrelevant");
        assert!(!cfg.ignore_old_mismatch());
    }

    #[test]
    fn with_ignore_old_mismatch_toggles_flag_both_ways() {
        let cfg = ApplyConfig::new("/irrelevant").with_ignore_old_mismatch(true);
        assert!(cfg.ignore_old_mismatch());
        let cfg = cfg.with_ignore_old_mismatch(false);
        assert!(!cfg.ignore_old_mismatch());
    }

    // --- BufWriter cache ---

    #[test]
    fn buffered_writes_are_invisible_before_flush() {
        use std::io::Write;

        let tmp = tempfile::tempdir().unwrap();
        let mut s = session(tmp.path());
        let path = tmp.path().join("buffered.dat");

        let writer = s.open_cached(&path).unwrap();
        writer.write_all(&[0xAB]).unwrap();

        assert_eq!(std::fs::metadata(&path).unwrap().len(), 0);

        s.flush().unwrap();
        assert_eq!(std::fs::read(&path).unwrap(), vec![0xAB]);
    }

    #[test]
    fn flush_keeps_handles_in_cache() {
        let tmp = tempfile::tempdir().unwrap();
        let mut s = session(tmp.path());
        s.open_cached(&tmp.path().join("a.dat")).unwrap();
        s.open_cached(&tmp.path().join("b.dat")).unwrap();
        s.flush().unwrap();
        assert_eq!(s.file_cache.len(), 2);
    }

    #[test]
    fn evict_cached_flushes_pending_writes_to_disk() {
        use std::io::Write;

        let tmp = tempfile::tempdir().unwrap();
        let mut s = session(tmp.path());
        let path = tmp.path().join("evict.dat");

        let writer = s.open_cached(&path).unwrap();
        writer.write_all(b"queued").unwrap();
        assert_eq!(std::fs::metadata(&path).unwrap().len(), 0);

        s.evict_cached(&path).unwrap();
        assert_eq!(std::fs::read(&path).unwrap(), b"queued");
        assert!(!s.file_cache.contains_key(&path));
    }

    #[test]
    fn clear_file_cache_flushes_every_pending_write() {
        use std::io::Write;

        let tmp = tempfile::tempdir().unwrap();
        let mut s = session(tmp.path());
        let a = tmp.path().join("a.dat");
        let b = tmp.path().join("b.dat");

        s.open_cached(&a).unwrap().write_all(b"AA").unwrap();
        s.open_cached(&b).unwrap().write_all(b"BB").unwrap();

        s.clear_file_cache().unwrap();

        assert_eq!(std::fs::read(&a).unwrap(), b"AA");
        assert_eq!(std::fs::read(&b).unwrap(), b"BB");
        assert!(s.file_cache.is_empty());
    }

    // --- Debug impl ---

    #[test]
    fn apply_session_debug_renders_all_fields() {
        let tmp = tempfile::tempdir().unwrap();
        let s = ApplyConfig::new(tmp.path())
            .with_platform(Platform::Ps4)
            .with_ignore_missing(true)
            .into_session();

        let rendered = format!("{s:?}");
        for needle in [
            "ApplySession",
            "ApplyConfig",
            "game_path",
            "platform",
            "Ps4",
            "ignore_missing",
            "true",
            "ignore_old_mismatch",
            "file_cache_len",
            "path_cache_len",
            "decompressor",
        ] {
            assert!(
                rendered.contains(needle),
                "Debug output must mention {needle:?}; got: {rendered}"
            );
        }
    }

    // --- DeleteDirectory happy path ---

    #[test]
    fn delete_directory_success_removes_existing_dir() {
        let tmp = tempfile::tempdir().unwrap();
        let target = tmp.path().join("to_remove");
        std::fs::create_dir(&target).unwrap();

        let mut s = session(tmp.path());
        apply_delete_directory(
            &DeleteDirectory {
                name: "to_remove".into(),
            },
            &mut s,
        )
        .expect("delete on an existing directory must succeed");

        assert!(!target.exists());
    }

    // --- ensure_dir_all cache-hit branch ---

    #[test]
    fn ensure_dir_all_cache_hit_returns_early_without_syscall() {
        let tmp = tempfile::tempdir().unwrap();
        let mut s = session(tmp.path());

        let path = tmp.path().join("cached_dir");
        s.ensure_dir_all(&path).unwrap();
        assert!(path.is_dir());
        assert_eq!(s.dirs_created.len(), 1);

        let p2 = tmp.path().join("cached_dir");
        s.ensure_dir_all(&p2).unwrap();
        assert_eq!(s.dirs_created.len(), 1);
    }

    // --- InMemoryFs end-to-end ---

    #[test]
    fn in_memory_fs_records_directory_creation() {
        let fs = InMemoryFs::new();
        let mut s = ApplyConfig::new("/g").with_vfs(fs.clone()).into_session();
        apply_add_directory(&AddDirectory { name: "sub".into() }, &mut s).unwrap();
        assert!(fs.snapshot_dirs().contains(&PathBuf::from("/g/sub")));
    }
}