flint-lean 0.11.0

//! Content-defined chunking of the manifest's key stream
//! (`docs/plans/flint-lean-chunked-manifest-design.md` §3).
//!
//! The pointer layout made a TAKEOVER cheap; it did nothing for a
//! publish, which still writes every entry of the project to change one
//! file. Splitting the sorted key stream into immutable, content-
//! addressed chunks makes a publish rewrite only what it touched.
//!
//! Everything here is pure and synchronous on purpose: the boundary
//! rule is part of the on-the-wire format, and a format decision that
//! can only be exercised through a store is one nobody tests properly.

use flint_store::crc64_nvme;

/// Entries per chunk, in expectation. ~1 MiB of compact JSON at the
/// ~277 B/entry this manifest measures: big enough that per-object
/// overhead is noise, small enough that one changed file rewrites ~1
/// MiB rather than the whole project.
pub const CHUNK_TARGET: usize = 4096;
/// Floor. Hash boundaries are geometrically distributed, so without one
/// a project accumulates many tiny objects.
pub const CHUNK_MIN: usize = CHUNK_TARGET / 4;
/// Ceiling, so one pathological run cannot recreate the single-object
/// problem this exists to remove.
pub const CHUNK_MAX: usize = CHUNK_TARGET * 4;

/// Does a chunk end after this key?
///
/// A function of the KEY ALONE — never of the entry's contents, and
/// never of the key's position. Both matter:
///
/// - Position would be fixed-count chunking, where inserting one key
///   near the front shifts every later key by one slot, moves every
///   boundary and rewrites every chunk. That restores O(entries) on
///   exactly the operation being optimised (§3).
/// - Contents would move a boundary whenever a file's size or etag
///   changed, so editing a file in place would re-split the project
///   instead of rewriting one chunk.
fn cuts_after_with(key: &str, target: usize) -> bool {
    crc64_nvme(key.as_bytes()).is_multiple_of(target as u64)
}

/// Split a SORTED key stream into content-defined runs, as end-exclusive
/// indices. `[]` for no keys; the last run always ends at `keys.len()`.
///
/// `min`/`max` are where the pure-function property leaks: a suppressed
/// or forced cut depends on where the previous boundary fell, so a
/// change can cascade — but only to the next natural boundary that
/// satisfies `min`, which is one chunk in expectation. The guarantee is
/// "O(changed) in expectation", not worst case, and this is why.
pub fn chunk_ranges_with(keys: &[&str], target: usize, min: usize, max: usize) -> Vec<usize> {
    debug_assert!(target > 0 && min > 0 && max >= min, "degenerate chunk sizing");
    let mut cuts = Vec::new();
    let mut run = 0usize;
    for (i, k) in keys.iter().enumerate() {
        run += 1;
        if (run >= min && cuts_after_with(k, target)) || run >= max {
            cuts.push(i + 1);
            run = 0;
        }
    }
    // The tail is a chunk even when it never hit a boundary. Dropping it
    // would silently lose every entry after the last cut.
    if run > 0 {
        cuts.push(keys.len());
    }
    cuts
}

/// The shipped sizing.
pub fn chunk_ranges(keys: &[&str]) -> Vec<usize> {
    chunk_ranges_with(keys, CHUNK_TARGET, CHUNK_MIN, CHUNK_MAX)
}

/// The content address of a chunk body: SHA-256, truncated to 128 bits.
///
/// NOT `crc64_nvme`, which the boundary rule above uses. The two hashes
/// are answering different questions and carry different consequences:
/// a boundary collision merely moves a cut and costs nothing, while an
/// ADDRESS collision means two different chunks share an object key and
/// one silently shadows the other — a data-loss class, and a silent one.
/// 64 bits is not enough margin to take that risk for; 128 is past any
/// practical concern.
pub fn chunk_address(body: &[u8]) -> String {
    use sha2::{Digest, Sha256};
    let d = Sha256::digest(body);
    d.iter().take(16).map(|b| format!("{b:02x}")).collect()
}

