znippy-common 0.9.14

// index.rs — v0.6 format: blobs stored inline, Arrow IPC is a pure metadata index.
//
// File layout:
//   [blob_0][blob_1]...[blob_N]  — compressed/raw chunk bytes, written as produced
//   [Arrow IPC stream]           — metadata index, written after all blobs
//   [8 bytes LE u64]             — byte offset where Arrow IPC starts (footer)
//
// Arrow schema columns:
//   relative_path, chunk_seq, fdata_offset, checksum_group,
//   compressed, uncompressed_size, blob_offset, blob_size, checksum

use std::collections::HashMap;
use std::fs::File;
use std::io::{Read, Seek, SeekFrom};
use std::path::{Path, PathBuf};
use std::sync::Arc;

use crate::common_config::StrategicConfig;
use crate::meta::BlobMeta;
use crate::plugin::ExtensionRow;
use crate::{decompress_archive};
use anyhow::Result;
use arrow::array::{
    Array, ArrayRef, BooleanBuilder, FixedSizeBinaryBuilder, Int8Builder, StringBuilder,
    UInt32Builder, UInt64Builder,
};
use arrow::datatypes::{DataType, Field, Schema};
use arrow::record_batch::RecordBatch;
use once_cell::sync::Lazy;

/// Per-file extension metadata carried into the Arrow index.
/// (plugin_type_id, extracted fields) — None for files with no matching plugin.
pub type FileExtMeta = Option<(i8, ExtensionRow)>;

/// v0.6 schema: Arrow IPC is a pure metadata index; blobs are stored inline before it.
/// Base index columns — present in every archive, type-agnostic.
/// Package-type modules contribute their own columns on top via `schema_fields()`;
/// the writer composes the on-disk schema with [`compose_index_schema`].
pub static ZNIPPY_INDEX_SCHEMA: Lazy<Arc<Schema>> = Lazy::new(|| {
    Arc::new(Schema::new(base_index_fields()))
});

fn base_index_fields() -> Vec<Field> {
    vec![
        Field::new("relative_path", DataType::Utf8, false),
        Field::new("chunk_seq", DataType::UInt32, false),
        Field::new("fdata_offset", DataType::UInt64, false),
        Field::new("compressed", DataType::Boolean, false),
        Field::new("uncompressed_size", DataType::UInt64, false),
        Field::new("blob_offset", DataType::UInt64, false),
        Field::new("blob_size", DataType::UInt64, false),
        Field::new("checksum", DataType::FixedSizeBinary(32), false),
    ]
}

pub fn znippy_index_schema() -> &'static Arc<Schema> {
    &ZNIPPY_INDEX_SCHEMA
}

/// Compose the on-disk index schema: base columns, plus — when a module contributes columns —
/// a `pkg_type` discriminator followed by the module's own `ext_fields`.
/// With no module fields, this is exactly the base schema (v0.6 layout, directly DuckDB-queryable).
pub fn compose_index_schema(ext_fields: &[Field]) -> Arc<Schema> {
    let mut fields = base_index_fields();
    if !ext_fields.is_empty() {
        fields.push(Field::new("pkg_type", DataType::Int8, true));
        fields.extend(ext_fields.iter().cloned());
    }
    Arc::new(Schema::new(fields))
}

/// On-disk archive format version, recorded in the Arrow index schema metadata
/// under `znippy_format_version`. The reader refuses any archive whose recorded
/// version is greater than this — a newer format it cannot safely parse — with a
/// clear error instead of panicking or mis-parsing. Bump only on a real on-disk
/// format change.
pub const ZNIPPY_FORMAT_VERSION: u32 = 3;

/// Schema-metadata key holding the on-disk [`ZNIPPY_FORMAT_VERSION`].
pub const FORMAT_VERSION_KEY: &str = "znippy_format_version";

/// Enforce that an archive's recorded format version is one this reader supports.
/// `metadata` is an Arrow schema's key/value metadata. A *newer* version yields a
/// clear error instead of a panic or silent mis-parse; equal/older versions — and
/// archives with no recorded version — read exactly as before.
pub fn check_format_version(metadata: &HashMap<String, String>) -> Result<()> {
    if let Some(raw) = metadata.get(FORMAT_VERSION_KEY) {
        let version: u32 = raw
            .parse()
            .map_err(|_| anyhow::anyhow!("invalid znippy archive format version {raw:?}"))?;
        anyhow::ensure!(
            version <= ZNIPPY_FORMAT_VERSION,
            "znippy archive format v{version} is newer than this reader supports \
             (max v{ZNIPPY_FORMAT_VERSION}) — upgrade znippy",
        );
    }
    Ok(())
}

/// Build Arrow schema metadata containing config (no checksum entries — those live in column).
pub fn build_arrow_metadata_for_config(config: &StrategicConfig) -> HashMap<String, String> {
    let mut m = HashMap::new();
    m.insert(FORMAT_VERSION_KEY.into(), ZNIPPY_FORMAT_VERSION.to_string());
    m.insert("max_core_in_flight".into(), config.max_core_in_flight.to_string());
    m.insert("max_core_in_compress".into(), config.max_core_in_compress.to_string());
    m.insert("max_mem_allowed".into(), config.max_mem_allowed.to_string());
    m.insert("min_free_memory_ratio".into(), config.min_free_memory_ratio.to_string());
    m.insert("file_split_block_size".into(), config.file_split_block_size.to_string());
    m.insert("max_chunks".into(), config.max_chunks.to_string());
    m.insert("compression_level".into(), config.compression_level.to_string());
    m.insert("zstd_output_buffer_size".into(), config.zstd_output_buffer_size.to_string());
    m
}

pub fn extract_config_from_arrow_metadata(
    metadata: &HashMap<String, String>,
) -> anyhow::Result<StrategicConfig> {
    Ok(StrategicConfig {
        max_core_allowed: 0,
        max_core_in_flight: metadata
            .get("max_core_in_flight")
            .ok_or_else(|| anyhow::anyhow!("Missing 'max_core_in_flight'"))?
            .parse()?,
        max_core_in_compress: metadata
            .get("max_core_in_compress")
            .ok_or_else(|| anyhow::anyhow!("Missing 'max_core_in_compress'"))?
            .parse()?,
        max_mem_allowed: metadata
            .get("max_mem_allowed")
            .ok_or_else(|| anyhow::anyhow!("Missing 'max_mem_allowed'"))?
            .parse()?,
        min_free_memory_ratio: metadata
            .get("min_free_memory_ratio")
            .ok_or_else(|| anyhow::anyhow!("Missing 'min_free_memory_ratio'"))?
            .parse()?,
        file_split_block_size: metadata
            .get("file_split_block_size")
            .ok_or_else(|| anyhow::anyhow!("Missing 'file_split_block_size'"))?
            .parse()?,
        max_chunks: metadata
            .get("max_chunks")
            .ok_or_else(|| anyhow::anyhow!("Missing 'max_chunks'"))?
            .parse()?,
        compression_level: metadata
            .get("compression_level")
            .ok_or_else(|| anyhow::anyhow!("Missing 'compression_level'"))?
            .parse()?,
        zstd_output_buffer_size: metadata
            .get("zstd_output_buffer_size")
            .ok_or_else(|| anyhow::anyhow!("Missing 'zstd_output_buffer_size'"))?
            .parse()?,
    })
}

/// Build the Arrow metadata index batch from blob positions.
///
/// Every row carries its own per-slice BLAKE3 in the `checksum` column
/// (over the chunk's uncompressed bytes).
pub fn build_metadata_batch<F>(
    blobs: &[BlobMeta],
    path_resolver: F,
    ext_meta: &[FileExtMeta],
    ext_fields: &[Field],
) -> arrow::error::Result<RecordBatch>
where
    F: Fn(u64) -> String,
{
    let len = blobs.len();

    let mut path_builder = StringBuilder::with_capacity(len, len * 64);
    let mut seq_builder = UInt32Builder::with_capacity(len);
    let mut fdata_builder = UInt64Builder::with_capacity(len);
    let mut compressed_builder = BooleanBuilder::with_capacity(len);
    let mut size_builder = UInt64Builder::with_capacity(len);
    let mut blob_offset_builder = UInt64Builder::with_capacity(len);
    let mut blob_size_builder = UInt64Builder::with_capacity(len);
    let mut checksum_builder = FixedSizeBinaryBuilder::with_capacity(len, 32);

    for blob in blobs {
        let m = &blob.chunk_meta;
        path_builder.append_value(path_resolver(m.file_index));
        seq_builder.append_value(m.chunk_seq);
        fdata_builder.append_value(m.fdata_offset);
        compressed_builder.append_value(m.compressed);
        size_builder.append_value(m.uncompressed_size);
        blob_offset_builder.append_value(blob.blob_offset);
        blob_size_builder.append_value(blob.blob_size);
        checksum_builder.append_value(m.checksum)?;
    }

    let mut columns: Vec<ArrayRef> = vec![
        Arc::new(path_builder.finish()),
        Arc::new(seq_builder.finish()),
        Arc::new(fdata_builder.finish()),
        Arc::new(compressed_builder.finish()),
        Arc::new(size_builder.finish()),
        Arc::new(blob_offset_builder.finish()),
        Arc::new(blob_size_builder.finish()),
        Arc::new(checksum_builder.finish()),
    ];

    // Module-contributed columns: a pkg_type discriminator + one column per ext field.
    if !ext_fields.is_empty() {
        let mut pkg_type_builder = Int8Builder::with_capacity(len);
        for blob in blobs {
            match ext_meta.get(blob.chunk_meta.file_index as usize).and_then(|x| x.as_ref()) {
                Some((type_id, _)) => pkg_type_builder.append_value(*type_id),
                None => pkg_type_builder.append_null(),
            }
        }
        columns.push(Arc::new(pkg_type_builder.finish()));

        for field in ext_fields {
            columns.push(build_ext_column(field, blobs, ext_meta));
        }
    }

    RecordBatch::try_new(compose_index_schema(ext_fields), columns)
}

