znippy-common 0.9.13

//! `ArrowIpcSinkAppend` — the **append/resume-capable v2 clone** of
//! [`ArrowIpcSink`](crate::ArrowIpcSink).
//!
//! ## Why a clone and not a refactor
//! The original [`ArrowIpcSink`] is the A baseline of the arrow-ipc write/seal
//! A/B audit — its hot path (`push_subindex` + `finish`) must stay byte- and
//! perf-identical. So this is a *near-identical copy* that adds the two new
//! capabilities **without touching the original**:
//!
//!  * **fresh path** — [`ArrowIpcSinkAppend::new`] + `push_subindex` + `finish`
//!    is a line-for-line clone of `ArrowIpcSink`. It writes the **same v0.7
//!    on-disk bytes** (same sub-index serialisation, same sorted lookup, same
//!    fst trie, same manifest, same `ZNPYMIDX` footer). The A/B parity test
//!    proves this clone is a zero-cost superset on the normal write path.
//!
//!  * **resume/append path** — [`ArrowIpcSinkAppend::open_existing`] reopens an
//!    already-sealed `.znippy`, recovers its existing data rows, repositions the
//!    cursor at the **end of the blob region** (truncating the old metadata
//!    tail), and lets the caller `push_subindex` more blobs' rows. `finish()`
//!    then re-seals the merged old+new row set — the **first-class native blob
//!    append** that the iceberg lifecycle test (#23) previously had to do by
//!    hand.
//!
//! ### The resume mechanism, in bytes
//! A sealed v0.7 archive is:
//! ```text
//! [ blob_0 … blob_N ][ data sub-idx(es) ][ lookup sub-idx ][ trie ][ manifest ][ ZNPYMIDX ][off]
//! ^0                 ^blob_end           (the whole metadata tail is rebuildable)
//! ```
//! To append we must:
//!   1. read the full manifest (incl. reserved lookup/trie entries),
//!   2. find `blob_end` = the lowest `index_offset` over all sub-index entries
//!      (where the blob region stops and the rebuildable tail begins),
//!   3. recover the existing **data** rows from the sorted lookup sub-index (one
//!      cheap read of the already-sorted reserved section — no per-sub-index
//!      re-scan), seeding the lookup accumulator,
//!   4. set the cursor to `blob_end` so the caller's new blob bytes + new data
//!      sub-index overwrite the old (now-stale) metadata tail,
//!   5. on `finish()`, re-sort the merged rows and re-emit lookup + trie +
//!      manifest + footer.
//!
//! The new blob bytes are written by the caller (the compress pipeline / a
//! library append entrypoint such as [`append_files`]) to the file at
//! `blob_end()` exactly as a fresh archive
//! writes blobs at offset 0; this sink owns only the metadata tail, identical to
//! the original's contract.

use std::fs::File;
use std::os::unix::fs::FileExt;
use std::path::Path;
use std::sync::Arc;

use anyhow::{Result, anyhow};
use arrow::array::{
    BooleanArray, BooleanBuilder, FixedSizeBinaryArray, FixedSizeBinaryBuilder, StringArray,
    StringBuilder, UInt32Array, UInt32Builder, UInt64Array, UInt64Builder,
};
use arrow::datatypes::Schema;
use arrow::ipc::writer::StreamWriter;
use arrow::record_batch::RecordBatch;

use crate::index::{
    ChunkLoc, LOOKUP_MODULE, MULTI_INDEX_MAGIC, ManifestEntry, RESERVED_PKG_TYPE, TRIE_MODULE,
    data_subindex_schema, is_reserved_module, lookup_schema, read_znippy_full_manifest,
    write_manifest_bytes,
};
use crate::index::{META_MODULE, read_reserved_section_bytes};
use crate::meta_index::{
    MetaTable, build_meta_batch, decode_meta_section, meta_schema,
};
use crate::meta_sink::{ArchiveMetaSink, GroupKey};

/// Append/resume-capable v2 clone of [`ArrowIpcSink`](crate::ArrowIpcSink).
///
/// Field-for-field identical to the original; the only added surface is the
/// [`open_existing`](Self::open_existing) constructor and the [`blob_end`](Self::blob_end)
/// accessor. The fresh-write path is byte-identical to the original.
pub struct ArrowIpcSinkAppend {
    file: Arc<File>,
    cursor: u64,
    entries: Vec<ManifestEntry>,
    lookup_paths: Vec<String>,
    lookup_locs: Vec<ChunkLoc>,
    /// On resume, the recovered pre-existing data rows. They are re-emitted as a
    /// data sub-index in `finish()` (so the ordinary reader, which reads data
    /// sub-indexes, still sees them) AND merged into the rebuilt lookup. Empty
    /// for a fresh sink. Kept separate from `lookup_*` so they aren't
    /// double-counted before the re-emit.
    carried: Vec<(String, ChunkLoc)>,
    /// Searchable metadata to seal as the `META_MODULE` sub-index.
    ///
    /// `None` means "emit no section", which is what a reader later reports as
    /// `ArchiveMeta::NoMetadata`; `Some(empty)` means "emit a present, empty
    /// index". The two are different archives on disk and different answers on
    /// read, and this field is the only place that decision is made.
    ///
    /// On resume this is **seeded from the archive's existing section**, because
    /// `open_existing` truncates the whole metadata tail — without carrying it,
    /// every append would silently erase the metadata the archive already had.
    meta: Option<MetaTable>,
}

impl ArrowIpcSinkAppend {
    /// Fresh archive: identical to [`ArrowIpcSink::new`](crate::ArrowIpcSink::new).
    /// `blob_end_offset` is the byte offset just past the last blob.
    pub fn new(file: Arc<File>, blob_end_offset: u64) -> Self {
        Self {
            file,
            cursor: blob_end_offset,
            entries: Vec::new(),
            lookup_paths: Vec::new(),
            lookup_locs: Vec::new(),
            carried: Vec::new(),
            meta: None,
        }
    }

    /// Seal `meta` as this archive's searchable metadata sub-index, replacing
    /// anything carried from a resumed archive.
    pub fn with_meta(mut self, meta: MetaTable) -> Self {
        self.meta = Some(meta);
        self
    }

    /// Add rows to the metadata to be sealed, keeping whatever a resume carried.
    /// Creates the section if the archive had none.
    pub fn merge_meta(&mut self, rows: impl IntoIterator<Item = crate::meta_index::MetaEntry>) {
        self.meta.get_or_insert_with(MetaTable::new).extend(rows);
    }

    /// The metadata this sink will seal — `None` when it will emit no section.
    pub fn meta(&self) -> Option<&MetaTable> {
        self.meta.as_ref()
    }