/// Is `addr` an address `chunk_address` could have minted: 32 lowercase
/// hex digits? A pointer is read from the bucket, so an address in it is
/// remote-supplied text. Anything that turns one into a LOCAL path (the
/// chunk cache) must refuse every other shape, `..` and `/` above all.
pub fn is_chunk_address(addr: &str) -> bool {
    addr.len() == 32 && addr.bytes().all(|b| b.is_ascii_digit() || (b'a'..=b'f').contains(&b))
}

// ── the wire format ─────────────────────────────────────────────────

use std::collections::BTreeMap;

use serde::{Deserialize, Serialize};

use super::manifest::{LeanEntry, Tombstone};
use super::{LeanError, LeanResult};

/// One entry of the pointer's chunk list.
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
pub struct ChunkRef {
    /// Content address; the object lives at `.flint/lean/chunks/<addr>`.
    pub addr: String,
    /// The chunk's FIRST key. Carried so a reader can resolve a path to
    /// its chunk without fetching anything (design §5's partial reads),
    /// and so `assemble` can check the list covers the key space in
    /// order rather than assuming it.
    pub first: String,
    /// Entry count, so a truncated or substituted chunk is caught
    /// against a number the pointer committed to.
    pub n: usize,
}

/// A chunk object's body.
///
/// A struct rather than a bare map so the object is self-describing and
/// so a future field does not change what the address hashes over.
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct ChunkBody {
    pub entries: BTreeMap<String, LeanEntry>,
}

fn encode(body: &ChunkBody) -> LeanResult<Vec<u8>> {
    serde_json::to_vec(body).map_err(|e| LeanError::State(format!("chunk encode: {e}")))
}

/// The document's tombstones (review 2026-09-18, H1e) in the chunked
/// layout: one content-addressed object beside the chunks, named by the
/// pointer, so an unchanged set re-addresses to the object that already
/// exists and the chunk reaper keeps it as it keeps a referenced chunk.
/// The single-object layout carries them inline in the document.
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct TombstoneBody {
    pub tombstones: BTreeMap<String, Tombstone>,
}

pub fn encode_tombstones(tombstones: &BTreeMap<String, Tombstone>) -> LeanResult<(String, Vec<u8>)> {
    let body = serde_json::to_vec(&TombstoneBody { tombstones: tombstones.clone() })
        .map_err(|e| LeanError::State(format!("tombstones encode: {e}")))?;
    Ok((chunk_address(&body), body))
}

pub fn decode_tombstones(bytes: &[u8], addr: &str) -> LeanResult<BTreeMap<String, Tombstone>> {
    let got = chunk_address(bytes);
    if got != addr {
        return Err(LeanError::State(format!("tombstones {addr} do not match their address (body hashes to {got})")));
    }
    let body: TombstoneBody =
        serde_json::from_slice(bytes).map_err(|e| LeanError::State(format!("tombstones {addr} parse: {e}")))?;
    Ok(body.tombstones)
}

/// Split entries into chunk bodies and their references.
///
/// `BTreeMap` iterates in sorted order and `LeanEntry` holds no floats,
/// so identical entry sets encode to identical bytes and therefore to
/// identical addresses. That determinism is not a nicety — it is the
/// whole dedup mechanism: an untouched chunk must re-address to the
/// object that already exists, or every publish uploads everything.
pub fn split_with(
    entries: &BTreeMap<String, LeanEntry>,
    target: usize,
    min: usize,
    max: usize,
) -> LeanResult<Vec<(ChunkRef, Vec<u8>)>> {
    let keys: Vec<&str> = entries.keys().map(|s| s.as_str()).collect();
    let cuts = chunk_ranges_with(&keys, target, min, max);
    let mut out = Vec::with_capacity(cuts.len());
    let mut start = 0usize;
    for c in cuts {
        let slice: BTreeMap<String, LeanEntry> = keys[start..c]
            .iter()
            .map(|k| ((*k).to_string(), entries[*k].clone()))
            .collect();
        let bytes = encode(&ChunkBody { entries: slice })?;
        out.push((
            ChunkRef {
                addr: chunk_address(&bytes),
                first: keys[start].to_string(),
                n: c - start,
            },
            bytes,
        ));
        start = c;
    }
    Ok(out)
}