/// Build one extension column from the per-file `ExtensionRow`, keyed by the field name.
/// Supports the Arrow types modules currently declare (Utf8, UInt32); other types yield nulls.
fn build_ext_column(field: &Field, blobs: &[BlobMeta], ext_meta: &[FileExtMeta]) -> ArrayRef {
    use crate::plugin::ExtensionValue;
    let len = blobs.len();
    let value_for = |blob: &BlobMeta| -> Option<&ExtensionValue> {
        ext_meta
            .get(blob.chunk_meta.file_index as usize)
            .and_then(|x| x.as_ref())
            .and_then(|(_, row)| row.fields.get(field.name()))
    };

    match field.data_type() {
        DataType::UInt32 => {
            let mut b = UInt32Builder::with_capacity(len);
            for blob in blobs {
                match value_for(blob) {
                    Some(ExtensionValue::U32(n)) => b.append_value(*n),
                    _ => b.append_null(),
                }
            }
            Arc::new(b.finish())
        }
        // Default to Utf8 for string-like fields (Str / OptStr).
        _ => {
            let mut b = StringBuilder::with_capacity(len, len * 16);
            for blob in blobs {
                match value_for(blob) {
                    Some(ExtensionValue::Str(s)) => b.append_value(s),
                    Some(ExtensionValue::OptStr(Some(s))) => b.append_value(s),
                    _ => b.append_null(),
                }
            }
            Arc::new(b.finish())
        }
    }
}

// ─── Multi-index container codec (planned v0.7, see design.md §6) ──────────────
//
// A multi-type archive holds several Arrow IPC index streams (one per (pkg_type, repo)
// sub-znippy, each with its own narrow schema), followed by a manifest stream that points
// at them, and a footer. The footer layout still *distinguishes* the legacy v0.6 trailer
// from the v0.7 one, but v0.6 archives are no longer readable — they are detected only to
// emit a clear "unsupported, re-compress with v0.7" error (see `read_znippy_index_filtered`):
//
//   v0.6 single index:  [...index...] [8-byte LE u64 index_offset]
//   v0.7 multi index:   [...sub-indexes...][manifest] [8-byte MAGIC] [8-byte LE u64 manifest_offset]
//
// A reader peeks the 8 bytes preceding the trailing offset: if they equal MAGIC it's a
// multi-index (v0.7) archive; otherwise it's a legacy v0.6 single index, which the reader
// rejects rather than parses.

/// Magic preceding the trailing offset that marks a multi-index (v0.7) archive.
pub const MULTI_INDEX_MAGIC: [u8; 8] = *b"ZNPYMIDX";

/// One entry in the multi-index manifest: a sub-znippy's identity + byte range.
#[derive(Debug, Clone, PartialEq)]
pub struct ManifestEntry {
    pub pkg_type: i8,
    pub repo: String,
    pub module_name: String,
    pub index_offset: u64,
    pub index_len: u64,
    pub row_count: u64,
}

/// Reserved `module_name` for the sorted random-access lookup sub-index.
/// Its rows are the base index columns re-sorted by `(relative_path, chunk_seq)`,
/// so external tools can `SELECT … WHERE relative_path = …` in O(log n) and the
/// native reader can binary-search it. Manifest readers filter this entry out of
/// the data sub-index set; [`read_znippy_lookup`] reads it explicitly.
pub const LOOKUP_MODULE: &str = "__znippy_lookup__";

/// Reserved `module_name` for the fst trie blob: an `fst::Map` of
/// `relative_path → first row index in the lookup sub-index`. Not Arrow IPC —
/// raw fst bytes — so it must never be parsed as a sub-index.
pub const TRIE_MODULE: &str = "__znippy_trie__";

/// Reserved `module_name` for the per-artifact detached CMS signatures
/// (feature `sign`). An Arrow IPC sub-index of `(relative_path, cms)` rows — one
/// detached CMS `SignedData` per file, over that file's merkle-folded chunk
/// hashes. Additive: readers that don't know it simply skip it (it is reserved),
/// so unsigned archives are byte-identical and old readers ignore signed ones.
pub const SIGN_ARTIFACTS_MODULE: &str = "__znippy_sign_artifacts__";

/// Reserved `module_name` for the per-archive detached CMS signature (feature
/// `sign`): raw CMS `SignedData` (DER) over the archive root digest.
pub const SIGN_ARCHIVE_MODULE: &str = "__znippy_sign_archive__";

/// Reserved `module_name` for the **searchable metadata sub-index**: typed
/// key/value rows, per entry and per archive, sorted by `(key, relative_path)`.
/// An Arrow IPC sub-index like the lookup — so "which entries carry key `X`" is
/// answered by one seek to the footer plus one read of this section, at a cost
/// set by the number of metadata rows and **not** by the payload size.
///
/// Reserved, therefore additive: a reader that predates it skips the entry, and
/// an archive that predates it simply has no such entry — which
/// [`crate::meta_index::read_archive_meta`] reports as
/// [`ArchiveMeta::NoMetadata`](crate::meta_index::ArchiveMeta::NoMetadata),
/// a state distinct from a present-but-empty index.
pub const META_MODULE: &str = "__znippy_meta__";

/// Reserved `module_name` for the **git oid index** written by the `git` package
/// format (`znippy-plugin-git`): a raw (non-Arrow) `stree` payload over the first
/// eight bytes of every object id, mapping to that object's first lookup row.
/// Raw bytes, never an Arrow IPC stream — it must never be parsed as a sub-index.
pub const GUNNAR_OID_MODULE: &str = "__gunnar_oid__";

/// Reserved `module_name` for the **git commit graph** written by the `git`
/// package format: an Arrow IPC sub-index of
/// `(oid, parents[], tree, committer_time, generation)`. Reserved so ordinary
/// `list`/`decompress`/iceberg readers skip it; queryable from DuckDB by slicing
/// its manifest byte range.
pub const GUNNAR_GRAPH_MODULE: &str = "__gunnar_graph__";

/// Reserved `module_name` for the **git reachability bitmaps** written by the
/// `git` package format: an Arrow IPC sub-index of `(commit_oid, bitmap)` where
/// `bitmap` is a serialized roaring bitmap over object ordinals.
pub const GUNNAR_REACH_MODULE: &str = "__gunnar_reach__";

/// Reserved `module_name` for the **git ref log** written by the `git` package
/// format: an Arrow IPC sub-index carrying **one RecordBatch per push**.
///
/// The batch boundary *is* the transaction boundary. There is no database here
/// (D18: redb was a database wedged between two archive formats, and it is
/// gone). A push is durable exactly when its IPC frame is complete on disk; a
/// frame torn by a crash is not half a push, it is a trailing byte run that no
/// reader will accept. Recovery is therefore "read forward while frames parse",
/// never a journal replay.
pub const GUNNAR_REFS_MODULE: &str = "__gunnar_refs__";

/// Reserved `module_name` for the **git secrets log** written by the `git`
/// package format: the same one-RecordBatch-per-push shape as
/// [`GUNNAR_REFS_MODULE`], carrying already-encrypted material only. znippy
/// never sees plaintext, and ciphertext is never compressed.
pub const GUNNAR_SECRETS_MODULE: &str = "__gunnar_secrets__";

/// Reserved `module_name` for the **delta map**: which stored chunks are delta
/// chunks, and what each one's base entry is.
///
/// # Why a reserved section and not a column
///
/// The obvious place is a `delta_base` column on the data sub-index. It is the
/// wrong place, and the reason is worth recording because the column is the
/// first thing anyone reaches for. `base_index_fields()` is the schema of EVERY
/// archive znippy has ever written; adding a column to it means a
/// [`ZNIPPY_FORMAT_VERSION`] bump, a new column every existing writer must
/// supply, and a changed on-disk shape for the iceberg sink and for anyone
/// pointing DuckDB at an archive — all to describe a property that almost no row
/// has.
///
/// A reserved section is additive by construction: the manifest readers already
/// skip reserved modules, so an archive without this one is exactly what it is
/// today and needs no version bump. `ZnippyArchive::open` joins it onto the
/// chunks it names; absent, every chunk is [`Stored`](crate::archive) and the
/// read path is byte-for-byte the one the golden digest pins.
///
/// The payload is an Arrow IPC stream of `(relative_path, chunk_seq, base_path)`.
/// It is independent of the objects, so an append must carry it — but it travels
/// as **decoded rows** rather than raw bytes (like `__meta__`, and unlike
/// `__gunnar_refs__`), because an append may add rows to it and two sections with
/// one module name is not a thing the manifest can express. That is why it is
/// NOT in [`CARRIED_RESERVED_MODULES`].
pub const ZNIPPY_DELTA_MODULE: &str = "__znippy_delta__";