    /// Reopen an **already-sealed** v0.7 `.znippy` for append/resume.
    ///
    /// Recovers the existing data rows from the sorted lookup sub-index, drops
    /// the (rebuildable) metadata tail by positioning the cursor at the end of
    /// the blob region, and returns a sink ready to accept more `push_subindex`
    /// calls. The caller appends its new blob bytes to the same file starting at
    /// [`blob_end`](Self::blob_end) before pushing the matching index rows.
    ///
    /// The file is opened read+write; nothing is mutated until `push_subindex` /
    /// `finish` overwrite the old tail.
    pub fn open_existing(path: &Path) -> Result<Self> {
        let file = Arc::new(
            std::fs::OpenOptions::new()
                .read(true)
                .write(true)
                .open(path)
                .map_err(|e| anyhow!("append: open {} for resume: {e}", path.display()))?,
        );

        // 1. Full manifest (incl. reserved lookup/trie entries).
        let (entries, _manifest_offset) = read_znippy_full_manifest(path)?;
        if entries.is_empty() {
            return Err(anyhow!("append: archive {} has an empty manifest", path.display()));
        }

        // 2. blob_end = lowest index_offset over all sub-index/reserved sections.
        //    Everything from there to EOF is the rebuildable metadata tail.
        let blob_end = entries
            .iter()
            .map(|e| e.index_offset)
            .min()
            .ok_or_else(|| anyhow!("append: no sections in manifest"))?;

        // 3. Recover existing DATA rows from the sorted lookup sub-index (one
        //    read of the already-sorted reserved section). Fall back to scanning
        //    the data sub-indexes if (unexpectedly) no lookup section is present.
        let (paths, locs) = recover_rows(path, &entries)?;
        let carried: Vec<(String, ChunkLoc)> = paths.into_iter().zip(locs).collect();

        // 4. Carry the searchable metadata section forward. The tail we are about
        //    to overwrite contains it, so a resume that did not recover it would
        //    quietly turn an archive WITH metadata into one without — and the
        //    reader would then honestly report `NoMetadata` about an archive that
        //    used to have some. Absent stays absent; present-but-empty stays
        //    present-but-empty.
        let meta = match read_reserved_section_bytes(path, META_MODULE)? {
            None => None,
            Some(bytes) => Some(decode_meta_section(&bytes)?.to_table()),
        };

        Ok(Self {
            file,
            cursor: blob_end,
            entries: Vec::new(), // rebuilt fresh by push_subindex + finish
            lookup_paths: Vec::new(),
            lookup_locs: Vec::new(),
            carried,
            meta,
        })
    }

    /// Byte offset where the blob region ends in a resumed archive — where the
    /// caller writes its newly-appended blob bytes (and where the first new data
    /// sub-index will be placed). For a fresh sink this is the `blob_end_offset`
    /// passed to [`new`](Self::new) until the first `push_subindex`.
    pub fn blob_end(&self) -> u64 {
        self.cursor
    }

    /// Number of pre-existing data rows recovered on resume (0 for a fresh sink).
    pub fn recovered_rows(&self) -> usize {
        self.carried.len()
    }

    /// Drop every carried (pre-existing) row whose `relative_path` is about to be
    /// re-written by this append, giving last-writer-wins **replace** semantics.
    /// Returns the number of rows dropped.
    ///
    /// Without this, appending a path the archive already contains left TWO row
    /// sets for it in the re-sealed index — the stale one and the new one — and
    /// nothing downstream treated that as an error: the reader that concatenates
    /// chunks returned both copies back to back at twice the real length, and the
    /// reader that places chunks at `fdata_offset` wrote both to offset 0, so the
    /// carried STALE copy (re-emitted last, in `finish`) won and `znippy get`
    /// silently handed back the old file. Every chunk's blake3 is individually
    /// correct in both cases, so even verified reads passed.
    fn drop_carried_paths(&mut self, replacing: &std::collections::HashSet<&str>) -> usize {
        if self.carried.is_empty() || replacing.is_empty() {
            return 0;
        }
        let before = self.carried.len();
        self.carried.retain(|(p, _)| !replacing.contains(p.as_str()));
        before - self.carried.len()
    }

    /// Re-emit the carried (recovered) rows as a single data sub-index so the
    /// ordinary reader — which reads data sub-indexes, not the lookup — still
    /// lists them after the re-seal. `push_subindex` also folds them into the
    /// rebuilt lookup accumulator. No-op for a fresh sink.
    fn emit_carried(&mut self) -> Result<()> {
        if self.carried.is_empty() {
            return Ok(());
        }
        let carried = std::mem::take(&mut self.carried);
        let (paths, locs): (Vec<String>, Vec<ChunkLoc>) = carried.into_iter().unzip();
        let batch = base_batch_from_rows(&paths, &locs)?;
        self.push_subindex(data_subindex_schema().as_ref(), &[batch], GroupKey {
            pkg_type: 0,
            repo: String::new(),
            module_name: String::new(),
        })
    }

    // ── below: a line-for-line clone of ArrowIpcSink's private machinery ──

    fn accumulate_lookup(&mut self, batch: &RecordBatch) {
        let cols = (|| {
            Some((
                batch.column_by_name("relative_path")?.as_any().downcast_ref::<StringArray>()?,
                batch.column_by_name("chunk_seq")?.as_any().downcast_ref::<UInt32Array>()?,
                batch.column_by_name("fdata_offset")?.as_any().downcast_ref::<UInt64Array>()?,
                batch.column_by_name("compressed")?.as_any().downcast_ref::<BooleanArray>()?,
                batch.column_by_name("uncompressed_size")?.as_any().downcast_ref::<UInt64Array>()?,
                batch.column_by_name("blob_offset")?.as_any().downcast_ref::<UInt64Array>()?,
                batch.column_by_name("blob_size")?.as_any().downcast_ref::<UInt64Array>()?,
                batch.column_by_name("checksum")?.as_any().downcast_ref::<FixedSizeBinaryArray>()?,
            ))
        })();
        let Some((paths, chunk_seq, fdata, compressed, usz, blob_off, blob_sz, checksum)) = cols
        else { return; };
        for i in 0..batch.num_rows() {
            let mut ck = [0u8; 32];
            ck.copy_from_slice(checksum.value(i));
            self.lookup_paths.push(paths.value(i).to_string());
            self.lookup_locs.push(ChunkLoc {
                chunk_seq: chunk_seq.value(i),
                fdata_offset: fdata.value(i),
                blob_offset: blob_off.value(i),
                blob_size: blob_sz.value(i),
                uncompressed_size: usz.value(i),
                compressed: compressed.value(i),
                checksum: ck,
            });
        }
    }

    fn write_lookup_and_trie(&mut self) -> Result<()> {
        let n = self.lookup_paths.len();
        let mut order: Vec<usize> = (0..n).collect();
        order.sort_by(|&a, &b| {
            self.lookup_paths[a].cmp(&self.lookup_paths[b])
                .then(self.lookup_locs[a].chunk_seq.cmp(&self.lookup_locs[b].chunk_seq))
        });

        let schema = lookup_schema();
        let batch = base_batch_permuted(
            schema.clone(),
            &self.lookup_paths,
            &self.lookup_locs,
            &order,
        )?;
        self.push_subindex(&schema, &[batch], GroupKey {
            pkg_type: RESERVED_PKG_TYPE,
            repo: String::new(),
            module_name: LOOKUP_MODULE.to_string(),
        })?;

        let mut builder = fst::MapBuilder::memory();
        let mut prev: Option<&str> = None;
        for (sorted_idx, &orig) in order.iter().enumerate() {
            let p = self.lookup_paths[orig].as_str();
            if prev != Some(p) {
                builder.insert(p.as_bytes(), sorted_idx as u64)
                    .map_err(|e| anyhow!("trie insert: {e}"))?;
                prev = Some(p);
            }
        }
        let trie_bytes = builder.into_inner().map_err(|e| anyhow!("trie finish: {e}"))?;
        self.write_raw_section(&trie_bytes, GroupKey {
            pkg_type: RESERVED_PKG_TYPE,
            repo: String::new(),
            module_name: TRIE_MODULE.to_string(),
        })
    }

    /// Emit the searchable metadata sub-index, when there is one to emit.
    ///
    /// Reserved module, so the data readers skip it and an older znippy simply
    /// ignores the entry. `None` writes NOTHING — that absence is exactly what
    /// `ArchiveMeta::NoMetadata` reports, and it is why an archive sealed without
    /// metadata stays byte-identical to one sealed before this module existed.
    fn write_meta_subindex(&mut self) -> Result<()> {
        let Some(table) = self.meta.take() else {
            return Ok(());
        };
        let batch = build_meta_batch(&table)?;
        let schema = meta_schema();
        self.push_subindex(schema.as_ref(), &[batch], GroupKey {
            pkg_type: RESERVED_PKG_TYPE,
            repo: String::new(),
            module_name: META_MODULE.to_string(),
        })
    }