pub fn split(entries: &BTreeMap<String, LeanEntry>) -> LeanResult<Vec<(ChunkRef, Vec<u8>)>> {
    split_with(entries, CHUNK_TARGET, CHUNK_MIN, CHUNK_MAX)
}

/// The chunk list covers the key space in order: each chunk's first key is
/// past the one before it.
pub fn check_list_order(refs: &[ChunkRef]) -> LeanResult<()> {
    for w in refs.windows(2) {
        if w[0].first >= w[1].first {
            return Err(LeanError::State(format!(
                "chunk list is not in strictly increasing key order: {:?} then {:?}",
                w[0].first, w[1].first
            )));
        }
    }
    Ok(())
}

/// Parse chunk `i` of `refs` and check it against what the list committed
/// to: its entry count, its first key, and no key at or past the next
/// chunk's first. The body's address is the caller's to have checked. A
/// full load (`assemble`) and lever G's view (`docview`) both parse
/// through here, so a chunk passes the same checks however it is read.
pub fn parse_chunk(refs: &[ChunkRef], i: usize, b: &[u8]) -> LeanResult<BTreeMap<String, LeanEntry>> {
    let r = &refs[i];
    let body: ChunkBody =
        serde_json::from_slice(b).map_err(|e| LeanError::State(format!("chunk {} parse: {e}", r.addr)))?;
    if body.entries.len() != r.n {
        return Err(LeanError::State(format!(
            "chunk {} holds {} entries, the pointer says {}",
            r.addr,
            body.entries.len(),
            r.n
        )));
    }
    match body.entries.keys().next() {
        Some(k) if *k == r.first => {}
        Some(k) => {
            return Err(LeanError::State(format!("chunk {} starts at {k:?}, the pointer says {:?}", r.addr, r.first)))
        }
        None => {
            return Err(LeanError::State(format!(
                "chunk {} is empty; an empty chunk cannot be addressed by a first key",
                r.addr
            )))
        }
    }
    if let (Some(next), Some(last)) = (refs.get(i + 1), body.entries.keys().next_back()) {
        if *last >= next.first {
            return Err(LeanError::State(format!(
                "chunk {} runs to {last:?}, past the next chunk's first key {:?} — the \
                 chunks overlap and one shadows the other",
                r.addr, next.first
            )));
        }
    }
    Ok(body.entries)
}

/// One step of a re-chunk's key stream.
enum Step {
    /// A key of the changed document, its entry, and whether it is an old
    /// chunk's LAST key (the old chunking cut after it).
    Key(String, LeanEntry, bool),
    /// A deleted key, or a delete of a key the document lacks.
    Skip,
    End,
}

