sley-remote 0.0.2

//! Native `git://` transport over a raw TCP stream.
//!
//! This is the anonymous git-daemon protocol: send one service request pkt-line,
//! then speak the same upload-pack / receive-pack protocol v0 streams used by
//! SSH. Authentication and protocol-v2 negotiation are intentionally out of
//! scope for this transport.

use std::collections::HashMap;
use std::io::{Read, Write};
use std::net::{Shutdown, TcpStream};
use std::path::Path;

use sley_core::{Capability, GitError, ObjectFormat, ObjectId, Result};
use sley_fetch::{install_upload_pack_raw_promisor_response, install_upload_pack_raw_response};
use sley_odb::{FileObjectDatabase, build_reachable_pack, collect_reachable_object_ids};
use sley_protocol::{
    GitService, ProtocolV2FetchShallowInfo, ReceivePackCommand, ReceivePackFeatures,
    ReceivePackPushRequestOptions, RefAdvertisement, UploadPackFeatures,
    UploadPackNegotiationRequest, UploadPackRawPackfileResponse, UploadPackRequest,
    build_receive_pack_push_request, parse_receive_pack_features, parse_refspec,
    parse_upload_pack_features, plan_push_commands, read_receive_pack_report_status,
    read_ref_advertisement_set, read_upload_pack_raw_packfile_response,
    read_upload_pack_shallow_info_and_raw_packfile_response, write_receive_pack_push_request,
    write_upload_pack_negotiation_request, write_upload_pack_request,
};
use sley_refs::FileRefStore;
use sley_transport::{RemoteTransport, RemoteUrl, ServiceRequest, write_service_request};

use crate::{PushOutcome, PushRequest};

const GIT_DAEMON_PORT: u16 = 9418;

pub(crate) struct GitPushRequest<'a> {
    pub git_dir: &'a Path,
    pub common_git_dir: &'a Path,
    pub format: ObjectFormat,
    pub remote: &'a RemoteUrl,
    pub refspecs: &'a [String],
    pub force: bool,
}

pub(crate) struct GitPushCommandsRequest<'a> {
    pub common_git_dir: &'a Path,
    pub format: ObjectFormat,
    pub remote: &'a RemoteUrl,
    pub command_forces: Vec<(ReceivePackCommand, bool)>,
    pub pack_objects: Vec<ObjectId>,
}

pub(crate) struct GitPushPlan {
    pub(crate) commands: Vec<ReceivePackCommand>,
    pub(crate) pack_objects: Vec<ObjectId>,
    stream: Option<TcpStream>,
    features: ReceivePackFeatures,
    advertisements: Vec<RefAdvertisement>,
}

pub struct GitFetchPackRequest<'a> {
    pub git_dir: &'a Path,
    pub format: ObjectFormat,
    pub remote: &'a RemoteUrl,
    pub features: &'a UploadPackFeatures,
    pub wants: Vec<ObjectId>,
    pub shallow: Vec<ObjectId>,
    pub deepen: Option<u32>,
    pub promisor: bool,
}

pub(crate) fn ls_remote_git(
    remote: &RemoteUrl,
    filter: &crate::ls_remote::LsRemoteFilter,
    matches: &dyn Fn(&str) -> bool,
) -> Result<(Vec<crate::ls_remote::LsRemoteRecord>, ObjectFormat)> {
    let mut stream = connect_git_service(remote, GitService::UploadPack)?;
    let set = read_ref_advertisement_set(ObjectFormat::Sha1, &mut stream)?;
    let features = set
        .refs
        .first()
        .map(|advertisement| parse_upload_pack_features(&advertisement.capabilities))
        .transpose()?
        .unwrap_or_default();
    let format = features.object_format.unwrap_or(ObjectFormat::Sha1);
    if format != ObjectFormat::Sha1 {
        return Err(GitError::Unsupported(format!(
            "git:// ls-remote currently supports SHA-1 advertisements, got {}",
            format.name()
        )));
    }
    let symrefs = features
        .symrefs
        .iter()
        .filter_map(|symref| symref.split_once(':'))
        .map(|(name, target)| (name.to_string(), target.to_string()))
        .collect::<HashMap<_, _>>();
    let mut records = Vec::new();
    for advertisement in set.refs {
        if advertisement.oid.is_null() {
            continue;
        }
        if filter.refs_only && (advertisement.name == "HEAD" || advertisement.name.ends_with("^{}"))
        {
            continue;
        }
        let is_head = advertisement.name.starts_with("refs/heads/");
        let is_tag = advertisement.name.starts_with("refs/tags/");
        if (filter.heads || filter.tags) && !((filter.heads && is_head) || (filter.tags && is_tag))
        {
            continue;
        }
        if !matches(&advertisement.name) {
            continue;
        }
        records.push(crate::ls_remote::LsRemoteRecord {
            oid: advertisement.oid,
            symref: symrefs.get(&advertisement.name).cloned(),
            name: advertisement.name,
        });
    }
    Ok((records, format))
}

