alkgit 0.1.0

//! The consumer half's fetch workstream (ADR-017 §1): the V2
//! `ls-refs` listing command and the full client-side ack loop
//! ([`GitSession::fetch`]) — the client side of the loop we serve
//! ([`serve_fetch_round`](crate::serve_fetch_round)) — riding
//! gix-protocol's client machinery below the `Negotiate`/RefMap layer
//! (ADR-023 §6's recorded scope). Rounds are **full stateless
//! commands** — wants + haves re-sent every round (ADR-014 §7), the
//! `done` round terminates with its sideband pack streamed to the
//! caller's sink through the bounded band-drain pump (POC-X findings
//! 2/4/5, `session.rs`'s shared helper). The argument and response
//! grammars are the same the server's machines speak
//! ([`LsRefsArgs`](crate::LsRefsArgs), [`RefLine`]); the client speaks
//! the grammar the captures record and we serve
//! (negotiation-captures.md, ADR-014's basis): capability lines are
//! the crate agent + the peer's advertised `object-format` echoed
//! (upstream's echo discipline); the argument lines are the pinned set
//! (`thin-pack`/`ofs-delta` initial args, wants, haves, `done`).
//!
//! One workstream one file (the construction+handshake workstream and
//! the shared drill helpers live in the `session` module — the flat
//! sibling shape, convention 14); the session's budget kinds and the
//! terminal error-break discipline are the `session` module's story.

use bstr::{BString, ByteSlice as _};
use gix_hash::ObjectId;
use gix_protocol::command::Feature;
use gix_protocol::fetch::response::Acknowledgement;
use gix_protocol::fetch::{Arguments, Response};
use gix_transport::client::Capabilities;
use gix_transport::Protocol;

use crate::seam::RefLine;
use crate::session::{
    drain_banded_payload, drain_bounded_lines, map_client_error, AdvertisementBounds, GitSession,
    HandshakeResponse, SessionError, CLIENT_AGENT,
};
use crate::substrate::Service;
use crate::v2::LsRefsArgs;

impl GitSession {
    /// The V2 `ls-refs` command (ADR-017 §1's listing primitive): the
    /// [`LsRefsArgs`] argument grammar the server's
    /// [`serve_ls_refs`](crate::serve_ls_refs) machine parses drives
    /// the request (`peel`/`symrefs`/`unborn` tokens, `ref-prefix`
    /// prefixes — empty prefixes are a local refusal, fail before the
    /// wire), and the response's ref lines parse into the wire's
    /// [`RefLine`] rows. The unborn shapes all parse as `oid: None`
    /// (the literal `unborn` token our server emits, the zero-oid
    /// unborn line, and a peer's attribute-only unborn line — the
    /// capture-pinned shapes); a response line the grammar cannot
    /// explain is a session error, fail-closed.
    ///
    /// The response drain is bounded by
    /// [`max_advertisement_refs`](crate::Limits::max_advertisement_refs)
    /// and
    /// [`max_request_body`](crate::Limits::max_request_body) — the
    /// handshake advertisement's own client-direction bounds; the whole
    /// command is bounded by the session's remaining wall clock. The
    /// session must be the upload-pack service; anything else is a
    /// local refusal (`ls-refs` never rides a push stream). The
    /// handshake runs implicitly first when absent (one advertisement
    /// per session — the cached handshake never re-enters the wire).
    ///
    /// Contract stamp: ADR-017 §1 (ls_refs), ADR-016 §2 (the command's
    /// request grammar, via the transport's `invoke`), ADR-009 (the
    /// bounds), ADR-023 §2 (the stateless-command round); POC-X
    /// findings 2/5; core/w4-session-fetch.
    pub async fn ls_refs(&mut self, args: &LsRefsArgs) -> Result<Vec<RefLine>, SessionError> {
        if let Some(cached) = &self.terminal {
            return Err(cached.clone());
        }
        if self.service != Service::UploadPack {
            let error = SessionError::InvalidInput(format!(
                "ls-refs rides the upload-pack service; this session is {:?}",
                self.service
            ));
            self.terminal = Some(error.clone());
            return Err(error);
        }
        let handshake = self.handshake().await?;
        let bounds = AdvertisementBounds {
            max_refs: self.limits.max_advertisement_refs,
            max_bytes: self.limits.max_request_body,
        };
        let prepared = prepare_ls_refs_args(args)?;
        let remaining = self.remaining_wall_clock();
        match tokio::time::timeout(remaining, self.ls_refs_once(&handshake, prepared, bounds)).await
        {
            Ok(inner) => {
                if let Err(error) = &inner {
                    self.terminal = Some(error.clone());
                }
                inner
            }
            Err(_elapsed) => Err(self.typed_wall_clock_breach()),
        }
    }