/// Schema of the [`ZNIPPY_DELTA_MODULE`] section.
pub fn delta_map_schema() -> Arc<Schema> {
    Arc::new(Schema::new(vec![
        Field::new("relative_path", DataType::Utf8, false),
        Field::new("chunk_seq", DataType::UInt32, false),
        Field::new("base_path", DataType::Utf8, false),
    ]))
}

/// `pkg_type` discriminant carried by reserved (non-data) manifest entries.
pub const RESERVED_PKG_TYPE: i8 = i8::MIN;

/// Every reserved `module_name` this reader knows, in one place.
///
/// Kept as a slice rather than a hand-written `||` chain so that adding a
/// reserved module is a single edit that both [`is_reserved_module`] and the
/// guard test see — the failure mode being avoided is a new module name that is
/// written as reserved by the sink but classified as *data* by the manifest
/// readers, which silently corrupts `list`, `decompress`, the iceberg sink and
/// the manifest-count assertions.
pub const RESERVED_MODULES: &[&str] = &[
    LOOKUP_MODULE,
    TRIE_MODULE,
    SIGN_ARTIFACTS_MODULE,
    SIGN_ARCHIVE_MODULE,
    META_MODULE,
    GUNNAR_OID_MODULE,
    GUNNAR_GRAPH_MODULE,
    GUNNAR_REACH_MODULE,
    GUNNAR_REFS_MODULE,
    GUNNAR_SECRETS_MODULE,
    ZNIPPY_DELTA_MODULE,
];

/// A reserved manifest entry holds a derived structure (lookup / trie / detached
/// signatures / the git oid-index, commit-graph and reachability sections), not
/// file rows. The merge + manifest readers skip these so data consumers are
/// unaffected — which is exactly what makes those sections additive and
/// backward-compatible.
pub fn is_reserved_module(module_name: &str) -> bool {
    RESERVED_MODULES.contains(&module_name)
}

/// The reserved sections an **append must carry forward**, because they are
/// independent logs rather than derivations of the blobs beside them.
///
/// MEASURED 2026-08-04, gunnar: an object-carrying push removed
/// `__gunnar_refs__` from the archive (144 928 -> 146 396 bytes, section gone),
/// because [`crate::ArrowIpcSinkAppend::open_existing`] truncates the whole
/// metadata tail and re-supplies no reserved section it was not handed. All
/// reserved sections were treated alike, and they are not alike:
///
/// * **derived** — `__gunnar_oid__`, `__gunnar_graph__`, `__gunnar_reach__`,
///   plus `__lookup__` / `__trie__` — are functions of the objects in the
///   archive. When objects change they are *wrong*, and dropping them is
///   correct; the writer rebuilds them.
/// * **independent** — `__gunnar_refs__` and `__gunnar_secrets__` — are logs,
///   one RecordBatch per push. Nothing in a new push's objects reproduces the
///   ref history that came before it, so dropping them destroys data.
///
/// `__meta__` is carried by its own typed path (`open_existing` decodes it into
/// a `MetaTable`) and so is deliberately not listed here: it would then be
/// written twice.
///
/// This distinction is what the append path could not previously express, and it
/// blocks serving refs from the archive for most of a repository's life.
pub const CARRIED_RESERVED_MODULES: &[&str] = &[GUNNAR_REFS_MODULE, GUNNAR_SECRETS_MODULE];

/// Whether an append must carry this reserved section forward verbatim.
/// See [`CARRIED_RESERVED_MODULES`].
pub fn is_carried_reserved_module(module_name: &str) -> bool {
    CARRIED_RESERVED_MODULES.contains(&module_name)
}

/// Read the raw bytes of one reserved sub-section by module name, if present.
/// Public entry point for the signature layer (feature `sign`) and any other
/// reader that needs a reserved section's bytes without re-implementing the
/// manifest walk.
pub fn read_reserved_section_bytes(path: &Path, module_name: &str) -> Result<Option<Vec<u8>>> {
    read_reserved_section(path, module_name)
}

/// Schema of the lookup sub-index — identical to the base index columns. The
/// lookup is the same per-chunk rows, re-sorted by `(relative_path, chunk_seq)`
/// and stripped of any plugin columns, so one path's chunks are contiguous.
///
/// **Carries no schema metadata**, and that is correct for the *lookup*: the
/// lookup is a RESERVED sub-index, which every reader skips. It is NOT correct
/// for a **data** sub-index — use [`data_subindex_schema`] there, or the archive
/// records no format version and [`check_format_version`] has nothing to check.
pub fn lookup_schema() -> Arc<Schema> {
    Arc::new(Schema::new(base_index_fields()))
}

/// Schema for a base-column **data** sub-index: the same columns as
/// [`lookup_schema`], stamped with the archive metadata that
/// [`build_arrow_metadata_for_config`] produces — including
/// [`FORMAT_VERSION_KEY`].
///
/// The reader (and holger's independent reader-side pin) determines an archive's
/// on-disk format version from the Arrow schema metadata of the **first
/// non-reserved sub-index**. A writer that seals its data sub-index with the bare
/// [`lookup_schema`] therefore produces an archive that records no version at
/// all, and the version pin degrades to "undetermined → read as before" for
/// exactly those archives. Every data sub-index must be sealed with this.
pub fn data_subindex_schema() -> Arc<Schema> {
    Arc::new(Schema::new_with_metadata(
        base_index_fields(),
        build_arrow_metadata_for_config(&crate::common_config::CONFIG),
    ))
}

/// One chunk's location for single-file random access.
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct ChunkLoc {
    pub chunk_seq: u32,
    pub fdata_offset: u64,
    pub blob_offset: u64,
    pub blob_size: u64,
    pub uncompressed_size: u64,
    pub compressed: bool,
    pub checksum: [u8; 32],
}

/// What the trailing footer of an archive points at.
#[derive(Debug, Clone, PartialEq)]
pub enum IndexFooter {
    /// Legacy v0.6: a single Arrow IPC index began at this offset. Still detected
    /// so the reader can reject v0.6 archives with a clear "re-compress with v0.7"
    /// error — it is no longer a live read path.
    Single { index_offset: u64 },
    /// v0.7: the manifest stream begins at this offset.
    Multi { manifest_offset: u64 },
}

/// Interpret an archive's trailing bytes. `tail` must be the last 16 bytes of the file
/// (or last 8 for tiny v0.6 files — then it's always Single).
pub fn interpret_footer(tail: &[u8]) -> IndexFooter {
    let n = tail.len();
    let offset = u64::from_le_bytes(tail[n - 8..].try_into().unwrap());
    if n >= 16 && tail[n - 16..n - 8] == MULTI_INDEX_MAGIC {
        IndexFooter::Multi { manifest_offset: offset }
    } else {
        IndexFooter::Single { index_offset: offset }
    }
}

fn manifest_schema() -> Arc<Schema> {
    Arc::new(Schema::new(vec![
        Field::new("pkg_type", DataType::Int8, false),
        Field::new("repo", DataType::Utf8, false),
        Field::new("module_name", DataType::Utf8, false),
        Field::new("index_offset", DataType::UInt64, false),
        Field::new("index_len", DataType::UInt64, false),
        Field::new("row_count", DataType::UInt64, false),
    ]))
}

/// Serialize manifest entries to an Arrow IPC stream (itself DuckDB-readable).
pub fn write_manifest_bytes(entries: &[ManifestEntry]) -> Result<Vec<u8>> {
    use arrow::ipc::writer::StreamWriter;

    let len = entries.len();
    let mut pkg_type = Int8Builder::with_capacity(len);
    let mut repo = StringBuilder::with_capacity(len, len * 16);
    let mut module_name = StringBuilder::with_capacity(len, len * 16);
    let mut index_offset = UInt64Builder::with_capacity(len);
    let mut index_len = UInt64Builder::with_capacity(len);
    let mut row_count = UInt64Builder::with_capacity(len);
    for e in entries {
        pkg_type.append_value(e.pkg_type);
        repo.append_value(&e.repo);
        module_name.append_value(&e.module_name);
        index_offset.append_value(e.index_offset);
        index_len.append_value(e.index_len);
        row_count.append_value(e.row_count);
    }

    let schema = manifest_schema();
    let batch = RecordBatch::try_new(
        schema.clone(),
        vec![
            Arc::new(pkg_type.finish()),
            Arc::new(repo.finish()),
            Arc::new(module_name.finish()),
            Arc::new(index_offset.finish()),
            Arc::new(index_len.finish()),
            Arc::new(row_count.finish()),
        ],
    )?;

    let mut buf = Vec::new();
    {
        let mut w = StreamWriter::try_new(&mut buf, &schema)?;
        w.write(&batch)?;
        w.finish()?;
    }
    Ok(buf)
}

