grit-lib 0.5.21

//! Serving fetches and clones (`upload-pack`).

use super::shallow_upload::{compute_shallow_response, ShallowRequest, ShallowResponse};
use super::upload_pack_policy::{
    resolve_want_ref, validate_wants, validate_wants_filtered, UploadPackPolicy,
};
use super::{
    advertised_refs, check_object_format, head_info, parse_oid, read_packet, write_flush,
    write_line, AdvertisedRef, DeepenNotError, Packet, Peel, ProtocolVersion, Result, ServeError,
    ServeOptions,
};
use crate::config::ConfigSet;
use crate::objects::{ObjectId, ObjectKind};
use crate::pack_objects::{write_upload_pack_stream, PackProgress};
use crate::pkt_line::SidebandChannel1Writer;
use crate::repo::Repository;
use crate::rev_list::ObjectFilter;
use crate::transfer::PackBuildOptions;
use std::collections::{HashSet, VecDeque};
use std::io::{Read, Write};

/// Capabilities offered in a v0/v1 advertisement, before the per-repository ones.
const V0_CAPABILITIES: &[&str] = &[
    "multi_ack_detailed",
    "side-band-64k",
    "thin-pack",
    "no-progress",
    "include-tag",
    "ofs-delta",
    "shallow",
    "deepen",
    "deepen-since",
    "deepen-not",
];

/// Serve one fetch session: advertise refs, negotiate, and send a pack.
///
/// The version in [`ServeOptions::protocol`] selects the wire format. With
/// [`ServeOptions::advertise_refs`] only the advertisement is written. With
/// [`ServeOptions::stateless_rpc`] the advertisement is skipped and a single
/// request is answered, which is how smart HTTP drives the exchange.
///
/// Shallow fetches, partial-clone filters, and `want-ref` follow repository
/// upload-pack policy and the capabilities advertised for the session.
///
/// # Parameters
///
/// - `repo`: the repository being served.
/// - `input`: requests from the client.
/// - `output`: where responses are written.
/// - `opts`: framing, protocol version, agent string, and hidden refs.
///
/// # Errors
///
/// Returns [`ServeError::NotOurRef`] when the client wants an object outside the
/// allowed set (advertised refs in stateful mode; reachable from ref tips in
/// stateless RPC), [`ServeError::Protocol`] for malformed or unsupported requests,
/// and I/O or repository errors from reading objects and writing the pack.
pub fn upload_pack(
    repo: &Repository,
    input: &mut dyn Read,
    output: &mut dyn Write,
    opts: &ServeOptions,
) -> Result<()> {
    let result = match opts.protocol {
        ProtocolVersion::V2 => serve_v2(repo, input, output, opts),
        ProtocolVersion::V0 | ProtocolVersion::V1 => serve_v0(repo, input, output, opts),
    };
    match result {
        Ok(()) => {
            output.flush()?;
            Ok(())
        }
        Err(err) if opts.stateless_rpc && opts.protocol == ProtocolVersion::V2 => {
            let msg = serve_error_client_message(&err);
            write_line(output, &format!("ERR {msg}"))?;
            write_flush(output)?;
            output.flush()?;
            Ok(())
        }
        Err(err) => {
            output.flush()?;
            Err(err)
        }
    }
}

fn serve_error_client_message(err: &ServeError) -> String {
    match err {
        ServeError::Protocol(msg) => msg.clone(),
        ServeError::UploadFilter(e) => e.to_string(),
        ServeError::NotOurRef(oid) => format!("not our ref {}", oid.to_hex()),
        ServeError::UnknownCommand(cmd) => format!("unknown command {cmd}"),
        ServeError::InvalidDeepenNot { spec, reason } => format!("deepen-not {spec}: {reason}"),
        ServeError::ObjectFormatMismatch { client, repository } => {
            format!("object format mismatch: client wants {client}, repository uses {repository}")
        }
        ServeError::Repository(e) => e.to_string(),
        ServeError::Io(e) => e.to_string(),
    }
}

fn repo_config(repo: &Repository) -> ConfigSet {
    ConfigSet::load(repo.environment(), Some(repo.git_dir()), true).unwrap_or_default()
}

fn upload_policy(repo: &Repository) -> UploadPackPolicy {
    UploadPackPolicy::from_config(&repo_config(repo))
}