pub fn git_upload_pack_advertisements(
    remote: &RemoteUrl,
    format: ObjectFormat,
) -> Result<(Vec<RefAdvertisement>, UploadPackFeatures)> {
    let mut stream = connect_git_service(remote, GitService::UploadPack)?;
    let set = read_ref_advertisement_set(format, &mut stream)?;
    let features = set
        .refs
        .first()
        .map(|advertisement| parse_upload_pack_features(&advertisement.capabilities))
        .transpose()?
        .unwrap_or_default();
    Ok((set.refs, features))
}

pub fn install_fetch_pack_via_git_upload_pack(
    request: GitFetchPackRequest<'_>,
) -> Result<Vec<ProtocolV2FetchShallowInfo>> {
    if request.wants.is_empty() {
        return Ok(Vec::new());
    }
    let local_db = FileObjectDatabase::from_git_dir(request.git_dir, request.format);
    if request.deepen.is_none() && all_wants_present(&local_db, &request.wants)? {
        return Ok(Vec::new());
    }
    let upload_request = UploadPackRequest {
        wants: request.wants,
        capabilities: shallow_request_capabilities(request.deepen),
        shallow: request.shallow,
        deepen: request.deepen,
        ..UploadPackRequest::default()
    };
    let haves = crate::local::local_have_oids(request.git_dir, request.format)?;
    let (shallow_info, response) = if request.deepen.is_some() {
        git_upload_pack_shallow_fetch_response(
            request.remote,
            request.format,
            request.features,
            upload_request,
            haves,
        )?
    } else {
        let response = git_upload_pack_fetch_response(
            request.remote,
            request.format,
            request.features,
            upload_request,
            haves,
        )?;
        (Vec::new(), response)
    };
    if request.promisor {
        install_upload_pack_raw_promisor_response(&response, &local_db)?;
    } else {
        install_upload_pack_raw_response(&response, &local_db)?;
    }
    Ok(shallow_info)
}

pub fn git_upload_pack_fetch_response(
    remote: &RemoteUrl,
    format: ObjectFormat,
    _features: &UploadPackFeatures,
    request: UploadPackRequest,
    haves: Vec<ObjectId>,
) -> Result<UploadPackRawPackfileResponse> {
    let (_shallow, response) =
        git_upload_pack_fetch_response_inner(remote, format, request, haves, false)?;
    Ok(response)
}

pub fn git_upload_pack_shallow_fetch_response(
    remote: &RemoteUrl,
    format: ObjectFormat,
    _features: &UploadPackFeatures,
    request: UploadPackRequest,
    haves: Vec<ObjectId>,
) -> Result<(
    Vec<ProtocolV2FetchShallowInfo>,
    UploadPackRawPackfileResponse,
)> {
    git_upload_pack_fetch_response_inner(remote, format, request, haves, true)
}

fn git_upload_pack_fetch_response_inner(
    remote: &RemoteUrl,
    format: ObjectFormat,
    request: UploadPackRequest,
    haves: Vec<ObjectId>,
    expect_shallow_info: bool,
) -> Result<(
    Vec<ProtocolV2FetchShallowInfo>,
    UploadPackRawPackfileResponse,
)> {
    let mut stream = connect_git_service(remote, GitService::UploadPack)?;
    read_ref_advertisement_set(format, &mut stream)?;
    write_upload_pack_request(&mut stream, Some(&request))?;
    write_upload_pack_negotiation_request(
        &mut stream,
        &UploadPackNegotiationRequest { haves, done: true },
    )?;
    stream.flush()?;
    let _ = stream.shutdown(Shutdown::Write);
    if expect_shallow_info {
        read_upload_pack_shallow_info_and_raw_packfile_response(format, &mut stream)
    } else {
        Ok((
            Vec::new(),
            read_upload_pack_raw_packfile_response(format, &mut stream)?,
        ))
    }
}