/// Parse a manifest Arrow IPC stream back into entries.
pub fn read_manifest_bytes(bytes: &[u8]) -> Result<Vec<ManifestEntry>> {
    use arrow::array::{Int8Array, StringArray, UInt64Array};
    use arrow::ipc::reader::StreamReader;

    let reader = StreamReader::try_new(std::io::Cursor::new(bytes), None)?;
    let mut out = Vec::new();
    for batch in reader {
        let batch = batch?;
        let col = |name: &str| batch.column_by_name(name)
            .ok_or_else(|| anyhow::anyhow!("manifest missing column {name}"));
        let pkg_type = col("pkg_type")?.as_any().downcast_ref::<Int8Array>()
            .ok_or_else(|| anyhow::anyhow!("pkg_type type"))?;
        let repo = col("repo")?.as_any().downcast_ref::<StringArray>()
            .ok_or_else(|| anyhow::anyhow!("repo type"))?;
        let module_name = col("module_name")?.as_any().downcast_ref::<StringArray>()
            .ok_or_else(|| anyhow::anyhow!("module_name type"))?;
        let index_offset = col("index_offset")?.as_any().downcast_ref::<UInt64Array>()
            .ok_or_else(|| anyhow::anyhow!("index_offset type"))?;
        let index_len = col("index_len")?.as_any().downcast_ref::<UInt64Array>()
            .ok_or_else(|| anyhow::anyhow!("index_len type"))?;
        let row_count = col("row_count")?.as_any().downcast_ref::<UInt64Array>()
            .ok_or_else(|| anyhow::anyhow!("row_count type"))?;
        for i in 0..batch.num_rows() {
            out.push(ManifestEntry {
                pkg_type: pkg_type.value(i),
                repo: repo.value(i).to_string(),
                module_name: module_name.value(i).to_string(),
                index_offset: index_offset.value(i),
                index_len: index_len.value(i),
                row_count: row_count.value(i),
            });
        }
    }
    Ok(out)
}

/// Read the Arrow IPC index from a v0.7 .znippy file.
///
/// Reads the 16-byte footer (8-byte `ZNPYMIDX` magic + 8-byte LE u64 manifest_offset),
/// parses the manifest, reads every sub-index, and merges all batches into one so callers
/// need no format-version awareness.
pub fn read_znippy_index(path: &Path) -> Result<(Arc<Schema>, Vec<RecordBatch>)> {
    read_znippy_index_filtered(path, &IndexFilter::default())
}

/// Selective sub-index filter: keep only sub-indexes whose manifest entry
/// matches. A `None` field matches anything, so `IndexFilter::default()` reads
/// the whole archive (what [`read_znippy_index`] does). Filtering happens at the
/// `(pkg_type, repo)` sub-index granularity — whole Arrow IPC streams are
/// skipped, never read — so a selective extract touches only the matching rows.
#[derive(Debug, Clone, Default)]
pub struct IndexFilter {
    /// Keep only this package-type discriminant (`ManifestEntry::pkg_type`).
    pub pkg_type: Option<i8>,
    /// Keep only this repo (`ManifestEntry::repo`).
    pub repo: Option<String>,
}

impl IndexFilter {
    pub fn is_empty(&self) -> bool {
        self.pkg_type.is_none() && self.repo.is_none()
    }
    fn matches(&self, e: &ManifestEntry) -> bool {
        self.pkg_type.is_none_or(|t| e.pkg_type == t)
            && self.repo.as_deref().is_none_or(|r| e.repo == r)
    }
}

/// Like [`read_znippy_index`] but keeps only sub-indexes matching `filter`.
pub fn read_znippy_index_filtered(
    path: &Path,
    filter: &IndexFilter,
) -> Result<(Arc<Schema>, Vec<RecordBatch>)> {
    let mut file = File::open(path)?;
    let file_len = file.metadata()?.len();
    anyhow::ensure!(file_len >= 16, "file too small to be a v0.7 znippy archive");

    file.seek(SeekFrom::End(-16))?;
    let mut tail = [0u8; 16];
    file.read_exact(&mut tail)?;

    match interpret_footer(&tail) {
        IndexFooter::Multi { manifest_offset } => {
            read_multi_index(&mut file, file_len, manifest_offset, filter)
        }
        IndexFooter::Single { .. } => {
            anyhow::bail!("v0.6 archives are not supported; re-compress with v0.7")
        }
    }
}

/// Read all sub-indexes from a v0.7 multi-index archive and concatenate them.
fn read_multi_index(
    file: &mut File,
    file_len: u64,
    manifest_offset: u64,
    filter: &IndexFilter,
) -> Result<(Arc<Schema>, Vec<RecordBatch>)> {
    use arrow::ipc::reader::StreamReader;

    // manifest lives between manifest_offset and (file_len − 16): 8-byte magic + 8-byte offset
    let manifest_end = file_len.checked_sub(16)
        .ok_or_else(|| anyhow::anyhow!("v0.7 archive too small"))?;
    anyhow::ensure!(manifest_offset <= manifest_end, "corrupt v0.7 manifest_offset");
    let manifest_len = (manifest_end - manifest_offset) as usize;

    file.seek(SeekFrom::Start(manifest_offset))?;
    let mut manifest_bytes = vec![0u8; manifest_len];
    file.read_exact(&mut manifest_bytes)?;
    let entries = read_manifest_bytes(&manifest_bytes)?;

    let mut all_batches: Vec<RecordBatch> = Vec::new();
    let mut schema: Option<Arc<Schema>> = None;

    for entry in &entries {
        // Skip derived structures (lookup sub-index, trie blob) — they are not
        // file-row data and must not be merged into the index.
        if is_reserved_module(&entry.module_name) {
            continue;
        }
        // Selective read: skip whole sub-indexes that don't match the filter.
        if !filter.matches(entry) {
            continue;
        }
        // Bound the attacker-controlled declared length against the real file
        // size before allocating, so a corrupt manifest can't force a huge alloc.
        anyhow::ensure!(
            entry.index_offset.checked_add(entry.index_len)
                .is_some_and(|end| end <= file_len),
            "sub-index for module {} out of bounds (offset={}, len={}, file_len={})",
            entry.module_name, entry.index_offset, entry.index_len, file_len
        );
        file.seek(SeekFrom::Start(entry.index_offset))?;
        let mut sub_bytes = vec![0u8; entry.index_len as usize];
        file.read_exact(&mut sub_bytes)?;
        let cursor = std::io::Cursor::new(sub_bytes);
        let reader = StreamReader::try_new(cursor, None)?;
        if schema.is_none() {
            let sub_schema = reader.schema();
            // Refuse archives written by a newer znippy than this reader supports,
            // before parsing any sub-index batches we might mis-interpret.
            check_format_version(sub_schema.metadata())?;
            schema = Some(sub_schema);
        }
        for batch in reader {
            all_batches.push(batch.map_err(|e| anyhow::anyhow!("sub-index read error: {}", e))?);
        }
    }

    let schema = schema.unwrap_or_else(|| Arc::new(Schema::new(base_index_fields())));

    // Merge all sub-index batches into one so callers stay format-agnostic.
    let merged = if all_batches.len() <= 1 {
        all_batches
    } else {
        let batch = arrow_select::concat::concat_batches(&schema, all_batches.iter())
            .map_err(|e| anyhow::anyhow!("concat sub-indexes: {}", e))?;
        vec![batch]
    };

    Ok((schema, merged))
}

/// Read the manifest from a v0.7 multi-index archive.
/// Returns an error if the file is a plain v0.6 single-index archive.
pub fn read_znippy_manifest(path: &Path) -> Result<Vec<ManifestEntry>> {
    let mut file = File::open(path)?;
    let file_len = file.metadata()?.len();
    anyhow::ensure!(file_len >= 16, "file too small to be a v0.7 archive");

    file.seek(SeekFrom::End(-16))?;
    let mut tail = [0u8; 16];
    file.read_exact(&mut tail)?;

    match interpret_footer(&tail) {
        IndexFooter::Single { .. } => {
            anyhow::bail!("not a v0.7 multi-index archive (no MULTI_INDEX_MAGIC)")
        }
        IndexFooter::Multi { manifest_offset } => {
            let manifest_end = file_len - 16;
            anyhow::ensure!(manifest_offset <= manifest_end, "corrupt manifest_offset");
            let manifest_len = (manifest_end - manifest_offset) as usize;
            file.seek(SeekFrom::Start(manifest_offset))?;
            let mut manifest_bytes = vec![0u8; manifest_len];
            file.read_exact(&mut manifest_bytes)?;
            let mut entries = read_manifest_bytes(&manifest_bytes)?;
            // Hide reserved (lookup/trie) entries from data-manifest consumers.
            entries.retain(|e| !is_reserved_module(&e.module_name));
            Ok(entries)
        }
    }
}

/// Read **all** manifest entries of a sealed v0.7 archive, *including* the
/// reserved (lookup/trie) ones, plus the manifest's byte offset.
///
/// Public entrypoint used by the append/resume sink
/// ([`ArrowIpcSinkAppend::open_existing`](crate::ArrowIpcSinkAppend::open_existing)):
/// it needs the reserved sections (to find the blob-region end and to recover
/// the sorted lookup) which `read_znippy_manifest` hides. Read-only — does not
/// touch the original write path.
pub fn read_znippy_full_manifest(path: &Path) -> Result<(Vec<ManifestEntry>, u64)> {
    let mut file = File::open(path)?;
    let file_len = file.metadata()?.len();
    read_full_manifest(&mut file, file_len)
}

