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.
//!
//! The only bridge between the two layers is the [`Apply`] trait, which every
//! chunk type implements. Callers that only need to inspect patch contents can
//! use the parser without ever touching this module.
//!
//! # `ApplyContext`
//!
//! [`ApplyContext`] holds all mutable apply-time state:
//!
//! - **Install root** — the absolute path to the game installation directory.
//!   All `SqPack` paths (`sqpack/<expansion>/...`) are resolved relative to
//!   this root by the internal path submodule.
//! - **Target platform** — selects the `win32`/`ps3`/`ps4` subfolder suffix
//!   used in `SqPack` file paths. Defaults to [`Platform::Win32`] and can be
//!   overridden either at construction time with [`ApplyContext::with_platform`]
//!   or at apply time when a [`crate::chunk::sqpk::SqpkTargetInfo`] chunk is
//!   encountered.
//! - **Ignore flags** — control whether missing files and old-data mismatches
//!   produce errors or logged warnings. `SqpkTargetInfo` chunks set these via
//!   the stream; callers can also pre-configure them.
//! - **File-handle cache** — a bounded map of open file handles. Because a
//!   typical patch applies dozens of chunks to the same `.dat` file,
//!   re-opening that file for every chunk would be wasteful. The cache avoids
//!   this while bounding the number of simultaneously open file descriptors.
//!   See the [cache section](#file-handle-cache) below.
//!
//! # File-handle cache
//!
//! Every `Apply` impl that writes to a `SqPack` file calls an internal
//! `open_cached` method on `ApplyContext` rather than opening the file
//! directly. The cache transparently returns an existing writable handle or
//! opens a new one (with `write=true, create=true, truncate=false`).
//!
//! Cached handles are wrapped in a [`std::io::BufWriter`] with a 64 KiB
//! buffer to coalesce the many small writes the SQPK pipeline emits — block
//! headers, zero-fill runs, decompressed DEFLATE block output — into a
//! smaller number of `write(2)` syscalls. Apply functions interact with the
//! buffered writer transparently because `BufWriter` implements both `Write`
//! and `Seek`. Call [`ApplyContext::flush`] to force buffered data through
//! to the operating system at a checkpoint of your choosing;
//! [`ZiPatchReader::apply_to`](crate::ZiPatchReader::apply_to) calls it
//! automatically before returning.
//!
//! The cache is capped at 256 entries. 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. This is a simple eviction strategy — it
//! trades some re-open overhead at eviction boundaries for bounded
//! file-descriptor usage. Memory cost at the cap is 256 × 64 KiB = 16 MiB.
//!
//! Callers should not rely on cached handles persisting across arbitrary
//! chunks. In particular, [`crate::chunk::sqpk::SqpkFile`]'s `RemoveAll`
//! operation flushes all cached handles before bulk-deleting files to ensure
//! no open handles survive into the deletion window (which matters on
//! Windows). Similarly, `DeleteFile` evicts the cached handle for the
//! specific path being removed.
//!
//! # Ordering and idempotency
//!
//! Chunks **must** be applied in stream order. The `ZiPatch` format is a
//! sequential log, not a random-access manifest: later chunks may depend on
//! filesystem state produced by earlier ones (e.g. an `AddFile` that writes
//! blocks into a file created by an earlier `MakeDirTree` or `AddDirectory`).
//!
//! Apply operations are **not idempotent** in general. Seeking to an offset
//! and writing data is idempotent if the same data is written, but
//! `RemoveAll` is destructive and `DeleteFile` can fail if the file is
//! already gone (unless `ignore_missing` is set). Partial application
//! followed by a retry requires careful state tracking at a higher level;
//! this crate does not provide transactional semantics.
//!
//! # Errors
//!
//! Every [`Apply::apply`] call returns [`crate::Result`], which is
//! `Result<(), `[`crate::ZiPatchError`]`>`. Errors propagate from:
//!
//! - `std::io::Error` — filesystem failures (permissions, missing parent
//!   directories, disk full, etc.) wrapped as [`crate::ZiPatchError::Io`].
//! - [`crate::ZiPatchError::NegativeFileOffset`] — a `SqpkFile` chunk
//!   carried a negative `file_offset` that cannot be converted to a seek
//!   position.
//!
//! On error, the apply operation aborts at the failing chunk. Any changes
//! already applied to the filesystem are **not** rolled back.
//!
//! # Progress and cancellation
//!
//! Install an [`ApplyObserver`] via [`ApplyContext::with_observer`] to be
//! notified after each top-level chunk applies and to signal cancellation
//! mid-stream. The observer's
//! [`on_chunk_applied`](ApplyObserver::on_chunk_applied) method receives a
//! [`ChunkEvent`] with the chunk index, 4-byte tag, and running byte count;
//! its [`should_cancel`](ApplyObserver::should_cancel) predicate is polled
//! between blocks inside long-running chunks so that aborting a multi-
//! hundred-MB `SqpkFile` `AddFile` does not have to wait for the whole
//! chunk to finish. Without an explicit observer, [`ApplyContext`] uses
//! [`NoopObserver`] and the existing apply path pays nothing.
//!
//! # Example
//!
//! ```no_run
//! use std::fs::File;
//! use zipatch_rs::{ApplyContext, ZiPatchReader};
//!
//! let patch_file = File::open("game.patch").unwrap();
//! let mut ctx = ApplyContext::new("/opt/ffxiv/game");
//!
//! ZiPatchReader::new(patch_file)
//!     .unwrap()
//!     .apply_to(&mut ctx)
//!     .unwrap();
//! ```

pub mod checkpoint;
pub(crate) mod observer;
pub(crate) mod path;
pub(crate) mod sqpk;

pub use checkpoint::{
    Checkpoint, CheckpointPolicy, CheckpointSink, InFlightAddFile, IndexedCheckpoint,
    NoopCheckpointSink, SequentialCheckpoint,
};
pub use observer::{ApplyObserver, ChunkEvent, NoopObserver};

use crate::Platform;
use crate::Result;
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::fs::{File, OpenOptions};
use std::io::{BufWriter, Write};
use std::path::{Path, PathBuf};
use tracing::{trace, warn};

/// Panics if `policy` is `FsyncEveryN(0)`. Called from both
/// [`ApplyContext::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. The combination `(main_id, sub_id, file_id, kind)` is the full
/// cache key, alongside the current `platform` (handled via cache
/// invalidation rather than as a key component, since `platform` changes
/// only at `apply_target_info` boundaries).
#[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,
}

