znippy-plugin-git 0.1.3

//! `__gunnar_reach__` — reachability bitmaps.
//!
//! For each *selected* commit, the set of objects reachable from it — the commit
//! itself, its ancestors, and every tree and blob those commits point at —
//! recorded as a roaring bitmap over **object ordinals**.
//!
//! The ordinal space is the archive's oid-lexicographic ordering of its distinct
//! objects, which is also the order the sorted lookup sub-index is in and the
//! order `__gunnar_oid__` records in its `ordinal` column. So a bitmap position
//! is resolvable to an object with the oid index alone.
//!
//! What it buys: `want − have` becomes `reach(want) ANDNOT reach(have)` — a
//! bitmap operation instead of a graph traversal. **No speedup is claimed here.**
//! Nothing in this repository has measured it against a real repository with real
//! merge history; the structure is built and correct, the number is not earned
//! (LAW 7).
//!
//! Cost, stated plainly: the builder is O(commits × bitmap width) in time and
//! holds live bitmaps for commits whose children are not yet processed, plus one
//! memoized tree-closure bitmap per distinct tree. On a wide DAG that is not
//! cheap. `ReachPolicy::max_commits` bounds how many bitmaps are *kept*, not how
//! many are computed.

use std::collections::HashMap;
use std::path::Path;
use std::sync::Arc;

use anyhow::{Result, anyhow};
use roaring::RoaringBitmap;
use znippy_common::GUNNAR_REACH_MODULE;
use znippy_common::arrow::array::{Array, BinaryArray, BinaryBuilder, StringArray, StringBuilder};
use znippy_common::arrow::datatypes::{DataType, Field, Schema};
use znippy_common::arrow::ipc::reader::StreamReader;
use znippy_common::arrow::record_batch::RecordBatch;
use znippy_common::read_reserved_section_bytes;

use crate::graph::CommitNode;
use crate::object::{GitObjectKind, tree_entries};

/// Which commits get a bitmap.
#[derive(Debug, Clone, Copy)]
pub struct ReachPolicy {
    /// Upper bound on how many bitmaps are sealed. Tips are taken first (a
    /// commit with no child inside the archive — the branch heads a clone
    /// negotiates against), then the remaining commits are sampled at an even
    /// stride through generation order.
    pub max_commits: usize,
}

impl Default for ReachPolicy {
    fn default() -> Self {
        // git's own bitmap selection is of this order: cover the tips, then
        // sample the history so a `have` usually lands near a bitmapped commit.
        Self { max_commits: 512 }
    }
}

/// One sealed bitmap.
#[derive(Debug, Clone, PartialEq)]
pub struct ReachEntry {
    pub commit: String,
    pub bitmap: RoaringBitmap,
}

pub fn reach_schema() -> Arc<Schema> {
    Arc::new(Schema::new(vec![
        Field::new("commit_oid", DataType::Utf8, false),
        Field::new("cardinality", DataType::UInt64, false),
        Field::new("bitmap", DataType::Binary, false),
    ]))
}

/// **Where a tree's bytes come from** — a lookup, not a table.
///
/// It was `&HashMap<String, Vec<u8>>`: every tree payload in the repository,
/// resident, keyed by a hex `String`. MEASURED on oden 2026-08-30, the real
/// `torvalds/linux` mirror through [`crate::git_ops::GitStore`]: **28.24 GB**
/// of tree payloads — 3.24× the 8.13 GiB pack — held on disk in the eager
/// exploded pool so that this map could be rebuilt into RAM on **every** fold.
/// Peak RSS 50.9 GiB. The appliance has 16.
///
/// So the map became a lookup. The eager arm is still a `HashMap` and costs
/// exactly what it did (`Cow::Borrowed`, no copy); the store's arm resolves the
/// tree the way any other read does — from the exploded cache if it is there,
/// by inflating its own delta chain out of the mmapped pack if it is not
/// ([`crate::resolve::materialize_at`]) — and the walk then puts it in the
/// cache for the next one.
///
/// `None` is *"not a tree this archive holds"* and is an ordinary answer, not
/// a fault: a blob is a leaf, a gitlink is a commit in another repository, and
/// a thin push's base may be outside the archive. Every caller below treats it
/// as "stop here", which is what it always meant.
pub trait TreeSource: Sync {
    fn tree(&self, oid_hex: &str) -> Option<std::borrow::Cow<'_, [u8]>>;
}

/// The eager arm: a map that is already in memory. Unchanged in cost — the
/// `Cow` borrows, so no payload is copied to satisfy the trait.
impl TreeSource for HashMap<String, Vec<u8>> {
    fn tree(&self, oid_hex: &str) -> Option<std::borrow::Cow<'_, [u8]>> {
        self.get(oid_hex).map(|v| std::borrow::Cow::Borrowed(v.as_slice()))
    }
}