fn serve_v0(
    repo: &Repository,
    input: &mut dyn Read,
    output: &mut dyn Write,
    opts: &ServeOptions,
) -> Result<()> {
    let config = repo_config(repo);
    let policy = UploadPackPolicy::from_config(&config);
    let refs = advertised_refs(repo, &opts.hidden_refs, &[], Peel::Yes)?;
    let head = head_info(repo);

    if opts.advertise_refs || !opts.stateless_rpc {
        if opts.protocol == ProtocolVersion::V1 {
            write_line(output, "version 1")?;
        }
        write_v0_advertisement(repo, output, &refs, &head, opts)?;
        if opts.advertise_refs {
            return Ok(());
        }
        output.flush()?;
    }

    let Some(request) = read_v0_request(repo, input, &config, &policy)? else {
        return Ok(());
    };
    if request.filter.is_some() {
        validate_wants_filtered(repo, &request.wants)?;
    } else {
        validate_wants(
            repo,
            &refs,
            head.oid,
            &request.wants,
            &policy,
            opts.stateless_rpc,
        )?;
    }

    let shallow_resp = if request.shallow.is_active() {
        Some(compute_shallow_response(
            repo,
            &request.wants,
            &request.shallow,
        )?)
    } else {
        None
    };
    if !opts.stateless_rpc {
        if let Some(ref sh) = shallow_resp {
            write_v0_shallow_lines(output, sh)?;
            write_flush(output)?;
            output.flush()?;
        }
    }

    let detailed = request.caps.contains("multi_ack_detailed");
    let mut common: Vec<ObjectId> = Vec::new();
    let mut seen: HashSet<ObjectId> = HashSet::new();
    loop {
        match read_packet(input)? {
            None => {
                if opts.stateless_rpc {
                    if let Some(ref sh) = shallow_resp {
                        write_v0_shallow_lines(output, sh)?;
                        write_flush(output)?;
                    }
                }
                return Ok(());
            }
            Some(Packet::Data(line)) if line == "done" => break,
            Some(Packet::Data(line)) => {
                let Some(hex) = line.strip_prefix("have ") else {
                    return Err(ServeError::Protocol(format!(
                        "expected have or done, got '{line}'"
                    )));
                };
                let oid = parse_oid(repo, hex.trim())?;
                if !seen.insert(oid) || !repo.odb().contains(&oid)? {
                    continue;
                }
                common.push(oid);
                if detailed {
                    write_line(output, &format!("ACK {} common", oid.to_hex()))?;
                } else if common.len() == 1 {
                    write_line(output, &format!("ACK {}", oid.to_hex()))?;
                }
            }
            Some(Packet::Flush) => {
                if detailed || common.is_empty() {
                    write_line(output, "NAK")?;
                }
                output.flush()?;
            }
            Some(other) => {
                return Err(ServeError::Protocol(format!(
                    "unexpected packet during negotiation: {other:?}"
                )))
            }
        }
    }

    if opts.stateless_rpc {
        if let Some(ref sh) = shallow_resp {
            write_v0_shallow_lines(output, sh)?;
            write_flush(output)?;
        }
    }
    match common.last() {
        Some(last) => write_line(output, &format!("ACK {}", last.to_hex()))?,
        None => write_line(output, "NAK")?,
    }
    output.flush()?;
    let sideband = request.caps.contains("side-band-64k");
    write_response_pack_from_v0(
        repo,
        &refs,
        &request,
        &common,
        request.filter.as_ref(),
        shallow_resp.as_ref(),
        output,
        sideband,
    )
}

/// Write the v0 ref advertisement: `HEAD` and every ref, capabilities on the
/// first line, and a peeled `^{}` line after each annotated tag.
fn write_v0_advertisement(
    repo: &Repository,
    output: &mut dyn Write,
    refs: &[AdvertisedRef],
    head: &super::HeadInfo,
    opts: &ServeOptions,
) -> Result<()> {
    let policy = upload_policy(repo);
    let mut caps: Vec<String> = V0_CAPABILITIES.iter().map(|c| (*c).to_owned()).collect();
    for cap in policy.v0_capability_tokens() {
        if !caps.iter().any(|existing| existing == cap) {
            caps.push(cap.to_owned());
        }
    }
    if let (Some(target), Some(_)) = (&head.target, head.oid) {
        caps.push(format!("symref=HEAD:{target}"));
    }
    caps.push(format!("object-format={}", repo.odb().hash_algo().name()));
    if !opts.agent.is_empty() {
        caps.push(format!("agent={}", opts.agent));
    }
    let caps = caps.join(" ");

    let mut lines: Vec<(ObjectId, String)> = Vec::new();
    if let Some(oid) = head.oid {
        lines.push((oid, "HEAD".to_owned()));
    }
    for r in refs {
        lines.push((r.oid, r.name.clone()));
        if let Some(peeled) = r.peeled {
            lines.push((peeled, format!("{}^{{}}", r.name)));
        }
    }

    if lines.is_empty() {
        let zero = ObjectId::null(repo.odb().hash_algo());
        write_line(
            output,
            &format!("{} capabilities^{{}}\0{caps}", zero.to_hex()),
        )?;
    }
    for (i, (oid, name)) in lines.iter().enumerate() {
        if i == 0 {
            write_line(output, &format!("{} {name}\0{caps}", oid.to_hex()))?;
        } else {
            write_line(output, &format!("{} {name}", oid.to_hex()))?;
        }
    }
    write_flush(output)
}

