alkgit 0.1.0

//! The wave-2 protocol V2 machines over the duplex session substrate
//! with the wave-1 backend doubles (`core/w2-v2-advertisement`): the
//! capability advertisement and the ls-refs command machine, substrate-
//! agnostic (ADR-005 §3) — served through the duplex machinery's
//! framing helpers.
//!
//! Rows are names + `Contract stamp:` versions, per
//! `tests/common` §The row/stamp discipline.

mod common;

use std::collections::BTreeMap;

use common::MemRefs;
use gix_hash::ObjectId;
use tokio::io::{duplex, AsyncReadExt, AsyncWriteExt};

use alkgit::{
    advertise_v2, parse_ls_refs_args, serve_ls_refs, GitRefs, Limits, LsRefsArgs, RefUpdate,
    RepoRecord, Service, Session, SubstrateError, Visibility,
};

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

fn oid(b: u8) -> ObjectId {
    ObjectId::from([b; 20])
}

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

/// 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
}

/// Reads pkt-lines from the client end until (and including) a flush.
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
}

/// A duplex session + its client end; the session is
/// [`PktLineWriter`]-shaped — the machines meet the session through the
/// substrate-agnostic trait, not the Session type.
fn session(
    limits: Limits,
) -> (
    tokio::io::DuplexStream,
    Session<tokio::io::DuplexStream>,
    Limits,
) {
    let (peer, stream) = duplex(1 << 16);
    (
        peer,
        Session::new(stream, limits, Service::UploadPack),
        limits,
    )
}

/// A ls-refs request's wire bytes: the command line, the arg lines, the
/// terminating flush.
fn ls_refs_request(args: &[&str]) -> Vec<u8> {
    let mut out = pkt(b"command=ls-refs\n");
    out.extend_from_slice(DELIMITER);
    for arg in args {
        out.extend_from_slice(&pkt(arg.as_bytes()));
    }
    out.extend_from_slice(FLUSH);
    out
}

/// The advertisement is exactly the honest list: `version 2`, the three
/// capability lines, flush — no unimplemented capability appears
/// (declined by omission, POC-1-verified against real git).
///
/// Contract stamp: ADR-003 (the honest capability list, V2-first);
/// transport.md §Advertisement (the capability set); ADR-014 (the ack
/// loop is grammar, not capability — `fetch=wait-for-done` unchanged).
#[tokio::test]
async fn advertisement_emits_exactly_the_honest_capability_list() {
    let (mut peer, mut session, _) = session(Limits::default());

    advertise_v2(&mut session).await.unwrap();

    let packets = read_pkts(&mut peer).await;
    let text = |p: &Vec<u8>| String::from_utf8_lossy(p).trim_end_matches('\n').to_owned();
    assert_eq!(text(&packets[0]), "version 2");
    assert_eq!(text(&packets[1]), "ls-refs=unborn");
    assert_eq!(text(&packets[2]), "fetch=wait-for-done");
    assert_eq!(text(&packets[3]), "object-format=sha1");
    #[cfg(feature = "sha256")]
    assert_eq!(text(&packets[4]), "object-format=sha256");
    #[cfg(not(feature = "sha256"))]
    assert_eq!(packets.len(), 5, "the advertisement ends at flush");
    #[cfg(feature = "sha256")]
    assert_eq!(packets.len(), 6, "the advertisement ends at flush");
    for declined in [
        "shallow",
        "filter",
        "packfile-uris",
        "object-info",
        "server-option",
    ] {
        assert!(
            packets.iter().all(|p| !text(p).contains(declined)),
            "{declined} is declined by omission (ADR-003)"
        );
    }
}