    fn write_raw_section(&mut self, bytes: &[u8], key: GroupKey) -> Result<()> {
        let start = self.cursor;
        self.file.write_all_at(bytes, start)?;
        self.cursor += bytes.len() as u64;
        self.entries.push(ManifestEntry {
            pkg_type: key.pkg_type,
            repo: key.repo,
            module_name: key.module_name,
            index_offset: start,
            index_len: bytes.len() as u64,
            row_count: 0,
        });
        Ok(())
    }
}

impl ArchiveMetaSink for ArrowIpcSinkAppend {
    fn push_subindex(
        &mut self,
        schema: &Schema,
        batches: &[RecordBatch],
        key: GroupKey,
    ) -> Result<()> {
        let sub_start = self.cursor;
        let mut sub_bytes: Vec<u8> = Vec::new();
        let mut sw = StreamWriter::try_new(&mut sub_bytes, schema)
            .map_err(|e| anyhow!("sub-index writer: {e}"))?;
        let mut row_count = 0u64;
        for batch in batches {
            row_count += batch.num_rows() as u64;
            sw.write(batch).map_err(|e| anyhow!("sub-index write: {e}"))?;
        }
        sw.finish().map_err(|e| anyhow!("sub-index finish: {e}"))?;

        // Accumulate base columns for the lookup layer from DATA sub-indexes only.
        // Widened from "not lookup, not trie" to "not reserved" when META_MODULE
        // arrived: the metadata sub-index is Arrow IPC and does come through here,
        // and its rows are key/value facts, not chunk locations — folding them
        // into the lookup would corrupt random access. Behaviour for every
        // pre-existing module is unchanged (sign sections are raw, never pushed).
        if !is_reserved_module(&key.module_name) {
            for batch in batches {
                self.accumulate_lookup(batch);
            }
        }

        let sub_len = sub_bytes.len() as u64;
        self.file.write_all_at(&sub_bytes, sub_start)?;
        self.cursor += sub_len;

        self.entries.push(ManifestEntry {
            pkg_type: key.pkg_type,
            repo: key.repo,
            module_name: key.module_name,
            index_offset: sub_start,
            index_len: sub_len,
            row_count,
        });
        Ok(())
    }

    fn finish(mut self: Box<Self>) -> Result<u64> {
        // Resume: re-emit recovered rows as a data sub-index (the ordinary reader
        // reads data sub-indexes, not the lookup). No-op on the fresh path.
        self.emit_carried()?;
        self.write_lookup_and_trie()?;
        self.write_meta_subindex()?;

        let manifest_offset = self.cursor;
        let manifest_bytes =
            write_manifest_bytes(&self.entries).map_err(|e| anyhow!("manifest: {e}"))?;
        self.file.write_all_at(&manifest_bytes, manifest_offset)?;

        let after = manifest_offset + manifest_bytes.len() as u64;
        self.file.write_all_at(&MULTI_INDEX_MAGIC, after)?;
        self.file.write_all_at(
            &manifest_offset.to_le_bytes(),
            after + MULTI_INDEX_MAGIC.len() as u64,
        )?;
        // Resume overwrites the old (longer-or-shorter) tail in place; if the new
        // tail is shorter than the old one, truncate so no stale footer lingers.
        let final_len = after + MULTI_INDEX_MAGIC.len() as u64 + 8;
        self.file.set_len(final_len)?;
        self.file.sync_all()?;

        Ok(final_len)
    }
}

/// One base-schema data batch from `(path, ChunkLoc)` rows, in the order given.
///
/// The single builder for base-schema index rows. Four call sites used to carry
/// a copy of this loop — `emit_carried`, `write_lookup_and_trie`, `ArrowIpcSink`'s
/// two — and a column appended in one order in one of them and another order in
/// the next is a silent index corruption no checksum catches, because every
/// individual chunk still hashes correctly (LAW 5, by construction).
pub(crate) fn base_batch_from_rows(paths: &[String], locs: &[ChunkLoc]) -> Result<RecordBatch> {
    let order: Vec<usize> = (0..paths.len()).collect();
    base_batch_permuted(data_subindex_schema(), paths, locs, &order)
}

/// [`base_batch_from_rows`] emitting rows in `order` under an explicit `schema` —
/// the sorted-lookup case, which is the same columns in a different order.
pub(crate) fn base_batch_permuted(
    schema: Arc<Schema>,
    paths: &[String],
    locs: &[ChunkLoc],
    order: &[usize],
) -> Result<RecordBatch> {
    let n = order.len();
    let mut path_b = StringBuilder::with_capacity(n, n * 16);
    let mut seq_b = UInt32Builder::with_capacity(n);
    let mut fdata_b = UInt64Builder::with_capacity(n);
    let mut comp_b = BooleanBuilder::with_capacity(n);
    let mut usz_b = UInt64Builder::with_capacity(n);
    let mut boff_b = UInt64Builder::with_capacity(n);
    let mut bsz_b = UInt64Builder::with_capacity(n);
    let mut ck_b = FixedSizeBinaryBuilder::with_capacity(n, 32);
    for &i in order {
        let loc = &locs[i];
        path_b.append_value(&paths[i]);
        seq_b.append_value(loc.chunk_seq);
        fdata_b.append_value(loc.fdata_offset);
        comp_b.append_value(loc.compressed);
        usz_b.append_value(loc.uncompressed_size);
        boff_b.append_value(loc.blob_offset);
        bsz_b.append_value(loc.blob_size);
        ck_b.append_value(loc.checksum).expect("checksum is 32 bytes");
    }
    Ok(RecordBatch::try_new(
        schema,
        vec![
            Arc::new(path_b.finish()),
            Arc::new(seq_b.finish()),
            Arc::new(fdata_b.finish()),
            Arc::new(comp_b.finish()),
            Arc::new(usz_b.finish()),
            Arc::new(boff_b.finish()),
            Arc::new(bsz_b.finish()),
            Arc::new(ck_b.finish()),
        ],
    )?)
}

/// Compress-or-store each file and write its blob at `cursor`, returning the
/// index rows and the new cursor. **No metadata is touched** — this is the blob
/// half alone, shared by the re-sealing [`append_files`] path and by the hot
/// journal path, so the two cannot drift on the skip decision, the blake3 domain
/// (original bytes) or the store-raw rule.
pub(crate) fn write_blobs(
    file: &File,
    cursor: u64,
    files: &[(String, Vec<u8>)],
    ctx: &mut crate::codec::CompressCtx,
    policy: crate::SkipPolicy,
) -> Result<(Vec<String>, Vec<ChunkLoc>, u64)> {
    let mut paths = Vec::with_capacity(files.len());
    let mut locs = Vec::with_capacity(files.len());
    let mut cursor = cursor;
    for (rel, bytes) in files {
        let checksum = *blake3::hash(bytes).as_bytes();
        // The skip decision, which the append path did not make at all until
        // 2026-08-03: every byte went to the codec and the frame was thrown away
        // whenever it came out no smaller. For already-compressed input — a
        // `.pack`, a `.jar`, a `.crate` — that is the entire codec cost paid to
        // learn what the file's name already said.
        //
        // MEASURED on gunnar's cold tier (oden, 2026-08-03): appending a
        // 34.1 MiB consolidated packfile cost 0.34 s of CPU compressing and
        // 0.03 s skipping — 11.3x — for byte-identical output.
        let skip = policy.skip_by_path(std::path::Path::new(rel.as_str()));
        let frame = if skip { Vec::new() } else { ctx.compress(bytes)? };
        let (on_disk, compressed): (&[u8], bool) = if !skip && frame.len() < bytes.len() {
            (&frame, true)
        } else {
            (bytes, false)
        };
        let blob_offset = cursor;
        file.write_all_at(on_disk, blob_offset)?;
        cursor += on_disk.len() as u64;
        paths.push(rel.clone());
        locs.push(ChunkLoc {
            chunk_seq: 0,
            fdata_offset: 0,
            blob_offset,
            blob_size: on_disk.len() as u64,
            uncompressed_size: bytes.len() as u64,
            compressed,
            checksum,
        });
    }
    Ok((paths, locs, cursor))
}