struct V0Request {
    wants: Vec<ObjectId>,
    caps: HashSet<String>,
    shallow: ShallowRequest,
    filter: Option<ObjectFilter>,
}

fn read_v0_request(
    repo: &Repository,
    input: &mut dyn Read,
    config: &ConfigSet,
    policy: &UploadPackPolicy,
) -> Result<Option<V0Request>> {
    let mut wants = Vec::new();
    let mut caps = HashSet::new();
    let mut shallow = ShallowRequest::default();
    let mut filter: Option<ObjectFilter> = None;
    loop {
        match read_packet(input)? {
            None | Some(Packet::Flush) => break,
            Some(Packet::Data(line)) => {
                if let Some(rest) = line.strip_prefix("want ") {
                    let mut parts = rest.split(' ');
                    let hex = parts.next().unwrap_or_default();
                    wants.push(parse_oid(repo, hex)?);
                    if wants.len() == 1 {
                        caps.extend(parts.filter(|c| !c.is_empty()).map(str::to_owned));
                    }
                    continue;
                }
                if parse_shallow_line(repo, &line, &mut shallow, config, policy, &mut filter)? {
                    continue;
                }
                return Err(ServeError::Protocol(format!(
                    "unsupported request line '{line}'"
                )));
            }
            Some(other) => {
                return Err(ServeError::Protocol(format!(
                    "unexpected packet in want list: {other:?}"
                )))
            }
        }
    }
    if wants.is_empty() {
        return Ok(None);
    }
    let format = caps.iter().find_map(|c| c.strip_prefix("object-format="));
    check_object_format(repo, format)?;
    Ok(Some(V0Request {
        wants,
        caps,
        shallow,
        filter,
    }))
}

fn parse_shallow_line(
    repo: &Repository,
    line: &str,
    shallow: &mut ShallowRequest,
    config: &ConfigSet,
    policy: &UploadPackPolicy,
    filter: &mut Option<ObjectFilter>,
) -> Result<bool> {
    if let Some(rest) = line.strip_prefix("shallow ") {
        shallow.client_shallow.push(parse_oid(repo, rest.trim())?);
        return Ok(true);
    }
    if let Some(rest) = line.strip_prefix("deepen ") {
        shallow.depth = Some(
            rest.trim()
                .parse()
                .map_err(|_| ServeError::Protocol(format!("invalid deepen value '{rest}'")))?,
        );
        return Ok(true);
    }
    if let Some(rest) = line.strip_prefix("deepen-since ") {
        shallow.deepen_since = Some(
            rest.trim()
                .parse()
                .map_err(|_| ServeError::Protocol(format!("invalid deepen-since '{rest}'")))?,
        );
        return Ok(true);
    }
    if let Some(rest) = line.strip_prefix("deepen-not ") {
        shallow
            .deepen_not
            .push(resolve_deepen_not(repo, rest.trim())?);
        return Ok(true);
    }
    if line == "deepen-relative" {
        shallow.deepen_relative = true;
        return Ok(true);
    }
    if let Some(rest) = line.strip_prefix("filter ") {
        let spec = rest.trim();
        let parsed = policy.parse_filter(config, spec)?;
        *filter = Some(match filter.take() {
            Some(existing) => existing.merge_with(parsed),
            None => parsed,
        });
        return Ok(true);
    }
    Ok(false)
}

/// Resolve a `deepen-not` argument to the commit that bounds the shallow history.
///
/// Git clients send whatever the user gave `--shallow-exclude`, usually a ref
/// name such as `v1.0` or `refs/heads/main`. A full-width object id is accepted
/// too. Short names expand through the standard ref lookup rules and must match
/// exactly one ref. Annotated tags are peeled.
fn resolve_deepen_not(repo: &Repository, spec: &str) -> Result<ObjectId> {
    let invalid = |reason| ServeError::InvalidDeepenNot {
        spec: spec.to_owned(),
        reason,
    };
    let oid = match parse_oid(repo, spec) {
        Ok(oid) => oid,
        Err(_) => {
            let (matches, name) = crate::refs::shorten::expand_ref_dwim_full_name(repo, spec)?;
            let name = match (matches, name) {
                (1, Some(name)) => name,
                (0, _) | (_, None) => return Err(invalid(DeepenNotError::UnknownRef)),
                _ => return Err(invalid(DeepenNotError::AmbiguousRef)),
            };
            repo.resolve_ref_oid(&name)
                .map_err(|_| invalid(DeepenNotError::UnknownRef))?
        }
    };
    if !repo.odb().contains(&oid)? {
        return Err(invalid(DeepenNotError::UnknownRef));
    }
    match repo.odb().peel_tags(oid)? {
        (commit, ObjectKind::Commit) => Ok(commit),
        _ => Err(invalid(DeepenNotError::NotACommit)),
    }
}