/// Re-chunk a document after `changes`, reusing every chunk of `old` that
/// neither the changes nor their cascade reach (latency plan §12, claim 1).
///
/// `old` is the chunk list `split_with` made of the document, and
/// `entries_of(i)` hands back chunk `i`'s entries; it is asked only for the
/// chunks the re-chunk passes through. `changes` maps a path to its new
/// entry, `None` deleting it (a delete of a path the document lacks changes
/// nothing). Each ref comes back with the body this call encoded, or `None`
/// where it is `old`'s ref, reused.
///
/// The chunker's only state is the run since the last cut, 0 at every cut.
/// So the work starts at the chunk holding the next change, whose start is a
/// cut in both chunkings, and stops at the first new cut that is also an old
/// one with no change before the next old chunk: from there the key stream,
/// and so its chunking, is the old one. The result is `split_with` of the
/// changed entries, ref for ref and byte for byte.
pub fn rechunk_with<F>(
    old: &[ChunkRef],
    mut entries_of: F,
    changes: &BTreeMap<String, Option<LeanEntry>>,
    target: usize,
    min: usize,
    max: usize,
) -> LeanResult<Vec<(ChunkRef, Option<Vec<u8>>)>>
where
    F: FnMut(usize) -> LeanResult<std::sync::Arc<BTreeMap<String, LeanEntry>>>,
{
    debug_assert!(target > 0 && min > 0 && max >= min, "degenerate chunk sizing");
    let changes: Vec<(&String, &Option<LeanEntry>)> = changes.iter().collect();
    // The chunk a path falls in: the last whose first key is not after it.
    let home = |p: &str| old.partition_point(|r| r.first.as_str() <= p).saturating_sub(1);
    let mut load = |m: usize| -> LeanResult<Vec<(String, LeanEntry)>> {
        Ok(entries_of(m)?.iter().map(|(k, e)| (k.clone(), e.clone())).collect())
    };
    let mut out: Vec<(ChunkRef, Option<Vec<u8>>)> = Vec::with_capacity(old.len() + 1);
    let mut c = 0usize; // the next change
    let mut i = 0usize; // the next old chunk not yet reused or re-chunked
    while c < changes.len() {
        let start = home(changes[c].0);
        debug_assert!(old.is_empty() || start >= i, "a change before the old chunk the last region synced to");
        let start = start.max(i);
        out.extend(old[i..start.min(old.len())].iter().map(|r| (r.clone(), None)));
        let mut m = start;
        let mut cur = if m < old.len() { load(m)? } else { vec![] };
        let mut pos = 0usize;
        let mut run: BTreeMap<String, LeanEntry> = BTreeMap::new();
        let mut first: Option<String> = None;
        loop {
            // The changes below the next old chunk's first key are this one's.
            let bound = old.get(m + 1).map(|r| r.first.as_str());
            let change = changes.get(c).copied().filter(|(k, _)| bound.is_none_or(|b| k.as_str() < b));
            if pos >= cur.len() && change.is_none() && m + 1 < old.len() {
                m += 1;
                cur = load(m)?;
                pos = 0;
                continue;
            }
            let step = match (cur.get(pos), change) {
                (None, None) => Step::End,
                (Some((ok, _)), Some((ck, cv))) if ck.as_str() < ok.as_str() => {
                    c += 1;
                    cv.clone().map_or(Step::Skip, |e| Step::Key(ck.clone(), e, false))
                }
                (Some((ok, _)), Some((ck, cv))) if ck.as_str() == ok.as_str() => {
                    c += 1;
                    pos += 1;
                    cv.clone().map_or(Step::Skip, |e| Step::Key(ck.clone(), e, pos == cur.len()))
                }
                (Some((ok, oe)), _) => {
                    pos += 1;
                    Step::Key(ok.clone(), oe.clone(), pos == cur.len())
                }
                (None, Some((ck, cv))) => {
                    c += 1;
                    cv.clone().map_or(Step::Skip, |e| Step::Key(ck.clone(), e, false))
                }
            };
            let (k, e, old_last) = match step {
                Step::Key(k, e, last) => (k, e, last),
                Step::Skip => continue,
                Step::End => {
                    // The tail is a chunk even when it never hit a boundary.
                    if let Some(f) = first.take() {
                        out.push(encode_chunk(f, std::mem::take(&mut run))?);
                    }
                    i = old.len();
                    break;
                }
            };
            first.get_or_insert_with(|| k.clone());
            run.insert(k.clone(), e);
            if (run.len() >= min && cuts_after_with(&k, target)) || run.len() >= max {
                let f = first.take().expect("a run holds its first key");
                out.push(encode_chunk(f, std::mem::take(&mut run))?);
                // In sync: the old chunking cut here too, and no change falls
                // before the next old chunk.
                let synced = old_last
                    && match (changes.get(c), old.get(m + 1)) {
                        (None, _) => true,
                        (Some((ck, _)), Some(next)) => ck.as_str() >= next.first.as_str(),
                        (Some(_), None) => false,
                    };
                if synced {
                    i = m + 1;
                    break;
                }
            }
        }
    }
    out.extend(old[i.min(old.len())..].iter().map(|r| (r.clone(), None)));
    Ok(out)
}