/// Outcome of a native [`append_files`] call.
#[derive(Debug, Clone)]
pub struct AppendReport {
    /// Data rows that existed in the archive before the append (recovered),
    /// including any that the append then replaced.
    pub rows_before: u64,
    /// Pre-existing rows dropped because the append re-wrote the same
    /// `relative_path` (replace semantics). Always 0 on the fresh-create path.
    pub rows_replaced: u64,
    /// New rows (one per appended file/chunk) written by the append.
    pub rows_added: u64,
    /// Byte offset where the appended blob region started (old blob_end).
    pub blob_append_offset: u64,
    /// Bytes of new blob payload appended (compressed/stored).
    pub blob_bytes_added: u64,
    /// Final size of the re-sealed archive.
    pub sealed_total_bytes: u64,
}

/// First-class **native blob append** — the caller-facing library primitive that
/// the iceberg lifecycle test (#23) previously had to perform by hand. (There is
/// no `compress --append` CLI verb today; this is the in-process entry point.)
///
/// Opens an existing sealed v0.7 `.znippy`, compresses each `(relative_path,
/// bytes)` in `new_files` with the znippy codec, appends the resulting blobs to
/// the same file past the existing blob region, then re-seals (merged old+new
/// lookup + trie + manifest + footer) via [`ArrowIpcSinkAppend`]. The original
/// blob bytes and existing rows are reused verbatim — nothing is recompressed.
///
/// **Replace semantics:** a `relative_path` in `new_files` that the archive
/// already contains REPLACES the existing entry — the pre-existing rows for that
/// path are dropped from the re-sealed index (counted in
/// [`AppendReport::rows_replaced`]) and its old blob bytes become unreferenced
/// dead payload. Appending the same path twice never leaves two live copies.
///
/// Mirrors the real compress path's per-blob accounting: blake3 over the
/// ORIGINAL bytes, store-raw when the codec frame is not smaller, one chunk per
/// file (`chunk_seq = 0`). `compression_level` is the codec level (e.g. 3).
pub fn append_files(
    archive: &Path,
    new_files: &[(String, Vec<u8>)],
    compression_level: i32,
) -> Result<AppendReport> {
    append_files_with_meta(archive, new_files, compression_level, None)
}

/// [`append_files`] with an explicit [`SkipPolicy`](crate::SkipPolicy).
///
/// The batch-level form, for a caller that already knows what it is appending —
/// `SkipPolicy::already_compressed()` stores every entry raw without inspecting
/// a byte, which is exact, free, and better informed than any probe. gunnar's
/// cold tier appends packfiles and uses it.
pub fn append_files_with_policy(
    archive: &Path,
    new_files: &[(String, Vec<u8>)],
    compression_level: i32,
    policy: crate::SkipPolicy,
) -> Result<AppendReport> {
    let sink = ArrowIpcSinkAppend::open_existing(archive)?;
    let rows_before = sink.recovered_rows() as u64;
    let blob_append_offset = sink.blob_end();
    write_files_into_sink(
        sink,
        new_files,
        compression_level,
        rows_before,
        blob_append_offset,
        policy,
    )
}

/// [`append_files`], additionally merging `meta` rows into the archive's
/// searchable metadata sub-index.
///
/// `None` leaves the metadata exactly as the archive had it — including having
/// none. `Some(rows)` merges into whatever was already there (the resume carries
/// the old section forward), creating the section if the archive had none.
pub fn append_files_with_meta(
    archive: &Path,
    new_files: &[(String, Vec<u8>)],
    compression_level: i32,
    meta: Option<MetaTable>,
) -> Result<AppendReport> {
    let mut sink = ArrowIpcSinkAppend::open_existing(archive)?;
    if let Some(table) = meta {
        sink.merge_meta(table.rows().to_vec());
    }
    let rows_before = sink.recovered_rows() as u64;
    let blob_append_offset = sink.blob_end();
    write_files_into_sink(
        sink,
        new_files,
        compression_level,
        rows_before,
        blob_append_offset,
        // Resolve per entry from its name, then from its bytes — the same
        // default `compress_dir` has. Deliberately NOT "always compress", which
        // is what this path did before and which is never the right answer for
        // an entry whose extension already says it is compressed.
        crate::SkipPolicy::resolve(),
    )
}

/// Create a fresh `.znippy` archive from in-memory `files` — the bootstrap inverse
/// of [`append_files`], which requires an already-sealed archive (it rejects an
/// empty manifest). Seed a new writable archive with this, then grow it with
/// [`append_files`]. Overwrites `archive` if it already exists.
pub fn create_archive(
    archive: &Path,
    files: &[(String, Vec<u8>)],
    compression_level: i32,
) -> Result<AppendReport> {
    create_archive_with_meta(archive, files, compression_level, None)
}

/// [`create_archive`], additionally sealing a searchable metadata sub-index.
///
/// `None` seals **no section** — the archive reads back as
/// `ArchiveMeta::NoMetadata` and is byte-identical to one from
/// [`create_archive`]. `Some(table)` seals the section even when the table is
/// empty, which reads back as a present-but-empty index: "searched, records
/// nothing", a different statement from "never had an index".
pub fn create_archive_with_meta(
    archive: &Path,
    files: &[(String, Vec<u8>)],
    compression_level: i32,
    meta: Option<MetaTable>,
) -> Result<AppendReport> {
    let blob_file = Arc::new(
        File::create(archive)
            .map_err(|e| anyhow!("create archive {}: {e}", archive.display()))?,
    );
    let mut sink = ArrowIpcSinkAppend::new(blob_file, 0);
    sink.meta = meta;
    write_files_into_sink(sink, files, compression_level, 0, 0, crate::SkipPolicy::resolve())
}

/// Build a complete `.znippy` archive **entirely in memory** from in-memory
/// `files` and return its bytes — no staging directory, no named output file. The
/// sink needs a positioned-write fd, so this seals into an **anonymous temp file**
/// (`O_TMPFILE` where the OS supports it → never linked into the filesystem
/// namespace), then reads the sealed bytes straight back out. The returned `Vec`
/// is byte-identical to what [`create_archive`] would write to a path.
///
/// Use this for zero-disk pipelines: build a release/airgap archive from product
/// bytes held in RAM and stream it across the gap without ever touching disk on
/// the build side. Pair with [`append_files`] (path) or hold the bytes and re-seal.
pub fn create_archive_to_vec(
    files: &[(String, Vec<u8>)],
    compression_level: i32,
) -> Result<(Vec<u8>, AppendReport)> {
    let anon = Arc::new(
        tempfile::tempfile().map_err(|e| anyhow!("anonymous archive fd: {e}"))?,
    );
    let sink = ArrowIpcSinkAppend::new(anon.clone(), 0);
    let report =
        write_files_into_sink(sink, files, compression_level, 0, 0, crate::SkipPolicy::resolve())?;
    // The Arc keeps the anonymous fd alive past `finish()`; read the sealed bytes.
    let mut bytes = vec![0u8; report.sealed_total_bytes as usize];
    anon.read_exact_at(&mut bytes, 0)
        .map_err(|e| anyhow!("read back anonymous archive: {e}"))?;
    Ok((bytes, report))
}