/// ls-refs parses the wire arguments, streams the client-prefix-
/// filtered ref lines from the doubles' listing, and ends at flush —
/// the full round over the duplex request reader.
///
/// Contract stamp: transport.md §ls-refs (parse, stream, flush);
/// ADR-005 §3 (the machine core both serving entries feed — here over
/// the duplex machinery); ADR-009 (the request shape's substrates).
#[tokio::test]
async fn ls_refs_round_streams_prefix_filtered_lines_and_ends_at_flush() {
    let refs = MemRefs::default();
    let doubled = repo("owner/doubled");
    for (name, id) in [
        ("refs/heads/main", oid(1)),
        ("refs/heads/topic", oid(2)),
        ("refs/tags/v1", oid(3)),
    ] {
        refs.apply_updates(
            &doubled,
            vec![RefUpdate {
                name: name.to_string(),
                expected: None,
                new: Some(id),
            }],
            false,
        )
        .await
        .unwrap();
    }

    let (mut peer, mut session, _) = session(Limits::default());
    peer.write_all(&ls_refs_request(&["ref-prefix refs/heads/"]))
        .await
        .unwrap();

    let request = session.read_request().await.unwrap().expect("a request");
    assert_eq!(request.command, "ls-refs");
    let args = parse_ls_refs_args(&request.args).unwrap();
    let limits = *session.limits();
    serve_ls_refs(
        &mut session,
        refs.list_refs(&doubled).await.unwrap(),
        &args,
        &limits,
    )
    .await
    .unwrap();

    let packets = read_pkts(&mut peer).await;
    let text = |p: &Vec<u8>| String::from_utf8_lossy(p).trim_end_matches('\n').to_owned();
    assert_eq!(text(&packets[0]), format!("{} refs/heads/main", oid(1)));
    assert_eq!(text(&packets[1]), format!("{} refs/heads/topic", oid(2)));
    assert_eq!(packets.len(), 3, "the response ends at flush");
    assert!(packets[2].is_empty(), "the terminating flush");
}

/// `peel` and `symrefs` are per-request attribute flags: the attributes
/// ride the ref line (the observed wire shape — symref-target attribute
/// first, then peeled) and appear only when requested.
///
/// Contract stamp: transport.md §ls-refs (peel/symrefs per-request
/// flags); POC-1 §2 (the observed line shape
/// `"<oid> <ref>[ symref-target:…][ peeled:…]"`); ADR-018 §5 (peeled
/// tags and symref targets are ls-refs-arg data, not advertisement
/// projection data).
#[tokio::test]
async fn peel_and_symrefs_attributes_appear_only_when_requested() {
    let refs = MemRefs::default();
    let doubled = repo("owner/doubled");
    let seeded = refs
        .apply_updates(
            &doubled,
            vec![
                RefUpdate {
                    name: "refs/heads/main".to_string(),
                    expected: None,
                    new: Some(oid(1)),
                },
                RefUpdate {
                    name: "refs/tags/v1".to_string(),
                    expected: None,
                    new: Some(oid(2)),
                },
            ],
            false,
        )
        .await
        .unwrap();
    assert!(seeded.iter().all(|outcome| outcome.result.is_ok()));

    let listed = refs.list_refs(&doubled).await.unwrap();
    let mut annotated: Vec<_> = listed;
    for line in annotated.iter_mut() {
        if line.name == "refs/heads/main" {
            line.symref_target = Some("refs/heads/display".to_owned());
        }
        if line.name == "refs/tags/v1" {
            line.peeled = Some(oid(1));
        }
    }

    // No attribute args: bare oid+name lines only.
    {
        let (mut peer, mut session, _) = session(Limits::default());
        peer.write_all(&ls_refs_request(&[])).await.unwrap();
        let request = session.read_request().await.unwrap().expect("a request");
        let args = parse_ls_refs_args(&request.args).unwrap();
        let limits = *session.limits();
        serve_ls_refs(&mut session, annotated.clone(), &args, &limits)
            .await
            .unwrap();
        let packets = read_pkts(&mut peer).await;
        let text = |p: &Vec<u8>| String::from_utf8_lossy(p).trim_end_matches('\n').to_owned();
        assert_eq!(text(&packets[0]), format!("{} refs/heads/main", oid(1)));
        assert_eq!(text(&packets[1]), format!("{} refs/tags/v1", oid(2)));
        assert_eq!(packets.len(), 3);
    }

    // Both args: the attributes render, symref-target before peeled.
    {
        let (mut peer, mut session, _) = session(Limits::default());
        peer.write_all(&ls_refs_request(&["peel", "symrefs"]))
            .await
            .unwrap();
        let request = session.read_request().await.unwrap().expect("a request");
        let args = parse_ls_refs_args(&request.args).unwrap();
        let limits = *session.limits();
        serve_ls_refs(&mut session, annotated, &args, &limits)
            .await
            .unwrap();
        let packets = read_pkts(&mut peer).await;
        let text = |p: &Vec<u8>| String::from_utf8_lossy(p).trim_end_matches('\n').to_owned();
        assert_eq!(
            text(&packets[0]),
            format!(
                "{} refs/heads/main symref-target:refs/heads/display",
                oid(1)
            )
        );
        assert_eq!(
            text(&packets[1]),
            format!("{} refs/tags/v1 peeled:{}", oid(2), oid(1))
        );
    }
}