/// **oid → ordinal, and ordinal → oid, over one flat buffer and nothing else.**
///
/// The ordinal space is `oids_raw` — every distinct oid in the archive,
/// fixed-width and back to back, in **oid-lexicographic (= raw byte) order**,
/// which is exactly the order the ordinal numbers it. Because that buffer is
/// already sorted and already the inverse map (ordinal `o` is the slice at
/// `o * oid_len`), both directions are answered *from it*:
///
/// - `ordinal_of_raw` / `ordinal_of_hex` — a **binary search** over the buffer,
///   O(log n) with no probe table to keep in step with it;
/// - `oid` — a slice.
///
/// This is what replaced [`Derived::oids`] (a hex `Vec<String>`) and
/// [`Derived::ordinal`] (a `HashMap<String, u32>`) — two per-oid `String`
/// allocations, built at every fold, that a comment then had to promise agreed
/// with the flat buffer. One representation, no promise needed (LAW 5). It owns
/// nothing: it is a `Copy` view (`&[u8]` + a length), so constructing it at open
/// allocates zero bytes.
///
/// [`Derived::oids`]: crate::git_ops
/// [`Derived::ordinal`]: crate::git_ops
#[derive(Clone, Copy)]
pub struct OrdinalSpace<'a> {
    /// The distinct oids, fixed-width, sorted, back to back. `oids_raw`.
    raw: &'a [u8],
    /// Width of one oid: 20 (sha1) or 32 (sha256).
    oid_len: usize,
}

impl<'a> OrdinalSpace<'a> {
    /// Borrow a flat, **sorted**, fixed-width oid buffer as the ordinal space.
    /// The sort is the caller's invariant (every builder produces it in
    /// oid-lexicographic order); the binary search below rests on it.
    pub fn new(raw: &'a [u8], oid_len: usize) -> Self {
        debug_assert!(oid_len > 0 && raw.len() % oid_len == 0, "ragged ordinal space");
        Self { raw, oid_len }
    }

    /// How many oids the space holds.
    pub fn len(&self) -> usize {
        if self.oid_len == 0 { 0 } else { self.raw.len() / self.oid_len }
    }

    pub fn is_empty(&self) -> bool {
        self.raw.is_empty()
    }

    /// ordinal → raw oid, a slice into the flat buffer. `None` for an ordinal
    /// outside the space.
    pub fn oid(&self, ordinal: u32) -> Option<&'a [u8]> {
        let at = (ordinal as usize).checked_mul(self.oid_len)?;
        self.raw.get(at..at + self.oid_len)
    }

    /// raw oid → ordinal, by binary search over the sorted flat buffer.
    ///
    /// This is the exact inverse of [`Self::oid`]: the buffer is sorted, so the
    /// index a search lands on IS the ordinal, and it equals what the old
    /// `HashMap<String, u32>` returned for the same oid — that map keyed the
    /// same sorted list by its own hex spelling.
    pub fn ordinal_of_raw(&self, oid: &[u8]) -> Option<u32> {
        if oid.len() != self.oid_len {
            return None;
        }
        let (mut lo, mut hi) = (0usize, self.len());
        while lo < hi {
            let mid = lo + (hi - lo) / 2;
            let at = mid * self.oid_len;
            match self.raw[at..at + self.oid_len].cmp(oid) {
                std::cmp::Ordering::Less => lo = mid + 1,
                std::cmp::Ordering::Greater => hi = mid,
                std::cmp::Ordering::Equal => return Some(mid as u32),
            }
        }
        None
    }

    /// hex oid → ordinal. Decodes the hex into a **stack** buffer (≤ 32 bytes),
    /// so a lookup allocates nothing, then binary-searches the raw form. A hex
    /// string that is the wrong length or not hex is simply absent.
    pub fn ordinal_of_hex(&self, hex: &str) -> Option<u32> {
        let mut buf = [0u8; 32];
        let dst = buf.get_mut(..self.oid_len)?;
        hex::decode_to_slice(hex, dst).ok()?;
        self.ordinal_of_raw(dst)
    }
}