/// Shared core of [`append_files`] / [`create_archive`]: compress each file's bytes
/// (store-raw if not smaller), write the blobs at the sink's running blob cursor,
/// then push one base-schema data sub-index and seal. `sink` is either a fresh
/// [`ArrowIpcSinkAppend::new`] or an [`ArrowIpcSinkAppend::open_existing`].
fn write_files_into_sink(
    mut sink: ArrowIpcSinkAppend,
    new_files: &[(String, Vec<u8>)],
    compression_level: i32,
    rows_before: u64,
    blob_append_offset: u64,
    policy: crate::SkipPolicy,
) -> Result<AppendReport> {
    use crate::codec::CompressCtx;

    // Replace, don't duplicate: a path being (re-)written now supersedes whatever
    // rows the archive already held for it. The old blob bytes stay in the file as
    // dead payload — they are simply no longer referenced by any index row.
    let incoming: std::collections::HashSet<&str> =
        new_files.iter().map(|(rel, _)| rel.as_str()).collect();
    let rows_replaced = sink.drop_carried_paths(&incoming) as u64;

    // Append the new blob bytes to the file at the running blob cursor, mirroring
    // the compress pipeline (hash original bytes; store-raw if not smaller). One
    // writer, shared with the hot journal path.
    let blob_file = sink.file.clone();
    let mut ctx = CompressCtx::new(compression_level)?;
    let (paths, locs, cursor) =
        write_blobs(&blob_file, blob_append_offset, new_files, &mut ctx, policy)?;
    let blob_bytes_added = cursor - blob_append_offset;
    blob_file.sync_all()?;

    // Advance the sink's cursor past the freshly-written blob region so the new
    // data sub-index lands after the appended blobs (not over them).
    sink.cursor = cursor;

    // A DATA sub-index: seal it with the format-version-stamped schema. With the
    // bare `lookup_schema()` every archive this path writes — i.e. every archive
    // a writable holger repo or `cargo publish` produces — recorded no format
    // version, so the reader-side version pin had nothing to check.
    let batch = base_batch_from_rows(&paths, &locs)?;
    let schema = data_subindex_schema();
    let rows_added = batch.num_rows() as u64;
    sink.push_subindex(
        schema.as_ref(),
        &[batch],
        GroupKey { pkg_type: 0, repo: String::new(), module_name: String::new() },
    )?;

    let sealed_total_bytes = Box::new(sink).finish()?;

    Ok(AppendReport {
        rows_before,
        rows_replaced,
        rows_added,
        blob_append_offset,
        blob_bytes_added,
        sealed_total_bytes,
    })
}

/// Recover existing DATA rows for the resume accumulator. Reads the sorted
/// lookup reserved section (the cheapest source — already the exact per-chunk
/// rows, sorted) when present; otherwise concatenates the data sub-indexes.
pub(crate) fn recover_rows(
    path: &Path,
    entries: &[ManifestEntry],
) -> Result<(Vec<String>, Vec<ChunkLoc>)> {
    use std::io::{Read, Seek, SeekFrom};

    let mut file = File::open(path)?;

    // Prefer the sorted lookup section: it is exactly the base-schema per-chunk
    // rows, one cheap stream decode.
    if let Some(lk) = entries.iter().find(|e| e.module_name == LOOKUP_MODULE) {
        file.seek(SeekFrom::Start(lk.index_offset))?;
        let mut bytes = vec![0u8; lk.index_len as usize];
        file.read_exact(&mut bytes)?;
        return decode_base_rows(&bytes);
    }

    // Fallback: read every NON-reserved data sub-index and concatenate its rows.
    let mut paths = Vec::new();
    let mut locs = Vec::new();
    for e in entries {
        if is_reserved_module(&e.module_name) {
            continue;
        }
        file.seek(SeekFrom::Start(e.index_offset))?;
        let mut bytes = vec![0u8; e.index_len as usize];
        file.read_exact(&mut bytes)?;
        let (mut p, mut l) = decode_base_rows(&bytes)?;
        paths.append(&mut p);
        locs.append(&mut l);
    }
    Ok((paths, locs))
}

// ── inject-assert tests (the "tests inject values, not just no-crash" LAW) ──
// Every test here seals a real archive through a `CompressCtx`, so all of them
// need the codec — see the note on `archive::tests`.
#[cfg(all(test, feature = "openzl"))]
mod tests {
    use super::*;
    use crate::codec::CompressCtx;
    use crate::meta::{BlobMeta, ChunkMeta};
    use crate::{ArrowIpcSink, ZnippyArchive, ZnippyReader};
    use std::time::{SystemTime, UNIX_EPOCH};

    fn unique_dir(tag: &str) -> std::path::PathBuf {
        let ns = SystemTime::now().duration_since(UNIX_EPOCH).unwrap().as_nanos();
        let d = std::env::temp_dir().join(format!("znippy_append_{tag}_{ns}_{:?}", std::thread::current().id()));
        std::fs::create_dir_all(&d).unwrap();
        d
    }

    /// **The append path takes the skip decision, and it did not used to.**
    ///
    /// Until 2026-08-03 `write_files_into_sink` ran `CompressCtx::compress` over
    /// every byte of every appended file and kept the frame only when it came
    /// out smaller. For already-compressed input that is the entire codec cost
    /// paid to learn nothing — `compress_dir` has consulted `SkipPolicy` since
    /// it existed, and this path silently did not.
    ///
    /// The observable is chosen so it cannot be faked. A *highly compressible*
    /// payload is appended under a `.pack` name, which the extension table
    /// declares already-compressed. If the policy is honoured the blob is stored
    /// RAW and `blob_bytes_added` equals the input length; if the codec runs, the
    /// frame is far smaller and the number collapses. Asserting on CPU time
    /// would have been the honest measure of the bug but is not a test; this is
    /// the same decision made visible in a byte count.
    ///
    /// Seen RED by restoring the unconditional `ctx.compress(bytes)`:
    /// `blob_bytes_added` drops to a few hundred bytes and the assertion fires.
    #[test]
    fn an_append_honours_the_skip_policy_instead_of_compressing_everything() {
        let dir = unique_dir("skip_policy");
        let archive = dir.join("a.znippy");
        create_archive(&archive, &[("seed.txt".into(), b"seed".to_vec())], 3).unwrap();

        // 256 KiB of one byte: the codec would crush this to almost nothing.
        let squishy = vec![b'A'; 256 * 1024];
        let name = format!("pack-{}.pack", "0f".repeat(20));
        let report =
            append_files(&archive, &[(name.clone(), squishy.clone())], 3).unwrap();

        assert_eq!(
            report.blob_bytes_added,
            squishy.len() as u64,
            "a `.pack` entry must be stored RAW. {} bytes were written for a {}-byte input, so \
             the codec ran over a file the extension table already said was compressed",
            report.blob_bytes_added,
            squishy.len()
        );
        // …and it still reads back byte-exact, which is what stored-raw has to mean.
        assert_eq!(crate::get_file(&archive, &name).unwrap(), squishy);

        // The MIRROR, so the test above is not merely asserting that nothing is
        // ever compressed: the identical bytes under an ordinary name DO go
        // through the codec.
        let report2 =
            append_files(&archive, &[("plain.txt".into(), squishy.clone())], 3).unwrap();
        assert!(
            report2.blob_bytes_added < squishy.len() as u64 / 10,
            "an ordinary name must still be compressed; {} bytes for {}",
            report2.blob_bytes_added,
            squishy.len()
        );
        assert_eq!(crate::get_file(&archive, "plain.txt").unwrap(), squishy);

        // And an explicit batch-level claim overrules the name entirely.
        let report3 = append_files_with_policy(
            &archive,
            &[("also-plain.txt".into(), squishy.clone())],
            3,
            crate::SkipPolicy::already_compressed(),
        )
        .unwrap();
        assert_eq!(
            report3.blob_bytes_added,
            squishy.len() as u64,
            "`already_compressed()` must store raw whatever the name says"
        );

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

    /// Deterministic synthetic rows — distinct, lexicographically-spread paths
    /// (same flavour as the bench's `synth_blobs`, so the sort/fst do real work).
    fn synth(n: usize, salt: u64) -> Vec<(String, Vec<u8>)> {
        (0..n)
            .map(|i| {
                let g = (i.wrapping_mul(2_654_435_761) ^ salt as usize) % 1000;
                let p = format!("repo/grp{g:03}/file{:08}_{salt}.bin", i);
                let body = format!("payload {i} salt {salt} {}\n", "z".repeat(8 + (i % 40)));
                (p, body.into_bytes())
            })
            .collect()
    }

    /// Write a fresh sealed archive with `sink` (codec-compressed blobs + one
    /// base-schema data sub-index + the seal). Returns the sealed length. Generic
    /// over a closure so we can drive BOTH ArrowIpcSink (A) and the clone (B)
    /// through the *identical* fresh path and compare bytes.
    fn write_fresh<S: ArchiveMetaSink + 'static>(
        path: &Path,
        files: &[(String, Vec<u8>)],
        make_sink: impl FnOnce(Arc<File>, u64) -> S,
    ) -> u64 {
        let file = Arc::new(File::create(path).unwrap());
        let mut ctx = CompressCtx::new(3).unwrap();
        let mut blobs = Vec::new();
        let mut paths = Vec::new();
        let mut cursor = 0u64;
        for (fi, (rel, bytes)) in files.iter().enumerate() {
            let checksum = *blake3::hash(bytes).as_bytes();
            let frame = ctx.compress(bytes).unwrap();
            let (on_disk, compressed): (&[u8], bool) =
                if frame.len() < bytes.len() { (&frame, true) } else { (bytes, false) };
            file.write_all_at(on_disk, cursor).unwrap();
            let blob_offset = cursor;
            cursor += on_disk.len() as u64;
            paths.push(rel.clone());
            blobs.push(BlobMeta {
                blob_offset,
                blob_size: on_disk.len() as u64,
                chunk_meta: ChunkMeta {
                    fdata_offset: 0,
                    file_index: fi as u64,
                    chunk_seq: 0,
                    checksum,
                    compressed,
                    uncompressed_size: bytes.len() as u64,
                    compressed_size: on_disk.len() as u64,
                },
            });
        }
        let resolver = { let p = paths.clone(); move |fi: u64| p[fi as usize].clone() };
        let batch = crate::build_metadata_batch(&blobs, resolver, &[], &[]).unwrap();
        // Same DATA sub-index schema `write_files_into_sink` seals with, so this
        // helper stays the byte-for-byte reference for `create_archive`.
        let schema = data_subindex_schema();
        let mut sink = make_sink(file.clone(), cursor);
        sink.push_subindex(
            schema.as_ref(),
            &[batch],
            GroupKey { pkg_type: 0, repo: String::new(), module_name: String::new() },
        )
        .unwrap();
        Box::new(sink).finish().unwrap()
    }