/// The unborn symref row (the N-5 rider's shape) maps to the
/// literal-`unborn` line when the client requested `unborn`, and is
/// omitted without the request — the advertised token's serving shape
/// (validated by the real-client sessions OQ-19 rider (a), wave 5:
/// `git ls-remote` + the unborn-HEAD clone consume the line; the
/// zero-oid shape the row served before the sessions was a real-client
/// error — the client wants the null oid, `ERR object <zero> does not
/// exist`).
///
/// Contract stamp: review 001 N-5 (the unborn-HEAD row shape, `oid:
/// None` + symref target — the wire shape corrected to the upstream
/// `unborn` emission by the rider-1 capture); ADR-003 (`ls-refs=unborn`
/// advertised, the advertisement is honest only if the shape serves);
/// transport.md §ls-refs (the unborn rider's grammar line
/// `unborn HEAD symref-target:…`).
#[tokio::test]
async fn unborn_head_emits_the_unborn_symref_line_only_when_requested() {
    let refs = MemRefs::default();
    let doubled = repo("owner/doubled");
    refs.set_head_symref(&doubled, Some("refs/heads/main".to_string()));

    // Requested: the literal-unborn HEAD line with its symref-target.
    {
        let (mut peer, mut session, _) = session(Limits::default());
        peer.write_all(&ls_refs_request(&["symrefs", "unborn"]))
            .await
            .unwrap();
        let request = session.read_request().await.unwrap().expect("a request");
        let args = parse_ls_refs_args(&request.args).unwrap();
        assert!(args.unborn);
        let limits = *session.limits();
        serve_ls_refs(
            &mut session,
            refs.list_refs(&doubled).await.unwrap(),
            &args,
            &limits,
        )
        .await
        .unwrap();
        let packets = read_pkts(&mut peer).await;
        let text = |p: &Vec<u8>| String::from_utf8_lossy(p).trim_end_matches('\n').to_owned();
        assert_eq!(
            text(&packets[0]),
            "unborn HEAD symref-target:refs/heads/main"
        );
        assert_eq!(packets.len(), 2, "no refs — the repo is unborn-empty");
    }

    // Not requested: the row is omitted (real git's behavior), the
    // response is flush-only.
    {
        let (mut peer, mut session, _) = session(Limits::default());
        peer.write_all(&ls_refs_request(&["symrefs"]))
            .await
            .unwrap();
        let request = session.read_request().await.unwrap().expect("a request");
        let args = parse_ls_refs_args(&request.args).unwrap();
        let limits = *session.limits();
        serve_ls_refs(
            &mut session,
            refs.list_refs(&doubled).await.unwrap(),
            &args,
            &limits,
        )
        .await
        .unwrap();
        let packets = read_pkts(&mut peer).await;
        assert_eq!(packets.len(), 1, "flush-only response");
        assert!(packets[0].is_empty(), "the terminating flush");
    }
}

/// An unknown ls-refs argument is a session error — the machines fail
/// closed, never guess.
///
/// Contract stamp: ADR-009 (fail-closed budgets posture over protocol
/// grammar input); transport.md §Error taxonomy (protocol errors are
/// session errors).
#[tokio::test]
async fn an_unknown_ls_refs_argument_is_a_session_error() {
    let error = parse_ls_refs_args(&["shallow".to_owned()]).unwrap_err();
    assert!(error.is_terminal(), "protocol errors are terminal");
}

/// The advertisement ref cap is fail-closed: a `list_refs` result
/// exceeding `Limits::max_advertisement_refs` errors the session with
/// nothing emitted — no partial ref list can ever stream.
///
/// Contract stamp: ADR-009 (the advertisement ref cap, fail-closed —
/// breach is a session error, never silent truncation); review 001 N-1
/// (a partial ref list is the worst advertisement failure mode —
/// clones appear to succeed).
#[tokio::test]
async fn advertisement_ref_cap_breaches_fail_closed_with_nothing_emitted() {
    let refs = MemRefs::default();
    let doubled = repo("owner/doubled");
    refs.apply_updates(
        &doubled,
        vec![
            RefUpdate {
                name: "refs/heads/a".to_string(),
                expected: None,
                new: Some(oid(1)),
            },
            RefUpdate {
                name: "refs/heads/b".to_string(),
                expected: None,
                new: Some(oid(2)),
            },
        ],
        false,
    )
    .await
    .unwrap();

    let limits = Limits {
        max_advertisement_refs: 1,
        ..Limits::default()
    };
    let (mut peer, mut session, limits) = session(limits);
    assert_eq!(limits.max_advertisement_refs, 1);

    let error = serve_ls_refs(
        &mut session,
        refs.list_refs(&doubled).await.unwrap(),
        &LsRefsArgs::default(),
        &limits,
    )
    .await
    .unwrap_err();
    match error {
        SubstrateError::Protocol(message) => {
            assert!(message.contains("cap"), "{message}");
        }
        other => panic!("a cap breach is a session error: {other:?}"),
    }
    // Nothing was emitted: dropping the session closes the write end,
    // and the peer sees EOF with zero bytes.
    drop(session);
    let mut buf = Vec::new();
    peer.read_to_end(&mut buf).await.unwrap();
    assert_eq!(buf, Vec::<u8>::new(), "no partial list was ever written");
}

