znippy-plugin-git 0.1.1

//! End-to-end: a real `.znippy` archive of real git objects, sealed through the
//! ordinary compress path with the `git` handler and the three reserved
//! sections, then read back.
//!
//! Every assertion here is on **applied output** — the bytes that came out, the
//! rows the index actually points at, the paths `list` actually returns — never
//! on a value handed straight back to the code that set it (LAW 2).
//!
//! sha256 is the primary arm, because gunnar defaults to sha256 (D11). sha1 gets
//! its own pass so the 40-hex path is exercised too.

use std::collections::HashMap;
use std::fs;
use std::path::{Path, PathBuf};
use std::sync::Arc;

use tempfile::TempDir;
use znippy_common::arrow::array::StringArray;
use znippy_common::arrow::ipc::reader::StreamReader;
use znippy_common::plugin::PluginRegistry;
use znippy_common::{
    ArchiveMetaSink, ArrowIpcSink, GUNNAR_GRAPH_MODULE, GUNNAR_OID_MODULE, GUNNAR_REACH_MODULE,
    GUNNAR_REFS_MODULE, GUNNAR_SECRETS_MODULE, LOOKUP_MODULE, RESERVED_MODULES, TRIE_MODULE,
    ZnippyArchive, ZnippyReader, get_file, is_reserved_module, read_reserved_section_bytes,
    read_znippy_manifest,
};
use znippy_compress::compress_dir;
use znippy_plugin_git::{
    GitHashKind, GitIndexBuilder, GitObjectKind, GitOidIndex, NativeGitPlugin, ReachPolicy,
    RefLog, RefUpdate, SecretUpdate, SecretsLog, canonical, read_graph, read_reach, read_refs,
    read_secrets,
};

// ── fixture: a two-commit repository, built by hand ───────────────────────────

struct Fixture {
    hash: GitHashKind,
    /// oid hex → canonical bytes, in the order they were created.
    objects: Vec<(String, Vec<u8>)>,
    blob_a: String,
    blob_b: String,
    tree1: String,
    tree2: String,
    commit1: String,
    commit2: String,
    big: String,
}

fn tree_bytes(entries: &[(&str, &str)]) -> Vec<u8> {
    let mut payload = Vec::new();
    for (name, oid_hex) in entries {
        payload.extend_from_slice(b"100644 ");
        payload.extend_from_slice(name.as_bytes());
        payload.push(0);
        payload.extend_from_slice(&hex::decode(oid_hex).unwrap());
    }
    canonical(GitObjectKind::Tree, &payload)
}

fn build_fixture(hash: GitHashKind) -> Fixture {
    let mut objects: Vec<(String, Vec<u8>)> = Vec::new();
    let mut add = |bytes: Vec<u8>| -> String {
        let oid = hash.oid_hex_of(&bytes);
        objects.push((oid.clone(), bytes));
        oid
    };

    let blob_a = add(canonical(GitObjectKind::Blob, b"hello\n"));
    let blob_b = add(canonical(GitObjectKind::Blob, b"world\n"));
    // 24 MiB: larger than the 10 MiB `file_split_block_size`, so this object's
    // chunk run in the lookup is more than one row. That is the case where a
    // "row of the object" that is not the FIRST row silently reads garbage.
    let big = add(canonical(
        GitObjectKind::Blob,
        &b"the quick brown fox jumps over the lazy dog\n".repeat(600_000),
    ));

    let tree1 = add(tree_bytes(&[("a.txt", &blob_a)]));
    let tree2 = add(tree_bytes(&[
        ("a.txt", &blob_a),
        ("b.txt", &blob_b),
        ("big.txt", &big),
    ]));

    let commit1 = add(canonical(
        GitObjectKind::Commit,
        format!(
            "tree {tree1}\nauthor A <a@x> 1700000000 +0000\ncommitter A <a@x> 1700000000 +0000\n\nfirst\n"
        )
        .as_bytes(),
    ));
    let commit2 = add(canonical(
        GitObjectKind::Commit,
        format!(
            "tree {tree2}\nparent {commit1}\nauthor A <a@x> 1700000100 +0000\ncommitter A <a@x> 1700000100 +0000\n\nsecond\n"
        )
        .as_bytes(),
    ));

    Fixture { hash, objects, blob_a, blob_b, tree1, tree2, commit1, commit2, big }
}