    /// PARITY: the clone's fresh write+seal must produce a BYTE-IDENTICAL archive
    /// to the original `ArrowIpcSink` over the same rows. This is the structural
    /// proof that the clone is a zero-cost superset on the normal path (the
    /// throughput parity number lives in the bench; this asserts correctness).
    #[test]
    fn clone_fresh_path_is_byte_identical_to_original() {
        let dir = unique_dir("parity");
        let files = synth(2_000, 1);

        let a = dir.join("a.znippy");
        let b = dir.join("b.znippy");
        let len_a = write_fresh(&a, &files, ArrowIpcSink::new);
        let len_b = write_fresh(&b, &files, ArrowIpcSinkAppend::new);

        assert_eq!(len_a, len_b, "clone seal produced a different total length");
        let bytes_a = std::fs::read(&a).unwrap();
        let bytes_b = std::fs::read(&b).unwrap();
        assert_eq!(
            bytes_a, bytes_b,
            "clone's fresh write path is NOT byte-identical to ArrowIpcSink — parity broken"
        );
        let _ = std::fs::remove_dir_all(&dir);
    }

    /// RESUME: native append → reopen with the ORDINARY arrow-ipc reader → both
    /// the original AND the appended files read back byte-exact, and the index
    /// lists exactly old+new. Inject real bytes, assert real bytes out.
    #[test]
    fn native_append_roundtrips_old_and_new_files() {
        let dir = unique_dir("resume");
        let archive = dir.join("store.znippy");
        let orig = synth(1_500, 7);
        write_fresh(&archive, &orig, ArrowIpcSink::new);

        let added = synth(300, 99);
        let report = append_files(&archive, &added, 3).unwrap();
        assert_eq!(report.rows_before, orig.len() as u64, "must recover all original rows");
        assert_eq!(report.rows_added, added.len() as u64);
        assert!(report.blob_bytes_added > 0, "append must write new blob bytes");
        assert!(
            report.sealed_total_bytes > report.blob_append_offset,
            "re-sealed file must be larger than the old blob region"
        );

        // Reopen with the plain reader (no append awareness) and verify EVERY
        // file — original and appended — comes back byte-exact.
        let ar = ZnippyArchive::open(&archive).unwrap();
        let mut listed = ar.list_files().unwrap();
        listed.sort();
        let mut expected: Vec<String> =
            orig.iter().chain(added.iter()).map(|(p, _)| p.clone()).collect();
        expected.sort();
        assert_eq!(listed, expected, "index must list exactly old+new files after append");

        for (p, bytes) in orig.iter().chain(added.iter()) {
            let got = ar.extract_file(p).unwrap();
            assert_eq!(&got, bytes, "byte mismatch after append for {p}");
        }

        // Random-access lookup of an appended file via the rebuilt trie+lookup.
        let probe = &added[123].0;
        let chunks = crate::locate_file(&archive, probe).unwrap();
        assert!(!chunks.is_empty(), "appended file must be locatable via the re-sealed lookup");
        let _ = std::fs::remove_dir_all(&dir);
    }

    /// BOOTSTRAP: `create_archive` seeds a fresh archive that (a) is byte-identical
    /// to the proven fresh write path, (b) reads back byte-exact via the ordinary
    /// reader, and (c) can then be grown by `append_files` — the create→append flow
    /// holger's writable `put` relies on. Inject real bytes, assert real bytes out.
    #[test]
    fn create_archive_seeds_then_grows() {
        let dir = unique_dir("create");
        let seed = synth(40, 5);

        // (a) create_archive == write_fresh(_, ArrowIpcSinkAppend::new) byte-for-byte.
        let made = dir.join("made.znippy");
        let report = create_archive(&made, &seed, 3).unwrap();
        assert_eq!(report.rows_before, 0, "fresh archive has no prior rows");
        assert_eq!(report.rows_added, seed.len() as u64);

        let ref_path = dir.join("ref.znippy");
        write_fresh(&ref_path, &seed, ArrowIpcSinkAppend::new);
        assert_eq!(
            std::fs::read(&made).unwrap(),
            std::fs::read(&ref_path).unwrap(),
            "create_archive must be byte-identical to the proven fresh write path"
        );

        // (b) seeded files read back byte-exact via the plain reader.
        let ar = ZnippyArchive::open(&made).unwrap();
        for (p, bytes) in &seed {
            assert_eq!(&ar.extract_file(p).unwrap(), bytes, "seed byte mismatch for {p}");
        }

        // (c) append_files grows the bootstrapped archive; old+new read back exact.
        let added = synth(15, 88);
        let rep2 = append_files(&made, &added, 3).unwrap();
        assert_eq!(rep2.rows_before, seed.len() as u64, "append must recover seeded rows");
        assert_eq!(rep2.rows_added, added.len() as u64);

        let ar2 = ZnippyArchive::open(&made).unwrap();
        for (p, bytes) in seed.iter().chain(added.iter()) {
            assert_eq!(&ar2.extract_file(p).unwrap(), bytes, "byte mismatch after grow for {p}");
        }
        let _ = std::fs::remove_dir_all(&dir);
    }