    async fn ls_refs_once(
        &mut self,
        handshake: &HandshakeResponse,
        prepared: Vec<BString>,
        bounds: AdvertisementBounds,
    ) -> Result<Vec<RefLine>, SessionError> {
        use gix_transport::client::async_io::TransportV2Ext as _;

        let empty = prepared.is_empty();
        let transport = self.transport_mut();
        let mut reader = transport
            .invoke(
                "ls-refs",
                command_features(&handshake.capabilities).into_iter(),
                if empty {
                    None
                } else {
                    Some(prepared.into_iter())
                },
                false,
            )
            .await
            .map_err(map_client_error)?;
        drain_bounded_lines(&mut reader, bounds, "the ls-refs response", |line| {
            parse_ref_line(line.trim_end())
        })
        .await
    }

    /// The V2 `fetch` command — the full client-side ack loop (ADR-017
    /// §1, ADR-014's grammar, the client side of the
    /// [`serve_fetch_round`](crate::serve_fetch_round) loop): the
    /// `rounds` plan is the caller's negotiation (rounds are full
    /// stateless commands — wants + haves re-sent every round, ADR-014
    /// §7; ADR-023 §6's recorded below-`Negotiate` scope). A no-`done`
    /// round carries at least one have (the grammar's shape) and its
    /// report carries the parsed acknowledgements (`ACK` commons —
    /// empty for a NAK-only round; a peer's `ready` is refused
    /// fail-closed, ADR-014 §4: never `ready`). The done round (with
    /// wants) terminates the fetch: its sideband-unwrapped pack bytes
    /// stream to the caller's `pack` sink
    /// ([`tokio::io::AsyncWrite`]) through the POC-X pump, bounded by
    /// [`max_pack_size`](crate::Limits::max_pack_size) — the sink never
    /// receives a byte past the budget, the breach is typed and the
    /// session ends. The done round's response **must** carry the
    /// packfile section (a done round answered otherwise violates the
    /// wait-for-done grammar — ADR-014 §2/§4 — and fails closed); a
    /// packfile section on a no-`done` round is the same refusal.
    ///
    /// Budgets (ADR-017 §6, ADR-009): the plan's round count against
    /// [`max_negotiation_rounds`](crate::Limits::max_negotiation_rounds)
    /// (checked before anything touches the wire), the whole
    /// operation under the session's remaining wall clock, the
    /// response pack under
    /// [`max_pack_size`](crate::Limits::max_pack_size). An empty plan,
    /// a no-`done` round without haves, a round after the done round,
    /// an empty `ref-prefix`, or a fetch on a push session is a local
    /// [`SessionError::InvalidInput`] — nothing on the wire. Any
    /// session error is terminal (error-break): later calls re-report
    /// without re-entry. The handshake runs implicitly first when
    /// absent.
    ///
    /// Returns one report per supplied round, in order.
    ///
    /// Contract stamp: ADR-017 §1/§4/§6, ADR-014 §1/§2/§4/§7, ADR-023
    /// §2/§6, ADR-009; POC-X findings 2/4/5 (the validated round
    /// structure, the pump, the persistence); core/w4-session-fetch.
    pub async fn fetch<S>(
        &mut self,
        rounds: Vec<FetchRound>,
        pack: &mut S,
    ) -> Result<Vec<FetchRoundReport>, SessionError>
    where
        S: tokio::io::AsyncWrite + Unpin,
    {
        if let Some(cached) = &self.terminal {
            return Err(cached.clone());
        }
        if self.service != Service::UploadPack {
            let error = SessionError::InvalidInput(format!(
                "fetch rides the upload-pack service; this session is {:?}",
                self.service
            ));
            self.terminal = Some(error.clone());
            return Err(error);
        }
        if rounds.is_empty() {
            let error = SessionError::InvalidInput(
                "fetch: the round plan is empty — one round is the minimum grammar shape"
                    .to_owned(),
            );
            self.terminal = Some(error.clone());
            return Err(error);
        }
        let cap = self.limits.max_negotiation_rounds;
        if rounds.len() > cap {
            let error = SessionError::BudgetExceeded {
                what: "the fetch negotiation".to_owned(),
                budget_name: "max_negotiation_rounds",
                limit: cap as u64,
            };
            self.terminal = Some(error.clone());
            return Err(error);
        }
        for (index, round) in rounds.iter().enumerate() {
            if !round.done && round.haves.is_empty() {
                let error = SessionError::InvalidInput(
                    "fetch: a no-done round carries at least one have (the ack-loop grammar's shape)"
                        .to_owned(),
                );
                self.terminal = Some(error.clone());
                return Err(error);
            }
            if round.done && index + 1 < rounds.len() {
                let error = SessionError::InvalidInput(
                    "fetch: the done round terminates the negotiation — no rounds may follow it"
                        .to_owned(),
                );
                self.terminal = Some(error.clone());
                return Err(error);
            }
        }
        let handshake = self.handshake().await?;
        let features = command_features(&handshake.capabilities);
        let max_pack_size = self.limits.max_pack_size;
        let remaining = self.remaining_wall_clock();
        match tokio::time::timeout(
            remaining,
            self.fetch_once(rounds, pack, features, max_pack_size),
        )
        .await
        {
            Ok(inner) => {
                if let Err(error) = &inner {
                    self.terminal = Some(error.clone());
                }
                inner
            }
            Err(_elapsed) => Err(self.typed_wall_clock_breach()),
        }
    }