/// Lay the objects out as a directory of oid-named files and seal it through the
/// ordinary `compress_dir` path with the `git` handler + the reserved sections.
fn seal(fx: &Fixture, dir: &Path, archive: &Path, policy: ReachPolicy) {
    fs::create_dir_all(dir).unwrap();
    let mut builder = GitIndexBuilder::new(fx.hash).with_reach_policy(policy);
    for (oid, bytes) in &fx.objects {
        fs::write(dir.join(oid), bytes).unwrap();
        let derived = builder.push_canonical(bytes).unwrap();
        assert_eq!(&derived, oid, "builder must derive the same oid the fixture did");
    }
    let reserved = builder.into_reserved_builder();
    let registry = PluginRegistry::with_plugin(Box::new(NativeGitPlugin::new()));

    let report = compress_dir(
        &dir.to_path_buf(),
        &archive.to_path_buf(),
        false,
        Some(&registry),
        None,
        Some(Box::new(move |f, b| {
            Box::new(ArrowIpcSink::new(f, b).with_reserved_builder(reserved))
                as Box<dyn ArchiveMetaSink>
        })),
    )
    .expect("compress_dir");
    assert_eq!(report.total_files, fx.objects.len() as u64);
}

/// `relative_path` at row `row` of the sealed lookup sub-index — decoded from the
/// archive, not from anything the writer kept in memory.
fn lookup_path_at(archive: &Path, row: u64) -> String {
    let bytes = read_reserved_section_bytes(archive, LOOKUP_MODULE)
        .unwrap()
        .expect("archive must carry a lookup sub-index");
    let reader = StreamReader::try_new(std::io::Cursor::new(bytes), None).unwrap();
    let mut seen = 0u64;
    for batch in reader {
        let batch = batch.unwrap();
        let paths = batch
            .column_by_name("relative_path")
            .unwrap()
            .as_any()
            .downcast_ref::<StringArray>()
            .unwrap();
        let n = batch.num_rows() as u64;
        if row < seen + n {
            return paths.value((row - seen) as usize).to_string();
        }
        seen += n;
    }
    panic!("lookup row {row} out of range ({seen} rows)");
}

/// How many lookup rows carry `path` — i.e. the object's chunk count.
fn lookup_rows_for(archive: &Path, path: &str) -> usize {
    let bytes = read_reserved_section_bytes(archive, LOOKUP_MODULE).unwrap().unwrap();
    let reader = StreamReader::try_new(std::io::Cursor::new(bytes), None).unwrap();
    let mut n = 0usize;
    for batch in reader {
        let batch = batch.unwrap();
        let paths = batch
            .column_by_name("relative_path")
            .unwrap()
            .as_any()
            .downcast_ref::<StringArray>()
            .unwrap();
        n += (0..batch.num_rows()).filter(|&i| paths.value(i) == path).count();
    }
    n
}

fn paths() -> (TempDir, PathBuf, PathBuf) {
    let tmp = TempDir::new().unwrap();
    let src = tmp.path().join("objects");
    let archive = tmp.path().join("repo.znippy");
    (tmp, src, archive)
}

// ── the tests ─────────────────────────────────────────────────────────────────

/// The headline claim: objects go in, and come back out **byte-exact through the
/// stree index**, with their oid re-derived from the bytes that came back.
///
/// Re-hashing is what makes this an applied-output assertion. A read that
/// returned the right length of the wrong bytes, or the right bytes of the wrong
/// object, fails here.
#[test]
fn sha256_objects_roundtrip_byte_exact_through_the_stree_index() {
    let fx = build_fixture(GitHashKind::Sha256);
    let (_tmp, src, archive) = paths();
    seal(&fx, &src, &archive, ReachPolicy::default());

    let index = GitOidIndex::open(&archive).unwrap().expect("archive must carry __gunnar_oid__");
    assert_eq!(index.len(), fx.objects.len());
    assert_eq!(index.hash_kind(), GitHashKind::Sha256);

    for (oid, original) in &fx.objects {
        assert_eq!(oid.len(), 64, "sha256 oids are 64 hex chars");
        let hit = index.lookup_hex(oid).unwrap_or_else(|| panic!("oid {oid} missed the index"));

        // The row the index points at is really this object's FIRST lookup row.
        assert_eq!(
            lookup_path_at(&archive, hit.lookup_row),
            *oid,
            "lookup_row {} does not point at {oid}",
            hit.lookup_row
        );

        // And the bytes come back exactly, re-hashing to the same oid.
        let got = get_file(&archive, oid).unwrap();
        assert_eq!(&got, original, "bytes differ for {oid}");
        assert_eq!(
            GitHashKind::Sha256.oid_hex_of(&got),
            *oid,
            "read-back bytes do not hash to their own oid"
        );
    }

    // The multi-chunk object really is multi-chunk — otherwise the "first row of
    // the run" assertion above is testing nothing.
    assert!(
        lookup_rows_for(&archive, &fx.big) > 1,
        "the 24 MiB object should span several lookup rows"
    );
}