/// **Carry a reachability table across a fold** — the same bitmaps, renumbered
/// into the new ordinal space, for the commits the new graph still holds.
///
/// A bitmap is a fact about an immutable commit: the set of objects reachable
/// from it never changes while the commit exists. What a fold changes is the
/// NUMBERING — the ordinal space is the sorted oid buffer, and every oid a push
/// adds shifts the ordinals above it — so an entry survives a fold exactly by
/// mapping each set bit through `old ordinal → new ordinal`. That map is
/// monotone (both spaces are sorted, and the new one is a superset), so it is
/// one merge pass over the two buffers and one in-order pass over each bitmap.
///
/// Returns `None` — *rebuild from scratch* — when the carry cannot be exact:
/// an old oid that the new space lacks (an object was removed under the table,
/// which a retain/gc does) or an entry whose commit `keep` no longer knows.
/// An entry for a commit that is gone is dropped, not carried; a BIT that is
/// gone refuses the whole carry, because an entry claiming an object the store
/// no longer holds would be a silent over-claim on the next negotiation.
///
/// Commits added since the carried build have no entry; [`accumulate`] walks
/// a bitmapless commit down to the nearest carried ancestor, so a request
/// stays exact and pays only for the new commits' trees. The caller bounds how
/// far that is allowed to grow (`git_ops::REACH_CARRY_MAX_FRACTION`).
pub fn carry_reach(
    old: &[ReachEntry],
    old_space: &OrdinalSpace<'_>,
    new_space: &OrdinalSpace<'_>,
    keep: impl Fn(&str) -> bool,
) -> Option<Vec<ReachEntry>> {
    if old.is_empty() || old_space.is_empty() {
        return None;
    }
    // old ordinal → new ordinal, by one merge over the two sorted buffers.
    let mut remap: Vec<Option<u32>> = Vec::with_capacity(old_space.len());
    let mut j = 0u32;
    let new_len = new_space.len() as u32;
    for o in 0..old_space.len() as u32 {
        let oid = old_space.oid(o)?;
        let mut hit = None;
        while j < new_len {
            let cand = new_space.oid(j)?;
            match cand.cmp(oid) {
                std::cmp::Ordering::Less => j += 1,
                std::cmp::Ordering::Equal => {
                    hit = Some(j);
                    j += 1;
                    break;
                }
                std::cmp::Ordering::Greater => break,
            }
        }
        remap.push(hit);
    }
    let mut out = Vec::with_capacity(old.len());
    for e in old {
        if !keep(&e.commit) {
            continue;
        }
        let mut mapped = Vec::with_capacity(e.bitmap.len() as usize);
        for b in e.bitmap.iter() {
            match remap.get(b as usize).copied().flatten() {
                Some(n) => mapped.push(n),
                None => return None,
            }
        }
        // Monotone map ⇒ `mapped` is ascending; the sorted constructor is the
        // cheap one and asserts exactly that.
        let bitmap = RoaringBitmap::from_sorted_iter(mapped).ok()?;
        out.push(ReachEntry { commit: e.commit.clone(), bitmap });
    }
    Some(out)
}

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

    fn space(oids: &[u8]) -> Vec<u8> {
        // 1-byte "oids" are enough for the arithmetic; the space is sorted.
        let mut v = oids.to_vec();
        v.sort_unstable();
        v
    }

    /// The carry renumbers every bit through the merge, drops an entry whose
    /// commit is gone, and refuses when a set bit's object is gone — red in
    /// every arm before the function existed, since a fold used to drop the
    /// table outright.
    #[test]
    fn a_carry_renumbers_bits_drops_dead_commits_and_refuses_a_missing_object() {
        let old_raw = space(&[10, 20, 30, 40]);
        let old = OrdinalSpace::new(&old_raw, 1);
        // Two oids inserted below 30 and one above 40: ordinals shift by 2 and 2.
        let new_raw = space(&[10, 15, 20, 25, 30, 40, 50]);
        let new = OrdinalSpace::new(&new_raw, 1);
        let entries = vec![
            ReachEntry { commit: "keep".into(), bitmap: [0u32, 2, 3].into_iter().collect() },
            ReachEntry { commit: "gone".into(), bitmap: [1u32].into_iter().collect() },
        ];
        let carried = carry_reach(&entries, &old, &new, |c| c == "keep").expect("exact carry");
        assert_eq!(carried.len(), 1, "the dead commit's entry is dropped, not carried");
        assert_eq!(carried[0].commit, "keep");
        let bits: Vec<u32> = carried[0].bitmap.iter().collect();
        assert_eq!(bits, vec![0, 4, 5], "10→0, 30→4, 40→5 in the new numbering");

        // An object removed from the space under a kept entry: refuse the carry.
        let shrunk_raw = space(&[10, 20, 40]);
        let shrunk = OrdinalSpace::new(&shrunk_raw, 1);
        assert!(carry_reach(&entries, &old, &shrunk, |_| true).is_none(), "30 is gone; a bitmap naming it must not survive");
        // Nothing to carry is not a carry.
        assert!(carry_reach(&[], &old, &new, |_| true).is_none());
    }
}

/// What the builder needs to know about every object in the archive.
pub struct ObjectFacts<'a> {
    /// oid ⇄ ordinal — the bitmap space, over the flat `oids_raw` buffer.
    pub ordinal: OrdinalSpace<'a>,
    /// Tree payloads, by oid hex. Blobs are not needed (they are leaves) and
    /// are deliberately not asked for.
    pub trees: &'a dyn TreeSource,
    /// Raw oid width, needed to walk tree records.
    pub oid_len: usize,
}