const MAX_CACHED_FDS: usize = 256;

// 64 KiB chosen to comfortably absorb the largest single writes the SQPK
// pipeline emits (1 KiB header chunks, DEFLATE block outputs up to ~32 KiB,
// zero-fill runs from `write_zeros`) without splitting them. Memory ceiling
// at the FD cap is 256 * 64 KiB = 16 MiB, which is trivial for a desktop
// launcher and the only realistic consumer of this library.
const WRITE_BUFFER_CAPACITY: usize = 64 * 1024;

/// Apply-time state: install root, target platform, flag toggles, and the
/// internal file-handle cache used by SQPK writers.
///
/// # Construction
///
/// Build with [`ApplyContext::new`], then chain the `with_*` builder methods
/// to override defaults:
///
/// ```
/// use zipatch_rs::{ApplyContext, Platform};
///
/// let ctx = ApplyContext::new("/opt/ffxiv/game")
///     .with_platform(Platform::Win32)
///     .with_ignore_missing(true);
///
/// assert_eq!(ctx.game_path().to_str().unwrap(), "/opt/ffxiv/game");
/// assert_eq!(ctx.platform(), Platform::Win32);
/// assert!(ctx.ignore_missing());
/// ```
///
/// # Platform mutation
///
/// The platform defaults to [`Platform::Win32`]. If the patch stream contains
/// a [`crate::chunk::sqpk::SqpkTargetInfo`] chunk, applying it overwrites
/// [`ApplyContext::platform`] with the platform declared in the chunk. This is
/// the normal case: real FFXIV patches begin with a `TargetInfo` chunk that
/// pins the platform, so the default is rarely used in practice.
///
/// Set the platform explicitly with [`ApplyContext::with_platform`] when you
/// know the target in advance or are processing a synthetic patch.
///
/// # Flag mutation
///
/// [`ApplyContext::ignore_missing`] and [`ApplyContext::ignore_old_mismatch`]
/// can also be overwritten mid-stream by `ApplyOption` chunks embedded in the
/// patch file. Set initial values with the `with_ignore_*` builder methods.
///
/// # File-handle cache
///
/// Internally, `ApplyContext` maintains a bounded map of open file handles
/// keyed by absolute path. The cache is an optimisation: a patch that writes
/// many chunks into the same `.dat` file re-uses a single handle rather than
/// opening and closing the file for every write.
///
/// The cache is capped at 256 entries. When that limit is reached and a new
/// path is needed, **all** entries are evicted at once. Handles are also
/// evicted explicitly before deleting a file (see `DeleteFile`) and before a
/// `RemoveAll` bulk operation.
pub struct ApplyContext {
    pub(crate) game_path: PathBuf,
    /// The target platform. Defaults to `Win32`. Note: `SqpkTargetInfo` chunks
    /// in the patch stream will override this value when applied.
    pub(crate) platform: Platform,
    pub(crate) ignore_missing: bool,
    pub(crate) ignore_old_mismatch: bool,
    // Capped at MAX_CACHED_FDS entries; cleared wholesale when full to bound open FD count.
    // Each handle is wrapped in a `BufWriter` to coalesce the many small
    // writes the SQPK pipeline emits (block headers, zero-fill runs) into a
    // smaller number of `write(2)` syscalls. `BufWriter` implements both
    // `Write` and `Seek`, so apply functions interact with it transparently;
    // operations that need the raw `File` (currently only `set_len`) call
    // `get_mut()` after an explicit `flush()`.
    //
    // `pub(crate)` so the SQPK `AddFile` apply loop can split-borrow it next
    // to `observer` for between-block cancellation polling.
    pub(crate) file_cache: HashMap<PathBuf, BufWriter<File>>,
    // Memoised set of directories already created via `ensure_dir_all`.
    // Avoids reissuing `mkdir -p` syscalls for shared parent directories
    // across long chains of `AddFile` / `MakeDirTree` / `ADIR` chunks. Keys
    // are the exact `PathBuf` handed to `create_dir_all`; no canonicalisation
    // is performed (which would itself cost a syscall and was never done by
    // the previous unconditional path). Destructive ops that remove
    // directories (`SqpkFile::RemoveAll`, `DeleteDirectory`) clear the entire
    // set — simpler than tracking exact entries, and correct because a
    // subsequent `create_dir_all` will re-establish whichever directories
    // are still needed at the next syscall cost.
    pub(crate) dirs_created: HashSet<PathBuf>,
    // Memoised SqPack `.dat`/`.index` path resolutions. A typical patch
    // dispatches thousands of `AddData`/`DeleteData`/`ExpandData`/`Header`
    // chunks targeting a small set of files, each of which would otherwise
    // recompute `expansion_folder_id` (one `String` alloc), the filename
    // (one `format!` alloc), and three chained `PathBuf::join` calls. The
    // cache short-circuits repeat lookups for the same
    // `(main_id, sub_id, file_id, kind)` tuple to a single `PathBuf` clone.
    // Invalidated by `apply_target_info` when the platform changes, since
    // the platform string is baked into the cached path.
    pub(crate) path_cache: HashMap<PathCacheKey, PathBuf>,
    // Reusable DEFLATE state shared across all `SqpkCompressedBlock` payloads
    // in a single `SqpkFile::AddFile` chunk (and across chunks). Constructing
    // a fresh decoder per block allocates ~100 KiB of internal zlib state;
    // a multi-block `AddFile` chunk can contain hundreds of blocks. Reset
    // between blocks via `Decompress::reset(false)` so the underlying
    // buffers are reused. `false` = raw DEFLATE, no zlib wrapper, matching
    // the SqPack on-the-wire layout.
    pub(crate) decompressor: flate2::Decompress,
    // Observer for progress/cancellation. Defaults to a no-op; replaced by
    // `with_observer`. Stored as a boxed trait object so that the public
    // `ApplyContext` type stays non-generic and remains nameable in downstream
    // signatures (`gaveloc-patcher` passes contexts across module boundaries).
    pub(crate) observer: Box<dyn ApplyObserver>,
    // Checkpoint sink installed via `with_checkpoint_sink`. Defaults to
    // `NoopCheckpointSink` so consumers that never opt in pay nothing (one
    // virtual call per emission site that immediately returns `Ok(())`).
    pub(crate) checkpoint_sink: Box<dyn CheckpointSink>,
    // Records-since-last-fsync counter used by `FsyncEveryN`. Reset to zero
    // whenever an fsync actually runs (either because the cadence fired or
    // because the policy was `Fsync`).
    pub(crate) checkpoints_since_fsync: u32,
    /// Test-only flush counter. Incremented each time [`Self::flush`] is called.
    ///
    /// Present only under the `test-utils` feature so integration tests can
    /// assert the exact flush cadence under each [`CheckpointPolicy`] without
    /// patching production code. Not part of the stable public API.
    #[cfg(any(test, feature = "test-utils"))]
    pub test_flush_count: usize,
    /// Test-only `sync_all` counter. Incremented each time [`Self::sync_all`] is called.
    ///
    /// Present only under the `test-utils` feature. See [`Self::test_flush_count`].
    #[cfg(any(test, feature = "test-utils"))]
    pub test_sync_count: usize,
    // Sequential apply driver's per-chunk progress, set by `apply_to` before
    // dispatching each `Chunk::apply` call so deep emission sites (e.g. the
    // per-DEFLATE-block emission inside `SqpkFile::AddFile`) can carry the
    // stream-relative chunk index and byte position without threading them
    // through every trait signature.
    pub(crate) current_chunk_index: u64,
    pub(crate) current_chunk_bytes_read: u64,
    // Identifying metadata for the patch stream currently being applied.
    // Populated by the sequential driver (`apply_to` / `resume_apply_to`)
    // before dispatching the first chunk so deep emission sites can stamp
    // each `SequentialCheckpoint` with the same identity the consumer
    // persisted. `patch_size` is only set when the driver has a `Seek`
    // source (i.e. `resume_apply_to`); the plain `apply_to` path leaves it
    // as `None`.
    pub(crate) patch_name: Option<String>,
    pub(crate) patch_size: Option<u64>,
}