/// Read all manifest entries, *including* reserved (lookup/trie) ones.
fn read_full_manifest(file: &mut File, file_len: u64) -> Result<(Vec<ManifestEntry>, u64)> {
    anyhow::ensure!(file_len >= 16, "file too small to be a v0.7 archive");
    file.seek(SeekFrom::End(-16))?;
    let mut tail = [0u8; 16];
    file.read_exact(&mut tail)?;
    let manifest_offset = match interpret_footer(&tail) {
        IndexFooter::Multi { manifest_offset } => manifest_offset,
        IndexFooter::Single { .. } => anyhow::bail!("not a v0.7 multi-index archive"),
    };
    let manifest_end = file_len.checked_sub(16)
        .ok_or_else(|| anyhow::anyhow!("v0.7 archive too small"))?;
    anyhow::ensure!(manifest_offset <= manifest_end, "corrupt manifest_offset");
    let manifest_len = (manifest_end - manifest_offset) as usize;
    file.seek(SeekFrom::Start(manifest_offset))?;
    let mut manifest_bytes = vec![0u8; manifest_len];
    file.read_exact(&mut manifest_bytes)?;
    Ok((read_manifest_bytes(&manifest_bytes)?, manifest_offset))
}

/// Read the raw bytes of one reserved sub-section (lookup or trie), if present.
fn read_reserved_section(path: &Path, module_name: &str) -> Result<Option<Vec<u8>>> {
    let mut file = File::open(path)?;
    let file_len = file.metadata()?.len();
    let (entries, _) = read_full_manifest(&mut file, file_len)?;
    let Some(entry) = entries.iter().find(|e| e.module_name == module_name) else {
        return Ok(None);
    };
    // Bound the attacker-controlled declared length against the real file size
    // before allocating, so a corrupt manifest can't force a huge allocation.
    anyhow::ensure!(
        entry.index_offset.checked_add(entry.index_len)
            .is_some_and(|end| end <= file_len),
        "reserved section {} out of bounds (offset={}, len={}, file_len={})",
        entry.module_name, entry.index_offset, entry.index_len, file_len
    );
    file.seek(SeekFrom::Start(entry.index_offset))?;
    let mut bytes = vec![0u8; entry.index_len as usize];
    file.read_exact(&mut bytes)?;
    Ok(Some(bytes))
}

/// Decode the lookup sub-index bytes into parallel column vectors (already sorted
/// by `(relative_path, chunk_seq)` on disk).
/// Read row `i` of an index/lookup `checksum` column as a 32-byte blake3 digest.
///
/// `FixedSizeBinaryArray` is one concrete Arrow type for *every* width, so a
/// `downcast_ref::<FixedSizeBinaryArray>()` succeeds just as happily for
/// `FixedSizeBinary(16)` as for `(32)`. Copying such a row straight into a
/// `[u8; 32]` panics ("source slice length (16) does not match destination slice
/// length (32)") — a process abort driven by an attacker-supplied schema, which
/// is exactly what the downcast-or-error style everywhere else in this file
/// exists to prevent. Check the declared width and return a clean `Err` instead.
/// (`ZnippyArchive::build_file_index` already guards this with the same test.)
fn checksum32(col: &arrow::array::FixedSizeBinaryArray, i: usize) -> Result<[u8; 32]> {
    if col.value_length() != 32 {
        return Err(anyhow::anyhow!(
            "checksum column has width {}, expected 32",
            col.value_length()
        ));
    }
    let mut ck = [0u8; 32];
    ck.copy_from_slice(col.value(i));
    Ok(ck)
}

fn decode_lookup(bytes: &[u8]) -> Result<LookupColumns> {
    use arrow::array::{BooleanArray, FixedSizeBinaryArray, StringArray, UInt32Array, UInt64Array};
    use arrow::ipc::reader::StreamReader;

    let reader = StreamReader::try_new(std::io::Cursor::new(bytes), None)?;
    let mut cols = LookupColumns::default();
    for batch in reader {
        let batch = batch?;
        let get = |n: &str| batch.column_by_name(n)
            .ok_or_else(|| anyhow::anyhow!("lookup missing column {n}"));
        let paths = get("relative_path")?.as_any().downcast_ref::<StringArray>()
            .ok_or_else(|| anyhow::anyhow!("relative_path type"))?;
        let chunk_seq = get("chunk_seq")?.as_any().downcast_ref::<UInt32Array>()
            .ok_or_else(|| anyhow::anyhow!("chunk_seq type"))?;
        let fdata = get("fdata_offset")?.as_any().downcast_ref::<UInt64Array>()
            .ok_or_else(|| anyhow::anyhow!("fdata_offset type"))?;
        let compressed = get("compressed")?.as_any().downcast_ref::<BooleanArray>()
            .ok_or_else(|| anyhow::anyhow!("compressed type"))?;
        let usize_col = get("uncompressed_size")?.as_any().downcast_ref::<UInt64Array>()
            .ok_or_else(|| anyhow::anyhow!("uncompressed_size type"))?;
        let blob_offset = get("blob_offset")?.as_any().downcast_ref::<UInt64Array>()
            .ok_or_else(|| anyhow::anyhow!("blob_offset type"))?;
        let blob_size = get("blob_size")?.as_any().downcast_ref::<UInt64Array>()
            .ok_or_else(|| anyhow::anyhow!("blob_size type"))?;
        let checksum = get("checksum")?.as_any().downcast_ref::<FixedSizeBinaryArray>()
            .ok_or_else(|| anyhow::anyhow!("checksum type"))?;
        for i in 0..batch.num_rows() {
            cols.paths.push(paths.value(i).to_string());
            let ck = checksum32(checksum, i)?;
            cols.locs.push(ChunkLoc {
                chunk_seq: chunk_seq.value(i),
                fdata_offset: fdata.value(i),
                blob_offset: blob_offset.value(i),
                blob_size: blob_size.value(i),
                uncompressed_size: usize_col.value(i),
                compressed: compressed.value(i),
                checksum: ck,
            });
        }
    }
    Ok(cols)
}

#[derive(Default)]
struct LookupColumns {
    paths: Vec<String>,
    locs: Vec<ChunkLoc>,
}

/// Locate every chunk of `target` for single-file random access.
///
/// Fast paths, in order: the fst trie (O(key length)) → binary search of the
/// sorted lookup sub-index (O(log n)) → linear scan of the merged main index
/// (O(n), for archives written before the lookup layer existed). Returns the
/// chunks sorted by `chunk_seq`, or an empty vec if `target` is not in the archive.
pub fn locate_file(path: &Path, target: &str) -> Result<Vec<ChunkLoc>> {
    if let Some(lookup_bytes) = read_reserved_section(path, LOOKUP_MODULE)? {
        let cols = decode_lookup(&lookup_bytes)?;
        let n = cols.paths.len();

        // Find any row whose path == target.
        let hit = if let Some(trie_bytes) = read_reserved_section(path, TRIE_MODULE)? {
            let map = fst::Map::new(trie_bytes)
                .map_err(|e| anyhow::anyhow!("trie open: {e}"))?;
            map.get(target.as_bytes()).map(|v| v as usize)
        } else {
            // Lookup is sorted by path; binary-search the path column.
            match cols.paths.binary_search_by(|p| p.as_str().cmp(target)) {
                Ok(i) => Some(i),
                Err(_) => None,
            }
        };

        let Some(hit) = hit else { return Ok(Vec::new()); };

        // The trie value is attacker-controlled: a corrupt/malicious archive can
        // map `target` to a row index past the (smaller) lookup table. Treat an
        // out-of-range hit as "not found" instead of indexing OOB below — this is
        // a remote-DoS guard on the random-access read path.
        if hit >= n { return Ok(Vec::new()); }

        // Expand to the contiguous run of rows sharing this path.
        let mut start = hit;
        while start > 0 && cols.paths[start - 1] == target { start -= 1; }
        let mut end = hit + 1;
        while end < n && cols.paths[end] == target { end += 1; }

        let mut out: Vec<ChunkLoc> = cols.locs[start..end].to_vec();
        out.sort_by_key(|c| c.chunk_seq);
        return Ok(out);
    }

    // Fallback: no lookup layer — scan the merged main index.
    locate_file_via_index(path, target)
}

/// Slim per-file (per-artifact) metadata for browsing an archive without
/// reading file bytes. One row per file, aggregated across its chunks.
#[derive(Debug, Clone, PartialEq)]
pub struct ArtifactMeta {
    pub relative_path: String,
    /// Total uncompressed size across all of the file's chunks.
    pub uncompressed_size: u64,
    pub chunk_count: u32,
    /// Whether the file's data was compressed (false on the stored-raw skip path).
    pub compressed: bool,
}

/// Metadata for **every** file in the archive, sorted by `relative_path`.
///
/// Reads only the lookup sub-index (not the file bytes); falls back to the main
/// index for archives written before the lookup layer.
pub fn get_all_files_meta(path: &Path) -> Result<Vec<ArtifactMeta>> {
    files_meta_impl(path, None)
}

/// Metadata for the files whose `relative_path` starts with `prefix` — for when
/// you don't want all files (e.g. one `group/artifact/` subtree).
///
/// When the trie is present this jumps straight to the first matching key via the
/// fst's ordered range (O(prefix) to seek, then O(matches)); otherwise it binary
/// -searches the sorted lookup; otherwise it scans the legacy main index.
pub fn get_files_meta_with_prefix(path: &Path, prefix: &str) -> Result<Vec<ArtifactMeta>> {
    files_meta_impl(path, Some(prefix))
}