fn write_v0_shallow_lines(output: &mut dyn Write, sh: &ShallowResponse) -> Result<()> {
    for oid in &sh.shallow {
        write_line(output, &format!("shallow {}", oid.to_hex()))?;
    }
    for oid in &sh.unshallow {
        write_line(output, &format!("unshallow {}", oid.to_hex()))?;
    }
    Ok(())
}

#[allow(clippy::too_many_arguments)]
fn write_response_pack_from_v0(
    repo: &Repository,
    refs: &[AdvertisedRef],
    request: &V0Request,
    common: &[ObjectId],
    filter: Option<&ObjectFilter>,
    shallow: Option<&ShallowResponse>,
    output: &mut dyn Write,
    sideband: bool,
) -> Result<()> {
    let mut pack_wants = request.wants.clone();
    if let Some(sh) = shallow {
        pack_wants.extend(sh.extra_wants.iter().copied());
    }
    pack_wants.sort();
    pack_wants.dedup();

    let pack_common: Vec<ObjectId> = if let Some(sh) = shallow {
        common
            .iter()
            .filter(|oid| !sh.unshallow.contains(oid))
            .copied()
            .collect()
    } else {
        common.to_vec()
    };

    let have_shallow: HashSet<ObjectId> = request.shallow.client_shallow.iter().copied().collect();
    let source_shallow = shallow
        .map(|s| s.pack_shallow_grafts.clone())
        .unwrap_or_default();

    write_response_pack(
        ResponsePackInput {
            repo,
            refs,
            wants: &pack_wants,
            common: &pack_common,
            caps: &request.caps,
            filter,
            have_shallow: &have_shallow,
            source_shallow: &source_shallow,
        },
        output,
        sideband,
    )
}

fn serve_v2(
    repo: &Repository,
    input: &mut dyn Read,
    output: &mut dyn Write,
    opts: &ServeOptions,
) -> Result<()> {
    if opts.advertise_refs || !opts.stateless_rpc {
        write_line(output, "version 2")?;
        if !opts.agent.is_empty() {
            write_line(output, &format!("agent={}", opts.agent))?;
        }
        write_line(output, "ls-refs=unborn")?;
        let policy = upload_policy(repo);
        let fetch_caps = policy.v2_fetch_features().join(" ");
        write_line(output, &format!("fetch={fetch_caps}"))?;
        write_line(
            output,
            &format!("object-format={}", repo.odb().hash_algo().name()),
        )?;
        write_flush(output)?;
        if opts.advertise_refs {
            return Ok(());
        }
        output.flush()?;
    }

    loop {
        let Some(request) = read_v2_request(input)? else {
            return Ok(());
        };
        check_object_format(repo, request.object_format.as_deref())?;
        match request.command.as_str() {
            "ls-refs" => ls_refs(repo, output, &request.args, opts)?,
            "fetch" => fetch_v2(repo, output, &request.args, opts)?,
            other => return Err(ServeError::UnknownCommand(other.to_owned())),
        }
        output.flush()?;
        if opts.stateless_rpc {
            return Ok(());
        }
    }
}

/// One v2 command request.
struct V2Request {
    command: String,
    object_format: Option<String>,
    args: Vec<String>,
}

/// Read a v2 command request. Returns `None` at end of input or on a bare
/// flush, which ends the session.
fn read_v2_request(input: &mut dyn Read) -> Result<Option<V2Request>> {
    let mut command = None;
    let mut object_format = None;
    let mut args = Vec::new();
    let mut in_args = false;
    loop {
        match read_packet(input)? {
            None => {
                if command.is_none() {
                    return Ok(None);
                }
                return Err(ServeError::Protocol("truncated command request".into()));
            }
            Some(Packet::Flush) => break,
            Some(Packet::Delim) if !in_args => in_args = true,
            Some(Packet::Data(line)) if in_args => args.push(line),
            Some(Packet::Data(line)) => {
                if let Some(c) = line.strip_prefix("command=") {
                    command = Some(c.to_owned());
                } else if let Some(f) = line.strip_prefix("object-format=") {
                    object_format = Some(f.to_owned());
                }
                // agent=, server-option= and unknown capabilities are ignored.
            }
            Some(other) => {
                return Err(ServeError::Protocol(format!(
                    "unexpected packet in command request: {other:?}"
                )))
            }
        }
    }
    match command {
        Some(command) => Ok(Some(V2Request {
            command,
            object_format,
            args,
        })),
        None if args.is_empty() => Ok(None),
        None => Err(ServeError::Protocol(
            "command request without command".into(),
        )),
    }
}