    async fn fetch_once<S>(
        &mut self,
        rounds: Vec<FetchRound>,
        pack: &mut S,
        features: Vec<Feature>,
        max_pack_size: u64,
    ) -> Result<Vec<FetchRoundReport>, SessionError>
    where
        S: tokio::io::AsyncWrite + Unpin,
    {
        use tokio_util::compat::TokioAsyncWriteCompatExt as _;

        let mut sink = pack.compat_write();
        let mut reports = Vec::with_capacity(rounds.len());
        for round in rounds {
            let mut arguments = Arguments::new(Protocol::V2, features.clone(), false);
            for want in &round.wants {
                arguments.want(*want);
            }
            for have in &round.haves {
                arguments.have(*have);
            }
            let mut reader = arguments
                .send(self.transport_mut(), round.done)
                .await
                .map_err(map_client_error)?;
            let response =
                Response::from_line_reader(Protocol::V2, &mut reader, round.done, !round.done)
                    .await
                    .map_err(|error| {
                        SessionError::Protocol(format!("malformed fetch response: {error}"))
                    })?;
            let mut acks = Vec::new();
            for ack in response.acknowledgements() {
                match ack {
                    Acknowledgement::Common(oid) => acks.push(*oid),
                    Acknowledgement::Nak => {}
                    Acknowledgement::Ready => {
                        return Err(SessionError::Protocol(
                            "the peer sent `ready` — never in the negotiation grammar, the \
                             wait-for-done posture has no ready (ADR-014 §4)"
                                .to_owned(),
                        ));
                    }
                }
            }
            if response.has_pack() {
                if !round.done {
                    return Err(SessionError::Protocol(
                        "the peer sent a packfile section on a no-done round — the \
                         wait-for-done grammar's pack is the done round's alone (ADR-014 §2/§4)"
                            .to_owned(),
                    ));
                }
                let pack_bytes = pump_pack(&mut reader, &mut sink, max_pack_size).await?;
                reports.push(FetchRoundReport {
                    acks,
                    pack_bytes: Some(pack_bytes),
                });
            } else {
                if round.done && !round.wants.is_empty() {
                    return Err(SessionError::Protocol(
                        "the done round's response carried no packfile section — the peer \
                         violated the wait-for-done grammar (ADR-014 §2/§4)"
                            .to_owned(),
                    ));
                }
                reports.push(FetchRoundReport {
                    acks,
                    pack_bytes: None,
                });
            }
        }
        Ok(reports)
    }
}