/// Build the reachability bitmaps.
///
/// `commits` must already carry generation numbers (see
/// [`crate::graph::assign_generations`]) and be ordered parents-before-children,
/// which that function guarantees.
pub fn build_reach(
    commits: &[CommitNode],
    facts: &ObjectFacts<'_>,
    policy: ReachPolicy,
) -> Vec<ReachEntry> {
    let n = commits.len();
    if n == 0 || policy.max_commits == 0 {
        return Vec::new();
    }
    let pos: HashMap<&str, usize> = commits
        .iter()
        .enumerate()
        .map(|(i, c)| (c.oid.as_str(), i))
        .collect();

    let selected = select_commits(commits, &pos, policy);
    let keep: Vec<bool> = {
        let mut k = vec![false; n];
        for &i in &selected {
            k[i] = true;
        }
        k
    };

    // How many children still need each commit's bitmap. Once it hits zero and
    // the commit is not selected, the bitmap is dropped — that is what keeps
    // peak memory to the DAG's frontier rather than the whole history.
    let mut pending: Vec<usize> = vec![0; n];
    let mut parent_idx: Vec<Vec<usize>> = Vec::with_capacity(n);
    for c in commits {
        let mut ps: Vec<usize> = c.parents.iter().filter_map(|p| pos.get(p.as_str()).copied()).collect();
        ps.sort_unstable();
        ps.dedup();
        for &p in &ps {
            pending[p] += 1;
        }
        parent_idx.push(ps);
    }

    let mut tree_memo: HashMap<String, RoaringBitmap> = HashMap::new();
    let mut live: HashMap<usize, RoaringBitmap> = HashMap::new();

    for i in 0..n {
        let c = &commits[i];
        let mut bm = RoaringBitmap::new();
        if let Some(o) = facts.ordinal.ordinal_of_hex(c.oid.as_str()) {
            bm.insert(o);
        }
        if let Some(t) = &c.tree {
            // Borrowed out of the memo, not cloned out of it. See
            // [`tree_closure`]'s own note: this is once per commit and the thing
            // being copied is a whole tree closure, which on a repository-shaped
            // history is most of the object graph.
            if let Some(tc) = tree_closure(t, facts, &mut tree_memo) {
                bm |= tc;
            }
        }
        for &p in &parent_idx[i] {
            if let Some(pb) = live.get(&p) {
                bm |= pb;
            }
            pending[p] -= 1;
            if pending[p] == 0 && !keep[p] {
                live.remove(&p);
            }
        }
        // **Nothing is cloned here.** A kept commit's bitmap is inserted into
        // `live` like any other and harvested below; the `out.push(ReachEntry {
        // bitmap: bm.clone() })` that used to stand here made a second, deep
        // copy of every selected commit's roaring bitmap while `live` held the
        // first — and on the live selection path `keep` is *every* commit, so
        // that was one full duplicate of the entire index. MEASURED on oden
        // 2026-08-14, a 1 581-commit / 12 455-object store: the whole build was
        // 530 218 allocations.
        //
        // Safe by the retention rule directly above: an entry is dropped from
        // `live` only when `pending[p] == 0 && !keep[p]`, so a kept one is never
        // dropped and is still there when the harvest runs.
        if pending[i] > 0 || keep[i] {
            live.insert(i, bm);
        }
    }

    // The harvest: **move** each selected commit's bitmap out of `live`.
    //
    // `selected` is sorted ascending (`select_commits` sorts and dedups), and
    // the loop above pushed in `0..n` order filtered by `keep`, so this produces
    // the identical order the clone-per-iteration version did. That is not a
    // detail to leave implicit — `decode_reach`/`build_reach_batch` round-trip
    // this vector positionally.
    let mut out: Vec<ReachEntry> = Vec::with_capacity(selected.len());
    for &i in &selected {
        // `expect` rather than `unwrap_or_default`: a missing entry here means
        // the retention rule above let a kept commit be dropped, and answering
        // with an EMPTY bitmap would be a silent under-send — a clone served
        // one object per branch, exiting zero. Panicking names the commit.
        let bitmap = live.remove(&i).unwrap_or_else(|| {
            panic!(
                "commit {} was selected for a reachability bitmap and its bitmap is not live at \
                 the end of the build; an empty one here would under-send a clone",
                commits[i].oid
            )
        });
        out.push(ReachEntry {
            commit: commits[i].oid.clone(),
            bitmap,
        });
    }
    out
}

/// Tips first, then an even sample through the rest, capped at `max_commits`.
fn select_commits(
    commits: &[CommitNode],
    pos: &HashMap<&str, usize>,
    policy: ReachPolicy,
) -> Vec<usize> {
    let n = commits.len();
    let mut has_child = vec![false; n];
    for c in commits {
        for p in &c.parents {
            if let Some(&pi) = pos.get(p.as_str()) {
                has_child[pi] = true;
            }
        }
    }
    let mut chosen: Vec<usize> = (0..n).filter(|&i| !has_child[i]).collect();
    chosen.truncate(policy.max_commits);

    if chosen.len() < policy.max_commits {
        let room = policy.max_commits - chosen.len();
        let rest: Vec<usize> = (0..n).filter(|&i| has_child[i]).collect();
        if !rest.is_empty() {
            let stride = rest.len().div_ceil(room).max(1);
            for &i in rest.iter().step_by(stride).take(room) {
                chosen.push(i);
            }
        }
    }
    chosen.sort_unstable();
    chosen.dedup();
    chosen
}

/// Every object reachable from a tree, memoized per tree oid.
///
/// Iterative (an explicit post-order stack), not recursive: a deep source tree
/// would otherwise be a stack-depth bet.
///
/// # It **borrows** out of the memo rather than cloning out of it
///
/// It returned `RoaringBitmap` by value until 2026-08-14, which meant a deep
/// copy of a whole tree closure on every call — including, and especially, the
/// memo *hit* on the first line, which is the common case: consecutive commits
/// in a history overwhelmingly share a root tree or find one already computed.
/// The thing being copied is not small. On a repository-shaped history a root
/// tree's closure is most of the object graph, so this was one full copy of it
/// per commit, 1 581 times on the store this was measured against.
///
/// `None` means the memo has no entry for `root` even after the walk, which the
/// walk only produces for an oid this archive does not hold. The caller ORs
/// nothing in, which is what `unwrap_or_default()` did and is the same answer —
/// it is `Option` rather than an empty bitmap so that "absent" and "reaches
/// nothing" stay different words at the call site.
pub(crate) fn tree_closure<'m>(
    root: &str,
    facts: &ObjectFacts<'_>,
    memo: &'m mut HashMap<String, RoaringBitmap>,
) -> Option<&'m RoaringBitmap> {
    if memo.contains_key(root) {
        // Deliberately re-looked-up at the bottom rather than returned from
        // here: NLL cannot see that the early borrow ends on this path, and the
        // alternative is a second `HashMap` probe on the hit path against a deep
        // copy of the whole closure. The probe wins by orders of magnitude.
        return memo.get(root);
    }
    // (oid, children_already_expanded)
    let mut stack: Vec<(String, bool)> = vec![(root.to_string(), false)];
    // Trees currently on the stack. A tree graph cannot contain a cycle, but a
    // *corrupt* one can, and "cannot happen" is not a termination argument for
    // bytes that arrived over the wire.
    let mut visiting: std::collections::HashSet<String> = std::collections::HashSet::new();

    while let Some((oid, expanded)) = stack.pop() {
        if memo.contains_key(&oid) {
            visiting.remove(&oid);
            continue;
        }
        let Some(payload) = facts.trees.tree(&oid) else {
            // Not a tree we hold (a blob, or an object outside the archive).
            let mut bm = RoaringBitmap::new();
            if let Some(o) = facts.ordinal.ordinal_of_hex(oid.as_str()) {
                bm.insert(o);
            }
            memo.insert(oid, bm);
            continue;
        };
        let children: Vec<String> = tree_entries(&payload, facts.oid_len)
            .iter()
            .map(|e| hex::encode(e.oid))
            .collect();
        if !expanded {
            visiting.insert(oid.clone());
            // A child that is already `visiting` is a back-edge: do not push it
            // again, or the stack never drains.
            let unresolved: Vec<String> = children
                .iter()
                .filter(|c| !memo.contains_key(*c) && !visiting.contains(*c))
                .cloned()
                .collect();
            if !unresolved.is_empty() {
                stack.push((oid, true));
                for c in unresolved {
                    stack.push((c, false));
                }
                continue;
            }
        }
        let mut bm = RoaringBitmap::new();
        if let Some(o) = facts.ordinal.ordinal_of_hex(oid.as_str()) {
            bm.insert(o);
        }
        for c in &children {
            if let Some(cb) = memo.get(c) {
                bm |= cb;
            } else if let Some(o) = facts.ordinal.ordinal_of_hex(c.as_str()) {
                // Back-edge or an object outside the archive: take the object
                // itself and stop rather than loop forever.
                bm.insert(o);
            }
        }
        memo.insert(oid.clone(), bm);
        visiting.remove(&oid);
    }
    memo.get(root)
}