fn ls_refs(
    repo: &Repository,
    output: &mut dyn Write,
    args: &[String],
    opts: &ServeOptions,
) -> Result<()> {
    let mut symrefs = false;
    let mut peel = Peel::No;
    let mut unborn = false;
    let mut prefixes: Vec<&str> = Vec::new();
    for arg in args {
        match arg.as_str() {
            "symrefs" => symrefs = true,
            "peel" => peel = Peel::Yes,
            "unborn" => unborn = true,
            other => {
                if let Some(p) = other.strip_prefix("ref-prefix ") {
                    prefixes.push(p);
                }
            }
        }
    }
    if prefixes.is_empty() || prefixes.iter().any(|p| "HEAD".starts_with(p)) {
        let head = head_info(repo);
        let target = head.target.as_deref().filter(|_| symrefs);
        match (head.oid, target) {
            (Some(oid), Some(t)) => {
                write_line(output, &format!("{} HEAD symref-target:{t}", oid.to_hex()))?
            }
            (Some(oid), None) => write_line(output, &format!("{} HEAD", oid.to_hex()))?,
            (None, Some(t)) if unborn => {
                write_line(output, &format!("unborn HEAD symref-target:{t}"))?
            }
            (None, _) => {}
        }
    }
    for r in advertised_refs(repo, &opts.hidden_refs, &prefixes, peel)? {
        let mut line = format!("{} {}", r.oid.to_hex(), r.name);
        if let Some(p) = r.peeled {
            line.push_str(&format!(" peeled:{}", p.to_hex()));
        }
        write_line(output, &line)?;
    }
    write_flush(output)
}

