girt 0.10.1

use std::collections::HashSet;
use std::io::{Read, Write};
use std::num::NonZeroU32;
use std::ops::ControlFlow;
use std::sync::atomic::AtomicBool;

use super::known::KnownObjects as _;
use super::v2::Capabilities;
use super::{
    FetchError as Error, FetchLimits, FetchOptions, KnownHistory, ReceivedFetch, check_cancelled,
};
use crate::packet::{Frame, Wire, packet, put};
use crate::refs::RefName;
use crate::{IdSet, ObjectFormat, ObjectId};

/// Whether an advertisement identifies a stored tip or a peeled tag hint.
#[derive(Clone, Copy, Debug, Eq, PartialEq, Hash)]
#[non_exhaustive]
pub enum AdvertisementKind {
    /// Stored reference tip; eligible for selection as a want.
    Tip,
    /// Peeled tag target hint; never a selectable reference tip.
    PeeledHint,
}

/// One byte-preserving reference advertisement. Peeling hints cannot be selected as wants.
#[derive(Debug, Clone)]
#[non_exhaustive]
pub struct AdvertisedRef {
    /// Full validated reference name (or HEAD); a peeled hint shares its tag's name.
    pub name: RefName,
    /// Advertised SHA-1 or SHA-256 tip or peeled target.
    pub id: ObjectId,
    /// Meaning of the advertised object ID.
    pub kind: AdvertisementKind,
}

impl AdvertisedRef {
    /// Creates an advertised reference with explicit tip or peeled-hint meaning.
    pub fn new(name: RefName, id: ObjectId, kind: AdvertisementKind) -> Self {
        Self { name, id, kind }
    }

    /// Sets [`Self::name`].
    #[must_use]
    pub fn with_name(mut self, name: RefName) -> Self {
        self.name = name;
        self
    }

    /// Sets [`Self::id`].
    #[must_use]
    pub fn with_id(mut self, id: ObjectId) -> Self {
        self.id = id;
        self
    }

    /// Sets [`Self::kind`].
    #[must_use]
    pub fn with_kind(mut self, kind: AdvertisementKind) -> Self {
        self.kind = kind;
        self
    }
}

/// References and capabilities from a v0/v1 advertisement or a v2 `ls-refs` listing, independent of
/// local references and policy. A prefix-selected v2 listing may omit references.
#[derive(Debug, Clone)]
#[non_exhaustive]
pub struct Advertisement {
    /// Entries in server order, including optional HEAD and peeled hints.
    pub refs: Vec<AdvertisedRef>,
    /// Raw capability tokens. Unknown optional capabilities are retained but never requested.
    pub capabilities: Vec<Vec<u8>>,
}

impl Advertisement {
    /// Creates an advertisement from references and raw capability bytes.
    pub fn new(refs: Vec<AdvertisedRef>, capabilities: Vec<Vec<u8>>) -> Self {
        Self { refs, capabilities }
    }

    /// Sets [`Self::refs`].
    #[must_use]
    pub fn with_refs(mut self, refs: Vec<AdvertisedRef>) -> Self {
        self.refs = refs;
        self
    }

    /// Sets [`Self::capabilities`].
    #[must_use]
    pub fn with_capabilities(mut self, capabilities: Vec<Vec<u8>>) -> Self {
        self.capabilities = capabilities;
        self
    }
}

/// Receives a complete, validated fetch without changing storage.
///
/// Streams must already speak upload-pack v0, v1 or v2 and terminate at EOF after the final flush.
/// This is a single session, not stateless HTTP or a reusable connection. A v2 session lists every
/// reference with one `ls-refs` command, sends one `fetch` command, then ends the session with a
/// flush so the server exits. `select` runs once after a
/// complete advertisement and returns explicit advertised tip IDs. Empty selection writes only a
/// flush and expects EOF. No haves are sent; a nonempty selection requests the complete reachable
/// history. Progress channel bytes are passed to `progress`; returning `Break(())` cancels the
/// transfer.
///
/// Cancellation is checked between I/O calls, packets, objects, and graph steps. It cannot
/// interrupt a blocked caller-owned stream or a single object hash/parse. Pack inflation checks
/// cancellation between chunks. Callers needing deadlines
/// must provide timeout/interrupt-capable streams. Interrupted I/O is returned immediately. The
/// caller must close both streams on any failure; they cannot be reused. Progress is advisory,
/// never a success signal.
///
/// # Errors
///
/// Rejects unsupported protocol versions or object formats, missing sideband or v2 `fetch`
/// capability, unexpected v2 response sections, invalid
/// advertisements, unadvertised wants, unexpected ACKs, truncation, trailing bytes, peer errors,
/// corruption, missing delta bases or reachable objects, and exceeded limits. No partial fetch
/// is returned. All received object identities are computed; selected tip connectivity and edge
/// kinds are checked within the pack. Gitlinks are external submodule roots and are not followed.
/// Payload checks use girt's supported parsers and tree validation, not a claim of complete `git
/// fsck` equivalence.
pub fn receive(
    reader: &mut impl Read,
    writer: &mut impl Write,
    select: impl FnOnce(&Advertisement) -> Vec<ObjectId>,
    limits: FetchLimits,
    cancel: &AtomicBool,
    progress: impl FnMut(&[u8]) -> ControlFlow<()>,
) -> Result<ReceivedFetch, Error> {
    receive_with_known(
        reader,
        writer,
        select,
        &KnownHistory::default(),
        limits,
        cancel,
        progress,
    )
}