/// **The bounded walk: OR into `acc` everything reachable from `tip`, stopping
/// at every commit that already has a bitmap.**
///
/// # Why this function is what makes a sampled bitmap table legal
///
/// Until 2026-08-14 this file had no walk at all, and the consequence ran all
/// the way up the stack. [`crate::git_ops::GitStore::reachable_oids`] answered a
/// `want` with no bitmap by contributing *the object itself and nothing else* —
/// a silent under-send, a clone that exits zero having served one object per
/// branch. `crate::serve`'s `select` therefore had to refuse any request whose
/// tips were not all bitmapped, and the only way to make that refusal never fire
/// was to bitmap **every** commit: `ReachPolicy { max_commits: usize::MAX }`, on
/// the request thread, holding the derived write lock, after every push (because
/// `refold` clears the table). MEASURED: 530 218 of a fetch's 575 349
/// allocations were that build, for a request that sent four objects.
///
/// With a walk, a missing bitmap is a *bounded amount of work* instead of a
/// wrong answer, and the table can be sampled like the sealed archive's already
/// is.
///
/// # Why it is bounded, stated as the property it rests on
///
/// **A stored bitmap is closed under reachability.** So reaching one accounts
/// for the entire history behind it, and the walk stops there rather than
/// descending. What it covers is therefore only what has been pushed since the
/// last time the table was built — not the repository — whatever the repository's
/// size. This is git's `add_to_include_set` rule, arrived at independently here
/// over a different object store: nothing of gitoxide's or of git's is copied,
/// and this crate links neither (LAW 5 applies to code, and these are different
/// codebases with different indexes).
///
/// # The two phases are an ordering, not a style
///
/// The commit walk runs to **completion** before a single tree is opened. Git
/// applies its skip rule one level down as well — `should_include_obj`, do not
/// descend into a tree whose bit is already set — and that rule is worth nothing
/// unless every stored bitmap this request will ever OR in is already in `acc`
/// when the tree walk starts. Get the order wrong and the first unbitmapped
/// commit costs a walk of every tree and blob in the repository, and the bounded
/// commit walk buys exactly nothing.
/// **The two per-fold lookup tables the reach walk keys into, paired as one
/// view.** `bitmap` is the stored closure of a commit (the stopping rule);
/// `node` is a commit's graph row (its parents and tree). Both were rebuilt as
/// `HashMap<&str, _>` on every negotiation — O(commits) per fetch, for tables
/// that cannot change between folds — and are now folded once and borrowed
/// through this view. Both are keyed by **hex** oid, the form the walk carries
/// (a [`CommitNode`]'s parents and tree are hex), so no lookup here allocates.
///
/// The index maps are `oid → row`; the view resolves that to a borrow of the
/// row so the caller never indexes a table by hand and never pairs an index
/// with the wrong table — the pair is built and handed out together by
/// [`crate::git_ops`]'s `reach_tables_with`.
#[derive(Clone, Copy)]
pub(crate) struct ReachView<'a> {
    pub reach: &'a [ReachEntry],
    pub reach_index: &'a HashMap<String, u32>,
    pub graph: &'a [CommitNode],
    pub graph_index: &'a HashMap<String, u32>,
}

impl<'a> ReachView<'a> {
    /// The stored bitmap of a commit, or `None` if it has none.
    #[inline]
    pub(crate) fn bitmap(&self, oid: &str) -> Option<&'a RoaringBitmap> {
        self.reach_index
            .get(oid)
            .map(|&i| &self.reach[i as usize].bitmap)
    }
    /// A commit's graph row, or `None` if this archive holds no such commit.
    #[inline]
    pub(crate) fn node(&self, oid: &str) -> Option<&'a CommitNode> {
        self.graph_index.get(oid).map(|&i| &self.graph[i as usize])
    }
    /// Whether the graph holds `oid` as a commit — the `want`/`have` test the
    /// negotiation makes before it decides to walk.
    #[inline]
    pub(crate) fn is_commit(&self, oid: &str) -> bool {
        self.graph_index.contains_key(oid)
    }
}