fn fetch_v2(
    repo: &Repository,
    output: &mut dyn Write,
    args: &[String],
    opts: &ServeOptions,
) -> Result<()> {
    let config = repo_config(repo);
    let policy = UploadPackPolicy::from_config(&config);
    let mut wants = Vec::new();
    let mut haves = Vec::new();
    let mut done = false;
    let mut caps = HashSet::new();
    let mut deepen = None;
    let mut deepen_since = None;
    let mut deepen_not: Vec<ObjectId> = Vec::new();
    let mut deepen_relative = false;
    let mut filter_spec: Option<String> = None;
    let mut want_ref_names: Vec<String> = Vec::new();
    for arg in args {
        if let Some(hex) = arg.strip_prefix("want ") {
            wants.push(parse_oid(repo, hex.trim())?);
        } else if let Some(hex) = arg.strip_prefix("have ") {
            haves.push(parse_oid(repo, hex.trim())?);
        } else if arg.starts_with("shallow ") {
        } else if let Some(rest) = arg.strip_prefix("deepen ") {
            deepen = Some(
                rest.trim()
                    .parse()
                    .map_err(|_| ServeError::Protocol(format!("bad deepen '{arg}'")))?,
            );
        } else if let Some(rest) = arg.strip_prefix("deepen-since ") {
            deepen_since = Some(
                rest.trim()
                    .parse()
                    .map_err(|_| ServeError::Protocol(format!("bad deepen-since '{arg}'")))?,
            );
        } else if let Some(rest) = arg.strip_prefix("deepen-not ") {
            deepen_not.push(resolve_deepen_not(repo, rest.trim())?);
        } else if let Some(spec) = arg.strip_prefix("filter ") {
            filter_spec = Some(spec.trim().to_owned());
        } else if let Some(name) = arg.strip_prefix("want-ref ").filter(|_| policy.ref_in_want) {
            let name = name.trim();
            if want_ref_names.iter().any(|n| n == name) {
                return Err(ServeError::Protocol(format!("duplicate want-ref {name}")));
            }
            want_ref_names.push(name.to_owned());
        } else if arg == "done" {
            done = true;
        } else if arg == "deepen-relative" {
            deepen_relative = true;
        } else if matches!(
            arg.as_str(),
            "thin-pack" | "no-progress" | "include-tag" | "ofs-delta"
        ) || (arg == "sideband-all" && policy.allow_sideband_all)
        {
            caps.insert(arg.clone());
        } else {
            return Err(ServeError::Protocol(format!(
                "unsupported fetch argument '{arg}'"
            )));
        }
    }

    let refs = advertised_refs(repo, &opts.hidden_refs, &[], Peel::Yes)?;
    let head = head_info(repo);
    let wanted_refs = want_ref_names
        .into_iter()
        .map(|name| {
            let oid = resolve_want_ref(repo, &refs, &opts.hidden_refs, &name)?;
            Ok((name, oid))
        })
        .collect::<Result<Vec<_>>>()?;
    wants.extend(wanted_refs.iter().map(|(_, oid)| *oid));

    let filter: Option<ObjectFilter> = match filter_spec {
        None => None,
        Some(spec) => Some(policy.parse_filter(&config, &spec)?),
    };

    if filter.is_some() {
        validate_wants_filtered(repo, &wants)?;
    } else {
        validate_wants(repo, &refs, head.oid, &wants, &policy, opts.stateless_rpc)?;
    }

    let sideband_all = caps.contains("sideband-all");
    let mut seen = HashSet::new();
    let common = repo
        .odb()
        .present_oids(haves.into_iter().filter(|h| seen.insert(*h)))?;

    if !done {
        write_v2_fetch_line(output, "acknowledgments", sideband_all)?;
        if common.is_empty() {
            write_v2_fetch_line(output, "NAK", sideband_all)?;
        }
        for oid in &common {
            write_v2_fetch_line(output, &format!("ACK {}", oid.to_hex()), sideband_all)?;
        }
        return write_flush(output);
    }

    let client_shallow: Vec<ObjectId> = args
        .iter()
        .filter_map(|a| a.strip_prefix("shallow "))
        .map(|hex| parse_oid(repo, hex.trim()))
        .collect::<Result<Vec<_>>>()?;

    let shallow_req = ShallowRequest {
        client_shallow: client_shallow.clone(),
        depth: deepen,
        deepen_since,
        deepen_not,
        deepen_relative,
    };
    let shallow_active = shallow_req.is_active();
    let shallow_resp = if shallow_active {
        compute_shallow_response(repo, &wants, &shallow_req)?
    } else {
        Default::default()
    };

    if !shallow_resp.shallow.is_empty() || !shallow_resp.unshallow.is_empty() {
        write_v2_fetch_line(output, "shallow-info", sideband_all)?;
        for oid in &shallow_resp.unshallow {
            write_v2_fetch_line(output, &format!("unshallow {}", oid.to_hex()), sideband_all)?;
        }
        for oid in &shallow_resp.shallow {
            write_v2_fetch_line(output, &format!("shallow {}", oid.to_hex()), sideband_all)?;
        }
        write_v2_section_delim(output)?;
    }

    if !wanted_refs.is_empty() {
        write_v2_fetch_line(output, "wanted-refs", sideband_all)?;
        for (name, oid) in &wanted_refs {
            write_v2_fetch_line(output, &format!("{} {name}", oid.to_hex()), sideband_all)?;
        }
        write_v2_section_delim(output)?;
    }

    let mut pack_wants = wants;
    pack_wants.extend(shallow_resp.extra_wants);
    pack_wants.sort();
    pack_wants.dedup();

    let have_shallow: HashSet<ObjectId> = client_shallow.iter().copied().collect();
    write_v2_fetch_line(output, "packfile", sideband_all)?;
    // Protocol v2 always side-band-64k frames the packfile section payload (band 1),
    // even when `sideband-all` was not negotiated.
    write_response_pack(
        ResponsePackInput {
            repo,
            refs: &refs,
            wants: &pack_wants,
            common: &common,
            caps: &caps,
            filter: filter.as_ref(),
            have_shallow: &have_shallow,
            source_shallow: &shallow_resp.pack_shallow_grafts,
        },
        output,
        true,
    )
}

fn write_v2_section_delim(output: &mut dyn Write) -> Result<()> {
    write!(output, "0001").map_err(crate::error::Error::Io)?;
    Ok(())
}

/// Write one v2 fetch response line, using sideband channel 1 when `sideband-all` was negotiated.
fn write_v2_fetch_line(output: &mut dyn Write, line: &str, sideband_all: bool) -> Result<()> {
    if sideband_all {
        let mut payload = Vec::with_capacity(line.len() + 1);
        payload.extend_from_slice(line.as_bytes());
        payload.push(b'\n');
        crate::pkt_line::write_sideband_packet(output, 1, &payload).map_err(ServeError::Io)?;
        Ok(())
    } else {
        write_line(output, line)
    }
}