/// One round of the client-side fetch negotiation ([`GitSession::fetch`]):
/// wants + haves + the `done` flag — a full stateless command (the
/// caller re-sends its accumulated commons every round; ADR-014 §7,
/// ADR-023 §6's recorded round structure).
#[derive(Clone, Debug)]
pub struct FetchRound {
    /// The `want` lines — the objects this side still needs.
    pub wants: Vec<ObjectId>,
    /// The `have` lines — the complete common + negotiator view for
    /// this round (stateless: re-sent in full every round); a no-`done`
    /// round carries at least one have.
    pub haves: Vec<ObjectId>,
    /// `done` — terminates negotiation; the response is the packfile
    /// section (the sideband pack streams to [`fetch`]'s sink).
    ///
    /// [`fetch`]: GitSession::fetch
    pub done: bool,
}

/// One round's parsed outcome ([`GitSession::fetch`]).
#[derive(Clone, Debug, Default, PartialEq, Eq)]
pub struct FetchRoundReport {
    /// The positive acknowledgements — the peer's recognized commons
    /// (`ACK <oid>` lines; ADR-014 §1). Empty for a NAK-only round.
    pub acks: Vec<ObjectId>,
    /// The done round's delivered pack size, in sideband-unwrapped
    /// bytes — `None` on a round answered with an acknowledgements
    /// section (never a pack).
    pub pack_bytes: Option<u64>,
}

/// The V2 commands' capability lines (the upstream client's echo
/// discipline): the crate agent plus the peer's advertised
/// `object-format` value echoed. Bare value-less capabilities are
/// never sent (gix's value-bearing filter — e.g. the peer's advertised
/// `wait-for-done` feature is never echoed as a capability line; the
/// acknowledgment loop is grammar, not capability, ADR-014).
fn command_features(capabilities: &Capabilities) -> Vec<Feature> {
    let mut features = vec![("agent", Some(CLIENT_AGENT.to_owned()))];
    if let Some(object_format) = capabilities
        .capability("object-format")
        .and_then(|capability| capability.value())
        .and_then(|value| value.to_str().ok().map(ToOwned::to_owned))
    {
        features.push(("object-format", Some(object_format)));
    }
    features
}

/// Builds the `ls-refs` command's argument lines from the caller's
/// [`LsRefsArgs`] — the same argument grammar the server's
/// [`parse_ls_refs_args`](crate::parse_ls_refs_args) parses back. An
/// empty ref-prefix is a grammar violation (the server's session error
/// class) refused locally, fail before the wire.
fn prepare_ls_refs_args(args: &LsRefsArgs) -> Result<Vec<BString>, SessionError> {
    let mut arguments: Vec<BString> = Vec::new();
    if args.symrefs {
        arguments.push("symrefs".into());
    }
    if args.peel {
        arguments.push("peel".into());
    }
    if args.unborn {
        arguments.push("unborn".into());
    }
    for prefix in &args.ref_prefixes {
        if prefix.is_empty() {
            return Err(SessionError::InvalidInput(
                "ls-refs: a ref-prefix is a value-bearing argument; the empty prefix is out of grammar"
                    .to_owned(),
            ));
        }
        arguments.push(format!("ref-prefix {prefix}").into());
    }
    Ok(arguments)
}