pub(crate) fn accumulate(
    tip: &str,
    view: &ReachView<'_>,
    facts: &ObjectFacts<'_>,
    acc: &mut RoaringBitmap,
) {
    // ── phase 1: the commit walk, to completion ────────────────────────────
    //
    // Collect the commits that have no bitmap, ORing in every stored bitmap the
    // frontier touches. Nothing here opens a tree.
    let mut unbitmapped: Vec<&str> = Vec::new();
    let mut seen: std::collections::HashSet<&str> = std::collections::HashSet::new();
    let mut stack: Vec<&str> = vec![tip];
    while let Some(oid) = stack.pop() {
        if !seen.insert(oid) {
            continue;
        }
        // The stopping rule. A stored bitmap is closed, so everything behind
        // this commit is already accounted for and its parents are not walked.
        if let Some(bm) = view.bitmap(oid) {
            *acc |= bm;
            continue;
        }
        let Some(node) = view.node(oid) else {
            // Not a commit this archive holds as a graph row. It contributes
            // itself if the index knows it, and nothing else — there is no
            // history here to walk. A `want` that reaches this and is genuinely
            // absent is refused by the caller, which is the one place that can
            // tell "absent" from "not a commit".
            if let Some(o) = facts.ordinal.ordinal_of_hex(oid) {
                acc.insert(o);
            }
            continue;
        };
        unbitmapped.push(oid);
        for p in &node.parents {
            stack.push(p.as_str());
        }
    }

    // ── phase 2: the trees, with the skip rule ─────────────────────────────
    for oid in unbitmapped {
        if let Some(o) = facts.ordinal.ordinal_of_hex(oid) {
            acc.insert(o);
        }
        let Some(node) = view.node(oid) else { continue };
        if let Some(tree) = &node.tree {
            accumulate_tree(tree.as_str(), facts, acc);
        }
    }
}

/// [`accumulate`]'s second phase for one root tree: every object under it that
/// is not already in `acc`.
///
/// The skip is git's `should_include_obj` and it is what keeps phase 2 bounded:
/// a set bit means that object — and, for a tree, everything under it — is
/// already accounted for, either by a stored bitmap ORed in during phase 1 or by
/// an earlier commit in this same walk. Descending anyway would be correct and
/// would also make the whole exercise pointless.
///
/// Iterative for the reason [`tree_closure`] is: a deep source tree is not a
/// stack-depth bet. It needs no `visiting` set of its own — a corrupt cycle
/// terminates here because a revisited oid's bit is already set.
pub(crate) fn accumulate_tree(root: &str, facts: &ObjectFacts<'_>, acc: &mut RoaringBitmap) {
    let mut stack: Vec<String> = vec![root.to_string()];
    while let Some(oid) = stack.pop() {
        let Some(o) = facts.ordinal.ordinal_of_hex(oid.as_str()) else {
            // An object this archive does not hold. A thin push's base, or a
            // submodule gitlink, which is a commit oid that lives in another
            // repository entirely and must never be selected for emission.
            continue;
        };
        if !acc.insert(o) {
            // Already set: this object, and everything under it, is accounted
            // for. `RoaringBitmap::insert` returns false when the bit was
            // already present, so the test and the set are one operation.
            continue;
        }
        // Only trees carry pointers. A blob is a leaf and `facts.trees`
        // deliberately does not hold one.
        if let Some(payload) = facts.trees.tree(oid.as_str()) {
            for e in tree_entries(&payload, facts.oid_len) {
                stack.push(hex::encode(e.oid));
            }
        }
    }
}

/// Which objects a repack must hold to make the reachability pass exact.
///
/// Only commits and trees carry pointers; blobs are leaves. The builder
/// therefore needs tree payloads and nothing else, which is why
/// [`ObjectFacts::trees`] exists and there is no `blobs` field.
pub fn needs_payload(kind: GitObjectKind) -> bool {
    matches!(kind, GitObjectKind::Commit | GitObjectKind::Tree)
}

pub fn build_reach_batch(entries: &[ReachEntry]) -> Result<RecordBatch> {
    use znippy_common::arrow::array::UInt64Builder;
    let n = entries.len();
    let mut oid_b = StringBuilder::with_capacity(n, n * 64);
    let mut card_b = UInt64Builder::with_capacity(n);
    let mut bm_b = BinaryBuilder::with_capacity(n, n * 128);
    for e in entries {
        oid_b.append_value(&e.commit);
        card_b.append_value(e.bitmap.len());
        let mut buf = Vec::new();
        e.bitmap
            .serialize_into(&mut buf)
            .map_err(|err| anyhow!("reach bitmap serialize: {err}"))?;
        bm_b.append_value(&buf);
    }
    RecordBatch::try_new(
        reach_schema(),
        vec![Arc::new(oid_b.finish()), Arc::new(card_b.finish()), Arc::new(bm_b.finish())],
    )
    .map_err(|e| anyhow!("reach batch: {e}"))
}