fn files_meta_impl(path: &Path, prefix: Option<&str>) -> Result<Vec<ArtifactMeta>> {
    use fst::{IntoStreamer, Streamer};

    if let Some(lookup_bytes) = read_reserved_section(path, LOOKUP_MODULE)? {
        let cols = decode_lookup(&lookup_bytes)?; // sorted by (relative_path, chunk_seq)
        let n = cols.paths.len();

        // Resolve the contiguous row window [lo, hi) to aggregate.
        let (lo, hi) = match prefix {
            None | Some("") => (0, n),
            Some(pre) => {
                // Seek the first key >= prefix. Use the trie's ordered range when
                // present (the "trie search"); else binary-search the sorted paths.
                let lo = if let Some(trie_bytes) = read_reserved_section(path, TRIE_MODULE)? {
                    let map = fst::Map::new(trie_bytes)
                        .map_err(|e| anyhow::anyhow!("trie open: {e}"))?;
                    let mut stream = map.range().ge(pre.as_bytes()).into_stream();
                    match stream.next() {
                        Some((k, v)) if k.starts_with(pre.as_bytes()) => v as usize,
                        _ => return Ok(Vec::new()),
                    }
                } else {
                    cols.paths.partition_point(|p| p.as_str() < pre)
                };
                if lo >= n || !cols.paths[lo].starts_with(pre) {
                    return Ok(Vec::new());
                }
                let mut hi = lo;
                while hi < n && cols.paths[hi].starts_with(pre) {
                    hi += 1;
                }
                (lo, hi)
            }
        };

        // Aggregate contiguous chunk rows per path.
        let mut out = Vec::new();
        let mut i = lo;
        while i < hi {
            let p = &cols.paths[i];
            let mut total = 0u64;
            let mut count = 0u32;
            let mut compressed = false;
            while i < hi && &cols.paths[i] == p {
                total += cols.locs[i].uncompressed_size;
                count += 1;
                compressed |= cols.locs[i].compressed;
                i += 1;
            }
            out.push(ArtifactMeta {
                relative_path: p.clone(),
                uncompressed_size: total,
                chunk_count: count,
                compressed,
            });
        }
        return Ok(out);
    }

    files_meta_via_index(path, prefix)
}

/// A cached, open-once random-access reader over a sealed v0.7 archive.
///
/// The free functions [`locate_file`] and [`get_files_meta_with_prefix`] re-open
/// the file and re-read the footer + manifest + both reserved sections (the
/// lookup sub-index and the trie) on **every** call, and re-`decode_lookup` every
/// row each time — three manifest reads per lookup (analysis §1.5). For a
/// long-lived reader (a browsing UI, holger's dynamic side, selective restore)
/// that is pure repeated I/O.
///
/// `ArchiveReader::open` reads the footer, manifest, lookup sub-index and trie
/// **once**, decodes the lookup columns once, and builds the fst map once. Every
/// subsequent [`locate`](ArchiveReader::locate),
/// [`files_meta`](ArchiveReader::files_meta) and
/// [`files_meta_with_prefix`](ArchiveReader::files_meta_with_prefix) query is then
/// served from memory — an fst `get` (O(key)) or a binary search (O(log n)) over
/// the already-decoded columns — with **zero** further I/O and **zero**
/// per-call manifest reads. The on-disk format is untouched; results are
/// identical to the equivalent free function.
pub struct ArchiveReader {
    /// Relative paths, sorted by `(relative_path, chunk_seq)` (lookup order).
    paths: Vec<String>,
    /// Chunk locators, index-aligned with `paths`.
    locs: Vec<ChunkLoc>,
    /// `path -> first row index` fst over the sorted paths, when the archive
    /// carries a trie section (it always does for v0.7 seals). Absent ⇒ fall back
    /// to binary search over `paths`.
    trie: Option<fst::Map<Vec<u8>>>,
    /// The archive handle kept open across calls, so [`read_file`](Self::read_file)
    /// serves selective restore with **one** open + cached locate (positioned
    /// `pread`s only — never seeks a shared cursor).
    file: File,
    /// Real archive length, for the same pre-alloc bounds check the free
    /// `get_file` does against attacker-controlled blob offsets/sizes.
    file_len: u64,
}

impl ArchiveReader {
    /// Open a sealed v0.7 archive and cache its manifest + lookup + trie. Reads
    /// the manifest exactly once (vs three re-reads per free-function call).
    pub fn open(path: &Path) -> Result<Self> {
        let mut file = File::open(path)?;
        let file_len = file.metadata()?.len();
        let (entries, _) = read_full_manifest(&mut file, file_len)?;

        // Read one reserved section by module name from the already-open handle,
        // bounding the attacker-controlled declared length against the real file
        // size before allocating (same guard as `read_reserved_section`).
        let mut read_section = |module: &str| -> Result<Option<Vec<u8>>> {
            let Some(entry) = entries.iter().find(|e| e.module_name == module) else {
                return Ok(None);
            };
            anyhow::ensure!(
                entry.index_offset.checked_add(entry.index_len)
                    .is_some_and(|end| end <= file_len),
                "reserved section {} out of bounds (offset={}, len={}, file_len={})",
                entry.module_name, entry.index_offset, entry.index_len, file_len
            );
            file.seek(SeekFrom::Start(entry.index_offset))?;
            let mut bytes = vec![0u8; entry.index_len as usize];
            file.read_exact(&mut bytes)?;
            Ok(Some(bytes))
        };

        let lookup_bytes = read_section(LOOKUP_MODULE)?.ok_or_else(|| {
            anyhow::anyhow!("archive has no lookup sub-index (not a v0.7 sealed archive)")
        })?;
        let cols = decode_lookup(&lookup_bytes)?;
        let trie = match read_section(TRIE_MODULE)? {
            Some(tb) => Some(fst::Map::new(tb).map_err(|e| anyhow::anyhow!("trie open: {e}"))?),
            None => None,
        };

        Ok(Self { paths: cols.paths, locs: cols.locs, trie, file, file_len })
    }

    /// Total number of chunk rows cached (across all files).
    pub fn row_count(&self) -> usize {
        self.paths.len()
    }

    /// Read a single file's bytes by relative path — the cached, held-open
    /// equivalent of [`get_file`](crate::get_file). Uses the in-memory locate
    /// (no manifest re-read, no per-call `decode_lookup`) and the archive handle
    /// opened once in [`open`](Self::open), then `pread`s + decompresses +
    /// blake3-verifies each chunk. This is the selective-restore entry point:
    /// hold one reader and call `read_file` per artifact instead of paying a full
    /// index re-parse on every `get_file`.
    ///
    /// Returns an error if `target` is not present in the archive.
    pub fn read_file(&self, target: &str) -> Result<Vec<u8>> {
        let chunks = self.locate(target);
        anyhow::ensure!(!chunks.is_empty(), "file not found in archive: {target}");
        crate::decompress::reassemble_file(&self.file, self.file_len, target, &chunks)
    }

    /// Locate every chunk of `target`, sorted by `chunk_seq` — the cached
    /// equivalent of [`locate_file`], but with no I/O after `open`.
    pub fn locate(&self, target: &str) -> Vec<ChunkLoc> {
        let n = self.paths.len();
        let hit = match &self.trie {
            Some(map) => map.get(target.as_bytes()).map(|v| v as usize),
            None => self.paths.binary_search_by(|p| p.as_str().cmp(target)).ok(),
        };
        let Some(hit) = hit else { return Vec::new(); };

        // The trie value is attacker-controlled (see `locate_file`): a malicious
        // archive can map `target` to a row index past the cached lookup table.
        // Treat an out-of-range hit as "not found" instead of panicking below.
        if hit >= n { return Vec::new(); }

        // Expand to the contiguous run of rows sharing this path.
        let mut start = hit;
        while start > 0 && self.paths[start - 1] == target { start -= 1; }
        let mut end = hit + 1;
        while end < n && self.paths[end] == target { end += 1; }

        let mut out: Vec<ChunkLoc> = self.locs[start..end].to_vec();
        out.sort_by_key(|c| c.chunk_seq);
        out
    }

    /// Per-file metadata for **every** file, sorted by `relative_path` — the
    /// cached equivalent of [`get_all_files_meta`].
    pub fn files_meta(&self) -> Vec<ArtifactMeta> {
        self.aggregate(0, self.paths.len())
    }

    /// Per-file metadata for files whose path starts with `prefix` — the cached
    /// equivalent of [`get_files_meta_with_prefix`].
    pub fn files_meta_with_prefix(&self, prefix: &str) -> Vec<ArtifactMeta> {
        let (lo, hi) = self.window(prefix);
        self.aggregate(lo, hi)
    }

    /// Resolve the contiguous `[lo, hi)` row window matching `prefix` — trie
    /// range-seek when present, else binary search over the sorted paths. Empty
    /// prefix selects all rows; a non-matching prefix returns `(0, 0)`.
    fn window(&self, prefix: &str) -> (usize, usize) {
        use fst::{IntoStreamer, Streamer};
        let n = self.paths.len();
        if prefix.is_empty() {
            return (0, n);
        }
        let lo = match &self.trie {
            Some(map) => {
                let mut stream = map.range().ge(prefix.as_bytes()).into_stream();
                match stream.next() {
                    Some((k, v)) if k.starts_with(prefix.as_bytes()) => v as usize,
                    _ => return (0, 0),
                }
            }
            None => self.paths.partition_point(|p| p.as_str() < prefix),
        };
        if lo >= n || !self.paths[lo].starts_with(prefix) {
            return (0, 0);
        }
        let mut hi = lo;
        while hi < n && self.paths[hi].starts_with(prefix) {
            hi += 1;
        }
        (lo, hi)
    }