#[test]
fn sha1_objects_roundtrip_too() {
    let fx = build_fixture(GitHashKind::Sha1);
    let (_tmp, src, archive) = paths();
    seal(&fx, &src, &archive, ReachPolicy::default());

    let index = GitOidIndex::open(&archive).unwrap().unwrap();
    assert_eq!(index.hash_kind(), GitHashKind::Sha1);
    for (oid, original) in &fx.objects {
        assert_eq!(oid.len(), 40);
        assert!(index.lookup_hex(oid).is_some(), "oid {oid} missed");
        let got = get_file(&archive, oid).unwrap();
        assert_eq!(&got, original);
        assert_eq!(GitHashKind::Sha1.oid_hex_of(&got), *oid);
    }
    // An oid of the other width must not resolve.
    assert!(index.lookup_hex(&"a".repeat(64)).is_none());
}

/// The batch path is the one that matters for pack serving. It must agree with
/// the serial path on every object, and must miss on an absent oid.
#[test]
fn batch_lookup_resolves_every_object_and_misses_the_absent_one() {
    let fx = build_fixture(GitHashKind::Sha256);
    let (_tmp, src, archive) = paths();
    seal(&fx, &src, &archive, ReachPolicy::default());
    let index = GitOidIndex::open(&archive).unwrap().unwrap();

    let mut queries: Vec<String> = fx.objects.iter().map(|(o, _)| o.clone()).collect();
    let absent = "f".repeat(64);
    queries.push(absent.clone());
    let refs: Vec<&str> = queries.iter().map(|s| s.as_str()).collect();

    let batched = index.lookup_batch_hex(&refs);
    assert_eq!(batched.len(), queries.len());
    for (i, q) in queries.iter().enumerate() {
        assert_eq!(batched[i], index.lookup_hex(q), "batch/serial disagree on {q}");
    }
    assert!(batched.last().unwrap().is_none(), "an absent oid must miss");
    assert!(batched[..fx.objects.len()].iter().all(|h| h.is_some()));
}

/// The reserved sections must be invisible to every ordinary reader. If they are
/// not, `list` gains phantom entries, `decompress` tries to write them as files
/// and the manifest counts drift.
#[test]
fn the_reserved_git_modules_are_invisible_to_ordinary_readers() {
    let fx = build_fixture(GitHashKind::Sha256);
    let (_tmp, src, archive) = paths();
    seal(&fx, &src, &archive, ReachPolicy::default());

    // `znippy list` / ZnippyArchive: exactly the objects, nothing else.
    let listed = ZnippyArchive::open(&archive).unwrap().list_files().unwrap();
    let mut listed_sorted = listed.clone();
    listed_sorted.sort();
    let mut expected: Vec<String> = fx.objects.iter().map(|(o, _)| o.clone()).collect();
    expected.sort();
    assert_eq!(listed_sorted, expected, "`list` must show the objects and only the objects");
    for p in &listed {
        assert!(!p.starts_with("__"), "reserved section leaked into the file list: {p}");
    }

    // The data manifest hides every reserved entry.
    let data_entries = read_znippy_manifest(&archive).unwrap();
    for e in &data_entries {
        assert!(
            !is_reserved_module(&e.module_name),
            "reserved module '{}' surfaced as a DATA sub-index",
            e.module_name
        );
    }
    for m in [GUNNAR_OID_MODULE, GUNNAR_GRAPH_MODULE, GUNNAR_REACH_MODULE, LOOKUP_MODULE, TRIE_MODULE]
    {
        assert!(
            !data_entries.iter().any(|e| e.module_name == m),
            "'{m}' must not appear in the data manifest"
        );
    }

    // …but the sections are genuinely there, in the FULL manifest.
    let (full, _) = znippy_common::read_znippy_full_manifest(&archive).unwrap();
    for m in [GUNNAR_OID_MODULE, GUNNAR_GRAPH_MODULE, GUNNAR_REACH_MODULE] {
        let e = full
            .iter()
            .find(|e| e.module_name == m)
            .unwrap_or_else(|| panic!("'{m}' missing from the full manifest"));
        assert!(e.index_len > 0, "'{m}' section is empty");
        assert!(RESERVED_MODULES.contains(&m.as_ref()));
    }

    // And a full decompress writes only the objects — no `__gunnar_*` files.
    let out = TempDir::new().unwrap();
    znippy_common::decompress_archive(&archive, true, out.path()).unwrap();
    let mut written: Vec<String> = walk(out.path())
        .into_iter()
        .map(|p| p.file_name().unwrap().to_string_lossy().to_string())
        .collect();
    written.sort();
    assert_eq!(written, expected, "decompress must reconstruct the objects and nothing else");
}