/// Receives a pack negotiated against explicitly verified local history.
///
/// Uses the same stream, cancellation and capability contract as [`receive`]. Sends at most
/// `min(max_haves, 32)` verified commits in one batch followed immediately by `done`, requesting
/// `multi_ack` when advertised. Without it, sends at most one have; v2 always sends the full batch
/// and expects no acknowledgments after `done`. ACKs must name offered IDs;
/// duplicate continuation ACKs, unsupported states and excess replies fail. No extra rounds or
/// unbounded ancestry walks occur. Selected IDs already verified locally are omitted from wire
/// wants; an entirely known selection sends only a flush. Tags and blobs can therefore be no-ops
/// too. When advertised, `thin-pack` permits a REF_DELTA base in verified local history; such a
/// pack is rewritten as a self-contained pack before installation. Missing or corrupt external
/// bases fail validation. Connectivity is checked across received and known objects.
/// [`ReceivedFetch::install`] rechecks local dependencies in the destination.
///
/// # Errors
///
/// Returns [`receive`]'s errors, including invalid ACKs and combined-graph connectivity failures.
/// Knowledge exceeding this call's local count or byte limits is rejected before transmission.
pub fn receive_with_known(
    reader: &mut impl Read,
    writer: &mut impl Write,
    select: impl FnOnce(&Advertisement) -> Vec<ObjectId>,
    known: &KnownHistory,
    limits: FetchLimits,
    cancel: &AtomicBool,
    progress: impl FnMut(&[u8]) -> ControlFlow<()>,
) -> Result<ReceivedFetch, Error> {
    receive_with_known_depth(
        reader,
        writer,
        select,
        known,
        FetchOptions {
            limits,
            depth: None,
        },
        cancel,
        progress,
    )
}

/// Receives a depth-limited history or deepens verified shallow history.
///
/// `depth` counts commits from selected tips. The result reports the server's resulting shallow
/// boundaries; it does not install them. Direct [`ReceivedFetch::install`] refuses shallow
/// results; [`super::FetchRequest::with_depth`] coordinates boundary publication.
///
/// # Errors
///
/// Returns the ordinary receive errors and rejects peers without `shallow` support.
pub fn receive_with_known_depth(
    reader: &mut impl Read,
    writer: &mut impl Write,
    select: impl FnOnce(&Advertisement) -> Vec<ObjectId>,
    known: &KnownHistory,
    options: FetchOptions,
    cancel: &AtomicBool,
    progress: impl FnMut(&[u8]) -> ControlFlow<()>,
) -> Result<ReceivedFetch, Error> {
    let FetchOptions { limits, depth } = options;
    #[cfg(feature = "tracing")]
    let span = tracing::debug_span!(
        target: "girt",
        "fetch.receive",
        outcome = "incomplete",
        failure_class = tracing::field::Empty,
        effects = tracing::field::Empty,
    );

    let operation = || {
        validate_known(known, limits, cancel)?;
        let mut wire = Wire {
            reader,
            remaining: limits.max_wire_bytes,
            cancel,
        };
        let (advertisement, dialect) = open(&mut wire, writer, limits)?;
        check_cancelled(cancel)?;
        let wants = select(&advertisement);
        let empty = known.shallow_only();
        let known = if depth.is_some() || known.applies_to(&wants) {
            known
        } else {
            &empty
        };
        let negotiation = request(
            writer,
            &advertisement,
            &dialect,
            wants,
            known,
            depth,
            limits,
            cancel,
        )?;
        if negotiation.v2 && negotiation.needs_pack {
            // End the v2 session after the one command; the server exits after its response.
            put(writer, b"0000", cancel)?;
            writer.flush()?;
        }
        if !negotiation.needs_pack {
            wire.end()?;
            return ReceivedFetch::without_pack(
                advertisement,
                negotiation.wants,
                known,
                limits,
                cancel,
            );
        }
        response(
            &mut wire,
            advertisement,
            negotiation,
            known,
            limits,
            cancel,
            progress,
        )
    };
    #[cfg(feature = "tracing")]
    let result = span.in_scope(operation);
    #[cfg(not(feature = "tracing"))]
    let result = { operation }();
    #[cfg(feature = "tracing")]
    crate::trace::finish(&span, &result, crate::trace::fetch);

    result
}

pub(super) struct Negotiation {
    pub(super) wants: Vec<ObjectId>,
    haves: IdSet,
    multi_ack: bool,
    pub(super) needs_pack: bool,
    shallow: Vec<ObjectId>,
    depth: Option<NonZeroU32>,
    v2: bool,
}