fn encode_chunk(first: String, entries: BTreeMap<String, LeanEntry>) -> LeanResult<(ChunkRef, Option<Vec<u8>>)> {
    let n = entries.len();
    let bytes = encode(&ChunkBody { entries })?;
    Ok((ChunkRef { addr: chunk_address(&bytes), first, n }, Some(bytes)))
}

/// A body that does not hash to the address it was fetched by: the store
/// handed back something else (a stale read, a wrong key, a truncated
/// body), or a local copy was damaged.
pub fn check_address_of(body: &[u8], addr: &str) -> LeanResult<()> {
    let got = chunk_address(body);
    if got != addr {
        return Err(LeanError::State(format!(
            "chunk {addr} does not match its address (body hashes to {got})"
        )));
    }
    Ok(())
}

/// Rebuild the entry map from a chunk list and the fetched bodies.
///
/// Every invariant the format depends on is CHECKED here rather than
/// assumed, because each one fails silently if it is not. A manifest
/// used to be one object: it was either there or it was not. Split
/// across N objects, a wrong list produces a well-formed manifest that
/// is quietly missing entries — and a missing entry reads to every
/// consumer as a file the agent deleted.
///
/// Checked: the list is in strictly increasing key order; each body
/// addresses to the reference that named it; each body's length and
/// first key match what the pointer committed to; no chunk spills past
/// the next chunk's first key; and no key arrives twice.
pub fn assemble(refs: &[ChunkRef], bodies: &[Vec<u8>]) -> LeanResult<BTreeMap<String, LeanEntry>> {
    assemble_checked(refs, bodies, true)
}

/// [`assemble`] over bodies whose addresses the caller has ALREADY checked
/// one by one (`manifest::load_from` checks each as it arrives, so it can
/// tell a stale cache entry from a bad fetch): every other check runs, and
/// no body is hashed twice. A caller that has not checked must use
/// `assemble`.
pub fn assemble_verified<B: AsRef<[u8]>>(refs: &[ChunkRef], bodies: &[B]) -> LeanResult<BTreeMap<String, LeanEntry>> {
    assemble_checked(refs, bodies, false)
}

fn assemble_checked<B: AsRef<[u8]>>(
    refs: &[ChunkRef],
    bodies: &[B],
    check_address: bool,
) -> LeanResult<BTreeMap<String, LeanEntry>> {
    if refs.len() != bodies.len() {
        return Err(LeanError::State(format!(
            "chunk list names {} chunks but {} bodies were fetched",
            refs.len(),
            bodies.len()
        )));
    }
    check_list_order(refs)?;
    let mut out: BTreeMap<String, LeanEntry> = BTreeMap::new();
    for (i, b) in bodies.iter().enumerate() {
        // The object was fetched BY address, so a mismatch means the
        // store handed back something else — a stale read, a wrong key,
        // a truncated body. Content addressing makes this checkable for
        // one hash per chunk, and declining to check it would be
        // choosing not to know.
        let b = b.as_ref();
        if check_address {
            check_address_of(b, &refs[i].addr)?;
        }
        let body = ChunkBody { entries: parse_chunk(refs, i, b)? };
        for (k, v) in body.entries {
            if out.insert(k.clone(), v).is_some() {
                return Err(LeanError::State(format!(
                    "key {k:?} appears in more than one chunk"
                )));
            }
        }
    }
    let expect: usize = refs.iter().map(|r| r.n).sum();
    if out.len() != expect {
        return Err(LeanError::State(format!(
            "assembled {} entries, the chunk list accounts for {expect}",
            out.len()
        )));
    }
    Ok(out)
}