fn walk(root: &Path) -> Vec<PathBuf> {
    let mut out = Vec::new();
    let mut stack = vec![root.to_path_buf()];
    while let Some(d) = stack.pop() {
        for e in fs::read_dir(&d).unwrap() {
            let p = e.unwrap().path();
            if p.is_dir() {
                stack.push(p);
            } else {
                out.push(p);
            }
        }
    }
    out
}

/// The commit graph, read back out of the sealed archive.
#[test]
fn the_commit_graph_carries_generations_and_parents() {
    let fx = build_fixture(GitHashKind::Sha256);
    let (_tmp, src, archive) = paths();
    seal(&fx, &src, &archive, ReachPolicy::default());

    let nodes = read_graph(&archive).unwrap().expect("archive must carry __gunnar_graph__");
    assert_eq!(nodes.len(), 2, "two commits, and no blobs or trees");
    let by: HashMap<&str, _> = nodes.iter().map(|n| (n.oid.as_str(), n)).collect();

    let c1 = by[fx.commit1.as_str()];
    let c2 = by[fx.commit2.as_str()];
    assert_eq!(c1.generation, 1);
    assert_eq!(c2.generation, 2);
    assert_eq!(c1.parents, Vec::<String>::new());
    assert_eq!(c2.parents, vec![fx.commit1.clone()]);
    assert_eq!(c1.tree.as_deref(), Some(fx.tree1.as_str()));
    assert_eq!(c2.tree.as_deref(), Some(fx.tree2.as_str()));
    assert_eq!(c1.committer_time, Some(1_700_000_000));
    assert_eq!(c2.committer_time, Some(1_700_000_100));

    // The generation ordering is the thing the graph is FOR: an ancestry test
    // becomes an integer comparison.
    assert!(c1.generation < c2.generation, "an ancestor must have a lower generation");
}

