alkgit-substrate 0.1.0

//! The OQ-19 rider-2 substrate facet (`substrate/w7-oq19-substrate-acks`):
//! the positive-ack capture re-run over the **real** substrate Find —
//! the machinery-side wave-5 rows
//! (`tests/duplex_serving.rs` / `tests/stateless_serving.rs` in the core
//! crate, over the `MemFind` double) re-landed on the substrate-backed
//! backend. The core crate's published serving machines
//! ([`serve_duplex`], [`serve_stateless_post`]) drive the POC-X
//! composition shape — the core is the published anchor, the substrate
//! backend rides the frozen seam — with the substrate [`PackGen`]'s real
//! `common_haves` (exists + is-commit over the alkblobs family store)
//! fronting the recognition.
//!
//! The session (the wave-5 rows' shape, verbatim): a multi-have fetch on
//! both serving substrates (duplex + stateless) — one no-`done` round
//! carrying two recognized commit haves plus one missing and one
//! exists-but-not-a-commit (blob) have emits `acknowledgments` +
//! `ACK <oid>` per recognized have in request order (`NAK` absent,
//! flush-terminated, never `ready`), and the follow-up done round over
//! the same mixed have set packs exactly closure(want) −
//! closure(recognized subset). The recognition contract (ADR-014 §5)
//! is asserted at the real seam: the blob have is **in** the store and
//! still unrecognized; the missing have subtracts nothing.
//!
//! Rows are names + `Contract stamp:` versions, per the core's
//! `tests/common` §The row/stamp discipline.

use std::collections::{BTreeMap, BTreeSet};
use std::pin::Pin;
use std::sync::Arc;
use std::task::{Context, Poll};

use alkblobs::Store;
use alkcall::core::Identity;
use alkgit::{
    serve_duplex, serve_stateless_post, AccessAction, AuthorizedRepo, BodyReader, GitPackGen,
    GitPackIngest, GitRefs, Limits, PreparedPush, PushOptions, RefUpdate, RepoRecord, Service,
    StorageError, Visibility,
};
use async_trait::async_trait;
use futures::io::AsyncWrite as FuturesAsyncWrite;
use futures::stream;
use gix_hash::ObjectId;
use gix_object::Kind;
use tokio::io::{duplex, AsyncReadExt, AsyncWriteExt};
use tokio::sync::Semaphore;

use alkgit_substrate::{InMemoryRefTable, ObjectFront, PackGen};

const FLUSH: &[u8] = b"0000";
const DELIMITER: &[u8] = b"0001";

fn repo(id: &str) -> RepoRecord {
    RepoRecord {
        repo_id: id.to_string(),
        visibility: Visibility::Private,
        grants: BTreeMap::new(),
    }
}

/// The marker the door hands the entries — the checked path
/// (`AuthorizedRepo::authorize`); here the holder identity is
/// read-granted.
fn marker(record: RepoRecord, identity: &Identity, action: AccessAction) -> AuthorizedRepo {
    AuthorizedRepo::authorize(record, Some(identity), action).expect("the door's check passed")
}

/// Encodes one pkt-line, client-side framing style.
fn pkt(payload: &[u8]) -> Vec<u8> {
    let total = payload.len() + 4;
    let mut out = Vec::with_capacity(total);
    out.extend_from_slice(format!("{total:04x}").as_bytes());
    out.extend_from_slice(payload);
    out
}

/// A V2 command request, client-side framing: the `command=` line, the
/// `0001` delimiter, the args, the terminating flush.
fn command_request(command: &[u8], pre: &[&str], args: &[&str]) -> Vec<u8> {
    [
        pkt(command),
        pre.iter()
            .flat_map(|l| pkt(l.as_bytes()))
            .collect::<Vec<u8>>(),
        DELIMITER.to_vec(),
        args.iter()
            .flat_map(|l| pkt(l.as_bytes()))
            .collect::<Vec<u8>>(),
        FLUSH.to_vec(),
    ]
    .concat()
}