pub(crate) fn plan_push_git(request: GitPushRequest<'_>) -> Result<GitPushPlan> {
    let GitPushRequest {
        git_dir,
        common_git_dir,
        format,
        remote,
        refspecs,
        force,
    } = request;
    let (stream, advertisements, features) = receive_pack_advertisements(remote, format)?;

    let local_store = FileRefStore::new(git_dir, format);
    let local_refs = crate::push::local_push_source_refs(&local_store, format)?;
    let parsed_refspecs = refspecs
        .iter()
        .map(|refspec| parse_refspec(&crate::push::normalize_push_refspec(refspec)))
        .collect::<Result<Vec<_>>>()?;
    let mut command_forces = Vec::new();
    for refspec in &parsed_refspecs {
        for command in plan_push_commands(
            format,
            &local_refs,
            &advertisements,
            std::slice::from_ref(refspec),
        )? {
            command_forces.push((command, force || refspec.force));
        }
    }
    let commands = command_forces
        .iter()
        .map(|(command, _)| command.clone())
        .collect::<Vec<_>>();
    if commands.is_empty() {
        let _ = stream.shutdown(Shutdown::Both);
        return Ok(GitPushPlan {
            commands,
            pack_objects: Vec::new(),
            stream: None,
            features,
            advertisements,
        });
    }

    let local_db = FileObjectDatabase::from_git_dir(common_git_dir, format);
    crate::push::reject_non_fast_forward_pushes(&local_db, format, &command_forces)?;
    Ok(GitPushPlan {
        commands,
        pack_objects: Vec::new(),
        stream: Some(stream),
        features,
        advertisements,
    })
}

pub(crate) fn plan_push_git_commands(request: GitPushCommandsRequest<'_>) -> Result<GitPushPlan> {
    let GitPushCommandsRequest {
        common_git_dir,
        format,
        remote,
        command_forces,
        pack_objects,
    } = request;
    let (stream, advertisements, features) = receive_pack_advertisements(remote, format)?;
    let commands = command_forces
        .iter()
        .map(|(command, _)| command.clone())
        .collect::<Vec<_>>();
    if commands.is_empty() {
        let _ = stream.shutdown(Shutdown::Both);
        return Ok(GitPushPlan {
            commands,
            pack_objects,
            stream: None,
            features,
            advertisements,
        });
    }

    let local_db = FileObjectDatabase::from_git_dir(common_git_dir, format);
    crate::push::reject_non_fast_forward_pushes(&local_db, format, &command_forces)?;
    Ok(GitPushPlan {
        commands,
        pack_objects,
        stream: Some(stream),
        features,
        advertisements,
    })
}

pub(crate) fn execute_push_git_plan(
    request: PushRequest<'_>,
    mut plan: GitPushPlan,
) -> Result<PushOutcome> {
    if plan.commands.is_empty() {
        return Ok(PushOutcome::default());
    }
    let mut stream = plan
        .stream
        .take()
        .ok_or_else(|| GitError::Command("git:// receive-pack stream was not available".into()))?;
    let commands = plan.commands.clone();
    let local_db = FileObjectDatabase::from_git_dir(request.common_git_dir, request.format);
    let remote_excluded_tips =
        crate::remote_advertisement_tips_known_to_local(&local_db, &plan.advertisements)?;
    let remote_excluded =
        collect_reachable_object_ids(&local_db, request.format, remote_excluded_tips)?;
    let starts = crate::pack::push_pack_roots(&commands, &plan.pack_objects);
    let packfile = build_reachable_pack(&local_db, request.format, starts, &remote_excluded)?
        .map(|pack| pack.pack)
        .unwrap_or_default();
    let receive_request = build_receive_pack_push_request(
        &plan.features,
        commands.clone(),
        packfile,
        ReceivePackPushRequestOptions {
            report_status: plan.features.report_status,
            ofs_delta: plan.features.ofs_delta,
            quiet: request.options.quiet && plan.features.quiet,
            object_format: plan
                .features
                .object_format
                .filter(|_| request.format != ObjectFormat::Sha1),
            ..ReceivePackPushRequestOptions::default()
        },
    )?;
    write_receive_pack_push_request(&mut stream, &receive_request)?;
    stream.flush()?;
    let _ = stream.shutdown(Shutdown::Write);

    let report = if plan.features.report_status {
        let report = read_receive_pack_report_status(&mut stream)?;
        crate::push::validate_receive_pack_report(&report)?;
        Some(report)
    } else {
        let mut sink = Vec::new();
        stream.read_to_end(&mut sink)?;
        None
    };
    Ok(PushOutcome { commands, report })
}

fn receive_pack_advertisements(
    remote: &RemoteUrl,
    format: ObjectFormat,
) -> Result<(TcpStream, Vec<RefAdvertisement>, ReceivePackFeatures)> {
    let mut stream = connect_git_service(remote, GitService::ReceivePack)?;
    let advertisement_set = read_ref_advertisement_set(format, &mut stream)?;
    let features = advertisement_set
        .refs
        .first()
        .map(|advertisement| parse_receive_pack_features(&advertisement.capabilities))
        .transpose()?
        .unwrap_or_default();
    if let Some(remote_format) = features.object_format {
        if remote_format != format {
            return Err(GitError::InvalidObjectId(format!(
                "remote repository uses {}, local repository uses {}",
                remote_format.name(),
                format.name()
            )));
        }
    } else if format != ObjectFormat::Sha1 {
        return Err(GitError::InvalidObjectId(format!(
            "remote repository did not advertise object-format for {} push",
            format.name()
        )));
    }
    Ok((stream, advertisement_set.refs, features))
}