/// Inputs for [`write_response_pack`].
struct ResponsePackInput<'a> {
    repo: &'a Repository,
    refs: &'a [AdvertisedRef],
    wants: &'a [ObjectId],
    common: &'a [ObjectId],
    caps: &'a HashSet<String>,
    filter: Option<&'a ObjectFilter>,
    have_shallow: &'a HashSet<ObjectId>,
    source_shallow: &'a HashSet<ObjectId>,
}

/// Stream the pack answering a fetch to `output`, adding annotated tags when
/// `include-tag` was negotiated.
///
/// With `sideband` the pack is framed on side-band channel 1 and followed by a
/// flush packet. The pack is written as it is produced, never held in memory whole.
fn write_response_pack(
    input: ResponsePackInput<'_>,
    output: &mut dyn Write,
    sideband: bool,
) -> Result<()> {
    let mut pack_wants = input.wants.to_vec();
    if input.caps.contains("include-tag") {
        pack_wants.extend(included_tags(&input)?);
    }
    pack_wants.sort();
    pack_wants.dedup();

    let cfg = input.repo.config().ok();
    let pack_opts = PackBuildOptions::for_serve(
        cfg.as_deref(),
        input.caps.contains("thin-pack") && !input.common.is_empty(),
        input.caps.contains("ofs-delta"),
    );
    let mut progress = |_ev: PackProgress| {};
    let mut write = |w: &mut dyn Write| {
        write_upload_pack_stream(
            input.repo,
            &pack_wants,
            input.common,
            input.source_shallow,
            input.have_shallow,
            input.filter,
            &pack_opts,
            &mut progress,
            w,
        )
    };
    if !sideband {
        write(output)?;
        return Ok(());
    }
    let mut band = SidebandChannel1Writer::new(output);
    write(&mut band)?;
    band.finish()?;
    write_flush(output)
}

/// Annotated tags the client did not ask for but whose target commit is being
/// sent, so the client can follow tags without a second request.
///
/// Only tags that peel to commits are considered. Both walks honor the shallow
/// grafts of their side, so a shallow fetch only picks up tags on the commits
/// it actually sends.
fn included_tags(input: &ResponsePackInput<'_>) -> Result<Vec<ObjectId>> {
    let repo = input.repo;
    let want_set: HashSet<ObjectId> = input.wants.iter().copied().collect();
    let candidates: Vec<&AdvertisedRef> = input
        .refs
        .iter()
        .filter(|r| r.name.starts_with("refs/tags/") && r.peeled.is_some())
        .filter(|r| !want_set.contains(&r.oid))
        .collect();
    if candidates.is_empty() {
        return Ok(Vec::new());
    }
    let have_commits = commit_closure(repo, input.common, &HashSet::new(), input.have_shallow)?;
    let sent_commits = commit_closure(repo, input.wants, &have_commits, input.source_shallow)?;
    Ok(candidates
        .into_iter()
        .filter(|r| r.peeled.is_some_and(|p| sent_commits.contains(&p)))
        .map(|r| r.oid)
        .collect())
}

/// Commits reachable from `tips` (peeling tags), not descending into `stop`.
/// Commits in `grafts` are included but their parents are not walked.
fn commit_closure(
    repo: &Repository,
    tips: &[ObjectId],
    stop: &HashSet<ObjectId>,
    grafts: &HashSet<ObjectId>,
) -> Result<HashSet<ObjectId>> {
    let mut seen = HashSet::new();
    let mut queue: VecDeque<ObjectId> = VecDeque::new();
    for tip in tips {
        let Ok(oid) = repo.odb().peel_tags(*tip).map(|(p, _)| p) else {
            continue;
        };
        queue.push_back(oid);
    }
    while let Some(oid) = queue.pop_front() {
        if stop.contains(&oid) || !seen.insert(oid) {
            continue;
        }
        let Ok(obj) = repo.odb().read(&oid) else {
            continue;
        };
        if obj.kind != ObjectKind::Commit {
            seen.remove(&oid);
            continue;
        }
        if grafts.contains(&oid) {
            continue;
        }
        let commit = crate::objects::parse_commit(&obj.data)?;
        queue.extend(commit.parents);
    }
    Ok(seen)
}

#[cfg(test)]
#[allow(clippy::unwrap_used, clippy::expect_used)]
mod tests {
    use super::*;
    use crate::config::environment::RepositoryOptions;
    use crate::repo::InitOptions;