    /// Aggregate the contiguous chunk rows in `[lo, hi)` into one `ArtifactMeta`
    /// per file (rows are already grouped by path in lookup order).
    fn aggregate(&self, lo: usize, hi: usize) -> Vec<ArtifactMeta> {
        let mut out = Vec::new();
        let mut i = lo;
        while i < hi {
            let p = &self.paths[i];
            let mut total = 0u64;
            let mut count = 0u32;
            let mut compressed = false;
            while i < hi && &self.paths[i] == p {
                total += self.locs[i].uncompressed_size;
                count += 1;
                compressed |= self.locs[i].compressed;
                i += 1;
            }
            out.push(ArtifactMeta {
                relative_path: p.clone(),
                uncompressed_size: total,
                chunk_count: count,
                compressed,
            });
        }
        out
    }
}

/// Legacy fallback: aggregate per-file metadata from the merged main index.
fn files_meta_via_index(path: &Path, prefix: Option<&str>) -> Result<Vec<ArtifactMeta>> {
    use arrow::array::{BooleanArray, StringArray, UInt64Array};

    let (schema, batches) = read_znippy_index(path)?;
    let batch = match batches.len() {
        0 => return Ok(Vec::new()),
        1 => batches.into_iter().next().unwrap(),
        _ => arrow_select::concat::concat_batches(&schema, batches.iter())?,
    };
    // Downcast-or-error: the schema is attacker-controlled, so a wrong column
    // type must surface as an `Err`, never an `unwrap` panic (DoS on `get`/meta).
    // Downcast-or-error: the schema is attacker-controlled, so a wrong column
    // type must surface as an `Err`, never an `unwrap` panic (DoS on `get`/meta).
    let col = |n: &str| batch.column_by_name(n)
        .ok_or_else(|| anyhow::anyhow!("index missing column {n}"));
    let paths = col("relative_path")?.as_any().downcast_ref::<StringArray>()
        .ok_or_else(|| anyhow::anyhow!("index column relative_path has unexpected type"))?;
    let usize_col = col("uncompressed_size")?.as_any().downcast_ref::<UInt64Array>()
        .ok_or_else(|| anyhow::anyhow!("index column uncompressed_size has unexpected type"))?;
    let compressed = col("compressed")?.as_any().downcast_ref::<BooleanArray>()
        .ok_or_else(|| anyhow::anyhow!("index column compressed has unexpected type"))?;

    // Main index rows are not path-sorted, so aggregate via a map.
    let mut agg: HashMap<&str, (u64, u32, bool)> = HashMap::new();
    for i in 0..batch.num_rows() {
        let p = paths.value(i);
        if let Some(pre) = prefix {
            if !p.starts_with(pre) { continue; }
        }
        let e = agg.entry(p).or_insert((0, 0, false));
        e.0 += usize_col.value(i);
        e.1 += 1;
        e.2 |= compressed.value(i);
    }
    let mut out: Vec<ArtifactMeta> = agg.into_iter().map(|(p, (sz, c, comp))| ArtifactMeta {
        relative_path: p.to_string(),
        uncompressed_size: sz,
        chunk_count: c,
        compressed: comp,
    }).collect();
    out.sort_by(|a, b| a.relative_path.cmp(&b.relative_path));
    Ok(out)
}

/// O(n) fallback for archives written before the lookup sub-index existed.
fn locate_file_via_index(path: &Path, target: &str) -> Result<Vec<ChunkLoc>> {
    use arrow::array::{BooleanArray, FixedSizeBinaryArray, StringArray, UInt32Array, UInt64Array};

    let (schema, batches) = read_znippy_index(path)?;
    let batch = match batches.len() {
        0 => return Ok(Vec::new()),
        1 => batches.into_iter().next().unwrap(),
        _ => arrow_select::concat::concat_batches(&schema, batches.iter())?,
    };
    // Downcast-or-error: the schema is attacker-controlled, so a wrong column
    // type must surface as an `Err`, never an `unwrap` panic (DoS on `get`).
    let col = |n: &str| batch.column_by_name(n)
        .ok_or_else(|| anyhow::anyhow!("index missing column {n}"));
    let paths = col("relative_path")?.as_any().downcast_ref::<StringArray>()
        .ok_or_else(|| anyhow::anyhow!("index column relative_path has unexpected type"))?;
    let chunk_seq = col("chunk_seq")?.as_any().downcast_ref::<UInt32Array>()
        .ok_or_else(|| anyhow::anyhow!("index column chunk_seq has unexpected type"))?;
    let fdata = col("fdata_offset")?.as_any().downcast_ref::<UInt64Array>()
        .ok_or_else(|| anyhow::anyhow!("index column fdata_offset has unexpected type"))?;
    let compressed = col("compressed")?.as_any().downcast_ref::<BooleanArray>()
        .ok_or_else(|| anyhow::anyhow!("index column compressed has unexpected type"))?;
    let usize_col = col("uncompressed_size")?.as_any().downcast_ref::<UInt64Array>()
        .ok_or_else(|| anyhow::anyhow!("index column uncompressed_size has unexpected type"))?;
    let blob_offset = col("blob_offset")?.as_any().downcast_ref::<UInt64Array>()
        .ok_or_else(|| anyhow::anyhow!("index column blob_offset has unexpected type"))?;
    let blob_size = col("blob_size")?.as_any().downcast_ref::<UInt64Array>()
        .ok_or_else(|| anyhow::anyhow!("index column blob_size has unexpected type"))?;
    let checksum = col("checksum")?.as_any().downcast_ref::<FixedSizeBinaryArray>()
        .ok_or_else(|| anyhow::anyhow!("index column checksum has unexpected type"))?;

    let mut out = Vec::new();
    for i in 0..batch.num_rows() {
        if paths.value(i) == target {
            let ck = checksum32(checksum, i)?;
            out.push(ChunkLoc {
                chunk_seq: chunk_seq.value(i),
                fdata_offset: fdata.value(i),
                blob_offset: blob_offset.value(i),
                blob_size: blob_size.value(i),
                uncompressed_size: usize_col.value(i),
                compressed: compressed.value(i),
                checksum: ck,
            });
        }
    }
    out.sort_by_key(|c| c.chunk_seq);
    Ok(out)
}

pub fn is_probably_compressed(path: &Path) -> bool {
    if let Some(ext) = path.extension().and_then(|e| e.to_str()) {
        let ext = ext.to_ascii_lowercase();
        matches!(
            ext.as_str(),
            "zip" | "gz" | "bz2" | "xz" | "lz" | "lzma" | "7z" | "rar" | "cab"
                | "jar" | "war" | "ear" | "zst" | "sz" | "lz4" | "tgz" | "txz"
                | "tbz" | "apk" | "dmg" | "deb" | "rpm" | "arrow" | "mpeg" | "mpg"
                | "jpeg" | "jpg" | "gif" | "bmp" | "png" | "crate" | "znippy"
                | "zdata" | "parquet" | "webp" | "webm"
                // already-compressed / high-entropy payloads (Skidbladnir bundles
                // carry these): age = encrypted (never shrinks), iso = squashfs
                // (already xz/zstd), pdf = Flate streams. NB: pg_dump plain-SQL
                // (`.dump`/`.sql`) is deliberately NOT here — it compresses well.
                | "age" | "iso" | "pdf"
                // git's pack directory — the bulk of any real `.git`, and of
                // gunnar's cold tier. `pack` holds deflated objects; `idx`,
                // `midx` and `bitmap` hold object ids (hash output) and EWAH
                // bitmaps, all incompressible. NB: `.rev` is deliberately NOT
                // here — it is a permutation of 0..N as big-endian u32s, whose
                // high bytes are mostly zero, so it genuinely does compress.
                | "pack" | "idx" | "midx" | "bitmap"
        )
    } else {
        false
    }
}

/// The name-only decision, kept as the fast path it has always been.
///
/// This is step 2 of the policy in [`crate::precompressed`] — it sees a path and
/// nothing else, so it cannot reach an entry that carries no extension (a git
/// loose object is named for its oid) and it believes a name that lies. Callers
/// holding real bytes should go through
/// [`SkipPolicy`](crate::precompressed::SkipPolicy), which runs this first and
/// then refines it with a magic-byte probe.
pub fn should_skip_compression(path: &Path) -> bool {
    is_probably_compressed(path)
}

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

    #[test]
    fn skips_already_compressed_and_encrypted() {
        for p in ["secrets/secrets.age", "infra/talos/talos-1.13.iso", "s3/doc.pdf", "x.PDF"] {
            assert!(should_skip_compression(Path::new(p)), "should skip {p}");
        }
    }

    #[test]
    fn compresses_plain_dump_and_config() {
        // pg_dump plain-SQL + declarative config must still be compressed.
        for p in ["dbdump/njord.dump", "njord.sql", "lakespec.json", "njordconf/serverspec.json"] {
            assert!(!should_skip_compression(Path::new(p)), "should compress {p}");
        }
    }

    /// A packfile and its index are the two largest things in any real `.git`,
    /// and neither was on the extension table.
    #[test]
    fn skips_git_pack_directory_artefacts() {
        for p in [
            ".git/objects/pack/pack-9f2c.pack",
            ".git/objects/pack/pack-9f2c.idx",
            ".git/objects/pack/multi-pack-index.midx",
            ".git/objects/pack/pack-9f2c.bitmap",
        ] {
            assert!(should_skip_compression(Path::new(p)), "should skip {p}");
        }
    }

    /// The pack directory is not uniformly incompressible, and the table must
    /// not be widened to the whole directory. `.rev` is a permutation of
    /// `0..N` as big-endian u32s — its high bytes are mostly zero.
    #[test]
    fn still_compresses_the_compressible_git_artefacts() {
        for p in [
            ".git/objects/pack/pack-9f2c.rev",
            ".git/objects/pack/pack-9f2c.promisor",
            ".git/COMMIT_EDITMSG",
            ".git/config",
        ] {
            assert!(!should_skip_compression(Path::new(p)), "should compress {p}");
        }
    }
}