/// Reads pkt-lines from the client end until (and including) a flush —
/// one advertisement or one command response.
async fn read_pkts(reader: &mut tokio::io::DuplexStream) -> Vec<Vec<u8>> {
    let mut packets = Vec::new();
    loop {
        let mut header = [0u8; 4];
        reader
            .read_exact(&mut header)
            .await
            .expect("pkt-line header");
        let len = usize::from_str_radix(std::str::from_utf8(&header).expect("hex header"), 16)
            .expect("hex length");
        if len < 4 {
            packets.push(Vec::new());
            break;
        }
        let mut payload = vec![0u8; len - 4];
        reader
            .read_exact(&mut payload)
            .await
            .expect("pkt-line payload");
        packets.push(payload);
    }
    packets
}

fn text(payload: &[u8]) -> String {
    String::from_utf8_lossy(payload)
        .trim_end_matches('\n')
        .to_owned()
}

fn text_lines(pkts: &[Vec<u8>]) -> Vec<String> {
    pkts.iter().map(|p| text(p)).collect()
}

/// The honest advertisement's shape as the substrate-backed session
/// serves it: the pinned four lines, plus the sha256 rider line when
/// the core crate's `sha256` feature rides the build (the feature is
/// the core crate's — OQ-05's passthrough — so the row probes the
/// emission rather than cfg-mirroring it).
fn assert_advertisement(mut lines: Vec<String>) {
    assert!(
        !lines.is_empty() && lines[lines.len() - 1].is_empty(),
        "the advertisement is flush-terminated: {lines:?}"
    );
    lines.pop();
    let base = ["version 2", "ls-refs=unborn", "fetch=wait-for-done"];
    assert_eq!(
        &lines[..base.len()],
        &base,
        "the honest advertisement's pinned prefix"
    );
    assert_eq!(lines[base.len()], "object-format=sha1");
    match lines.len() {
        4 => {}
        5 => {
            assert_eq!(
                lines[4], "object-format=sha256",
                "the sha256 rider rides the core crate's feature (OQ-05 passthrough)"
            );
        }
        _ => panic!("an unexpected advertisement shape: {lines:?}"),
    }
}

/// A chain of three commits (each its own tree + blob) put through the
/// real object front: the same fixture the wave-5 rows drove over the
/// doubles, with the loose objects now under the alkblobs family pool.
struct Fixture {
    record: RepoRecord,
    commit_a: ObjectId,
    commit_b: ObjectId,
    commit_c: ObjectId,
    blob_a: ObjectId,
    blob_c: ObjectId,
    tree_c: ObjectId,
}

async fn put_object(front: &ObjectFront, kind: Kind, content: &[u8]) -> ObjectId {
    let (key, _pin) = front.put(kind, content).await.unwrap();
    alkgit_substrate::key_to_oid(&key).expect("the put's key is the family key")
}

fn tree_bytes(entries: &[(&str, &ObjectId)]) -> Vec<u8> {
    let mut tree = Vec::new();
    for (name, oid) in entries {
        tree.extend_from_slice(format!("100644 {name}\0").as_bytes());
        tree.extend_from_slice(oid.as_bytes());
    }
    tree
}

fn commit_bytes(tree: &ObjectId, parents: &[&ObjectId]) -> Vec<u8> {
    let mut body = format!("tree {tree}\n");
    for parent in parents {
        body.push_str(&format!("parent {parent}\n"));
    }
    body.push_str(
        "author A <a@example.invalid> 1750000000 +0000\n\
         committer A <a@example.invalid> 1750000000 +0000\n\nthe message\n",
    );
    body.into_bytes()
}