    /// ZERO-COPY / no-filesystem: `create_archive_to_vec` builds a valid archive
    /// from in-memory bytes with NO staging tree and NO named output file (it seals
    /// into an anonymous fd). Assert the returned bytes are byte-identical to
    /// `create_archive`'s file output AND read back byte-exact through the reader.
    #[test]
    fn create_archive_to_vec_is_filesystem_free_and_round_trips() {
        let files = synth(24, 7);
        let (bytes, report) = create_archive_to_vec(&files, 3).unwrap();
        assert_eq!(report.rows_before, 0, "fresh in-memory archive has no prior rows");
        assert_eq!(report.rows_added, files.len() as u64);
        assert_eq!(bytes.len() as u64, report.sealed_total_bytes, "vec len == sealed size");

        let dir = unique_dir("tovec");
        // (1) byte-identical to the proven path (`create_archive` → file).
        let ref_path = dir.join("ref.znippy");
        create_archive(&ref_path, &files, 3).unwrap();
        assert_eq!(
            bytes,
            std::fs::read(&ref_path).unwrap(),
            "in-memory archive must be byte-identical to create_archive's file output"
        );
        // (2) the in-memory bytes ARE a real archive: persist + read back byte-exact.
        let p = dir.join("from_mem.znippy");
        std::fs::write(&p, &bytes).unwrap();
        let ar = ZnippyArchive::open(&p).unwrap();
        for (name, content) in &files {
            assert_eq!(&ar.extract_file(name).unwrap(), content, "byte mismatch for {name}");
        }
        let _ = std::fs::remove_dir_all(&dir);
    }

    /// Read an archive's recorded on-disk format version **exactly the way a
    /// reader does** — manifest, then the Arrow schema metadata of the FIRST
    /// non-reserved sub-index. This is byte-for-byte the same walk as
    /// `znippy_common::read_znippy_index`'s `check_format_version` call and as
    /// holger's independent `traits::recorded_format_version` pin, so what this
    /// helper returns is what those two see.
    fn recorded_format_version(path: &Path) -> Option<String> {
        use std::io::{Read, Seek, SeekFrom};

        use arrow::ipc::reader::StreamReader;

        let entries = crate::index::read_znippy_manifest(path).ok()?;
        let mut file = File::open(path).ok()?;
        for e in &entries {
            if is_reserved_module(&e.module_name) {
                continue;
            }
            file.seek(SeekFrom::Start(e.index_offset)).ok()?;
            let mut bytes = vec![0u8; e.index_len as usize];
            file.read_exact(&mut bytes).ok()?;
            let reader = StreamReader::try_new(std::io::Cursor::new(bytes), None).ok()?;
            return reader
                .schema()
                .metadata()
                .get(crate::index::FORMAT_VERSION_KEY)
                .cloned();
        }
        None
    }

    /// FORMAT VERSION: every archive the append/create path writes must RECORD
    /// its on-disk format version — on the fresh create, on the in-memory create,
    /// and still after a re-seal by `append_files` (which rebuilds the whole
    /// metadata tail, so a stamp that only survives the fresh path is no stamp).
    ///
    /// This is what makes a reader-side version pin mean anything for the
    /// archives this path produces — a writable holger repo and `cargo publish`
    /// both write through here. Sealing the data sub-index with the bare
    /// `lookup_schema()` (no schema metadata) recorded NO version at all, and the
    /// pin silently degraded to "undetermined → read as before" for exactly those
    /// archives.
    #[test]
    fn every_appended_archive_records_the_format_version() {
        let dir = unique_dir("fmtver");
        let want = crate::index::ZNIPPY_FORMAT_VERSION.to_string();

        // (a) fresh create.
        let made = dir.join("made.znippy");
        create_archive(&made, &synth(12, 3), 3).unwrap();
        assert_eq!(
            recorded_format_version(&made).as_deref(),
            Some(want.as_str()),
            "create_archive must stamp the on-disk format version"
        );

        // (b) after a re-seal: `append_files` rebuilds the metadata tail from
        // scratch, so the stamp has to be re-emitted, not merely inherited.
        append_files(&made, &synth(7, 91), 3).unwrap();
        assert_eq!(
            recorded_format_version(&made).as_deref(),
            Some(want.as_str()),
            "append_files must re-stamp the format version on the re-sealed archive"
        );

        // (c) the in-memory create path seals through the same code.
        let (bytes, _) = create_archive_to_vec(&synth(9, 4), 3).unwrap();
        let mem = dir.join("mem.znippy");
        std::fs::write(&mem, &bytes).unwrap();
        assert_eq!(
            recorded_format_version(&mem).as_deref(),
            Some(want.as_str()),
            "create_archive_to_vec must stamp the on-disk format version"
        );

        // The stamp did not cost readability: rows still read back byte-exact.
        let ar = ZnippyArchive::open(&mem).unwrap();
        for (p, body) in &synth(9, 4) {
            assert_eq!(&ar.extract_file(p).unwrap(), body, "byte mismatch for {p}");
        }
        let _ = std::fs::remove_dir_all(&dir);
    }