pub fn decode_reach(bytes: &[u8]) -> Result<Vec<ReachEntry>> {
    let reader = StreamReader::try_new(std::io::Cursor::new(bytes), None)
        .map_err(|e| anyhow!("reach reader: {e}"))?;
    let mut out = Vec::new();
    for batch in reader {
        let batch = batch.map_err(|e| anyhow!("reach batch read: {e}"))?;
        let oids = batch
            .column_by_name("commit_oid")
            .and_then(|c| c.as_any().downcast_ref::<StringArray>())
            .ok_or_else(|| anyhow!("reach: no `commit_oid` column"))?;
        let bms = batch
            .column_by_name("bitmap")
            .and_then(|c| c.as_any().downcast_ref::<BinaryArray>())
            .ok_or_else(|| anyhow!("reach: no `bitmap` column"))?;
        for i in 0..batch.num_rows() {
            let bitmap = RoaringBitmap::deserialize_from(bms.value(i))
                .map_err(|e| anyhow!("reach bitmap deserialize: {e}"))?;
            out.push(ReachEntry { commit: oids.value(i).to_string(), bitmap });
        }
    }
    Ok(out)
}

/// Read the reachability bitmaps out of a sealed archive. `Ok(None)` when absent.
pub fn read_reach(archive: &Path) -> Result<Option<Vec<ReachEntry>>> {
    match read_reserved_section_bytes(archive, GUNNAR_REACH_MODULE)? {
        Some(b) => Ok(Some(decode_reach(&b)?)),
        None => Ok(None),
    }
}

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

    fn hexid(c: char) -> String {
        std::iter::repeat_n(c, 40).collect()
    }

    /// Build a tiny repo: two commits, the second adding a file.
    /// A flat, sorted, fixed-width ordinal buffer from oids **already in
    /// ordinal order** — the shape [`OrdinalSpace`] borrows. The old tests
    /// hand-built a `HashMap<String, u32>` with the same ordinals; because the
    /// ordinal IS the oid-lexicographic rank, that map and this buffer are two
    /// spellings of one list.
    fn ordinal_raw(oids_in_order: &[&str]) -> Vec<u8> {
        let mut v = Vec::new();
        for o in oids_in_order {
            v.extend_from_slice(&hex::decode(o).unwrap());
        }
        v
    }

    /// Returns (commits, facts-owned-data).
    #[allow(clippy::type_complexity)]
    fn tiny_repo() -> (Vec<CommitNode>, Vec<u8>, HashMap<String, Vec<u8>>) {
        let blob1 = hexid('1');
        let blob2 = hexid('2');
        let tree1 = hexid('3');
        let tree2 = hexid('4');
        let c1 = hexid('5');
        let c2 = hexid('6');

        let mut trees: HashMap<String, Vec<u8>> = HashMap::new();
        let mut t1 = Vec::new();
        t1.extend_from_slice(b"100644 a\0");
        t1.extend_from_slice(&hex::decode(&blob1).unwrap());
        trees.insert(tree1.clone(), t1);

        let mut t2 = Vec::new();
        t2.extend_from_slice(b"100644 a\0");
        t2.extend_from_slice(&hex::decode(&blob1).unwrap());
        t2.extend_from_slice(b"100644 b\0");
        t2.extend_from_slice(&hex::decode(&blob2).unwrap());
        trees.insert(tree2.clone(), t2);

        // Ordinal = oid-lexicographic rank: hexid('1') < … < hexid('6'), so the
        // sorted raw buffer is simply these six in this order.
        let ordinal = ordinal_raw(&[&blob1, &blob2, &tree1, &tree2, &c1, &c2]);

        let commits = assign_generations(vec![
            CommitNode {
                oid: c1.clone(),
                parents: vec![],
                tree: Some(hexid('3')),
                committer_time: Some(1),
                generation: 0,
            },
            CommitNode {
                oid: c2.clone(),
                parents: vec![c1.clone()],
                tree: Some(hexid('4')),
                committer_time: Some(2),
                generation: 0,
            },
        ]);
        (commits, ordinal, trees)
    }

    #[test]
    fn bitmap_contains_exactly_the_reachable_objects() {
        let (commits, ordinal, trees) = tiny_repo();
        let facts = ObjectFacts { ordinal: OrdinalSpace::new(&ordinal, 20), trees: &trees, oid_len: 20 };
        let entries = build_reach(&commits, &facts, ReachPolicy { max_commits: 16 });

        let by_commit: HashMap<&str, &RoaringBitmap> =
            entries.iter().map(|e| (e.commit.as_str(), &e.bitmap)).collect();

        // c1 reaches: blob1(0), tree1(2), c1(4).
        let b1: Vec<u32> = by_commit[hexid('5').as_str()].iter().collect();
        assert_eq!(b1, vec![0, 2, 4], "c1 must not reach blob2/tree2/c2");

        // c2 reaches everything.
        let b2: Vec<u32> = by_commit[hexid('6').as_str()].iter().collect();
        assert_eq!(b2, vec![0, 1, 2, 3, 4, 5]);
    }

    #[test]
    fn want_minus_have_is_an_andnot() {
        let (commits, ordinal, trees) = tiny_repo();
        let facts = ObjectFacts { ordinal: OrdinalSpace::new(&ordinal, 20), trees: &trees, oid_len: 20 };
        let entries = build_reach(&commits, &facts, ReachPolicy { max_commits: 16 });
        let by: HashMap<&str, &RoaringBitmap> =
            entries.iter().map(|e| (e.commit.as_str(), &e.bitmap)).collect();

        let want = by[hexid('6').as_str()].clone();
        let have = by[hexid('5').as_str()].clone();
        let delta: Vec<u32> = (want - have).iter().collect();
        // Exactly the objects the second commit introduced: blob2(1), tree2(3), c2(5).
        assert_eq!(delta, vec![1, 3, 5]);
    }

    #[test]
    fn selection_takes_tips_first_and_respects_the_cap() {
        let (commits, ordinal, trees) = tiny_repo();
        let facts = ObjectFacts { ordinal: OrdinalSpace::new(&ordinal, 20), trees: &trees, oid_len: 20 };
        let entries = build_reach(&commits, &facts, ReachPolicy { max_commits: 1 });
        assert_eq!(entries.len(), 1);
        assert_eq!(entries[0].commit, hexid('6'), "the tip, not the root");
    }

    #[test]
    fn batch_roundtrips_through_arrow_ipc() {
        let (commits, ordinal, trees) = tiny_repo();
        let facts = ObjectFacts { ordinal: OrdinalSpace::new(&ordinal, 20), trees: &trees, oid_len: 20 };
        let entries = build_reach(&commits, &facts, ReachPolicy::default());
        let batch = build_reach_batch(&entries).unwrap();
        let mut buf = Vec::new();
        {
            let mut w = znippy_common::arrow::ipc::writer::StreamWriter::try_new(
                &mut buf,
                &reach_schema(),
            )
            .unwrap();
            w.write(&batch).unwrap();
            w.finish().unwrap();
        }
        let back = decode_reach(&buf).unwrap();
        assert_eq!(back, entries);
    }

    #[test]
    fn a_corrupt_tree_cycle_terminates() {
        // t -> t (self-referential): must not hang or blow the stack.
        let t = hexid('a');
        let mut payload = Vec::new();
        payload.extend_from_slice(b"40000 self\0");
        payload.extend_from_slice(&hex::decode(&t).unwrap());
        let trees: HashMap<String, Vec<u8>> = [(t.clone(), payload)].into_iter().collect();
        let ordinal = ordinal_raw(&[&t, &hexid('b')]);
        let facts = ObjectFacts { ordinal: OrdinalSpace::new(&ordinal, 20), trees: &trees, oid_len: 20 };
        let commits = assign_generations(vec![CommitNode {
            oid: hexid('b'),
            parents: vec![],
            tree: Some(t),
            committer_time: None,
            generation: 0,
        }]);
        let entries = build_reach(&commits, &facts, ReachPolicy::default());
        assert_eq!(entries.len(), 1);
        let bits: Vec<u32> = entries[0].bitmap.iter().collect();
        assert_eq!(bits, vec![0, 1]);
    }

    /// **The binary search is the exact inverse of the slice, and it is the same
    /// answer a `HashMap<String, u32>` over the sorted list would give.**
    ///
    /// Over a sorted raw space this asserts, at every ordinal: `oid(o)` is the
    /// slice, `ordinal_of_raw`/`ordinal_of_hex` round-trip back to `o`, and an
    /// oid the space does not hold is `None` both ways (the search must not land
    /// on a neighbour). Seen RED by `Equal => Some(mid as u32 + 1)` in
    /// `ordinal_of_raw`: every round trip below then names the next ordinal.
    #[test]
    fn ordinal_space_is_the_inverse_of_the_slice() {
        // 32 distinct sha1-width oids, deliberately built OUT of order so the
        // sort is doing real work, then sorted into the flat space.
        let mut oids: Vec<[u8; 20]> = (0..32u8)
            .map(|i| {
                let mut o = [0u8; 20];
                o[0] = i.wrapping_mul(37);
                o[19] = i;
                o
            })
            .collect();
        oids.sort_unstable();
        oids.dedup();
        let mut raw = Vec::new();
        for o in &oids {
            raw.extend_from_slice(o);
        }
        let space = OrdinalSpace::new(&raw, 20);
        assert_eq!(space.len(), oids.len());
        for (o, oid) in oids.iter().enumerate() {
            assert_eq!(space.oid(o as u32), Some(oid.as_slice()));
            assert_eq!(space.ordinal_of_raw(oid), Some(o as u32));
            assert_eq!(space.ordinal_of_hex(&hex::encode(oid)), Some(o as u32));
        }
        // Absent, wrong width, and out of range are all None — not a neighbour.
        assert_eq!(space.ordinal_of_raw(&[0xffu8; 20]), None);
        assert_eq!(space.ordinal_of_raw(&[0u8; 19]), None);
        assert_eq!(space.oid(oids.len() as u32), None);
        assert_eq!(space.ordinal_of_hex("not-hex"), None);
    }

    /// **The single representation owns nothing.** `OrdinalSpace` is a `Copy`
    /// view of `&[u8] + usize` — a slice reference and a length — so building one
    /// at open allocates zero bytes, which is the whole point of removing the hex
    /// `Vec<String>` and the `HashMap<String, u32>`. Red if either an owning
    /// field creeps back in (the size grows) or `Copy` is lost.
    #[test]
    fn ordinal_space_is_a_borrowed_view() {
        assert_eq!(
            std::mem::size_of::<OrdinalSpace<'_>>(),
            std::mem::size_of::<&[u8]>() + std::mem::size_of::<usize>(),
            "OrdinalSpace grew an owned field — it must stay a borrow of oids_raw"
        );
        fn assert_copy<T: Copy>() {}
        assert_copy::<OrdinalSpace<'_>>();
    }
}