/// The upload-pack protocol a session speaks after its first response line.
pub(super) enum Dialect {
    /// V0, or v1's version marker followed by a v0 advertisement.
    V0,
    /// V2 capabilities; references came from `ls-refs`.
    V2(Capabilities),
}

/// Reads the server's opening and, for v2, lists every reference with `ls-refs`.
///
/// The capability advertisement and reference listing together stay within
/// `max_advertisement_bytes`, as a v0 advertisement does.
pub(super) fn open(
    wire: &mut Wire<'_, impl Read>,
    writer: &mut impl Write,
    limits: FetchLimits,
) -> Result<(Advertisement, Dialect), Error> {
    let saved = wire
        .remaining
        .saturating_sub(limits.max_advertisement_bytes);
    wire.remaining -= saved;
    let first = wire.packet()?;
    let result = if first.as_deref() == Some(b"version 2\n") {
        let capabilities = Capabilities::read(wire, limits)?;
        let advertisement = capabilities.ls_refs(wire, writer, &[], limits)?;
        advertisement.object_format()?;
        (advertisement, Dialect::V2(capabilities))
    } else {
        let (advertisement, _) = advertise_after(wire, Some(first), limits)?;
        (advertisement, Dialect::V0)
    };
    wire.remaining += saved;
    Ok(result)
}

#[cfg(any(
    feature = "http",
    all(feature = "ssh", any(target_os = "macos", target_os = "linux"))
))]
impl Negotiation {
    /// Reports whether the response uses v2 sections rather than v0 acknowledgements.
    pub(super) fn is_v2(&self) -> bool {
        self.v2
    }
}

pub(super) fn validate_known(
    known: &KnownHistory,
    limits: FetchLimits,
    cancel: &AtomicBool,
) -> Result<(), Error> {
    check_cancelled(cancel)?;
    if known.objects.len() > limits.max_known_objects {
        return Err(Error::Limit("known objects"));
    }
    if known.shallow.len() > limits.max_shallow_roots {
        return Err(Error::Limit("shallow roots"));
    }
    let mut remaining = limits.max_known_bytes;
    for object in known.objects.values() {
        check_cancelled(cancel)?;
        remaining = remaining
            .checked_sub(object.data().len())
            .ok_or(Error::Limit("known bytes"))?;
    }
    Ok(())
}

#[allow(clippy::too_many_arguments)]
pub(super) fn request(
    writer: &mut impl Write,
    advertisement: &Advertisement,
    dialect: &Dialect,
    selected: Vec<ObjectId>,
    known: &KnownHistory,
    depth: Option<NonZeroU32>,
    limits: FetchLimits,
    cancel: &AtomicBool,
) -> Result<Negotiation, Error> {
    check_cancelled(cancel)?;
    let format = advertisement.object_format()?;
    if known.objects.keys().any(|id| id.format() != format)
        || known.haves.iter().any(|id| id.format() != format)
        || known.shallow.iter().any(|id| id.format() != format)
    {
        return Err(Error::Unsupported("known object format differs"));
    }
    let shallow_supported = match dialect {
        Dialect::V0 => advertisement.has(b"shallow"),
        Dialect::V2(capabilities) => capabilities.has_feature(b"fetch", b"shallow"),
    };
    if (!known.shallow.is_empty() || depth.is_some()) && !shallow_supported {
        return Err(Error::Unsupported("shallow negotiation"));
    }
    if selected.len() > limits.max_wants {
        return Err(Error::Limit("wants"));
    }
    let advertised: HashSet<_> = advertisement
        .refs
        .iter()
        .filter(|reference| reference.kind == crate::fetch::AdvertisementKind::Tip)
        .map(|reference| reference.id)
        .collect();
    let mut wants = Vec::new();
    let mut seen = HashSet::new();
    for id in selected {
        id.require_format(format)?;
        if !advertised.contains(&id) {
            return Err(Error::Unadvertised(id));
        }
        if seen.insert(id) {
            wants.push(id);
        }
    }
    let wire_wants: Vec<_> = wants
        .iter()
        .filter(|id| depth.is_some() || known.get(**id).is_none())
        .copied()
        .collect();
    if wire_wants.is_empty() {
        put(writer, b"0000", cancel)?;
        writer.flush()?;
        return Ok(Negotiation {
            wants,
            haves: IdSet::default(),
            multi_ack: false,
            needs_pack: false,
            shallow: known.shallow.clone(),
            depth,
            v2: matches!(dialect, Dialect::V2(_)),
        });
    }
    if let Dialect::V2(capabilities) = dialect {
        return request_v2(
            writer,
            capabilities,
            wants,
            &wire_wants,
            known,
            depth,
            limits,
            cancel,
        );
    }
    if !advertisement.has(b"side-band-64k") {
        return Err(Error::Unsupported("side-band-64k is required"));
    }
    let multi_ack =
        !known.haves.is_empty() && limits.max_haves != 0 && advertisement.has(b"multi_ack");
    let have_count = limits.max_haves.min(if multi_ack { 32 } else { 1 });
    for (i, id) in wire_wants.iter().enumerate() {
        let mut caps = String::new();
        if i == 0 {
            caps.push_str(" side-band-64k");
            if advertisement.has(b"ofs-delta") {
                caps.push_str(" ofs-delta");
            }
            if !known.haves.is_empty() && have_count > 0 && advertisement.has(b"thin-pack") {
                caps.push_str(" thin-pack");
            }
            if format == ObjectFormat::Sha256 {
                caps.push_str(" object-format=sha256");
            }
        }
        let ack_cap = if i == 0 && multi_ack {
            " multi_ack"
        } else {
            ""
        };
        packet(
            writer,
            format!("want {id}{caps}{ack_cap}\n").as_bytes(),
            cancel,
        )?;
    }
    for id in &known.shallow {
        packet(writer, format!("shallow {id}\n").as_bytes(), cancel)?;
    }
    if let Some(depth) = depth {
        packet(writer, format!("deepen {depth}\n").as_bytes(), cancel)?;
    }
    put(writer, b"0000", cancel)?;
    let haves: IdSet = known.haves.iter().take(have_count).copied().collect();
    for id in known.haves.iter().take(have_count) {
        packet(writer, format!("have {id}\n").as_bytes(), cancel)?;
    }
    packet(writer, b"done\n", cancel)?;
    writer.flush()?;
    Ok(Negotiation {
        wants,
        haves,
        multi_ack,
        needs_pack: true,
        shallow: known.shallow.clone(),
        depth,
        v2: false,
    })
}