/// The machines are substrate-agnostic: the same advertisement and
/// ls-refs shapes run through the stateless response end (the http
/// door's surface) with identical wire text — no duplex- or
/// stateless-specific type in any machine signature.
///
/// Contract stamp: ADR-005 §3 (two session substrates feeding one state
/// machine core); ADR-003 (advertisement-once is each substrate's
/// serving discipline, not the machine's); transport.md §ls-refs
/// (the same capability list and ref-line shapes on the stateless
/// http framing).
#[tokio::test]
async fn the_machines_run_unchanged_through_the_stateless_response_end() {
    use std::sync::{Arc, Mutex as StdMutex};

    use alkgit::StatelessRequest;
    use futures::io::AsyncWrite as FuturesAsyncWrite;

    #[derive(Clone, Default)]
    struct VecSink(Arc<StdMutex<Vec<u8>>>);

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

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

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

    let sink = VecSink::default();

    // The advertisement rides its own flush-delimited response (the
    // info/refs-equivalent dump — the substrate's one-shot flush rule
    // structures http serving per-request-set), and the ls-refs POST's
    // response is a second `StatelessResponse`: two per-request-set
    // responses, one machine core.
    {
        let mut post = StatelessRequest::new(
            futures::io::empty(),
            sink.clone(),
            Limits::default(),
            Service::UploadPack,
        );
        advertise_v2(post.response()).await.unwrap();
        assert!(post.response().is_flushed(), "the dump ends at flush");
    }

    let refs = MemRefs::default();
    let doubled = repo("owner/doubled");
    refs.apply_updates(
        &doubled,
        vec![RefUpdate {
            name: "refs/heads/main".to_string(),
            expected: None,
            new: Some(oid(1)),
        }],
        false,
    )
    .await
    .unwrap();
    {
        let mut post = StatelessRequest::new(
            futures::io::empty(),
            sink.clone(),
            Limits::default(),
            Service::UploadPack,
        );
        serve_ls_refs(
            post.response(),
            refs.list_refs(&doubled).await.unwrap(),
            &LsRefsArgs::default(),
            &Limits::default(),
        )
        .await
        .unwrap();
        assert!(post.response().is_flushed(), "the response ends at flush");
    }

    // The framed body: the advertisement section, the ls-refs section —
    // identical wire text to the duplex serving, ending at the flush.
    let body = sink.0.lock().expect("sink is not poisoned").clone();
    let pkt_lines: Vec<Vec<u8>> = {
        let mut lines = Vec::new();
        let mut cursor = body.as_slice();
        loop {
            if cursor.is_empty() {
                break;
            }
            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
    };
    let text = |p: &Vec<u8>| String::from_utf8_lossy(p).trim_end_matches('\n').to_owned();
    assert_eq!(text(&pkt_lines[0]), "version 2");
    assert_eq!(text(&pkt_lines[1]), "ls-refs=unborn");
    assert_eq!(text(&pkt_lines[2]), "fetch=wait-for-done");
    assert_eq!(text(&pkt_lines[3]), "object-format=sha1");
    // The dump's flush, then the ls-refs POST's response section.
    #[cfg(feature = "sha256")]
    assert_eq!(text(&pkt_lines[4]), "object-format=sha256");
    #[cfg(not(feature = "sha256"))]
    assert_eq!(text(&pkt_lines[4]), "", "the dump ends at its flush");
    #[cfg(not(feature = "sha256"))]
    assert_eq!(text(&pkt_lines[5]), format!("{} refs/heads/main", oid(1)));
    #[cfg(feature = "sha256")]
    assert_eq!(text(&pkt_lines[5]), "", "the dump ends at its flush");
    #[cfg(feature = "sha256")]
    assert_eq!(text(&pkt_lines[6]), format!("{} refs/heads/main", oid(1)));
    #[cfg(feature = "sha256")]
    assert_eq!(pkt_lines.len(), 8, "flush-terminated response");
    #[cfg(not(feature = "sha256"))]
    assert_eq!(pkt_lines.len(), 7, "flush-terminated response");
    assert!(pkt_lines.last().expect("the terminating flush").is_empty());
}