/// Parses one ls-refs response ref line back into the wire's
/// [`RefLine`] row (the V2 ref grammar: `<oid> <name>[
/// symref-target:<target>][ peeled:<oid>]`). The unborn row's shapes
/// all map to `oid: None` — the upstream grammar's literal `unborn`
/// token in the oid position (gix's own fixture grammar, also our
/// server's emission since the rider-1 capture), a peer's
/// attribute-only line, and the zero-oid form the emission served
/// before the capture. A `symref-target:(null)` (the out-of-namespace
/// edge upstream serves) parses as no target; a peeled oid must
/// decode; a line the grammar cannot explain is a session error,
/// fail-closed.
fn parse_ref_line(line: &str) -> Result<RefLine, SessionError> {
    let invalid = |line: &str| {
        SessionError::Protocol(format!(
            "the ls-refs response line {line:?} is not a V2 ref line \
             (<'oid> <name>[ symref-target:..][ peeled:..]') — fail closed"
        ))
    };
    let mut segments = line.split(' ');
    let (name, oid, attributes): (String, Option<ObjectId>, Vec<&str>) = match segments.next() {
        Some("unborn") => match segments.next() {
            Some(name) if !name.is_empty() => (name.to_owned(), None, segments.collect()),
            _ => return Err(invalid(line)),
        },
        Some(hex)
            if !hex.is_empty()
                && hex.bytes().all(|byte| byte.is_ascii_hexdigit())
                && (hex.len() == 40 || hex.len() == 64) =>
        {
            let oid = ObjectId::from_hex(hex.as_bytes())
                .map_err(|_| invalid(line))
                .map(|oid| (!oid.as_bytes().iter().all(|byte| *byte == 0)).then_some(oid))?;
            match segments.next() {
                Some(name) if !name.is_empty() => (name.to_owned(), oid, segments.collect()),
                _ => return Err(invalid(line)),
            }
        }
        Some(name) => match segments.next() {
            Some(attribute) if !name.is_empty() => (
                name.to_owned(),
                None,
                Some(attribute).into_iter().chain(segments).collect(),
            ),
            _ => return Err(invalid(line)),
        },
        None => return Err(invalid(line)),
    };
    let peeled_oid = |hex: &str| ObjectId::from_hex(hex.as_bytes()).map_err(|_| invalid(line));
    let mut symref_target = None;
    let mut peeled = None;
    for segment in attributes {
        let (token, value) = match segment.split_once(':') {
            Some(pair) => pair,
            None => return Err(invalid(line)),
        };
        if value.is_empty() {
            return Err(invalid(line));
        }
        match token {
            "symref-target" => symref_target = (value != "(null)").then(|| value.to_owned()),
            "peeled" => peeled = Some(peeled_oid(value)?),
            _ => return Err(invalid(line)),
        }
    }
    Ok(RefLine {
        name,
        oid,
        peeled,
        symref_target,
    })
}