// V2 has no multi_ack: with `done` in the only request, the server sends no acknowledgments and
// the response starts at an optional `shallow-info` section followed by `packfile`.
#[allow(clippy::too_many_arguments)]
fn request_v2(
    writer: &mut impl Write,
    capabilities: &Capabilities,
    wants: Vec<ObjectId>,
    wire_wants: &[ObjectId],
    known: &KnownHistory,
    depth: Option<NonZeroU32>,
    limits: FetchLimits,
    cancel: &AtomicBool,
) -> Result<Negotiation, Error> {
    if !capabilities.has(b"fetch") {
        return Err(Error::Unsupported("v2 fetch"));
    }
    let have_count = limits.max_haves.min(32);
    capabilities.write_command(writer, b"fetch", cancel)?;
    if !known.haves.is_empty() && have_count > 0 {
        packet(writer, b"thin-pack\n", cancel)?;
    }
    packet(writer, b"ofs-delta\n", cancel)?;
    for id in wire_wants {
        packet(writer, format!("want {id}\n").as_bytes(), cancel)?;
    }
    for id in &known.shallow {
        packet(writer, format!("shallow {id}\n").as_bytes(), cancel)?;
    }
    if let Some(depth) = depth {
        packet(writer, format!("deepen {depth}\n").as_bytes(), cancel)?;
    }
    let haves: IdSet = known.haves.iter().take(have_count).copied().collect();
    for id in known.haves.iter().take(have_count) {
        packet(writer, format!("have {id}\n").as_bytes(), cancel)?;
    }
    packet(writer, b"done\n", cancel)?;
    put(writer, b"0000", cancel)?;
    writer.flush()?;
    Ok(Negotiation {
        wants,
        haves,
        multi_ack: false,
        needs_pack: true,
        shallow: known.shallow.clone(),
        depth,
        v2: true,
    })
}

pub(super) fn response(
    wire: &mut Wire<'_, impl Read>,
    advertisement: Advertisement,
    negotiation: Negotiation,
    known: &KnownHistory,
    limits: FetchLimits,
    cancel: &AtomicBool,
    progress: impl FnMut(&[u8]) -> ControlFlow<()>,
) -> Result<ReceivedFetch, Error> {
    response_observed(
        wire,
        advertisement,
        negotiation,
        known,
        limits,
        progress,
        &mut super::progress::ValidationObserver::new(cancel, &mut |_| {}),
    )
}

pub(super) fn response_observed(
    wire: &mut Wire<'_, impl Read>,
    advertisement: Advertisement,
    negotiation: Negotiation,
    known: &KnownHistory,
    limits: FetchLimits,
    mut progress: impl FnMut(&[u8]) -> ControlFlow<()>,
    observer: &mut super::progress::ValidationObserver<'_>,
) -> Result<ReceivedFetch, Error> {
    let shallow = read_response_prefix(wire, &negotiation, limits)?;
    let mut pack = super::bytes::Bytes::temporary()?;
    while let Some(bytes) = wire.packet()? {
        match bytes.split_first() {
            Some((1, data)) => {
                if data.len() > limits.max_pack_bytes.saturating_sub(pack.len()) {
                    return Err(Error::Limit("pack bytes"));
                }
                pack.append(data)?;
            }
            Some((2, data)) => {
                if progress(data).is_break() {
                    return Err(Error::Cancelled);
                }
            }
            Some((3, data)) => return Err(Error::Remote(data.to_vec())),
            _ => return Err(Error::Protocol("invalid sideband channel")),
        }
    }
    wire.end()?;
    ReceivedFetch::validate_known(
        advertisement,
        negotiation.wants,
        pack,
        known,
        shallow,
        FetchOptions {
            limits,
            depth: negotiation.depth,
        },
        observer,
    )
}