/// The substrate-backed fixture: the object pool is the real
/// [`ObjectFront`] over an in-memory alkblobs store, the pack generation
/// and the recognition seam are the real [`PackGen`] (direct shape —
/// through-store serves byte-identical packs, `substrate/w7-packgen`'s
/// row), the refs table is the substrate's [`InMemoryRefTable`] (the
/// fetch session lists no refs on either row). The backend's record
/// reads its refs against the table with no rows — the rider session
/// drives only wants and haves.
async fn fixture() -> (Fixture, Arc<InMemoryRefTable>, Arc<PackGen>) {
    let front = ObjectFront::new(Arc::new(Store::mem_only()));
    let blob_a = put_object(&front, Kind::Blob, b"alpha").await;
    let blob_b = put_object(&front, Kind::Blob, b"beta").await;
    let blob_c = put_object(&front, Kind::Blob, b"gamma").await;
    let tree_a = put_object(&front, Kind::Tree, &tree_bytes(&[("a.txt", &blob_a)])).await;
    let tree_b = put_object(&front, Kind::Tree, &tree_bytes(&[("b.txt", &blob_b)])).await;
    let tree_c = put_object(&front, Kind::Tree, &tree_bytes(&[("c.txt", &blob_c)])).await;
    let commit_a = put_object(&front, Kind::Commit, &commit_bytes(&tree_a, &[])).await;
    let commit_b = put_object(&front, Kind::Commit, &commit_bytes(&tree_b, &[&commit_a])).await;
    let commit_c = put_object(&front, Kind::Commit, &commit_bytes(&tree_c, &[&commit_b])).await;

    let mut record = repo("owner/substrate");
    record
        .grants
        .insert("u-7731".to_string(), vec![AccessAction::Read]);
    let refs = Arc::new(InMemoryRefTable::new());
    let packgen = Arc::new(PackGen::direct(front));
    (
        Fixture {
            record,
            commit_a,
            commit_b,
            commit_c,
            blob_a,
            blob_c,
            tree_c,
        },
        refs,
        packgen,
    )
}

fn identity() -> Identity {
    Identity {
        id: "u-7731".to_string(),
        scopes: Vec::new(),
        resources: std::collections::HashMap::new(),
    }
}

/// The seam the fetch session never reaches (ADR-010's one-seam-per-call
/// discipline; the substrate's ingestion implement is wave 8's, the
/// rider is fetch-side): an honest `Err` if it is ever called, never a
/// silent success. Nothing in the rider session calls it.
#[derive(Clone, Default)]
struct UnservicedIngest;

#[async_trait]
impl GitPackIngest for UnservicedIngest {
    async fn prepare(
        &self,
        _repo: &RepoRecord,
        _updates: Vec<RefUpdate>,
        _pack: Box<dyn std::io::Read + Send>,
        _push_options: Option<PushOptions>,
        _limits: Limits,
    ) -> Result<PreparedPush, StorageError> {
        Err(StorageError::Io(
            "the fetch session does not reach the ingestion seam".to_owned(),
        ))
    }
}

/// Decodes the sink's band-1 chunk stream back through the pack
/// machinery: `(oid, kind, body)` per entry plus the verified trailer
/// — every entry whole-base (the pipeline's `PackCopyAndBaseObjects`),
/// inflating and re-hashing exactly as the core rows' decoder does.
fn decode_pack(pack: &[u8]) -> (BTreeSet<(ObjectId, Kind)>, ObjectId) {
    let mut entries = gix_pack::data::input::BytesToEntriesIter::new_from_header(
        std::io::Cursor::new(pack),
        gix_pack::data::input::Mode::Verify,
        gix_pack::data::input::EntryDataMode::Keep,
        gix_hash::Kind::Sha1,
    )
    .expect("the pack stream starts with a decodable header");
    let mut objects = BTreeSet::new();
    let mut trailer = None;
    let mut decompressor = gix_zlib::Decompress::new();
    for entry in &mut entries {
        let entry = entry.expect("every pack entry decodes");
        if entry.trailer.is_some() {
            trailer = entry.trailer;
        }
        let kind = match entry.header {
            gix_pack::data::entry::Header::Commit => Kind::Commit,
            gix_pack::data::entry::Header::Tree => Kind::Tree,
            gix_pack::data::entry::Header::Blob => Kind::Blob,
            gix_pack::data::entry::Header::Tag => Kind::Tag,
            delta @ (gix_pack::data::entry::Header::OfsDelta { .. }
            | gix_pack::data::entry::Header::RefDelta { .. }) => {
                panic!("the pipeline packs whole bases, got {delta:?}")
            }
        };
        let compressed = entry.compressed.expect("keep mode retains the entry bytes");
        let mut object = vec![0; entry.decompressed_size as usize];
        let inflated = gix_zlib::stream::inflate::read(
            &mut std::io::Cursor::new(compressed),
            &mut decompressor,
            &mut object,
        )
        .expect("the entry inflates");
        decompressor.reset();
        assert_eq!(
            inflated,
            object.len(),
            "the inflate filled the entry's size"
        );
        let id = gix_object::compute_hash(gix_hash::Kind::Sha1, kind, &object)
            .expect("the object re-hashes");
        objects.insert((id, kind));
    }
    let trailer = trailer.unwrap_or_else(|| {
        let body_len = pack.len() - gix_hash::Kind::Sha1.len_in_bytes();
        let mut hasher = gix_hash::hasher(gix_hash::Kind::Sha1);
        hasher.update(&pack[..body_len]);
        let computed = hasher
            .try_finalize()
            .expect("the zero-entry trailer hashes over the header alone");
        let digest: [u8; 20] = pack[body_len..]
            .try_into()
            .expect("the trailer is a full digest");
        let stored = ObjectId::from(digest);
        assert_eq!(computed, stored, "the zero-entry pack's trailer verifies");
        stored
    });
    (objects, trailer)
}