/// The done round's packfile pump (POC-X finding 4): the response
/// reader's progress handler unwraps the sideband — band 1's chunks
/// stream to the caller's sink (byte-counted, never past
/// `max_pack_size` — the sink receives nothing beyond the budget), band
/// 2's progress text is skipped, band 3 is the peer's abort (recorded
/// and the pump breaks — the typed protocol error carries the peer's
/// text). The pump ends at the section's flush (the reader's stop).
///
/// The loop is the `session` module's shared band-drain pump
/// (review 005 R-1's fold-in — identical shape as the push client's
/// status-report read).
async fn pump_pack(
    reader: &mut Box<dyn gix_transport::client::async_io::ExtendedBufRead<'_> + Unpin + '_>,
    sink: &mut (impl futures::io::AsyncWrite + Unpin),
    max_pack_size: u64,
) -> Result<u64, SessionError> {
    drain_banded_payload(
        reader,
        sink,
        true,
        max_pack_size,
        "the response pack",
        "the packfile section",
    )
    .await
}

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

    /// The client ref-line parse round-trips the server's emissions
    /// (the machine's own [`serve_ls_refs`](crate::serve_ls_refs)
    /// grammar, POC-1-observed) and every unborn shape parses as
    /// `oid: None`; a `(null)` symref target is the no-target edge.
    ///
    /// Contract stamp: ADR-016 §2's V2 ref grammar (the response line
    /// both directions speak), gix's from_v2_refs fixture grammar
    /// (the literal `unborn` token); core/w4-session-fetch.
    #[test]
    fn ref_lines_parse_back_with_every_unborn_shape() {
        let oid = |byte: u8| ObjectId::from([byte; 20]);
        let line = format!("{} refs/heads/main symref-target:refs/heads/x", oid(0x1a));
        let parsed = parse_ref_line(line.trim_end()).expect("the ordinary line");
        assert_eq!(
            parsed,
            RefLine {
                name: "refs/heads/main".to_owned(),
                oid: Some(oid(0x1a)),
                peeled: None,
                symref_target: Some("refs/heads/x".to_owned()),
            }
        );
        let line = format!("{} refs/tags/v1 peeled:{}", oid(0x2b), oid(0x1a));
        assert_eq!(
            parse_ref_line(line.trim_end())
                .expect("the peeled line")
                .peeled,
            Some(oid(0x1a))
        );
        for shape in [
            "unborn HEAD symref-target:refs/heads/main".to_owned(),
            format!("{} HEAD symref-target:refs/heads/main", "0".repeat(40)),
            "HEAD symref-target:refs/heads/main".to_owned(),
        ] {
            let parsed = parse_ref_line(shape.trim_end()).expect(shape.as_str());
            assert_eq!(
                parsed.oid, None,
                "the unborn shape parses oid-less: {shape}"
            );
            assert_eq!(parsed.name, "HEAD");
            assert_eq!(parsed.symref_target, Some("refs/heads/main".to_owned()));
        }
        let parsed =
            parse_ref_line("0000000000000000000000000000000000000000 HEAD symref-target:(null)")
                .expect("the (null) symref edge");
        assert_eq!(parsed.name, "HEAD");
        assert_eq!(parsed.oid, None);
        assert_eq!(parsed.symref_target, None);
    }

    /// The client ref-line parse fails closed on off-grammar lines.
    ///
    /// Contract stamp: ADR-017 §6 (fail-closed); core/w4-session-fetch.
    #[test]
    fn ref_line_parse_refuses_off_grammar_lines_fail_closed() {
        for line in ["", "HEAD", "zz HEAD", &format!("{} x", "0".repeat(41))] {
            match parse_ref_line(line.trim_end()) {
                Err(SessionError::Protocol(_)) => {}
                other => panic!("expected the protocol error for {line:?}, got {other:?}"),
            }
        }
        let bad = format!("{} refs/heads/x peeled:zz", "1".repeat(40));
        match parse_ref_line(&bad) {
            Err(SessionError::Protocol(_)) => {}
            other => panic!("expected the protocol error for the undecodable peel, got {other:?}"),
        }
    }

    /// The commands' capability lines: the crate agent plus the peer's
    /// advertised `object-format` echoed; no value-less capability
    /// (`wait-for-done` is never echoed — the ack loop is grammar, not
    /// capability, ADR-014).
    ///
    /// Contract stamp: ADR-016 §2 (the command grammar), ADR-014 (the
    /// wait-for-done posture); core/w4-session-fetch.
    #[test]
    fn the_command_capabilities_are_the_agent_and_the_object_format_echo() {
        let fixture = Capabilities::from_lines(
            "version 2\nls-refs=unborn\nfetch=wait-for-done\nobject-format=sha1\n".into(),
        )
        .expect("the fixture capability list");
        assert_eq!(
            command_features(&fixture),
            vec![
                ("agent", Some(CLIENT_AGENT.to_owned())),
                ("object-format", Some("sha1".to_owned())),
            ],
            "the crate agent and the peer's advertised object-format echo"
        );
        let no_format = Capabilities::from_lines("version 2\nls-refs=unborn\n".into())
            .expect("the fixture capability list");
        assert_eq!(
            command_features(&no_format),
            vec![("agent", Some(CLIENT_AGENT.to_owned()))],
            "no object-format echo when the peer advertised none"
        );
        assert_eq!(
            prepare_ls_refs_args(&LsRefsArgs {
                symrefs: true,
                peel: true,
                ref_prefixes: vec!["refs/heads/".to_owned()],
                unborn: false,
            })
            .expect("the args")
            .iter()
            .map(|arg| String::from_utf8_lossy(arg).into_owned())
            .collect::<Vec<_>>(),
            vec!["symrefs", "peel", "ref-prefix refs/heads/"]
        );
        match prepare_ls_refs_args(&LsRefsArgs {
            ref_prefixes: vec![String::new()],
            ..LsRefsArgs::default()
        }) {
            Err(SessionError::InvalidInput(message)) => {
                assert!(message.contains("ref-prefix"), "{message}");
            }
            other => panic!("the empty prefix refuses locally, got {other:?}"),
        }
    }
}