// Live HTTP notification validates this prefix once before exposing sideband messages. Final
// validation repeats the same checks and retains ownership of the resulting shallow boundaries.
pub(super) fn read_response_prefix(
    wire: &mut Wire<'_, impl Read>,
    negotiation: &Negotiation,
    limits: FetchLimits,
) -> Result<Vec<ObjectId>, Error> {
    if negotiation.v2 {
        return read_v2_sections(wire, negotiation, limits);
    }
    let shallow = read_shallow(wire, negotiation, limits)?;
    read_ack(wire, &negotiation.haves, negotiation.multi_ack)?;
    Ok(shallow)
}

// Reads sections up to and including the `packfile` header. `acknowledgments` cannot appear after
// `done`; `wanted-refs` and `packfile-uris` were never requested.
fn read_v2_sections(
    wire: &mut Wire<'_, impl Read>,
    negotiation: &Negotiation,
    limits: FetchLimits,
) -> Result<Vec<ObjectId>, Error> {
    let mut roots: HashSet<_> = negotiation.shallow.iter().copied().collect();
    let mut header = wire.packet()?.ok_or(Error::Protocol("v2 fetch response"))?;
    if header == b"shallow-info\n" {
        if negotiation.depth.is_none() && negotiation.shallow.is_empty() {
            return Err(Error::Unsupported("shallow server"));
        }
        let format = negotiation.wants[0].format();
        let mut count = 0;
        loop {
            match wire.frame()? {
                Frame::Delim => break,
                Frame::Data(bytes) => {
                    count += 1;
                    if count > limits.max_shallow_roots {
                        return Err(Error::Limit("shallow response"));
                    }
                    apply_shallow_line(&mut roots, &negotiation.shallow, line(&bytes), format)?;
                }
                Frame::Flush | Frame::ResponseEnd => {
                    return Err(Error::Protocol("v2 shallow-info section"));
                }
            }
        }
        header = wire.packet()?.ok_or(Error::Protocol("v2 fetch response"))?;
    }
    if header != b"packfile\n" {
        return Err(Error::Protocol("v2 fetch response section"));
    }
    finish_roots(roots, limits)
}

fn read_shallow(
    wire: &mut Wire<'_, impl Read>,
    negotiation: &Negotiation,
    limits: FetchLimits,
) -> Result<Vec<ObjectId>, Error> {
    let mut roots: HashSet<_> = negotiation.shallow.iter().copied().collect();
    if negotiation.depth.is_none() {
        let mut roots: Vec<_> = roots.into_iter().collect();
        roots.sort_unstable();
        return Ok(roots);
    }
    let format = negotiation.wants[0].format();
    let mut count = 0;
    while let Some(bytes) = wire.packet()? {
        count += 1;
        if count > limits.max_shallow_roots {
            return Err(Error::Limit("shallow response"));
        }
        apply_shallow_line(&mut roots, &negotiation.shallow, line(&bytes), format)?;
    }
    finish_roots(roots, limits)
}

// `unshallow` may only name a root the client sent; repeating one removes nothing and fails.
fn apply_shallow_line(
    roots: &mut HashSet<ObjectId>,
    offered: &[ObjectId],
    bytes: &[u8],
    format: ObjectFormat,
) -> Result<(), Error> {
    let (add, raw) = if let Some(raw) = bytes.strip_prefix(b"shallow ") {
        (true, raw)
    } else if let Some(raw) = bytes.strip_prefix(b"unshallow ") {
        (false, raw)
    } else {
        return Err(Error::Protocol("shallow response"));
    };
    let id = std::str::from_utf8(raw)
        .ok()
        .and_then(|raw| ObjectId::from_hex(format, raw).ok())
        .ok_or(Error::Protocol("shallow object ID"))?;
    if add {
        roots.insert(id);
    } else if !offered.contains(&id) || !roots.remove(&id) {
        return Err(Error::Protocol("unknown unshallow root"));
    }
    Ok(())
}

fn finish_roots(roots: HashSet<ObjectId>, limits: FetchLimits) -> Result<Vec<ObjectId>, Error> {
    if roots.len() > limits.max_shallow_roots {
        return Err(Error::Limit("shallow roots"));
    }
    let mut roots: Vec<_> = roots.into_iter().collect();
    roots.sort_unstable();
    Ok(roots)
}