    /// Bare repository with `n` linear commits (oldest first) and an annotated
    /// tag `refs/tags/t<i>` on every commit index in `tagged`.
    struct Fixture {
        _dir: tempfile::TempDir,
        repo: Repository,
        commits: Vec<ObjectId>,
        tags: Vec<AdvertisedRef>,
    }

    fn fixture(n: usize, tagged: &[usize]) -> Fixture {
        let dir = tempfile::tempdir().unwrap();
        Repository::init(dir.path(), &InitOptions::new().with_bare(true)).unwrap();
        let repo = Repository::open(dir.path(), &RepositoryOptions::empty()).unwrap();
        let tree = repo
            .odb()
            .write_loose_materialize(ObjectKind::Tree, b"")
            .unwrap();
        let mut commits: Vec<ObjectId> = Vec::new();
        for i in 0..n {
            let parent = commits
                .last()
                .map(|p| format!("parent {p}\n"))
                .unwrap_or_default();
            let body = format!(
                "tree {tree}\n{parent}author T <t@x> {i} +0000\ncommitter T <t@x> {i} +0000\n\nc{i}\n"
            );
            commits.push(
                repo.odb()
                    .write_loose_materialize(ObjectKind::Commit, body.as_bytes())
                    .unwrap(),
            );
        }
        let tags = tagged
            .iter()
            .map(|&i| {
                let body = format!(
                    "object {}\ntype commit\ntag t{i}\ntagger T <t@x> 0 +0000\n\nt\n",
                    commits[i]
                );
                AdvertisedRef {
                    name: format!("refs/tags/t{i}"),
                    oid: repo
                        .odb()
                        .write_loose_materialize(ObjectKind::Tag, body.as_bytes())
                        .unwrap(),
                    peeled: Some(commits[i]),
                }
            })
            .collect();
        Fixture {
            _dir: dir,
            repo,
            commits,
            tags,
        }
    }

    fn depth_request(depth: u32) -> ShallowRequest {
        ShallowRequest {
            depth: Some(depth),
            ..ShallowRequest::default()
        }
    }

    #[test]
    fn shallow_response_for_tag_wants_names_only_commits() {
        let fx = fixture(5, &[1, 4]);
        let tip = fx.commits[4];
        let wants = [tip, fx.tags[0].oid, fx.tags[1].oid];
        let resp = compute_shallow_response(&fx.repo, &wants, &depth_request(1)).unwrap();
        let mut shallow = resp.shallow.clone();
        shallow.sort();
        let mut expected = vec![fx.commits[1], tip];
        expected.sort();
        assert_eq!(shallow, expected);
        for oid in &resp.pack_shallow_grafts {
            assert_eq!(
                fx.repo.odb().read_info(oid).unwrap().kind,
                ObjectKind::Commit
            );
        }
    }

    #[test]
    fn shallow_response_ignores_wants_that_are_not_commits() {
        let fx = fixture(2, &[]);
        let tree = fx
            .repo
            .odb()
            .write_loose_materialize(ObjectKind::Tree, b"")
            .unwrap();
        let resp =
            compute_shallow_response(&fx.repo, &[fx.commits[1], tree], &depth_request(1)).unwrap();
        assert_eq!(resp.shallow, vec![fx.commits[1]]);
    }

    #[test]
    fn included_tags_stop_at_the_shallow_boundary() {
        let fx = fixture(6, &[1, 3, 5]);
        let tip = fx.commits[5];
        let resp = compute_shallow_response(&fx.repo, &[tip], &depth_request(3)).unwrap();
        let caps = HashSet::from(["include-tag".to_owned()]);
        let tags = included_tags(&ResponsePackInput {
            repo: &fx.repo,
            refs: &fx.tags,
            wants: &[tip],
            common: &[],
            caps: &caps,
            filter: None,
            have_shallow: &HashSet::new(),
            source_shallow: &resp.pack_shallow_grafts,
        })
        .unwrap();
        let mut expected = vec![fx.tags[1].oid, fx.tags[2].oid];
        expected.sort();
        let mut tags = tags;
        tags.sort();
        assert_eq!(tags, expected, "t1 lies beyond the depth-3 cut");
    }

    #[test]
    fn included_tags_skip_history_the_client_already_has() {
        let fx = fixture(6, &[1, 3, 5]);
        let caps = HashSet::from(["include-tag".to_owned()]);
        let tags = included_tags(&ResponsePackInput {
            repo: &fx.repo,
            refs: &fx.tags,
            wants: &[fx.commits[5]],
            common: &[fx.commits[3]],
            caps: &caps,
            filter: None,
            have_shallow: &HashSet::new(),
            source_shallow: &HashSet::new(),
        })
        .unwrap();
        assert_eq!(tags, vec![fx.tags[2].oid]);
    }
}