    /// METADATA SEARCH, end to end through a real archive:
    ///  (a) an archive sealed WITHOUT metadata reads back as `NoMetadata` — the
    ///      backward-compatible case, since that is exactly what every archive
    ///      written before this module looks like;
    ///  (b) sealing metadata makes "which entries carry key X" answerable, and
    ///      the answer resolves to real bytes via the ordinary reader;
    ///  (c) a re-seal by `append_files` CARRIES the metadata forward and merges
    ///      new rows into it — without that, every append would silently erase it;
    ///  (d) the metadata rows never leak into the data index or the lookup.
    #[test]
    fn metadata_is_searchable_survives_a_reseal_and_absence_is_reported_as_absence() {
        use crate::meta_index::{ArchiveMeta, MetaSearch, MetaTable, MetaValue, read_archive_meta};

        let dir = unique_dir("meta");
        let files = synth(30, 2);

        // (a) BACKWARD COMPAT: no metadata sealed → NoMetadata, not "found nothing".
        let plain = dir.join("plain.znippy");
        create_archive(&plain, &files, 3).unwrap();
        let m = read_archive_meta(&plain).unwrap();
        assert_eq!(m, ArchiveMeta::NoMetadata, "an archive with no meta section must say so");
        assert!(!m.is_searchable());
        assert_eq!(m.find_by_key("build-thing"), MetaSearch::NoMetadata);
        assert!(
            m.find_by_key("build-thing").hits().is_none(),
            "absence must NOT present itself as an empty result set"
        );

        // A present-but-EMPTY index is the other state, and it is distinguishable.
        let empty = dir.join("empty.znippy");
        create_archive_with_meta(&empty, &files, 3, Some(MetaTable::new())).unwrap();
        let me = read_archive_meta(&empty).unwrap();
        assert!(me.is_searchable(), "a sealed empty index WAS searched");
        assert!(me.index().is_some_and(|i| i.is_empty()));
        assert_eq!(me.find_by_key("build-thing"), MetaSearch::Hits(&[]));

        // (b) SEARCHABLE: two entries carry a build-thing, one does not.
        let wasm = b"\0asm\x01\0\0\0".to_vec();
        let (p0, p1, p2) = (files[0].0.clone(), files[1].0.clone(), files[2].0.clone());
        let mut t = MetaTable::new();
        t.insert(p0.clone(), "build-thing", MetaValue::Bytes(wasm.clone()))
            .insert(p0.clone(), "build-thing.abi", "wasi-p2")
            .insert(p1.clone(), "build-thing", MetaValue::Bytes(wasm.clone()))
            .insert(p2.clone(), "coverage", 0.5f64)
            .insert_archive("producer", "znippy");

        let ar = dir.join("meta.znippy");
        create_archive_with_meta(&ar, &files, 3, Some(t)).unwrap();

        let m = read_archive_meta(&ar).unwrap();
        let idx = m.index().expect("sealed index is present");
        assert_eq!(idx.len(), 5);
        let hits = m.find_by_key("build-thing").hits().unwrap();
        assert_eq!(hits.len(), 2, "exactly the two entries that carry one");
        let mut got: Vec<&str> = hits.iter().filter_map(|h| h.path()).collect();
        got.sort();
        let mut want = vec![p0.as_str(), p1.as_str()];
        want.sort();
        assert_eq!(got, want, "the search names the right ENTRIES");
        assert_eq!(hits[0].value.as_bytes(), Some(&wasm[..]), "and the right VALUE");
        assert_eq!(idx.archive_value("producer").and_then(MetaValue::as_str), Some("znippy"));
        assert_eq!(
            idx.find_by_prefix("build-thing").len(),
            3,
            "prefix sweeps build-thing + build-thing.abi"
        );

        // The hit resolves to real bytes without extracting anything else.
        let reader = ZnippyArchive::open(&ar).unwrap();
        let want_bytes = &files.iter().find(|(p, _)| *p == p0).unwrap().1;
        assert_eq!(&reader.extract_file(hits[0].path().unwrap()).unwrap(), want_bytes);

        // (c) A RE-SEAL keeps it, and merges.
        let added = synth(6, 77);
        let mut more = MetaTable::new();
        more.insert(added[0].0.clone(), "build-thing", MetaValue::Bytes(wasm.clone()));
        append_files_with_meta(&ar, &added, 3, Some(more)).unwrap();

        let m2 = read_archive_meta(&ar).unwrap();
        let hits2 = m2.find_by_key("build-thing").hits().unwrap();
        assert_eq!(hits2.len(), 3, "the append merged, it did not replace");
        assert_eq!(
            m2.index().unwrap().archive_value("producer").and_then(MetaValue::as_str),
            Some("znippy"),
            "the archive-level row survived the re-seal"
        );

        // A plain `append_files` (no meta argument) must also preserve it.
        append_files(&ar, &synth(3, 91), 3).unwrap();
        assert_eq!(
            read_archive_meta(&ar).unwrap().find_by_key("build-thing").hits().unwrap().len(),
            3,
            "an append that says nothing about metadata must not erase it"
        );

        // (d) The metadata rows are NOT file rows: the data index and the lookup
        //     see only the real entries.
        let ar2 = ZnippyArchive::open(&ar).unwrap();
        let n_files = files.len() + added.len() + 3;
        assert_eq!(
            ar2.file_count(),
            n_files,
            "metadata rows leaked into the data index"
        );
        let lookup_bytes = read_reserved_section_bytes(&ar, LOOKUP_MODULE).unwrap().unwrap();
        assert_eq!(
            decode_base_rows(&lookup_bytes).unwrap().0.len(),
            n_files,
            "metadata rows leaked into the random-access lookup"
        );
        for (p, bytes) in files.iter().chain(added.iter()) {
            assert_eq!(&ar2.extract_file(p).unwrap(), bytes, "byte mismatch for {p}");
        }
        let _ = std::fs::remove_dir_all(&dir);
    }

    /// Cached `ArchiveReader` must return byte-for-byte the SAME chunks and
    /// per-file metadata as the re-reading free functions — for a present file,
    /// an absent file, and a prefix window. Proves idea (B) is a pure read-side
    /// cache with zero behavioural drift from `locate_file` /
    /// `get_files_meta_with_prefix`.
    #[test]
    fn archive_reader_matches_free_functions() {
        let dir = unique_dir("reader");
        let archive = dir.join("store.znippy");
        let files = synth(600, 13);
        write_fresh(&archive, &files, ArrowIpcSink::new);

        let reader = crate::ArchiveReader::open(&archive).unwrap();
        assert_eq!(reader.row_count(), files.len(), "one chunk per synth file");

        // Present files: cached locate == free-function locate.
        for (p, _) in files.iter().step_by(37) {
            let cached = reader.locate(p);
            let free = crate::locate_file(&archive, p).unwrap();
            assert!(!cached.is_empty(), "cached reader failed to locate {p}");
            assert_eq!(cached, free, "cached vs free locate diverged for {p}");
        }

        // Absent file: both return empty.
        let missing = "repo/does/not/exist.bin";
        assert!(reader.locate(missing).is_empty());
        assert!(crate::locate_file(&archive, missing).unwrap().is_empty());

        // Whole-archive metadata parity.
        assert_eq!(
            reader.files_meta(),
            crate::get_all_files_meta(&archive).unwrap(),
            "cached files_meta diverged from get_all_files_meta"
        );

        // Prefix window parity (synth spreads paths across repo/grpNNN/).
        for prefix in ["repo/grp001/", "repo/grp0", "repo/", ""] {
            assert_eq!(
                reader.files_meta_with_prefix(prefix),
                crate::get_files_meta_with_prefix(&archive, prefix).unwrap(),
                "cached prefix meta diverged for {prefix:?}"
            );
        }

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

/// Decode an Arrow-IPC sub-index stream of the base schema into parallel
/// `(paths, locs)` vectors (mirrors `index::decode_lookup`, kept here so the
/// original stays untouched).
pub(crate) fn decode_base_rows(bytes: &[u8]) -> Result<(Vec<String>, Vec<ChunkLoc>)> {
    use arrow::ipc::reader::StreamReader;

    let reader = StreamReader::try_new(std::io::Cursor::new(bytes), None)
        .map_err(|e| anyhow!("append: lookup ipc reader: {e}"))?;
    let mut paths = Vec::new();
    let mut locs = Vec::new();
    for batch in reader {
        let batch = batch.map_err(|e| anyhow!("append: lookup batch decode: {e}"))?;
        let get = |n: &str| batch.column_by_name(n)
            .ok_or_else(|| anyhow!("append: lookup missing column {n}"));
        let p = get("relative_path")?.as_any().downcast_ref::<StringArray>()
            .ok_or_else(|| anyhow!("relative_path type"))?;
        let seq = get("chunk_seq")?.as_any().downcast_ref::<UInt32Array>()
            .ok_or_else(|| anyhow!("chunk_seq type"))?;
        let fdata = get("fdata_offset")?.as_any().downcast_ref::<UInt64Array>()
            .ok_or_else(|| anyhow!("fdata_offset type"))?;
        let comp = get("compressed")?.as_any().downcast_ref::<BooleanArray>()
            .ok_or_else(|| anyhow!("compressed type"))?;
        let usz = get("uncompressed_size")?.as_any().downcast_ref::<UInt64Array>()
            .ok_or_else(|| anyhow!("uncompressed_size type"))?;
        let boff = get("blob_offset")?.as_any().downcast_ref::<UInt64Array>()
            .ok_or_else(|| anyhow!("blob_offset type"))?;
        let bsz = get("blob_size")?.as_any().downcast_ref::<UInt64Array>()
            .ok_or_else(|| anyhow!("blob_size type"))?;
        let ck = get("checksum")?.as_any().downcast_ref::<FixedSizeBinaryArray>()
            .ok_or_else(|| anyhow!("checksum type"))?;
        for i in 0..batch.num_rows() {
            let mut c = [0u8; 32];
            c.copy_from_slice(ck.value(i));
            paths.push(p.value(i).to_string());
            locs.push(ChunkLoc {
                chunk_seq: seq.value(i),
                fdata_offset: fdata.value(i),
                blob_offset: boff.value(i),
                blob_size: bsz.value(i),
                uncompressed_size: usz.value(i),
                compressed: comp.value(i),
                checksum: c,
            });
        }
    }
    Ok((paths, locs))
}