// One batch of at most 32 haves keeps both directions below ordinary pipe capacity. Reading
// the bounded ACK sequence after done avoids assuming a flush/NAK boundary in single-ACK mode.
fn read_ack(wire: &mut Wire<'_, impl Read>, haves: &IdSet, multi_ack: bool) -> Result<(), Error> {
    let mut continued = HashSet::new();
    for _ in 0..=haves.len() {
        let packet = wire
            .packet()?
            .ok_or(Error::Protocol("flush instead of ACK/NAK"))?;
        let bytes = line(&packet);
        if bytes == b"NAK" && continued.is_empty() {
            return Ok(());
        }
        let (raw, more) = if multi_ack && bytes.ends_with(b" continue") {
            (&bytes[..bytes.len() - 9], true)
        } else {
            (bytes, false)
        };
        let id = raw
            .strip_prefix(b"ACK ")
            .and_then(|id| std::str::from_utf8(id).ok())
            .and_then(|id| id.parse::<ObjectId>().ok())
            .filter(|id| haves.contains(id))
            .ok_or(Error::Protocol("expected ACK of an offered have"))?;
        if !more {
            return Ok(());
        }
        if !continued.insert(id) {
            return Err(Error::Protocol("duplicate ACK continue"));
        }
    }
    Err(Error::Limit("negotiation acknowledgements"))
}

impl Advertisement {
    pub(crate) fn has(&self, capability: &[u8]) -> bool {
        self.capabilities.iter().any(|value| value == capability)
    }

    pub(crate) fn object_format(&self) -> Result<ObjectFormat, Error> {
        let mut format = None;
        for capability in &self.capabilities {
            if let Some(value) = capability.strip_prefix(b"object-format=") {
                let value = match value {
                    b"sha1" => ObjectFormat::Sha1,
                    b"sha256" => ObjectFormat::Sha256,
                    _ => return Err(Error::Unsupported("object format")),
                };
                if format.replace(value).is_some() {
                    return Err(Error::Protocol("duplicate object format"));
                }
            }
        }
        let format = format.unwrap_or(ObjectFormat::Sha1);
        if self
            .refs
            .iter()
            .any(|reference| reference.id.format() != format)
        {
            return Err(Error::Protocol("advertised object format"));
        }
        Ok(format)
    }
}

pub(super) fn advertise_with_version(
    wire: &mut Wire<'_, impl Read>,
    limits: FetchLimits,
) -> Result<(Advertisement, super::ProtocolVersion), Error> {
    advertise_after(wire, None, limits)
}

/// Parses a v0/v1 advertisement whose first packet, if `first` is `Some`, was already read.
fn advertise_after(
    wire: &mut Wire<'_, impl Read>,
    mut first_packet: Option<Option<Vec<u8>>>,
    limits: FetchLimits,
) -> Result<(Advertisement, super::ProtocolVersion), Error> {
    let mut advertisement = Advertisement {
        refs: vec![],
        capabilities: vec![],
    };
    let start = wire.remaining;
    let mut first = true;
    let mut empty_marker = false;
    let mut names = HashSet::new();
    // Narrow the wire budget before reading, so even a huge first advertisement packet is bounded.
    let saved = wire
        .remaining
        .saturating_sub(limits.max_advertisement_bytes);
    wire.remaining -= saved;
    let mut version_checked = false;
    let mut version = super::ProtocolVersion::V0;
    while let Some(bytes) = match first_packet.take() {
        Some(packet) => packet,
        None => wire.packet()?,
    } {
        let bytes = line(&bytes);
        if !version_checked && bytes == b"version 1" {
            version_checked = true;
            version = super::ProtocolVersion::V1;
            continue;
        }
        version_checked = true;
        if bytes.starts_with(b"version ") {
            return Err(Error::Unsupported("protocol version"));
        }
        if bytes.starts_with(b"shallow ") {
            return Err(Error::Unsupported("shallow server"));
        }
        let reference = if first {
            let (reference, caps) =
                split_byte(bytes, 0).ok_or(Error::Protocol("missing capabilities"))?;
            if caps.contains(&0) || caps.iter().any(|b| b.is_ascii_control()) {
                return Err(Error::Protocol("invalid capabilities"));
            }
            advertisement.capabilities = caps
                .split(|b| *b == b' ')
                .filter(|c| !c.is_empty())
                .map(<[u8]>::to_vec)
                .collect();
            reference
        } else {
            bytes
        };
        let (raw_id, name) =
            split_byte(reference, b' ').ok_or(Error::Protocol("reference advertisement"))?;
        let id = std::str::from_utf8(raw_id)
            .ok()
            .and_then(|id| id.parse::<ObjectId>().ok())
            .ok_or(Error::Protocol("advertised object ID"))?;
        if id.is_null() {
            if !first || name != b"capabilities^{}" {
                return Err(Error::Protocol("zero advertised ID"));
            }
            empty_marker = true;
        } else {
            if empty_marker {
                return Err(Error::Protocol("refs after empty advertisement"));
            }
            if advertisement.refs.len() == limits.max_refs {
                return Err(Error::Limit("advertised refs"));
            }
            let peeled = name.ends_with(b"^{}");
            let name = if peeled {
                &name[..name.len() - 3]
            } else {
                name
            };
            let name = RefName::new(name).map_err(|_| Error::Protocol("advertised ref name"))?;
            if peeled
                && (!name.as_bytes().starts_with(b"refs/tags/")
                    || !names.contains(&(name.clone(), false)))
            {
                return Err(Error::Protocol("unmatched peeled hint"));
            }
            if !names.insert((name.clone(), peeled)) {
                return Err(Error::Protocol("duplicate ref"));
            }
            advertisement.refs.push(AdvertisedRef {
                name,
                id,
                kind: if peeled {
                    crate::fetch::AdvertisementKind::PeeledHint
                } else {
                    crate::fetch::AdvertisementKind::Tip
                },
            });
        }
        first = false;
    }
    wire.remaining += saved;
    debug_assert!(wire.remaining <= start);
    advertisement.object_format()?;
    Ok((advertisement, version))
}