// Hand-written so we don't need `ApplyObserver: Debug` as a supertrait —
// adding supertraits is a SemVer break, and forcing every user observer to
// implement `Debug` would be a needless ergonomic tax. The observer field
// is summarised as an opaque placeholder.
impl std::fmt::Debug for ApplyContext {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        let mut s = f.debug_struct("ApplyContext");
        s.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("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("observer", &"<dyn ApplyObserver>")
            .field("checkpoint_sink", &"<dyn CheckpointSink>")
            .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 ApplyContext {
    /// Create a context targeting the given game install directory.
    ///
    /// Defaults: platform is [`Platform::Win32`], both ignore-flags are off.
    ///
    /// Use the `with_*` builder methods to change these defaults before
    /// applying the first chunk.
    ///
    /// # Example
    ///
    /// ```
    /// use zipatch_rs::ApplyContext;
    ///
    /// let ctx = ApplyContext::new("/opt/ffxiv/game");
    /// assert_eq!(ctx.game_path().to_str().unwrap(), "/opt/ffxiv/game");
    /// ```
    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,
            file_cache: HashMap::new(),
            dirs_created: HashSet::new(),
            path_cache: HashMap::new(),
            // `false` = raw DEFLATE (no zlib header). SqPack `AddFile` blocks
            // store an RFC 1951 raw deflate stream with no wrapper.
            decompressor: flate2::Decompress::new(false),
            observer: Box::new(NoopObserver),
            checkpoint_sink: Box::new(NoopCheckpointSink),
            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 game installation directory.
    ///
    /// All file paths produced during apply are relative to this root.
    #[must_use]
    pub fn game_path(&self) -> &std::path::Path {
        &self.game_path
    }

    /// Returns the current target platform.
    ///
    /// This value may change during apply if the patch stream contains a
    /// [`crate::chunk::sqpk::SqpkTargetInfo`] chunk.
    #[must_use]
    pub fn platform(&self) -> Platform {
        self.platform
    }

    /// Returns whether missing files are silently ignored during apply.
    ///
    /// When `true`, operations that target a file that does not exist log a
    /// warning instead of returning an error. This flag may be overwritten
    /// mid-stream by an `ApplyOption` chunk.
    #[must_use]
    pub fn ignore_missing(&self) -> bool {
        self.ignore_missing
    }

    /// Returns whether old-data mismatches are silently ignored during apply.
    ///
    /// When `true`, apply operations that detect a checksum or data mismatch
    /// against the existing on-disk content proceed without error. This flag
    /// may be overwritten mid-stream by an `ApplyOption` chunk.
    #[must_use]
    pub fn ignore_old_mismatch(&self) -> bool {
        self.ignore_old_mismatch
    }

    /// Sets the target platform. Defaults to [`Platform::Win32`].
    ///
    /// The platform determines the directory suffix used when resolving `SqPack`
    /// file paths (`win32`, `ps3`, or `ps4`).
    ///
    /// Note: a [`crate::chunk::sqpk::SqpkTargetInfo`] chunk encountered during
    /// apply will override this value.
    #[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.
    ///
    /// When `false` (the default), any apply operation that cannot find its
    /// target file returns [`crate::ZiPatchError::Io`] with kind
    /// [`std::io::ErrorKind::NotFound`].
    ///
    /// When `true`, those failures are demoted to `warn!`-level tracing events.
    #[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.
    ///
    /// When `false` (the default), an apply operation that detects that the
    /// on-disk data does not match the expected "before" state returns an error.
    ///
    /// When `true`, the mismatch is logged at `warn!` level and the operation
    /// continues.
    #[must_use]
    pub fn with_ignore_old_mismatch(mut self, v: bool) -> Self {
        self.ignore_old_mismatch = v;
        self
    }

    /// Install an [`ApplyObserver`] for progress reporting and cancellation.
    ///
    /// The observer's [`on_chunk_applied`](ApplyObserver::on_chunk_applied)
    /// method is called after each top-level chunk; its
    /// [`should_cancel`](ApplyObserver::should_cancel) method is polled
    /// inside long-running chunks (currently the
    /// [`SqpkFile`](crate::chunk::sqpk::SqpkFile) block-write loop) so that
    /// cancellation is observable mid-chunk on multi-hundred-MB payloads.
    ///
    /// Returning [`ControlFlow::Break`](std::ops::ControlFlow::Break) from
    /// `on_chunk_applied`, or `true` from `should_cancel`, aborts the apply
    /// loop with [`crate::ZiPatchError::Cancelled`]. Filesystem changes already
    /// applied are **not** rolled back.
    ///
    /// The default observer is a no-op: parsing-only consumers and existing
    /// callers that never call `with_observer` pay nothing.
    ///
    /// # `'static` bound
    ///
    /// The `'static` bound follows from [`ApplyContext`] storing the
    /// observer in a `Box<dyn ApplyObserver>` — a trait object whose
    /// lifetime parameter defaults to `'static`. To pass an observer that
    /// holds a channel sender or similar handle, wrap it in
    /// `Arc<Mutex<...>>` (which is `'static`) or implement
    /// [`ApplyObserver`] on a struct that owns the handle directly.
    #[must_use]
    pub fn with_observer(mut self, observer: impl ApplyObserver + 'static) -> Self {
        self.observer = Box::new(observer);
        self
    }

    /// Install a [`CheckpointSink`] to receive apply-time checkpoints.
    ///
    /// The driver hands the sink a [`Checkpoint`] at each natural recovery
    /// boundary — after every top-level chunk in the sequential
    /// [`apply_to`](crate::ZiPatchReader::apply_to) loop, after every DEFLATE
    /// block inside a [`SqpkFile`](crate::chunk::sqpk::SqpkFile) `AddFile`,
    /// and at the per-target / every-64-regions cadence used by
    /// [`crate::index::IndexApplier::execute`]. The sink's
    /// [`CheckpointPolicy`] then decides whether the driver also flushes the
    /// file-handle cache or escalates to a full `fsync` via
    /// [`Self::sync_all`].
    ///
    /// Default is [`NoopCheckpointSink`]: consumers that never call this
    /// method pay nothing.
    ///
    /// # `'static` bound
    ///
    /// Mirrors [`Self::with_observer`]: the sink is boxed internally into a
    /// `Box<dyn CheckpointSink>` whose lifetime parameter defaults to
    /// `'static`. Wrap in `Arc<Mutex<...>>` or implement [`CheckpointSink`]
    /// on a struct that owns the handle directly if you need to share the
    /// sink across threads.
    ///
    /// # Panics
    ///
    /// Panics if the sink reports
    /// [`CheckpointPolicy::FsyncEveryN`] with `n == 0`. A zero cadence is
    /// programmer error — either the consumer meant
    /// [`CheckpointPolicy::Fsync`] (fsync every record) or the constant was
    /// computed from a runtime value that wasn't validated. Surfacing the
    /// check at install time keeps the failure adjacent to the bug rather
    /// than deep inside the apply loop where the cadence first matters.
    #[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
    }

    /// Flush every cached `BufWriter`, then `fsync` the underlying file
    /// handles.
    ///
    /// Stronger durability guarantee than [`Self::flush`]: a successful
    /// return implies all writes have not only reached the OS but been
    /// committed to durable storage (modulo what the underlying filesystem
    /// guarantees about `fsync`). Used by the apply drivers when an
    /// installed [`CheckpointSink`] requests
    /// [`CheckpointPolicy::Fsync`] or hits the
    /// [`CheckpointPolicy::FsyncEveryN`] cadence.
    ///
    /// Handles are retained in the cache after `sync_all`; subsequent
    /// writes reuse the existing buffered writer. Surfacing the first
    /// `std::io::Error` from any of the flushes or syncs aborts further
    /// processing; the rest of the cache is still attempted so that any
    /// other failing handle does not silently swallow its error.
    ///
    /// # Errors
    ///
    /// Returns the first `std::io::Error` produced by any writer's flush or
    /// any underlying handle's `sync_all`.
    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.flush() {
                first_err.get_or_insert(e);
                continue;
            }
            if let Err(e) = writer.get_ref().sync_all() {
                first_err.get_or_insert(e);
            }
        }
        match first_err {
            Some(e) => Err(e),
            None => Ok(()),
        }
    }

    /// Record a chunk-boundary `checkpoint` to the installed sink, then
    /// honour the sink's policy. Used by the apply drivers at every
    /// per-chunk / per-target / every-64-regions emission site.
    ///
    /// Errors from the sink are wrapped as [`crate::ZiPatchError::Io`]; a
    /// flush or fsync failure escalates with the same vocabulary the
    /// sequential driver already uses for its post-loop flush.
    ///
    /// Mid-DEFLATE-block emissions inside `SqpkFile::AddFile` must use
    /// [`Self::record_checkpoint_mid_block`] instead — those emissions are
    /// too frequent to interleave with a sync syscall, and the driver
    /// guarantees the next chunk-boundary checkpoint flushes the
    /// `BufWriter` bytes the mid-block emissions accumulated.
    pub(crate) fn record_checkpoint(&mut self, checkpoint: &Checkpoint) -> Result<()> {
        self.checkpoint_sink.record(checkpoint)?;
        match self.checkpoint_sink.policy() {
            CheckpointPolicy::Flush => {
                self.flush()?;
            }
            CheckpointPolicy::Fsync => {
                self.sync_all()?;
                self.checkpoints_since_fsync = 0;
            }
            CheckpointPolicy::FsyncEveryN(n) => {
                // `with_checkpoint_sink` rejects `FsyncEveryN(0)` at install
                // time, so `n` is always >= 1 by the time the cadence runs.
                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` to the installed
    /// sink. No flush, no fsync regardless of policy: per-block emissions
    /// inside an `AddFile` loop fire often enough that honouring the
    /// `Fsync` / `FsyncEveryN` policy here would gut throughput on
    /// multi-GB files. The next chunk-boundary
    /// [`Self::record_checkpoint`] flushes everything the mid-block run
    /// accumulated in the `BufWriter`.
    pub(crate) fn record_checkpoint_mid_block(&mut self, checkpoint: &Checkpoint) -> Result<()> {
        self.checkpoint_sink.record(checkpoint)?;
        Ok(())
    }

    /// Return a writable handle to `path`, opening it if not already cached.
    ///
    /// If the cache has reached 256 entries and `path` is not already present,
    /// all cached handles are flushed and dropped before opening the new one.
    /// The file is opened with `write=true, create=true, truncate=false` and
    /// wrapped in a [`BufWriter`] with a 64 KiB buffer.
    ///
    /// # Errors
    ///
    /// Returns `std::io::Error` if the file cannot be opened, or if the
    /// flush triggered by cache-full eviction fails on any of the dropped
    /// handles (e.g. disk full while persisting buffered writes).
    pub(crate) fn open_cached(&mut self, path: &Path) -> std::io::Result<&mut BufWriter<File>> {
        // Cache-hit fast path: avoid cloning the path into a `PathBuf` on every
        // call. The indexed apply path calls this once per region (often
        // millions of regions per chain), so skipping the allocation on the
        // common hit path is the win — at the cost of one extra HashMap lookup
        // on the rare miss.
        if self.file_cache.contains_key(path) {
            return Ok(self
                .file_cache
                .get_mut(path)
                .expect("contains_key returned true above"));
        }
        // Crude eviction: flush + clear all when full to bound open FD count.
        // Flushing first surfaces write errors (disk full, quota) that would
        // otherwise be silently swallowed by `BufWriter::drop`.
        if self.file_cache.len() >= MAX_CACHED_FDS {
            self.drain_and_flush()?;
        }
        let file = OpenOptions::new()
            .write(true)
            .create(true)
            .truncate(false)
            .open(path)?;
        Ok(self
            .file_cache
            .entry(path.to_path_buf())
            .or_insert_with(|| BufWriter::with_capacity(WRITE_BUFFER_CAPACITY, file)))
    }

    /// Flush and remove the cached handle for `path`, if any.
    ///
    /// Called before a file is deleted so that the OS handle is closed before
    /// the unlink (no-op on Linux, required on Windows where an open handle
    /// prevents deletion). The buffered writes are flushed first so that any
    /// pending data lands on disk before the close.
    ///
    /// If no handle is cached for `path`, returns `Ok(())`.
    ///
    /// # Errors
    ///
    /// Returns `std::io::Error` if flushing the buffered writes fails.
    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.
    ///
    /// Called by `RemoveAll` before bulk-deleting an expansion folder's files
    /// to ensure no lingering open handles survive into the deletion window,
    /// and by `flush()` to provide a public durability checkpoint.
    ///
    /// # Errors
    ///
    /// Returns the first `std::io::Error` produced by any handle's flush.
    /// Remaining handles are still flushed and cleared even if an earlier
    /// one failed; the cache is always empty on return.
    pub(crate) fn clear_file_cache(&mut self) -> std::io::Result<()> {
        self.drain_and_flush()
    }

    /// Create `path` and every missing ancestor, memoising the call.
    ///
    /// A real patch issues thousands of `create_dir_all` calls against a
    /// handful of shared parent directories (`sqpack/ffxiv`,
    /// `sqpack/ex1/...`, etc.). The first call for a given path falls through
    /// to [`std::fs::create_dir_all`] and inserts the path into an internal
    /// set; later calls for the same path return `Ok(())` without issuing the
    /// syscall.
    ///
    /// The cache is cleared by destructive ops that might remove a directory
    /// it tracks (`SqpkFile::RemoveAll`, `DeleteDirectory`). Cache misses
    /// after a clear cost exactly one `create_dir_all` syscall, the same as
    /// before this optimisation.
    ///
    /// # Errors
    ///
    /// Returns `std::io::Error` if [`std::fs::create_dir_all`] fails. On
    /// failure the path is **not** inserted into the cache, so a retry that
    /// fixes the underlying problem (e.g. permissions) will re-attempt the
    /// syscall.
    pub(crate) fn ensure_dir_all(&mut self, path: &Path) -> std::io::Result<()> {
        if self.dirs_created.contains(path) {
            return Ok(());
        }
        std::fs::create_dir_all(path)?;
        self.dirs_created.insert(path.to_path_buf());
        Ok(())
    }

    /// Drop every memoised entry in the created-directories set.
    ///
    /// Called by destructive operations that may remove a directory the
    /// cache claims still exists. Subsequent [`Self::ensure_dir_all`] calls
    /// fall back to one real `create_dir_all` syscall per path until the set
    /// repopulates.
    pub(crate) fn invalidate_dirs_created(&mut self) {
        self.dirs_created.clear();
    }

    /// Drop every memoised entry in the `SqPack` path cache.
    ///
    /// Called by `apply_target_info` when `ApplyContext::platform` changes,
    /// since the cached `PathBuf`s embed the platform string. Cache misses
    /// after a clear cost one full path resolution per `(main_id, sub_id,
    /// file_id, kind)` until the set repopulates.
    pub(crate) fn invalidate_path_cache(&mut self) {
        self.path_cache.clear();
    }

    /// Flush every cached writer's buffer, then drain the cache.
    ///
    /// Used by both [`Self::clear_file_cache`] and the cache-full path inside
    /// [`Self::open_cached`]. Distinct from [`Self::flush`], which flushes in
    /// place without dropping the handles.
    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 operating system.
    ///
    /// Forces any data still sitting in [`BufWriter`] buffers (one per cached
    /// `SqPack` file) to be written via `write(2)`. Open handles are retained
    /// — this is a durability checkpoint, not a cache eviction. Subsequent
    /// chunks targeting the same files reuse the existing handles.
    ///
    /// `apply_to` calls this automatically before returning so successful
    /// completion of a patch implies all writes have reached the OS. Explicit
    /// calls are useful when applying chunks one at a time via
    /// [`Apply::apply`] and reading the resulting file state in between, or
    /// when implementing a custom apply driver that wants intermediate
    /// commit points.
    ///
    /// This is **not** `fsync`. Data is handed off to the OS but may still
    /// reside in the page cache; survival across a crash requires
    /// `File::sync_all` on the underlying handles, which this method does
    /// not perform.
    ///
    /// # Errors
    ///
    /// Returns the first `std::io::Error` produced by any writer's flush.
    /// Remaining writers are still attempted even if an earlier one failed.
    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(()),
        }
    }
}

/// Applies a parsed chunk to the filesystem via an [`ApplyContext`].
///
/// Every top-level [`Chunk`] variant and every
/// [`crate::chunk::sqpk::SqpkCommand`] variant implements this trait. The
/// usual entry point is [`Chunk::apply`], which dispatches to the appropriate
/// implementation.
///
/// # Ordering
///
/// Chunks must be applied in the order they appear in the patch stream.
/// The format is a sequential log; later chunks may depend on state produced
/// by earlier ones.
///
/// # Idempotency
///
/// Apply operations are **not idempotent** in general. Write operations are
/// idempotent only if the data payload is identical to what is already on
/// disk. Destructive operations (`RemoveAll`, `DeleteFile`, `DeleteDirectory`)
/// are not repeatable without error unless `ignore_missing` is set.
///
/// # Errors
///
/// Returns [`crate::ZiPatchError`] on any filesystem or data error. The error
/// is not recovered from; the caller should treat it as fatal for the current
/// apply session.
///
/// # Panics
///
/// Implementations do not panic under normal operation. Panics would indicate
/// a bug in the parsing layer (e.g. a chunk with fields that violate internal
/// invariants established during parsing).
pub trait Apply {
    /// Apply this chunk to `ctx`.
    ///
    /// On success, any filesystem changes the chunk describes have been
    /// written. On error, changes may be partial; the caller is responsible
    /// for any recovery.
    fn apply(&self, ctx: &mut ApplyContext) -> Result<()>;
}

/// Dispatch table for top-level chunk variants.
///
/// `FileHeader`, `ApplyFreeSpace`, and `EndOfFile` are metadata or structural
/// chunks with no filesystem effect; they return `Ok(())` immediately.
/// All other variants delegate to their specific `Apply` implementation.
impl Apply for Chunk {
    fn apply(&self, ctx: &mut ApplyContext) -> Result<()> {
        match self {
            Chunk::FileHeader(_) | Chunk::ApplyFreeSpace(_) | Chunk::EndOfFile => Ok(()),
            Chunk::Sqpk(c) => c.apply(ctx),
            Chunk::ApplyOption(c) => c.apply(ctx),
            Chunk::AddDirectory(c) => c.apply(ctx),
            Chunk::DeleteDirectory(c) => c.apply(ctx),
        }
    }
}

/// Updates [`ApplyContext`] ignore-flags from the chunk payload.
///
/// `ApplyOption` chunks are embedded in the patch stream to toggle
/// [`ApplyContext::ignore_missing`] and [`ApplyContext::ignore_old_mismatch`]
/// at specific points during apply. Applying this chunk mutates `ctx` in
/// place; no filesystem I/O is performed.
impl Apply for ApplyOption {
    fn apply(&self, ctx: &mut ApplyContext) -> Result<()> {
        trace!(kind = ?self.kind, value = self.value, "apply option");
        match self.kind {
            ApplyOptionKind::IgnoreMissing => ctx.ignore_missing = self.value,
            ApplyOptionKind::IgnoreOldMismatch => ctx.ignore_old_mismatch = self.value,
        }
        Ok(())
    }
}

/// Creates a directory under the game install root.
///
/// Equivalent to `fs::create_dir_all(game_path / name)`. Intermediate
/// directories are created as needed; the call is idempotent if the directory
/// already exists.
///
/// # Errors
///
/// Returns [`crate::ZiPatchError::Io`] if directory creation fails for any
/// reason other than the directory already existing (e.g. a permission error
/// or a non-directory file at the path).
impl Apply for AddDirectory {
    fn apply(&self, ctx: &mut ApplyContext) -> Result<()> {
        trace!(name = %self.name, "create directory");
        let path = ctx.game_path.join(&self.name);
        ctx.ensure_dir_all(&path)?;
        Ok(())
    }
}

/// Removes a directory from the game install root.
///
/// The directory must be **empty**; `remove_dir` (not `remove_dir_all`) is
/// used intentionally so that stale files inside the directory cause a visible
/// error rather than silent data loss.
///
/// If the directory does not exist and [`ApplyContext::ignore_missing`] is
/// `true`, the missing directory is logged at `warn!` level and `Ok(())` is
/// returned. If `ignore_missing` is `false`, the `NotFound` I/O error is
/// propagated.
///
/// # Errors
///
/// Returns [`crate::ZiPatchError::Io`] if the removal fails for any reason
/// other than a missing directory with `ignore_missing = true`.
impl Apply for DeleteDirectory {
    fn apply(&self, ctx: &mut ApplyContext) -> Result<()> {
        match std::fs::remove_dir(ctx.game_path.join(&self.name)) {
            Ok(()) => {
                trace!(name = %self.name, "delete directory");
                // The just-removed directory (or an ancestor it co-occupies)
                // may sit in the created-dirs cache; clear the whole set so a
                // subsequent `ensure_dir_all` re-issues a real syscall.
                ctx.invalidate_dirs_created();
                Ok(())
            }
            Err(e) if e.kind() == std::io::ErrorKind::NotFound && ctx.ignore_missing => {
                warn!(name = %self.name, "delete directory: not found, ignored");
                Ok(())
            }
            Err(e) => Err(e.into()),
        }
    }
}

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

    // --- Cache semantics ---

    #[test]
    fn cache_eviction_clears_all_entries_when_at_capacity() {
        // Fill the cache to exactly MAX_CACHED_FDS, then request one new path.
        // The eviction must drop all 256 entries and leave only the new one.
        let tmp = tempfile::tempdir().unwrap();
        let mut ctx = ApplyContext::new(tmp.path());

        for i in 0..MAX_CACHED_FDS {
            ctx.open_cached(&tmp.path().join(format!("{i}.dat")))
                .unwrap();
        }
        assert_eq!(
            ctx.file_cache.len(),
            MAX_CACHED_FDS,
            "cache should be full before triggering eviction"
        );

        ctx.open_cached(&tmp.path().join("new.dat")).unwrap();
        assert_eq!(
            ctx.file_cache.len(),
            1,
            "eviction must clear all entries and leave only the new handle"
        );
    }

    #[test]
    fn cache_hit_does_not_trigger_eviction_when_full() {
        // With a full cache, requesting an *already-cached* path must NOT evict
        // — the size stays at MAX_CACHED_FDS.
        let tmp = tempfile::tempdir().unwrap();
        let mut ctx = ApplyContext::new(tmp.path());

        for i in 0..MAX_CACHED_FDS {
            ctx.open_cached(&tmp.path().join(format!("{i}.dat")))
                .unwrap();
        }
        // Re-request the first path — it is already in the cache.
        ctx.open_cached(&tmp.path().join("0.dat")).unwrap();
        assert_eq!(
            ctx.file_cache.len(),
            MAX_CACHED_FDS,
            "cache hit on a full cache must not evict anything"
        );
    }

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

        ctx.evict_cached(&a).unwrap();
        assert_eq!(
            ctx.file_cache.len(),
            1,
            "evict_cached must remove exactly the targeted path"
        );
        assert!(
            ctx.file_cache.contains_key(&b),
            "evict_cached must not remove the other path"
        );
    }

    #[test]
    fn evict_cached_is_noop_for_absent_path() {
        let tmp = tempfile::tempdir().unwrap();
        let mut ctx = ApplyContext::new(tmp.path());
        ctx.open_cached(&tmp.path().join("a.dat")).unwrap();
        // Evicting a path that was never inserted must not panic or change the cache.
        ctx.evict_cached(&tmp.path().join("nonexistent.dat"))
            .unwrap();
        assert_eq!(ctx.file_cache.len(), 1);
    }

    #[test]
    fn clear_file_cache_removes_all_handles() {
        let tmp = tempfile::tempdir().unwrap();
        let mut ctx = ApplyContext::new(tmp.path());
        ctx.open_cached(&tmp.path().join("a.dat")).unwrap();
        ctx.open_cached(&tmp.path().join("b.dat")).unwrap();
        assert_eq!(ctx.file_cache.len(), 2);
        ctx.clear_file_cache().unwrap();
        assert_eq!(
            ctx.file_cache.len(),
            0,
            "clear_file_cache must empty the cache"
        );
    }

    // --- Builder accessors ---

    #[test]
    fn game_path_returns_install_root_unchanged() {
        let tmp = tempfile::tempdir().unwrap();
        let ctx = ApplyContext::new(tmp.path());
        assert_eq!(
            ctx.game_path(),
            tmp.path(),
            "game_path() must return exactly the path passed to new()"
        );
    }

    #[test]
    fn default_platform_is_win32() {
        let ctx = ApplyContext::new("/irrelevant");
        assert_eq!(
            ctx.platform(),
            Platform::Win32,
            "default platform must be Win32"
        );
    }

    #[test]
    fn with_platform_overrides_default() {
        let ctx = ApplyContext::new("/irrelevant").with_platform(Platform::Ps4);
        assert_eq!(
            ctx.platform(),
            Platform::Ps4,
            "with_platform must override the Win32 default"
        );
    }

    #[test]
    fn default_ignore_missing_is_false() {
        let ctx = ApplyContext::new("/irrelevant");
        assert!(
            !ctx.ignore_missing(),
            "ignore_missing must default to false"
        );
    }

    #[test]
    fn with_ignore_missing_toggles_flag_both_ways() {
        let ctx = ApplyContext::new("/irrelevant").with_ignore_missing(true);
        assert!(
            ctx.ignore_missing(),
            "with_ignore_missing(true) must set the flag"
        );
        let ctx = ctx.with_ignore_missing(false);
        assert!(
            !ctx.ignore_missing(),
            "with_ignore_missing(false) must clear the flag"
        );
    }

    #[test]
    fn default_ignore_old_mismatch_is_false() {
        let ctx = ApplyContext::new("/irrelevant");
        assert!(
            !ctx.ignore_old_mismatch(),
            "ignore_old_mismatch must default to false"
        );
    }

    #[test]
    fn with_ignore_old_mismatch_toggles_flag_both_ways() {
        let ctx = ApplyContext::new("/irrelevant").with_ignore_old_mismatch(true);
        assert!(
            ctx.ignore_old_mismatch(),
            "with_ignore_old_mismatch(true) must set the flag"
        );
        let ctx = ctx.with_ignore_old_mismatch(false);
        assert!(
            !ctx.ignore_old_mismatch(),
            "with_ignore_old_mismatch(false) must clear the flag"
        );
    }

    // --- with_observer ---
    //
    // The end-to-end "default context uses NoopObserver" check lives in
    // `src/lib.rs` as `default_no_observer_apply_succeeds_as_before`, which
    // exercises the full `apply_to` driver path; duplicating it here would
    // only re-test the same behaviour through a slightly different lens.

    // --- BufWriter cache ---

    #[test]
    fn buffered_writes_are_invisible_before_flush() {
        // The whole point of wrapping cached handles in a BufWriter is to
        // hold small writes in user-space memory until enough have queued
        // up. Lock that down: a 1-byte write must not be visible on disk
        // until flush() is called.
        use std::io::Write;

        let tmp = tempfile::tempdir().unwrap();
        let mut ctx = ApplyContext::new(tmp.path());
        let path = tmp.path().join("buffered.dat");

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

        // File exists (open_cached opened it with create=true) but is empty
        // — the byte is sitting in the BufWriter, not on disk.
        assert_eq!(
            std::fs::metadata(&path).unwrap().len(),
            0,
            "buffered write must not reach disk before flush"
        );

        ctx.flush().unwrap();
        assert_eq!(
            std::fs::read(&path).unwrap(),
            vec![0xAB],
            "flush must drain the buffer to disk"
        );
    }

    #[test]
    fn flush_keeps_handles_in_cache() {
        // flush() is a durability checkpoint, not an eviction — handles
        // must survive so subsequent chunks targeting the same file reuse
        // them rather than reopening.
        let tmp = tempfile::tempdir().unwrap();
        let mut ctx = ApplyContext::new(tmp.path());
        ctx.open_cached(&tmp.path().join("a.dat")).unwrap();
        ctx.open_cached(&tmp.path().join("b.dat")).unwrap();
        assert_eq!(ctx.file_cache.len(), 2);

        ctx.flush().unwrap();
        assert_eq!(
            ctx.file_cache.len(),
            2,
            "flush must not drop cached handles"
        );
    }

    #[test]
    fn evict_cached_flushes_pending_writes_to_disk() {
        // evict_cached must flush before dropping — otherwise buffered
        // writes against the about-to-be-closed handle would be silently
        // discarded by BufWriter::drop's error-swallowing flush.
        use std::io::Write;

        let tmp = tempfile::tempdir().unwrap();
        let mut ctx = ApplyContext::new(tmp.path());
        let path = tmp.path().join("evict.dat");

        let writer = ctx.open_cached(&path).unwrap();
        writer.write_all(b"queued").unwrap();
        assert_eq!(
            std::fs::metadata(&path).unwrap().len(),
            0,
            "pre-condition: write is buffered, not on disk"
        );

        ctx.evict_cached(&path).unwrap();
        assert_eq!(
            std::fs::read(&path).unwrap(),
            b"queued",
            "evict_cached must flush before closing the handle"
        );
        assert!(
            !ctx.file_cache.contains_key(&path),
            "evict_cached must also remove the entry"
        );
    }

    #[test]
    fn clear_file_cache_flushes_every_pending_write() {
        // clear_file_cache must flush every buffered writer before dropping
        // — RemoveAll relies on this to avoid losing pending data against
        // the about-to-be-unlinked files.
        use std::io::Write;

        let tmp = tempfile::tempdir().unwrap();
        let mut ctx = ApplyContext::new(tmp.path());
        let a = tmp.path().join("a.dat");
        let b = tmp.path().join("b.dat");

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

        ctx.clear_file_cache().unwrap();

        assert_eq!(std::fs::read(&a).unwrap(), b"AA");
        assert_eq!(std::fs::read(&b).unwrap(), b"BB");
        assert!(ctx.file_cache.is_empty(), "cache must be empty after clear");
    }

    // --- Debug impl ---

    #[test]
    fn apply_context_debug_renders_all_fields() {
        // ApplyContext can't derive Debug because Box<dyn ApplyObserver> doesn't
        // implement it; the hand-written impl substitutes a placeholder for the
        // observer. Lock down the rendered shape so future refactors don't
        // accidentally drop fields (and so the impl itself is exercised by tests).
        let tmp = tempfile::tempdir().unwrap();
        let ctx = ApplyContext::new(tmp.path())
            .with_platform(Platform::Ps4)
            .with_ignore_missing(true);

        let rendered = format!("{ctx:?}");
        for needle in [
            "ApplyContext",
            "game_path",
            "platform",
            "Ps4",
            "ignore_missing",
            "true",
            "ignore_old_mismatch",
            "file_cache_len",
            "path_cache_len",
            "decompressor",
            "<flate2::Decompress>",
            "observer",
            "<dyn ApplyObserver>",
        ] {
            assert!(
                rendered.contains(needle),
                "Debug output must mention {needle:?}; got: {rendered}"
            );
        }
    }

    // --- DeleteDirectory happy path ---

    #[test]
    fn delete_directory_success_removes_existing_dir() {
        // Exercises the Ok(()) trace+return arm of DeleteDirectory::apply
        // (previously only the ignore_missing and propagate-error arms were
        // covered).
        let tmp = tempfile::tempdir().unwrap();
        let target = tmp.path().join("to_remove");
        std::fs::create_dir(&target).unwrap();
        assert!(target.is_dir(), "pre-condition: directory must exist");

        let mut ctx = ApplyContext::new(tmp.path());
        DeleteDirectory {
            name: "to_remove".into(),
        }
        .apply(&mut ctx)
        .expect("delete on an existing directory must succeed");

        assert!(!target.exists(), "directory must be removed");
    }

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

    #[test]
    fn ensure_dir_all_cache_hit_returns_early_without_syscall() {
        // The second call for the same path must take the early-return branch
        // at line 521 (`return Ok(())`).  We confirm this is hit by pre-seeding
        // `dirs_created` and then calling `ensure_dir_all` for a path that does
        // NOT actually exist on disk — if the cache-miss branch ran it would
        // call `create_dir_all` and either succeed (masking the bug) or fail.
        let tmp = tempfile::tempdir().unwrap();
        let mut ctx = ApplyContext::new(tmp.path());

        let path = tmp.path().join("cached_dir");
        // First call: misses the cache, creates the directory on disk, inserts.
        ctx.ensure_dir_all(&path).unwrap();
        assert!(path.is_dir(), "first call must create the directory");
        assert_eq!(
            ctx.dirs_created.len(),
            1,
            "path must be cached after first call"
        );

        // Remove the directory so a second real `create_dir_all` would see it
        // gone — if the cache-hit branch is NOT taken, the syscall would still
        // succeed (create_dir_all is idempotent for missing dirs), so instead
        // we verify the set length stays at 1, not 2.
        let p2 = tmp.path().join("cached_dir");
        ctx.ensure_dir_all(&p2).unwrap();
        assert_eq!(
            ctx.dirs_created.len(),
            1,
            "cache hit must not re-insert the path (set length must stay 1)"
        );
    }

    // --- drain_and_flush error branch ---

    #[test]
    fn drain_and_flush_error_propagates_first_io_error() {
        // Trigger the `Some(e) => Err(e)` arm in `drain_and_flush`.
        //
        // `/dev/full` always returns ENOSPC on write — use it as the backing
        // file so the BufWriter's flush (which actually calls write(2)) fails.
        // We open `/dev/full` directly and inject the handle into `file_cache`
        // via the `pub(crate)` field, bypassing `open_cached` which uses
        // `create=true` (incompatible with a character device).
        use std::io::Write;

        let dev_full = std::path::PathBuf::from("/dev/full");
        if !dev_full.exists() {
            // /dev/full is Linux-specific; skip on platforms that lack it.
            return;
        }

        let file = OpenOptions::new()
            .write(true)
            .open(&dev_full)
            .expect("/dev/full must be openable for writing");

        let mut ctx = ApplyContext::new("/irrelevant");
        let mut writer = BufWriter::with_capacity(WRITE_BUFFER_CAPACITY, file);
        // Write into the BufWriter's user-space buffer — this succeeds because
        // BufWriter holds the data in RAM.  The write only reaches /dev/full
        // (and fails with ENOSPC) when the buffer is flushed.
        writer.write_all(&[0xAB; 128]).unwrap();
        ctx.file_cache.insert(dev_full.clone(), writer);

        let result = ctx.clear_file_cache();

        assert!(
            result.is_err(),
            "drain_and_flush must propagate the ENOSPC error from /dev/full"
        );
        assert!(
            ctx.file_cache.is_empty(),
            "cache must be drained even when flush fails"
        );
    }

    // --- flush error branch ---

    #[test]
    fn flush_error_propagates_first_io_error() {
        // Trigger the `Some(e) => Err(e)` arm in `flush` using the same
        // `/dev/full` trick as `drain_and_flush_error_propagates_first_io_error`.
        // `flush` keeps handles in the cache (unlike `drain_and_flush`), so we
        // assert the cache still contains the entry after the failed flush.
        use std::io::Write;

        let dev_full = std::path::PathBuf::from("/dev/full");
        if !dev_full.exists() {
            return;
        }

        let file = OpenOptions::new()
            .write(true)
            .open(&dev_full)
            .expect("/dev/full must be openable for writing");

        let mut ctx = ApplyContext::new("/irrelevant");
        let mut writer = BufWriter::with_capacity(WRITE_BUFFER_CAPACITY, file);
        writer.write_all(&[0xCD; 128]).unwrap();
        ctx.file_cache.insert(dev_full.clone(), writer);

        let result = ctx.flush();

        assert!(
            result.is_err(),
            "flush must propagate the ENOSPC error from /dev/full"
        );
        assert_eq!(
            ctx.file_cache.len(),
            1,
            "flush must NOT evict handles — only drain_and_flush does that"
        );
    }
}