/// The reachability bitmaps, read back and used the way pack serving uses them:
/// `want − have` as an ANDNOT, resolved back to real oids through the oid index.
#[test]
fn reachability_bitmaps_answer_want_minus_have() {
    let fx = build_fixture(GitHashKind::Sha256);
    let (_tmp, src, archive) = paths();
    seal(&fx, &src, &archive, ReachPolicy { max_commits: 16 });

    let entries = read_reach(&archive).unwrap().expect("archive must carry __gunnar_reach__");
    let by: HashMap<&str, _> = entries.iter().map(|e| (e.commit.as_str(), &e.bitmap)).collect();
    let want: roaring::RoaringBitmap = (*by
        .get(fx.commit2.as_str())
        .unwrap_or_else(|| panic!("no bitmap for the tip commit")))
    .clone();
    let have: roaring::RoaringBitmap = (*by
        .get(fx.commit1.as_str())
        .unwrap_or_else(|| panic!("no bitmap for the root commit")))
    .clone();

    let index = GitOidIndex::open(&archive).unwrap().unwrap();
    let ordinal_to_oid: HashMap<u32, String> = fx
        .objects
        .iter()
        .map(|(o, _)| {
            let hit = index.lookup_hex(o).unwrap();
            (hit.ordinal, o.clone())
        })
        .collect();

    let resolve = |bm: &roaring::RoaringBitmap| -> Vec<String> {
        let mut v: Vec<String> = bm.iter().map(|o| ordinal_to_oid[&o].clone()).collect();
        v.sort();
        v
    };

    // c1 reaches: itself, tree1, blob_a. Nothing from the second commit.
    let mut expect_have = vec![fx.commit1.clone(), fx.tree1.clone(), fx.blob_a.clone()];
    expect_have.sort();
    assert_eq!(resolve(&have), expect_have);

    // want − have is exactly what the second commit introduced.
    let mut expect_delta = vec![
        fx.commit2.clone(),
        fx.tree2.clone(),
        fx.blob_b.clone(),
        fx.big.clone(),
    ];
    expect_delta.sort();
    assert_eq!(resolve(&(want.clone() - have)), expect_delta);

    // …and the tip reaches every object in the archive.
    assert_eq!(want.len() as usize, fx.objects.len());
}

/// Opting out of the bitmaps is a distinct, observable state — not an empty
/// section that a consumer would read as "nothing is reachable".
#[test]
fn without_reach_emits_no_section_at_all() {
    let fx = build_fixture(GitHashKind::Sha256);
    let (_tmp, src, archive) = paths();
    fs::create_dir_all(&src).unwrap();
    let mut builder = GitIndexBuilder::new(fx.hash).without_reach();
    for (oid, bytes) in &fx.objects {
        fs::write(src.join(oid), bytes).unwrap();
        builder.push_canonical(bytes).unwrap();
    }
    let reserved = builder.into_reserved_builder();
    let registry = PluginRegistry::with_plugin(Box::new(NativeGitPlugin::new()));
    compress_dir(
        &src,
        &archive,
        false,
        Some(&registry),
        None,
        Some(Box::new(move |f, b| {
            Box::new(ArrowIpcSink::new(f, b).with_reserved_builder(reserved))
                as Box<dyn ArchiveMetaSink>
        })),
    )
    .unwrap();

    assert!(read_reach(&archive).unwrap().is_none(), "no bitmaps means NO section");
    assert!(read_graph(&archive).unwrap().is_some(), "the graph is still there");
    assert!(GitOidIndex::open(&archive).unwrap().is_some(), "the oid index is still there");
}

/// A plain (non-git) archive carries none of these sections, and asking for them
/// is a clean `None` rather than an error — the sections are additive.
#[test]
fn a_plain_archive_reports_no_git_sections() {
    let tmp = TempDir::new().unwrap();
    let src = tmp.path().join("plain");
    fs::create_dir_all(&src).unwrap();
    fs::write(src.join("hello.txt"), b"not a git object").unwrap();
    let archive = tmp.path().join("plain.znippy");
    compress_dir(&src, &archive, false, None, None, None).unwrap();

    assert!(GitOidIndex::open(&archive).unwrap().is_none());
    assert!(read_graph(&archive).unwrap().is_none());
    assert!(read_reach(&archive).unwrap().is_none());
}

/// The sink refuses a non-reserved extra section. Without this guard the section
/// would be merged into the data index by `read_multi_index`.
#[test]
fn the_sink_refuses_a_non_reserved_extra_section() {
    use znippy_common::ReservedSection;
    let tmp = TempDir::new().unwrap();
    let src = tmp.path().join("plain");
    fs::create_dir_all(&src).unwrap();
    fs::write(src.join("a.txt"), b"x").unwrap();
    let archive = tmp.path().join("bad.znippy");

    let err = compress_dir(
        &src,
        &archive,
        false,
        None,
        None,
        Some(Box::new(|f: Arc<fs::File>, b: u64| {
            Box::new(ArrowIpcSink::new(f, b).with_reserved_builder(Box::new(|_view| {
                Ok(vec![ReservedSection::raw("totally_ordinary", vec![1, 2, 3])])
            }))) as Box<dyn ArchiveMetaSink>
        })),
    )
    .expect_err("a non-reserved extra section must be refused");
    let msg = format!("{err:#}");
    assert!(
        msg.contains("totally_ordinary") && msg.contains("reserved"),
        "error should name the offending module: {msg}"
    );
}