/// The OQ-19 rider-2 positive-ack capture over the duplex session, the
/// substrate-facet run: a multi-have negotiation session where the
/// no-`done` round acks **each** recognized have — per-have `ACK <oid>`
/// lines in the request's order, only the seam-recognized subset (the
/// missing and the exists-but-not-a-commit haves stay unacked), `NAK`
/// absent — and the done round's boundary set is the recognized subset:
/// the unrecognized haves subtract nothing, both commons bound the walk.
///
/// Contract stamp: OQ-19 rider 2 (the substrate-odb facet — the
/// both-runs wording's second run, tasks/architecture/
/// oq-19-wire-validations.md §2; the machinery-side run is
/// core/w5-oq19-positive-acks); ADR-014 §1 (ACK per recognized have,
/// NAK only when no ACKs); ADR-014 §2 as amended by review 001 A-4
/// (the boundary set is the recognized subset); ADR-014 §5 (exists +
/// is-commit — the blob have drives the negative leg over the real
/// front); ADR-005 §1 (both rounds ride one session over the command
/// loop); ADR-019 (the backend is the real substrate composition).
#[tokio::test]
async fn a_multi_have_session_acks_each_recognized_have_and_the_done_boundary_is_the_recognized_subset(
) {
    let (fx, refs, packgen) = fixture().await;
    let missing = ObjectId::from([9u8; 20]);
    let holder = identity();
    let read_marker = marker(fx.record.clone(), &holder, AccessAction::Read);
    let refs_dyn: Arc<dyn GitRefs> = refs;
    let gen_dyn: Arc<dyn GitPackGen> = packgen;
    let ingest: Arc<dyn GitPackIngest> = Arc::new(UnservicedIngest);

    let (mut peer, joined) = {
        let (peer, stream) = duplex(1 << 16);
        let joined = tokio::spawn({
            let marker = read_marker.clone();
            async move {
                serve_duplex(
                    &marker,
                    stream,
                    Limits::default(),
                    Service::UploadPack,
                    refs_dyn,
                    gen_dyn,
                    ingest,
                    false,
                    Arc::new(Semaphore::new(
                        Limits::default().max_concurrent_blocking_tasks,
                    )),
                )
                .await
            }
        });
        (peer, joined)
    };

    assert_advertisement(text_lines(&read_pkts(&mut peer).await));

    peer.write_all(&command_request(
        b"command=fetch\n",
        &["object-format=sha1", "agent=git/2.43"],
        &[
            &format!("want {}", fx.commit_c),
            &format!("have {}", fx.commit_a),
            &format!("have {}", missing),
            &format!("have {}", fx.commit_b),
        ],
    ))
    .await
    .unwrap();
    let ack_round = text_lines(&read_pkts(&mut peer).await);
    assert_eq!(
        ack_round,
        vec![
            "acknowledgments".to_owned(),
            format!("ACK {}", fx.commit_a),
            format!("ACK {}", fx.commit_b),
            String::new(),
        ],
        "per-have ACKs in request order, unrecognized haves silent, flush-terminated"
    );

    peer.write_all(&command_request(
        b"command=fetch\n",
        &["object-format=sha1", "agent=git/2.43"],
        &[
            "thin-pack",
            "no-progress",
            &format!("want {}", fx.commit_c),
            &format!("have {}", fx.commit_a),
            &format!("have {}", missing),
            &format!("have {}", fx.blob_a),
            &format!("have {}", fx.commit_b),
            "done",
        ],
    ))
    .await
    .unwrap();
    let done_round = read_pkts(&mut peer).await;
    assert_eq!(text(&done_round[0]), "packfile");
    let band_chunks: Vec<&[u8]> = done_round[1..done_round.len() - 1]
        .iter()
        .map(|p| &p[1..])
        .collect();
    let pack: Vec<u8> = band_chunks.concat();
    let (ids, _trailer) = decode_pack(&pack);
    assert_eq!(
        ids,
        BTreeSet::from([
            (fx.commit_c, Kind::Commit),
            (fx.tree_c, Kind::Tree),
            (fx.blob_c, Kind::Blob),
        ]),
        "the boundary set is the recognized subset: both commons bounded the \
         walk, the missing and the non-commit have subtracted nothing"
    );
    assert!(done_round.last().unwrap().is_empty(), "flush-terminated");

    drop(peer);
    joined
        .await
        .expect("the session task joins")
        .expect("the clean close ends the session quietly");
}