// Parse arbitrary reference/capability bytes without UTF-8 conversion.
use crate::bytes::split_once as split_byte;

fn line(bytes: &[u8]) -> &[u8] {
    bytes.strip_suffix(b"\n").unwrap_or(bytes)
}

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

    use super::*;

    fn shallow_response(lines: &[&str], limits: FetchLimits) -> Result<Vec<ObjectId>, Error> {
        let cancel = AtomicBool::new(false);
        let mut bytes = Vec::new();
        for line in lines {
            packet(&mut bytes, line.as_bytes(), &cancel).unwrap();
        }
        bytes.extend_from_slice(b"0000");
        let mut reader = bytes.as_slice();
        let mut wire = Wire {
            reader: &mut reader,
            remaining: 1024,
            cancel: &cancel,
        };
        let id = ObjectId::Sha1([1; 20]);
        read_shallow(
            &mut wire,
            &Negotiation {
                wants: vec![id],
                haves: IdSet::default(),
                multi_ack: false,
                needs_pack: true,
                shallow: vec![],
                depth: NonZeroU32::new(1),
                v2: false,
            },
            limits,
        )
    }

    #[test]
    fn bounds_and_validates_shallow_response() {
        let id = ObjectId::Sha1([1; 20]);
        assert_eq!(
            shallow_response(&[&format!("shallow {id}\n")], FetchLimits::default()).unwrap(),
            vec![id]
        );
        assert!(matches!(
            shallow_response(
                &[&format!("shallow {id}\n")],
                FetchLimits {
                    max_shallow_roots: 0,
                    ..FetchLimits::default()
                }
            ),
            Err(Error::Limit(_))
        ));
        assert!(matches!(
            shallow_response(&["shallow invalid\n"], FetchLimits::default()),
            Err(Error::Protocol("shallow object ID"))
        ));
        assert!(matches!(
            shallow_response(&[&format!("unshallow {id}\n")], FetchLimits::default()),
            Err(Error::Protocol("unknown unshallow root"))
        ));
    }

    fn acknowledgements(lines: &[&str], multi: bool) -> Result<(), Error> {
        let cancel = AtomicBool::new(false);
        let mut bytes = Vec::new();
        for line in lines {
            packet(&mut bytes, line.as_bytes(), &cancel).unwrap();
        }
        let mut reader = bytes.as_slice();
        let mut wire = Wire {
            reader: &mut reader,
            remaining: 1024,
            cancel: &cancel,
        };
        let id = "1111111111111111111111111111111111111111".parse().unwrap();
        read_ack(&mut wire, &IdSet::from_iter([id]), multi)
    }

    #[rstest]
    #[case::nak(&["NAK"])]
    #[case::plain(&["ACK 1111111111111111111111111111111111111111"])]
    #[case::multi(&["ACK 1111111111111111111111111111111111111111 continue", "ACK 1111111111111111111111111111111111111111"])]
    fn accepts_supported_ack_sequences(#[case] lines: &[&str]) {
        assert!(acknowledgements(lines, true).is_ok());
    }

    #[rstest]
    #[case::unknown(&["ACK 2222222222222222222222222222222222222222"])]
    #[case::ready(&["ACK 1111111111111111111111111111111111111111 ready"])]
    #[case::common(&["ACK 1111111111111111111111111111111111111111 common"])]
    #[case::truncated(&["ACK 1111111111111111111111111111111111111111 continue"])]
    #[case::nak_after_continue(&["ACK 1111111111111111111111111111111111111111 continue", "NAK"])]
    #[case::duplicate(&["ACK 1111111111111111111111111111111111111111 continue", "ACK 1111111111111111111111111111111111111111 continue"])]
    #[case::empty(&[])]
    fn rejects_invalid_or_unbounded_ack_sequences(#[case] lines: &[&str]) {
        assert!(acknowledgements(lines, true).is_err());
    }

    #[test]
    fn cancellation_after_advertisement_skips_selection() {
        struct CancelOnRead<'a>(&'a AtomicBool);
        impl Read for CancelOnRead<'_> {
            fn read(&mut self, out: &mut [u8]) -> std::io::Result<usize> {
                out.copy_from_slice(b"0000");
                self.0.store(true, std::sync::atomic::Ordering::Relaxed);
                Ok(4)
            }
        }
        let cancel = AtomicBool::new(false);
        let result = receive(
            &mut CancelOnRead(&cancel),
            &mut Vec::new(),
            |_| panic!("selection after cancellation"),
            FetchLimits::default(),
            &cancel,
            |_| ControlFlow::Continue(()),
        );
        assert!(matches!(result, Err(Error::Cancelled)));
    }

    #[test]
    fn rejects_unrequested_multi_ack() {
        assert!(
            acknowledgements(
                &["ACK 1111111111111111111111111111111111111111 continue"],
                false
            )
            .is_err()
        );
    }

    fn v2_sections(packets: &[&str], depth: Option<u32>) -> Result<Vec<ObjectId>, Error> {
        v2_sections_offering(packets, depth, &[])
    }

    fn v2_sections_offering(
        packets: &[&str],
        depth: Option<u32>,
        offered: &[ObjectId],
    ) -> Result<Vec<ObjectId>, Error> {
        let cancel = AtomicBool::new(false);
        let mut bytes = Vec::new();
        for packet_bytes in packets {
            match *packet_bytes {
                "0001" | "0000" => bytes.extend_from_slice(packet_bytes.as_bytes()),
                other => packet(&mut bytes, other.as_bytes(), &cancel).unwrap(),
            }
        }
        let mut reader = bytes.as_slice();
        let mut wire = Wire {
            reader: &mut reader,
            remaining: 1024,
            cancel: &cancel,
        };
        read_v2_sections(
            &mut wire,
            &Negotiation {
                wants: vec![ObjectId::Sha1([1; 20])],
                haves: IdSet::default(),
                multi_ack: false,
                needs_pack: true,
                shallow: offered.to_vec(),
                depth: depth.and_then(NonZeroU32::new),
                v2: true,
            },
            FetchLimits::default(),
        )
    }

    #[test]
    fn v2_sections_accept_packfile_and_shallow_info() {
        let id = ObjectId::Sha1([1; 20]);
        assert_eq!(v2_sections(&["packfile\n"], None).unwrap(), vec![]);
        let shallow = format!("shallow {id}\n");
        assert_eq!(
            v2_sections(
                &["shallow-info\n", shallow.as_str(), "0001", "packfile\n"],
                Some(1)
            )
            .unwrap(),
            vec![id]
        );
    }

    #[rstest]
    #[case::acknowledgments_after_done(&["acknowledgments\n", "NAK\n", "0001", "packfile\n"], Some(1), "v2 fetch response section")]
    #[case::wanted_refs(&["wanted-refs\n"], None, "v2 fetch response section")]
    #[case::packfile_uris(&["packfile-uris\n"], None, "v2 fetch response section")]
    #[case::flush(&["0000"], None, "v2 fetch response")]
    #[case::shallow_info_without_delim(&["shallow-info\n", "0000"], Some(1), "v2 shallow-info section")]
    #[case::bad_shallow_line(&["shallow-info\n", "deepen 1\n", "0001"], Some(1), "shallow response")]
    fn v2_sections_reject_unexpected_responses(
        #[case] packets: &[&str],
        #[case] depth: Option<u32>,
        #[case] reason: &str,
    ) {
        assert!(matches!(
            v2_sections(packets, depth),
            Err(Error::Protocol(actual)) if actual == reason
        ));
    }

    #[test]
    fn v2_shallow_info_from_shallow_server_is_unsupported() {
        let shallow = format!("shallow {}\n", ObjectId::Sha1([1; 20]));
        assert!(matches!(
            v2_sections(&["shallow-info\n", shallow.as_str(), "0001"], None),
            Err(Error::Unsupported("shallow server"))
        ));
    }

    #[test]
    fn unshallow_requires_a_root_the_client_sent() {
        let id = ObjectId::Sha1([1; 20]);
        let offered = ObjectId::Sha1([2; 20]);
        let shallow = format!("shallow {id}\n");
        let unshallow = format!("unshallow {id}\n");
        assert!(matches!(
            v2_sections(
                &["shallow-info\n", &shallow, &unshallow, "0001", "packfile\n"],
                Some(1)
            ),
            Err(Error::Protocol("unknown unshallow root"))
        ));
        let unshallow_offered = format!("unshallow {offered}\n");
        assert_eq!(
            v2_sections_offering(
                &[
                    "shallow-info\n",
                    &shallow,
                    &unshallow_offered,
                    "0001",
                    "packfile\n"
                ],
                Some(2),
                &[offered]
            )
            .unwrap(),
            vec![id]
        );
        assert!(matches!(
            v2_sections_offering(
                &[
                    "shallow-info\n",
                    &unshallow_offered,
                    &unshallow_offered,
                    "0001",
                    "packfile\n"
                ],
                Some(2),
                &[offered]
            ),
            Err(Error::Protocol("unknown unshallow root"))
        ));
    }
}