// ── refs + secrets: the push logs, sealed into the archive ────────────────────

/// Seal the fixture's objects **plus** a ref log and a secrets log, exactly the
/// way gunnar's `repack()` does: the logs are appended to live, then handed to
/// the one `GitIndexBuilder` that emits every reserved section.
fn seal_with_logs(
    fx: &Fixture,
    dir: &Path,
    archive: &Path,
    refs: &RefLog,
    secrets: &SecretsLog,
) {
    fs::create_dir_all(dir).unwrap();
    let mut builder = GitIndexBuilder::new(fx.hash);
    for (oid, bytes) in &fx.objects {
        fs::write(dir.join(oid), bytes).unwrap();
        builder.push_canonical(bytes).unwrap();
    }
    let builder = builder
        .with_section(refs.seal_section().unwrap())
        .unwrap()
        .with_section(secrets.seal_section().unwrap())
        .unwrap();

    let reserved = builder.into_reserved_builder();
    let registry = PluginRegistry::with_plugin(Box::new(NativeGitPlugin::new()));
    compress_dir(
        &dir.to_path_buf(),
        &archive.to_path_buf(),
        false,
        Some(&registry),
        None,
        Some(Box::new(move |f, b| {
            Box::new(ArrowIpcSink::new(f, b).with_reserved_builder(reserved))
                as Box<dyn ArchiveMetaSink>
        })),
    )
    .expect("compress_dir");
}

/// The headline claim for the push logs: refs and secrets pushed to a live log
/// survive into the **sealed archive** and come back with their per-push
/// boundaries and their fold intact — read from the archive on disk, never from
/// the writer's memory.
#[test]
fn refs_and_secrets_survive_the_seal_and_fold_correctly() {
    let fx = build_fixture(GitHashKind::Sha256);
    let (_tmp, src, archive) = paths();
    let logdir = _tmp.path().join("logs");
    fs::create_dir_all(&logdir).unwrap();

    let refs = RefLog::new(logdir.join("refs.log"));
    refs.push(&[
        RefUpdate::set("refs/heads/main", &fx.commit1),
        RefUpdate::set("refs/heads/doomed", &fx.commit1),
    ])
    .unwrap();
    refs.push(&[RefUpdate::set("refs/heads/main", &fx.commit2)]).unwrap();
    refs.push(&[RefUpdate::delete("refs/heads/doomed")]).unwrap();

    let secrets = SecretsLog::new(logdir.join("secrets.log"));
    secrets
        .push(&[SecretUpdate::new("deploy-key", b"AGE-CIPHERTEXT-v1".to_vec()).unwrap()])
        .unwrap();
    secrets
        .push(&[SecretUpdate::new("deploy-key", b"AGE-CIPHERTEXT-v2".to_vec())
            .unwrap()
            .for_recipient("age1qqq")])
        .unwrap();

    seal_with_logs(&fx, &src, &archive, &refs, &secrets);

    // Read back OUT OF THE ARCHIVE.
    let sealed_refs = read_refs(&archive).unwrap().expect("archive must carry __gunnar_refs__");
    assert_eq!(
        sealed_refs.get("refs/heads/main").and_then(|r| r.target.clone()),
        Some(fx.commit2.clone()),
        "the second push must win inside the sealed archive"
    );
    assert!(
        !sealed_refs.contains_key("refs/heads/doomed"),
        "a deletion pushed before the seal must not resurrect inside the archive"
    );
    assert_eq!(sealed_refs.len(), 1);

    let sealed_secrets =
        read_secrets(&archive).unwrap().expect("archive must carry __gunnar_secrets__");
    assert_eq!(
        sealed_secrets["deploy-key"].ciphertext,
        b"AGE-CIPHERTEXT-v2".to_vec(),
        "the rotation must have survived the seal"
    );
    assert_eq!(sealed_secrets["deploy-key"].recipient.as_deref(), Some("age1qqq"));

    // The per-push framing survives the seal: three ref pushes, two secret
    // pushes — not one merged batch each. If the seal flattened them, the
    // history would be gone even though the fold still looked right.
    let ref_batches = znippy_plugin_git::pushlog::read_sealed(&archive, GUNNAR_REFS_MODULE)
        .unwrap()
        .unwrap();
    assert_eq!(ref_batches.len(), 3, "one RecordBatch per push must survive sealing");
    let secret_batches = znippy_plugin_git::pushlog::read_sealed(&archive, GUNNAR_SECRETS_MODULE)
        .unwrap()
        .unwrap();
    assert_eq!(secret_batches.len(), 2, "one RecordBatch per push must survive sealing");
}