/// An async sink whose framed bytes the rows inspect after the fact —
/// the http door's response body surface shape.
#[derive(Clone, Default)]
struct VecSink(Arc<parking_lot::Mutex<Vec<u8>>>);

impl VecSink {
    fn body(&self) -> Vec<u8> {
        self.0.lock().clone()
    }
}

impl FuturesAsyncWrite for VecSink {
    fn poll_write(
        self: Pin<&mut Self>,
        _cx: &mut Context<'_>,
        buf: &[u8],
    ) -> Poll<std::io::Result<usize>> {
        self.0.lock().extend_from_slice(buf);
        Poll::Ready(Ok(buf.len()))
    }

    fn poll_flush(self: Pin<&mut Self>, _cx: &mut Context<'_>) -> Poll<std::io::Result<()>> {
        Poll::Ready(Ok(()))
    }

    fn poll_close(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<std::io::Result<()>> {
        self.poll_flush(cx)
    }
}

type PostBody =
    BodyReader<stream::Iter<std::vec::IntoIter<std::io::Result<bytes::Bytes>>>, std::io::Error>;

fn body_parts(parts: Vec<Vec<u8>>) -> PostBody {
    let items: Vec<std::io::Result<bytes::Bytes>> = parts
        .into_iter()
        .map(|c| Ok(bytes::Bytes::from(c)))
        .collect();
    BodyReader::new(stream::iter(items))
}

/// A subsequent POST body (negotiation rounds): only the command.
fn command_post_body(command: &[u8], pre: &[&str], args: &[&str]) -> Vec<u8> {
    [
        pkt(command),
        pre.iter()
            .flat_map(|l| pkt(l.as_bytes()))
            .collect::<Vec<u8>>(),
        DELIMITER.to_vec(),
        args.iter()
            .flat_map(|l| pkt(l.as_bytes()))
            .collect::<Vec<u8>>(),
        FLUSH.to_vec(),
    ]
    .concat()
}

fn parse_pkt_lines(body: &[u8]) -> Vec<Vec<u8>> {
    let mut lines = Vec::new();
    let mut cursor = body;
    while !cursor.is_empty() {
        let (header, rest) = cursor.split_at(4);
        let len = usize::from_str_radix(std::str::from_utf8(header).expect("hex header"), 16)
            .expect("hex length");
        cursor = rest;
        if len < 4 {
            lines.push(Vec::new());
            continue;
        }
        let (payload, rest) = cursor.split_at(len - 4);
        cursor = rest;
        lines.push(payload.to_vec());
    }
    lines
}

/// The OQ-19 rider-2 positive-ack capture over the stateless entry, the
/// substrate-facet run: a no-`done` POST with multiple real haves acks
/// **each** recognized have — per-have `ACK <oid>` lines in the
/// request's order, only the seam-recognized subset (the missing and
/// the exists-but-not-a-commit haves stay unacked), `NAK` absent — and
/// the done round POST's boundary set is the recognized subset: the
/// unrecognized haves subtract nothing, both commons bound the walk.
///
/// Contract stamp: OQ-19 rider 2 (the substrate-odb facet — the
/// both-runs wording's second run, tasks/architecture/
/// oq-19-wire-validations.md §2; the machinery-side run is
/// core/w5-oq19-positive-acks); ADR-014 §1 (ACK per recognized have,
/// NAK only when no ACKs); ADR-014 §2 as amended by review 001 A-4
/// (the boundary set is the recognized subset); ADR-014 §5 (exists +
/// is-commit over the real front); ADR-014 §7 (each POST is one round,
/// no cross-request state); ADR-019 (the backend is the real substrate
/// composition).
#[tokio::test]
async fn a_multi_have_post_acks_each_recognized_have_and_the_done_boundary_is_the_recognized_subset(
) {
    let (fx, refs, packgen) = fixture().await;
    let missing = ObjectId::from([9u8; 20]);
    let holder = identity();
    let read_marker = marker(fx.record.clone(), &holder, AccessAction::Read);
    let refs_dyn: Arc<dyn GitRefs> = refs;
    let gen_dyn: Arc<dyn GitPackGen> = packgen;
    let ingest: Arc<dyn GitPackIngest> = Arc::new(UnservicedIngest);
    let pipeline = Arc::new(Semaphore::new(
        Limits::default().max_concurrent_blocking_tasks,
    ));

    let body = body_parts(vec![command_post_body(
        b"command=fetch\n",
        &["object-format=sha1", "agent=git/2.43"],
        &[
            &format!("want {}", fx.commit_c),
            &format!("have {}", fx.commit_a),
            &format!("have {}", missing),
            &format!("have {}", fx.commit_b),
        ],
    )]);
    let sink = VecSink::default();
    serve_stateless_post(
        &read_marker,
        body,
        sink.clone(),
        Limits::default(),
        Service::UploadPack,
        Arc::clone(&refs_dyn),
        Arc::clone(&gen_dyn),
        Arc::clone(&ingest),
        false,
        Arc::clone(&pipeline),
    )
    .await
    .unwrap();
    let ack_round: Vec<String> = parse_pkt_lines(&sink.body())
        .iter()
        .map(|p| text(p))
        .collect();
    assert_eq!(
        ack_round,
        vec![
            "acknowledgments".to_owned(),
            format!("ACK {}", fx.commit_a),
            format!("ACK {}", fx.commit_b),
            String::new(),
        ],
        "per-have ACKs in request order, unrecognized haves silent, flush-terminated"
    );

    let body = body_parts(vec![command_post_body(
        b"command=fetch\n",
        &["object-format=sha1", "agent=git/2.43"],
        &[
            &format!("want {}", fx.commit_c),
            &format!("have {}", fx.commit_a),
            &format!("have {}", missing),
            &format!("have {}", fx.blob_a),
            &format!("have {}", fx.commit_b),
            "done",
        ],
    )]);
    let sink = VecSink::default();
    serve_stateless_post(
        &read_marker,
        body,
        sink.clone(),
        Limits::default(),
        Service::UploadPack,
        refs_dyn,
        gen_dyn,
        ingest,
        false,
        pipeline,
    )
    .await
    .unwrap();
    let lines = parse_pkt_lines(&sink.body());
    assert_eq!(text(&lines[0]), "packfile");
    let band_chunks: Vec<&[u8]> = lines[1..lines.len() - 1].iter().map(|p| &p[1..]).collect();
    let pack: Vec<u8> = band_chunks.concat();
    let (ids, _trailer) = decode_pack(&pack);
    assert_eq!(
        ids,
        BTreeSet::from([
            (fx.commit_c, Kind::Commit),
            (fx.tree_c, Kind::Tree),
            (fx.blob_c, Kind::Blob),
        ]),
        "the boundary set is the recognized subset: both commons bounded the \
         walk, the missing and the non-commit have subtracted nothing"
    );
    assert!(
        lines.last().unwrap().is_empty(),
        "the response ends at flush"
    );
}