fn connect_git_service(remote: &RemoteUrl, service: GitService) -> Result<TcpStream> {
    if remote.transport != RemoteTransport::Git {
        return Err(GitError::InvalidFormat(
            "git:// service requires a git remote".into(),
        ));
    }
    let host = remote
        .host
        .as_deref()
        .ok_or_else(|| GitError::InvalidFormat("git:// remote is missing a host".into()))?;
    let port = remote.port.unwrap_or(GIT_DAEMON_PORT);
    let mut stream = TcpStream::connect((host, port))?;
    let request = ServiceRequest {
        service,
        path: remote.path.clone(),
        host: Some(git_host_parameter(remote, host, port)),
        parameters: Vec::new(),
        protocol: None,
        extra_parameters: Vec::new(),
    };
    write_service_request(&mut stream, &request)?;
    stream.flush()?;
    Ok(stream)
}

fn git_host_parameter(remote: &RemoteUrl, host: &str, port: u16) -> String {
    match remote.port {
        Some(_) if host.contains(':') && !host.starts_with('[') => format!("[{host}]:{port}"),
        Some(_) => format!("{host}:{port}"),
        None => host.to_string(),
    }
}

fn shallow_request_capabilities(deepen: Option<u32>) -> Vec<Capability> {
    if deepen.is_some() {
        vec![Capability {
            name: "shallow".into(),
            value: None,
        }]
    } else {
        Vec::new()
    }
}

fn all_wants_present(db: &FileObjectDatabase, wants: &[ObjectId]) -> Result<bool> {
    for want in wants {
        if !db.contains(want)? {
            return Ok(false);
        }
    }
    Ok(true)
}

#[cfg(test)]
mod tests {
    use super::*;
    use std::net::TcpListener;
    use std::thread;

    use sley_protocol::{ProtocolVersion, RefAdvertisement, RefAdvertisementSet};
    use sley_transport::read_service_request;

    #[test]
    fn ls_remote_git_sends_daemon_request_and_reads_advertisements() {
        let listener = TcpListener::bind(("127.0.0.1", 0)).expect("bind daemon");
        let port = listener.local_addr().expect("local addr").port();
        let tip = ObjectId::from_hex(
            ObjectFormat::Sha1,
            "1111111111111111111111111111111111111111",
        )
        .expect("oid");
        let server = thread::spawn(move || {
            let (mut stream, _) = listener.accept().expect("accept daemon client");
            let request = read_service_request(&mut stream).expect("service request");
            assert_eq!(request.service, GitService::UploadPack);
            assert_eq!(request.path, "/repo.git");
            let expected_host = format!("127.0.0.1:{port}");
            assert_eq!(request.host.as_deref(), Some(expected_host.as_str()));
            sley_protocol::write_ref_advertisement_set(
                &mut stream,
                &RefAdvertisementSet {
                    protocol: ProtocolVersion::V0,
                    refs: vec![
                        RefAdvertisement {
                            oid: tip,
                            name: "HEAD".into(),
                            capabilities: vec![Capability {
                                name: "symref".into(),
                                value: Some("HEAD:refs/heads/main".into()),
                            }],
                        },
                        RefAdvertisement {
                            oid: tip,
                            name: "refs/heads/main".into(),
                            capabilities: Vec::new(),
                        },
                    ],
                    shallow: Vec::new(),
                },
            )
            .expect("write advertisements");
        });
        let remote = RemoteUrl {
            transport: RemoteTransport::Git,
            user: None,
            password: None,
            host: Some("127.0.0.1".into()),
            port: Some(port),
            path: "/repo.git".into(),
        };
        let (records, format) = ls_remote_git(
            &remote,
            &crate::ls_remote::LsRemoteFilter::default(),
            &|_| true,
        )
        .expect("ls-remote");

        server.join().expect("server thread");
        assert_eq!(format, ObjectFormat::Sha1);
        assert_eq!(records.len(), 2);
        assert_eq!(records[0].name, "HEAD");
        assert_eq!(records[0].symref.as_deref(), Some("refs/heads/main"));
        assert_eq!(records[1].name, "refs/heads/main");
        assert_eq!(records[1].oid, tip);
    }
}