#[derive(Debug, Default)]
pub struct VerifyReport {
    pub total_files: usize,
    pub verified_files: usize,
    pub corrupt_files: usize,
    pub total_bytes: u64,
    pub verified_bytes: u64,
    pub corrupt_bytes: u64,
    pub chunks: u64,
}

pub fn list_archive_contents(path: &Path) -> Result<()> {
    let (_schema, batches) = read_znippy_index(path)?;
    for batch in &batches {
        let paths = batch
            .column_by_name("relative_path")
            .and_then(|c| c.as_any().downcast_ref::<arrow::array::StringArray>())
            .ok_or_else(|| anyhow::anyhow!("missing relative_path column"))?;
        let sizes = batch
            .column_by_name("uncompressed_size")
            .and_then(|c| c.as_any().downcast_ref::<arrow::array::UInt64Array>())
            .ok_or_else(|| anyhow::anyhow!("missing uncompressed_size column"))?;
        let chunk_seqs = batch
            .column_by_name("chunk_seq")
            .and_then(|c| c.as_any().downcast_ref::<arrow::array::UInt32Array>());
        let group_ids = batch
            .column_by_name("group_id")
            .and_then(|c| c.as_any().downcast_ref::<arrow::array::StringArray>());
        let artifact_ids = batch
            .column_by_name("artifact_id")
            .and_then(|c| c.as_any().downcast_ref::<arrow::array::StringArray>());
        let versions = batch
            .column_by_name("version")
            .and_then(|c| c.as_any().downcast_ref::<arrow::array::StringArray>());
        for i in 0..batch.num_rows() {
            // Only print once per file (chunk_seq == 0)
            if let Some(seqs) = chunk_seqs {
                if seqs.value(i) != 0 {
                    continue;
                }
            }
            if let (Some(g), Some(a), Some(v)) = (group_ids, artifact_ids, versions) {
                if !g.is_null(i) {
                    println!(
                        "{}\t{}\t{}:{}:{}",
                        paths.value(i),
                        sizes.value(i),
                        g.value(i),
                        a.value(i),
                        v.value(i)
                    );
                    continue;
                }
            }
            println!("{}\t{}", paths.value(i), sizes.value(i));
        }
    }
    Ok(())
}

pub fn verify_archive_integrity(path: &Path) -> Result<VerifyReport> {
    let out_dir = PathBuf::from("/dev/null");
    decompress_archive(path, false, &out_dir)
}

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

    #[test]
    fn current_and_older_versions_are_accepted() {
        let mut m = HashMap::new();
        // No version recorded (legacy / pre-version archives): read as before.
        check_format_version(&m).unwrap();
        // Exactly the supported version.
        m.insert(FORMAT_VERSION_KEY.into(), ZNIPPY_FORMAT_VERSION.to_string());
        check_format_version(&m).unwrap();
        // An older version.
        m.insert(FORMAT_VERSION_KEY.into(), (ZNIPPY_FORMAT_VERSION - 1).to_string());
        check_format_version(&m).unwrap();
    }

    #[test]
    fn newer_version_is_rejected_clearly() {
        let mut m = HashMap::new();
        m.insert(FORMAT_VERSION_KEY.into(), (ZNIPPY_FORMAT_VERSION + 1).to_string());
        let err = check_format_version(&m).unwrap_err().to_string();
        assert!(err.contains("newer than this reader supports"), "got: {err}");
        assert!(err.contains("upgrade znippy"), "got: {err}");
    }

    #[test]
    fn writer_records_the_current_version() {
        let meta = build_arrow_metadata_for_config(&crate::common_config::CONFIG);
        assert_eq!(
            meta.get(FORMAT_VERSION_KEY).map(String::as_str),
            Some(ZNIPPY_FORMAT_VERSION.to_string().as_str())
        );
    }
}

#[cfg(test)]
mod checksum_width_tests {
    use super::*;
    use arrow::array::FixedSizeBinaryArray;

    /// `FixedSizeBinaryArray` is ONE concrete Arrow type for every width, so the
    /// `downcast_ref::<FixedSizeBinaryArray>()` in `decode_lookup` /
    /// `locate_file_via_index` succeeds just as happily for a hostile archive
    /// declaring `FixedSizeBinary(16)`. The old code then did
    /// `[0u8; 32].copy_from_slice(col.value(i))`, which panics — a process abort
    /// driven purely by an attacker-supplied schema, on the `znippy get` path.
    #[test]
    fn narrow_checksum_column_errors_instead_of_panicking() {
        let narrow = FixedSizeBinaryArray::try_from_iter([[0u8; 16]].into_iter()).unwrap();
        assert_eq!(narrow.value_length(), 16);
        let err = checksum32(&narrow, 0).expect_err("a 16-byte checksum column must be an Err");
        assert!(
            err.to_string().contains("width 16"),
            "error must name the bad width, got: {err}"
        );
    }

    #[test]
    fn wide_checksum_column_errors_instead_of_truncating() {
        let wide = FixedSizeBinaryArray::try_from_iter([[7u8; 64]].into_iter()).unwrap();
        assert!(checksum32(&wide, 0).is_err(), "a 64-byte checksum column must be an Err");
    }

    #[test]
    fn correct_width_checksum_is_read_verbatim() {
        let mut digest = [0u8; 32];
        for (i, b) in digest.iter_mut().enumerate() {
            *b = i as u8;
        }
        let col = FixedSizeBinaryArray::try_from_iter([digest].into_iter()).unwrap();
        assert_eq!(checksum32(&col, 0).unwrap(), digest);
    }
}

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

    /// The regression this guards is specific and it has teeth: a module name
    /// that the sink writes with `RESERVED_PKG_TYPE` but that
    /// [`is_reserved_module`] does not recognise is classified as a **data**
    /// sub-index. `read_multi_index` then merges its rows into the file index and
    /// `read_znippy_manifest` reports it as a data section — corrupting `list`,
    /// `decompress`, the iceberg sink and every manifest-count assertion.
    ///
    /// So the assertion is not "the constant exists"; it is "every reserved
    /// module name in the catalog is classified reserved, and nothing else is".
    #[test]
    fn every_reserved_module_is_classified_reserved() {
        for m in RESERVED_MODULES {
            assert!(
                is_reserved_module(m),
                "'{m}' is in RESERVED_MODULES but is_reserved_module says it is DATA — \
                 its rows would be merged into the file index"
            );
        }
    }

    /// The three `git` package-format modules, named literally. If someone
    /// removes one from [`RESERVED_MODULES`], the loop above still passes
    /// (it iterates whatever is left) — this is the test that goes red.
    #[test]
    fn the_git_format_modules_are_reserved() {
        for m in [
            GUNNAR_OID_MODULE,
            GUNNAR_GRAPH_MODULE,
            GUNNAR_REACH_MODULE,
            GUNNAR_REFS_MODULE,
            GUNNAR_SECRETS_MODULE,
        ] {
            assert!(is_reserved_module(m), "git package-format module '{m}' must be reserved");
            assert!(
                RESERVED_MODULES.contains(&m),
                "git package-format module '{m}' must be in the catalog"
            );
        }
        assert_eq!(GUNNAR_OID_MODULE, "__gunnar_oid__");
        assert_eq!(GUNNAR_GRAPH_MODULE, "__gunnar_graph__");
        assert_eq!(GUNNAR_REACH_MODULE, "__gunnar_reach__");
        assert_eq!(GUNNAR_REFS_MODULE, "__gunnar_refs__");
        assert_eq!(GUNNAR_SECRETS_MODULE, "__gunnar_secrets__");
    }

    /// The other direction: a data module must NOT be swept up as reserved, or
    /// its file rows would silently vanish from `list` and `decompress`.
    #[test]
    fn ordinary_module_names_stay_data() {
        for m in ["maven", "git", "rust", "", "__gunnar__", "gunnar_oid", "__gunnar_oid", "objects"] {
            assert!(!is_reserved_module(m), "'{m}' must be treated as a DATA sub-index");
        }
    }

    /// No duplicates and no accidental empty entry in the catalog.
    #[test]
    fn the_catalog_is_well_formed() {
        let mut seen = std::collections::HashSet::new();
        for m in RESERVED_MODULES {
            assert!(!m.is_empty(), "empty reserved module name");
            assert!(seen.insert(*m), "duplicate reserved module name '{m}'");
        }
        assert_eq!(seen.len(), RESERVED_MODULES.len());
    }
}