/// The push logs must be invisible to every ordinary reader. If either section
/// were classified as data, `list` would report phantom files and `decompress`
/// would try to write them — the exact corruption `is_reserved_module` exists to
/// prevent.
#[test]
fn the_push_log_sections_are_reserved_and_do_not_leak_into_the_data_index() {
    let fx = build_fixture(GitHashKind::Sha256);
    let (_tmp, src, archive) = paths();
    let logdir = _tmp.path().join("logs");
    fs::create_dir_all(&logdir).unwrap();

    let refs = RefLog::new(logdir.join("refs.log"));
    refs.push(&[RefUpdate::set("refs/heads/main", &fx.commit2)]).unwrap();
    let secrets = SecretsLog::new(logdir.join("secrets.log"));
    secrets.push(&[SecretUpdate::new("k", b"CIPHER".to_vec()).unwrap()]).unwrap();

    seal_with_logs(&fx, &src, &archive, &refs, &secrets);

    // Both directions matter, and only asserting one of them would be blind:
    // a section absent from the DATA manifest could simply not have been written
    // at all, and a section present in the FULL manifest could still be leaking
    // into the data view.
    let data_entries = read_znippy_manifest(&archive).unwrap();
    let (full, _) = znippy_common::read_znippy_full_manifest(&archive).unwrap();
    for m in [GUNNAR_REFS_MODULE, GUNNAR_SECRETS_MODULE] {
        assert!(is_reserved_module(m), "{m} must be reserved");
        assert!(RESERVED_MODULES.contains(&m), "{m} must be in the catalog");
        assert!(
            !data_entries.iter().any(|e| e.module_name == m),
            "'{m}' surfaced as a DATA sub-index — list/decompress would be corrupted"
        );
        let e = full
            .iter()
            .find(|e| e.module_name == m)
            .unwrap_or_else(|| panic!("'{m}' missing from the full manifest — it was never sealed"));
        assert!(e.index_len > 0, "'{m}' section is empty — nothing was actually written");
    }

    // The data view is exactly the objects — no ref or secret row anywhere.
    let ar = ZnippyArchive::open(&archive).unwrap();
    let listed = ar.list_files().unwrap();
    assert_eq!(
        listed.len(),
        fx.objects.len(),
        "the push logs must not add entries to the data index: {listed:?}"
    );
    for name in &listed {
        assert!(
            !name.contains("gunnar") && !name.contains("refs/") && !name.contains("deploy"),
            "a push-log row leaked into the data index as {name}"
        );
    }
}

/// A git archive sealed without any push logs reports `None` for both — a state
/// distinct from "present but empty". A repository with no refs at all and a
/// repository whose refs were never recorded are different facts.
#[test]
fn an_archive_without_push_logs_reports_none_not_empty() {
    let fx = build_fixture(GitHashKind::Sha256);
    let (_tmp, src, archive) = paths();
    seal(&fx, &src, &archive, ReachPolicy::default());

    assert!(read_refs(&archive).unwrap().is_none(), "no refs section means None, not empty");
    assert!(read_secrets(&archive).unwrap().is_none());

    // Contrast: a log that exists but has never been pushed to seals as a
    // present-and-empty section, which reads back as Some(empty).
    let (_tmp2, src2, archive2) = paths();
    let logdir = _tmp2.path().join("logs");
    fs::create_dir_all(&logdir).unwrap();
    let refs = RefLog::new(logdir.join("refs.log"));
    let secrets = SecretsLog::new(logdir.join("secrets.log"));
    seal_with_logs(&fx, &src2, &archive2, &refs, &secrets);
    let r = read_refs(&archive2).unwrap();
    assert!(
        r.as_ref().is_some_and(|m| m.is_empty()),
        "an empty log must seal as Some(empty), distinguishable from None: {r:?}"
    );
}