//! Single-ref receive-pack over standard WASI HTTP or in-process SSH exec channels.

use crate::Result;
use crate::{
    http::{WasiHttp, clean_url},
    remotes, repo,
    repo::error,
};
use gix::bstr::ByteSlice;
use gix_transport::client::blocking_io::http::{Http, PostBodyDataKind};
use std::{
    collections::{BTreeMap, BTreeSet},
    io::{Read, Write},
};

/// Push one branch or explicit single branch refspec over HTTPS or the SSH exec transport.
#[derive(serde::Serialize, serde::Deserialize)]
pub struct PushParams {
    /// Plain name under /repos (a-z0-9._-, no '..'), or absolute path under /plugins/ or /repos/ (no empty, '.' or '..' segments or trailing slash).
    pub repo: String,
    /// Configured remote, default origin (pushurl takes precedence).
    pub remote: Option<String>,
    /// Short local branch, default current branch.
    pub branch: Option<String>,
    /// Single branch refspec: source:destination (including HEAD:destination), source, or :destination to delete. No wildcards.
    pub refspec: Option<String>,
    /// Allow non-fast-forward replacement. A leading + in refspec also enables this.
    pub force: Option<bool>,
    /// Set source branch upstream to the remote destination after success.
    pub set_upstream: Option<bool>,
    /// Delete the destination instead of sending objects (default false).
    pub delete: Option<bool>,
}
/// Confirmed remote update; local tracking state is updated only after report-status success.
#[derive(serde::Serialize, serde::Deserialize)]
pub struct PushResult {
    /// Remote used.
    pub remote: String,
    /// Full destination ref.
    pub reference: String,
    /// Advertised old object ID, or null for a new ref.
    pub old: Option<String>,
    /// New object ID, or null for deletion.
    pub new: Option<String>,
    /// Whether the server already had the requested value.
    pub up_to_date: bool,
    /// Number of objects sent (no delta compression).
    pub objects: usize,
}
fn branch_ref(name: &str) -> Result<String> {
    let full = if name.starts_with("refs/") {
        name.to_owned()
    } else {
        format!("refs/heads/{name}")
    };
    if name.is_empty() || name.starts_with('-') || !full.starts_with("refs/heads/") {
        return Err(error(
            "push refspec",
            "expected a branch name or refs/heads/<name>; tags and wildcard/multi-ref pushes are not supported",
        ));
    }
    let _: gix::refs::FullName = full
        .as_str()
        .try_into()
        .map_err(|e| error("push refspec", e))?;
    Ok(full)
}
pub(crate) struct Plan {
    pub source: Option<String>,
    pub destination: String,
    pub new: Option<gix::ObjectId>,
    pub force: bool,
}
pub(crate) fn plan(repo: &gix::Repository, p: &PushParams) -> Result<Plan> {
    let mut force = p.force.unwrap_or(false);
    let (source, destination, deleting) = if let Some(spec) = &p.refspec {
        if p.branch.is_some() {
            return Err(error("push", "branch and refspec are mutually exclusive"));
        }
        let spec = if let Some(s) = spec.strip_prefix('+') {
            force = true;
            s
        } else {
            spec.as_str()
        };
        let (src, dst) = spec.split_once(':').unwrap_or((spec, spec));
        if dst.is_empty() || dst.contains(':') || spec.contains('*') {
            return Err(error(
                "push refspec",
                "expected one source:destination or :destination",
            ));
        }
        let deleting = p.delete.unwrap_or(false) || src.is_empty();
        let source = if deleting {
            None
        } else if src == "HEAD" {
            Some(
                crate::head_info(repo)?
                    .branch
                    .unwrap_or_else(|| "HEAD".into()),
            )
        } else {
            Some(branch_ref(src)?)
        };
        (source, branch_ref(dst)?, deleting)
    } else {
        let source = match &p.branch {
            Some(branch) => branch_ref(branch)?,
            None => crate::head_info(repo)?
                .branch
                .ok_or_else(|| error("push", "detached HEAD: specify branch or refspec"))?,
        };
        let deleting = p.delete.unwrap_or(false);
        (
            if deleting { None } else { Some(source.clone()) },
            source,
            deleting,
        )
    };
    if deleting && p.set_upstream.unwrap_or(false) {
        return Err(error("push", "set_upstream cannot be used with deletion"));
    }
    if source.as_deref() == Some("HEAD") && p.set_upstream.unwrap_or(false) {
        return Err(error(
            "push upstream",
            "detached HEAD has no branch to set upstream for",
        ));
    }
    let new = source
        .as_deref()
        .map(|src| -> Result<gix::ObjectId> {
            let id = if src == "HEAD" {
                repo.head_id()
                    .map_err(|e| error("push source", e))?
                    .detach()
            } else {
                let reference = repo
                    .find_reference(src)
                    .map_err(|e| error("push source", e))?;
                reference
                    .try_id()
                    .ok_or_else(|| error("push source", "source must directly name a commit"))?
                    .detach()
            };
            repo.find_commit(id)
                .map_err(|e| error("push source commit", e))?;
            Ok(id)
        })
        .transpose()?;
    Ok(Plan {
        source,
        destination,
        new,
        force,
    })
}

// pkt-line readers reject truncation and reserved lengths rather than treating missing status as success.
pub(crate) fn pkt(out: &mut impl Write, data: &[u8]) -> Result<()> {
    if data.len() > 65516 {
        return Err(error("push packet", "packet too large"));
    }
    write!(out, "{:04x}", data.len() + 4)
        .and_then(|_| out.write_all(data))
        .map_err(|e| error("push packet", e))
}
fn packet(input: &mut impl Read) -> Result<Option<Vec<u8>>> {
    let mut header = [0; 4];
    input
        .read_exact(&mut header)
        .map_err(|e| error("push packet header", e))?;
    let len = std::str::from_utf8(&header)
        .ok()
        .and_then(|s| u16::from_str_radix(s, 16).ok())
        .ok_or_else(|| error("push packet", "invalid length"))? as usize;
    if len == 0 {
        return Ok(None);
    }
    if !(4..=65520).contains(&len) {
        return Err(error("push packet", "invalid packet length"));
    }
    let mut bytes = vec![0; len - 4];
    input
        .read_exact(&mut bytes)
        .map_err(|e| error("push packet body", e))?;
    Ok(Some(bytes))
}
pub(crate) struct Advertisement {
    pub refs: BTreeMap<String, gix::ObjectId>,
    pub caps: BTreeSet<String>,
}
fn advertise(input: &mut impl Read) -> Result<Advertisement> {
    if packet(input)?.as_deref() != Some(b"# service=git-receive-pack\n")
        || packet(input)?.is_some()
    {
        return Err(error(
            "push advertisement",
            "expected smart HTTP git-receive-pack service",
        ));
    }
    advertise_refs(input)
}
/// SSH hook: git-receive-pack exec channels begin directly with these ref packets,
/// without the smart-HTTP service prelude. commands(), pack(), and report() are also transport-neutral.
pub(crate) fn advertise_refs(input: &mut impl Read) -> Result<Advertisement> {
    let mut refs = BTreeMap::new();
    let mut caps = BTreeSet::new();
    let mut first = true;
    let mut count = 0;
    while let Some(line) = packet(input)? {
        count += 1;
        if count > 100_000 {
            return Err(error("push advertisement", "too many refs"));
        }
        let text = std::str::from_utf8(&line)
            .map_err(|e| error("push advertisement", e))?
            .trim_end_matches('\n');
        if let Some(message) = text.strip_prefix("ERR ") {
            return Err(error("push remote", message));
        }
        if text.starts_with("shallow ") {
            return Err(error(
                "push advertisement",
                "shallow remotes are not supported",
            ));
        }
        let (reference, capabilities) = text
            .split_once('\0')
            .map(|(a, b)| (a, Some(b)))
            .unwrap_or((text, None));
        if let Some(capabilities) = capabilities {
            if !first {
                return Err(error("push advertisement", "capabilities after first ref"));
            }
            caps.extend(capabilities.split_ascii_whitespace().map(str::to_owned));
        }
        first = false;
        let (oid, name) = reference
            .split_once(' ')
            .ok_or_else(|| error("push advertisement", "invalid ref line"))?;
        let id =
            gix::ObjectId::from_hex(oid.as_bytes()).map_err(|e| error("push advertisement", e))?;
        if id.kind() != gix::hash::Kind::Sha1 {
            return Err(error("push", "only SHA-1 repositories supported"));
        }
        if !id.is_null() {
            refs.insert(name.to_owned(), id);
        }
    }
    if !caps.contains("report-status") {
        return Err(error("push", "remote does not support report-status"));
    }
    if caps.contains("object-format=sha256") {
        return Err(error("push", "SHA-256 remote is unsupported"));
    }
    Ok(Advertisement { refs, caps })
}
pub(crate) fn commands(
    old: Option<gix::ObjectId>,
    plan: &Plan,
    caps: &[&str],
    out: &mut impl Write,
) -> Result<()> {
    let zero = gix::hash::Kind::Sha1.null();
    let command = format!(
        "{} {} {}\0{}\n",
        old.unwrap_or(zero),
        plan.new.unwrap_or(zero),
        plan.destination,
        caps.join(" ")
    );
    pkt(out, command.as_bytes())?;
    out.write_all(b"0000")
        .map_err(|e| error("push commands", e))
}
/// Unknown remote tips are not assumed to be ancestors. Force is required until fetched.
pub(crate) fn is_ancestor(
    repo: &gix::Repository,
    old: gix::ObjectId,
    new: gix::ObjectId,
) -> Result<bool> {
    if old == new {
        return Ok(true);
    }
    if repo
        .try_find_object(old)
        .map_err(|e| error("push ancestry", e))?
        .is_none()
    {
        return Ok(false);
    }
    let walk = repo
        .find_commit(new)
        .map_err(|e| error("push ancestry", e))?
        .ancestors()
        .use_commit_graph(false)
        .all()
        .map_err(|e| error("push ancestry", e))?;
    for item in walk {
        if item.map_err(|e| error("push ancestry", e))?.id == old {
            return Ok(true);
        }
    }
    Ok(false)
}
/// Walk the object graph once, stopping BEFORE decoding seen objects or remote boundaries.
/// Returns the number of objects decoded, also used by the traversal-cost regression tests.
/// The boundary must be transitively closed (the reachable remote object set).
/// Gitlink commits belong to another repository and are never followed.
pub(crate) fn reachable(
    repo: &gix::Repository,
    tip: gix::ObjectId,
    seen: &mut BTreeSet<gix::ObjectId>,
    boundary: &BTreeSet<gix::ObjectId>,
) -> Result<usize> {
    let mut pending = vec![tip];
    let mut decoded = 0;
    while let Some(id) = pending.pop() {
        if boundary.contains(&id) || !seen.insert(id) {
            continue;
        }
        let obj = repo
            .find_object(id)
            .map_err(|e| error("push reachable object", e))?;
        decoded += 1;
        match obj.kind {
            gix::objs::Kind::Commit => {
                let commit = obj.try_into_commit().map_err(|e| error("push commit", e))?;
                pending.push(
                    commit
                        .tree_id()
                        .map_err(|e| error("push tree", e))?
                        .detach(),
                );
                pending.extend(commit.parent_ids().map(|id| id.detach()));
            }
            // Follow one tag at a time, including intermediate annotated tags in seen.
            gix::objs::Kind::Tag => {
                let tag = obj
                    .try_into_tag()
                    .map_err(|e| error("push remote tag", e))?;
                pending.push(
                    tag.target_id()
                        .map_err(|e| error("push remote tag target", e))?
                        .detach(),
                );
            }
            gix::objs::Kind::Tree => {
                let tree = obj.try_into_tree().map_err(|e| error("push tree", e))?;
                for entry in tree.iter() {
                    let entry = entry.map_err(|e| error("push tree entry", e))?;
                    if entry.mode().is_tree() {
                        pending.push(entry.object_id());
                    } else if !entry.mode().is_commit() && !boundary.contains(&entry.object_id()) {
                        seen.insert(entry.object_id());
                    }
                }
            }
            gix::objs::Kind::Blob => {}
        }
    }
    Ok(decoded)
}
pub(crate) fn objects_to_send(
    repo: &gix::Repository,
    new: gix::ObjectId,
    remote: impl IntoIterator<Item = gix::ObjectId>,
) -> Result<BTreeSet<gix::ObjectId>> {
    let mut theirs = BTreeSet::new();
    let empty = BTreeSet::new();
    for id in remote {
        if !theirs.contains(&id)
            && repo
                .try_find_object(id)
                .map_err(|e| error("push remote object", e))?
                .is_some()
        {
            reachable(repo, id, &mut theirs, &empty)?;
        }
    }
    let mut ours = BTreeSet::new();
    reachable(repo, new, &mut ours, &theirs)?;
    Ok(ours)
}
/// Compress one object at a time and stream pack bytes to the HTTP request spool.
pub(crate) fn pack(
    repo: &gix::Repository,
    ids: &BTreeSet<gix::ObjectId>,
    out: &mut impl Write,
) -> Result<()> {
    use gix_pack::data::output::{Count, Entry, bytes::FromEntriesIter};
    let count = u32::try_from(ids.len()).map_err(|e| error("push pack count", e))?;
    let entries = ids.iter().map(|id| {
        let object = repo.find_object(*id)?;
        Entry::from_data(
            &Count::from_data(*id, None),
            &gix::objs::Data {
                kind: object.kind,
                data: &object.data,
                object_hash: gix::hash::Kind::Sha1,
            },
            gix_zlib::Compression::default(),
        )
        .map(|entry| vec![entry])
    });
    for written in FromEntriesIter::new(
        entries,
        out,
        count,
        gix_pack::data::Version::V2,
        gix::hash::Kind::Sha1,
    ) {
        written.map_err(|e| error("push pack", format!("{e:#}")))?;
    }
    Ok(())
}
fn parse_status(input: &mut impl Read, reference: &str) -> Result<()> {
    let unpack = packet(input)?.ok_or_else(|| error("push status", "missing unpack status"))?;
    let unpack = String::from_utf8_lossy(&unpack);
    if unpack.trim_end() != "unpack ok" {
        return Err(error("push remote", unpack.trim_end()));
    }
    let mut confirmed = false;
    while let Some(line) = packet(input)? {
        let line = String::from_utf8_lossy(&line);
        let line = line.trim_end();
        if line == format!("ok {reference}") && !confirmed {
            confirmed = true;
        } else if let Some(message) = line.strip_prefix("ng ") {
            return Err(error("push rejected", message));
        } else {
            return Err(error("push status", format!("unexpected status: {line}")));
        }
    }
    if !confirmed {
        return Err(error(
            "push status",
            "remote did not confirm the destination ref",
        ));
    }
    Ok(())
}
// Hook explanations often arrive on band 2, with only a generic reason in ng.
// Keep a bounded byte tail (lossy UTF-8 conversion happens only when rendering it).
const REMOTE_DETAIL_LIMIT: usize = 8192;
fn with_remote_detail(result: Result<()>, tail: &[u8]) -> Result<()> {
    result.map_err(|mut e| {
        let detail = String::from_utf8_lossy(tail);
        if !detail.trim().is_empty() {
            e.message.push_str(&format!("\nremote: {}", detail.trim()));
        }
        e
    })
}
pub(crate) fn report(input: &mut impl Read, reference: &str, sideband: bool) -> Result<()> {
    if !sideband {
        return parse_status(input, reference);
    }
    let mut status = Vec::new();
    let mut remote_detail = Vec::new();
    while let Some(line) = packet(input)? {
        match line.split_first() {
            Some((1, bytes)) => {
                if status.len() + bytes.len() > 1024 * 1024 {
                    return Err(error("push status", "report-status too large"));
                }
                status.extend_from_slice(bytes);
            }
            Some((2, bytes)) => {
                remote_detail.extend_from_slice(bytes);
                if remote_detail.len() > REMOTE_DETAIL_LIMIT {
                    remote_detail.drain(..remote_detail.len() - REMOTE_DETAIL_LIMIT);
                }
            }
            Some((3, bytes)) => {
                return with_remote_detail(
                    Err(error(
                        "push remote",
                        String::from_utf8_lossy(bytes).trim_end(),
                    )),
                    &remote_detail,
                );
            }
            _ => return Err(error("push sideband", "invalid channel")),
        }
    }
    with_remote_detail(
        parse_status(&mut status.as_slice(), reference),
        &remote_detail,
    )
}
fn update_local(repo: &gix::Repository, remote: &str, plan: &Plan, upstream: bool) -> Result<()> {
    let short = plan
        .destination
        .strip_prefix("refs/heads/")
        .ok_or_else(|| error("push tracking", "not a branch"))?;
    let tracking = format!("refs/remotes/{remote}/{short}");
    use gix::refs::{
        Target,
        transaction::{Change, PreviousValue, RefEdit},
    };
    if let Some(new) = plan.new {
        repo.edit_references_as(
            [RefEdit {
                name: tracking
                    .as_str()
                    .try_into()
                    .map_err(|e| error("push tracking", e))?,
                deref: false,
                change: Change::Update {
                    log: Default::default(),
                    expected: PreviousValue::Any,
                    new: Target::Object(new),
                },
            }],
            None,
        )
        .map_err(|e| error("push tracking", e))?;
    } else if let Some(reference) = repo
        .try_find_reference(&tracking)
        .map_err(|e| error("push tracking", e))?
    {
        reference.delete().map_err(|e| error("push tracking", e))?;
    }
    if upstream {
        let branch = plan
            .source
            .as_deref()
            .and_then(|s| s.strip_prefix("refs/heads/"))
            .ok_or_else(|| error("push upstream", "missing source branch"))?;
        let mut config = remotes::load(repo)?;
        let ids: Vec<_> = config
            .sections_and_ids_by_name("branch")
            .into_iter()
            .flatten()
            .filter(|(s, _)| {
                s.header()
                    .subsection_name()
                    .is_some_and(|s| s == branch.as_bytes())
            })
            .map(|(_, id)| id)
            .collect();
        for id in ids {
            if let Some(mut section) = config.section_mut_by_id(id) {
                while section.remove("remote").is_some() {}
                while section.remove("merge").is_some() {}
            }
        }
        let mut section = config
            .section_mut_or_create_new("branch", branch)
            .map_err(|e| error("push upstream", e))?;
        section
            .push("remote", Some(remote.into()))
            .map_err(|e| error("push upstream", e))?;
        section
            .push("merge", Some(plan.destination.as_str().into()))
            .map_err(|e| error("push upstream", e))?;
        remotes::save(repo, &config)?;
    }
    Ok(())
}
/// Send one reference update through the smart-HTTP receive-pack service.
pub fn push(p: PushParams) -> Result<PushResult> {
    let mut repo = repo::open(&p.repo)?;
    repo.refs.write_reflog = gix::refs::store::WriteReflog::Disable;
    let remote = p.remote.as_deref().unwrap_or("origin");
    remotes::validate_name(remote)?;
    let plan = plan(&repo, &p)?;
    if plan.new.is_some()
        && repo
            .path()
            .join("shallow")
            .try_exists()
            .map_err(|e| error("push shallow", e))?
    {
        return Err(error(
            "push",
            "shallow pushes are not supported; fetch full history first",
        ));
    }
    let configured = repo
        .find_remote(remote)
        .map_err(|_| error("push", "configured remote not found"))?;
    let url = configured
        .url(gix::remote::Direction::Push)
        .ok_or_else(|| error("push", "remote has no push URL"))?
        .to_bstring()
        .to_str_lossy()
        .into_owned();
    if crate::ssh::is_ssh(&url) {
        return ssh_push(&repo, &url, remote, &plan, p.set_upstream.unwrap_or(false));
    }
    let base = clean_url(&url)
        .map_err(|e| error("push URL", e))?
        .to_string()
        .trim_end_matches('/')
        .to_owned();
    let mut http = WasiHttp::default();

    let result = (|| {
        let mut discovery = http
            .get(
                &format!("{base}/info/refs?service=git-receive-pack"),
                &base,
                ["Accept: application/x-git-receive-pack-advertisement"],
            )
            .map_err(|e| error("push discovery", format!("{e:#}")))?;
        let advertised = advertise(&mut discovery.body)?;
        drop(discovery);
        let old = advertised.refs.get(&plan.destination).copied();
        if let (Some(old), Some(new)) = (old, plan.new) {
            if old != new && !plan.force && !is_ancestor(&repo, old, new)? {
                return Err(error(
                    "push",
                    "non-fast-forward (or remote tip unknown locally); fetch first or use force",
                ));
            }
        }
        let up_to_date = old == plan.new;
        let mut objects = 0;
        if !up_to_date {
            if plan.new.is_none() && !advertised.caps.contains("delete-refs") {
                return Err(error("push", "remote does not allow deleting refs"));
            }
            let ids = match plan.new {
                Some(new) => objects_to_send(&repo, new, advertised.refs.values().copied())?,
                None => BTreeSet::new(),
            };
            objects = ids.len();
            let sideband = advertised.caps.contains("side-band-64k");
            let mut caps = vec!["report-status"];
            if sideband {
                caps.push("side-band-64k");
            }
            if advertised.caps.iter().any(|c| c.starts_with("agent=")) {
                caps.push("agent=sausage-git/0.1");
            }
            let base = http.redirected_base_url().unwrap_or(base);
            let mut post = http
                .post(
                    &format!("{base}/git-receive-pack"),
                    &base,
                    [
                        "Content-Type: application/x-git-receive-pack-request",
                        "Accept: application/x-git-receive-pack-result",
                    ],
                    PostBodyDataKind::Unbounded,
                )
                .map_err(|e| error("push request", format!("{e:#}")))?;
            commands(old, &plan, &caps, &mut post.post_body)?;
            if plan.new.is_some() {
                pack(&repo, &ids, &mut post.post_body)?;
            }
            drop(post.post_body);

            report(&mut post.body, &plan.destination, sideband)?;
        }
        update_local(&repo, remote, &plan, p.set_upstream.unwrap_or(false)).map_err(|e| {
            error(
                "push succeeded remotely, but local state update failed",
                e.message,
            )
        })?;

        Ok(PushResult {
            remote: remote.into(),
            reference: plan.destination,
            old: old.map(|id| id.to_string()),
            new: plan.new.map(|id| id.to_string()),
            up_to_date,
            objects,
        })
    })();
    result
}
fn ssh_push(
    repo: &gix::Repository,
    url: &str,
    remote: &str,
    plan: &Plan,
    upstream: bool,
) -> Result<PushResult> {
    let (mut read, mut write) =
        crate::ssh::connection(url, gix_transport::Service::ReceivePack)?.into_inner();
    let advertised = advertise_refs(&mut read)?;
    let old = advertised.refs.get(&plan.destination).copied();
    if let (Some(old), Some(new)) = (old, plan.new) {
        if old != new && !plan.force && !is_ancestor(repo, old, new)? {
            return Err(error("push", "non-fast-forward; fetch first or use force"));
        }
    }
    let up_to_date = old == plan.new;
    let mut objects = 0;
    if !up_to_date {
        if plan.new.is_none() && !advertised.caps.contains("delete-refs") {
            return Err(error("push", "remote does not allow deleting refs"));
        }
        let ids = match plan.new {
            Some(new) => objects_to_send(repo, new, advertised.refs.values().copied())?,
            None => BTreeSet::new(),
        };
        objects = ids.len();
        let sideband = advertised.caps.contains("side-band-64k");
        let mut caps = vec!["report-status"];
        if sideband {
            caps.push("side-band-64k");
        }
        commands(old, plan, &caps, &mut write)?;
        if plan.new.is_some() {
            pack(repo, &ids, &mut write)?;
        }
        write.flush().map_err(|e| error("push SSH flush", e))?;
        report(&mut read, &plan.destination, sideband)?;
    } else {
        write
            .write_all(b"0000")
            .and_then(|_| write.flush())
            .map_err(|e| error("push SSH finish", e))?;
    }
    update_local(repo, remote, plan, upstream)?;
    Ok(PushResult {
        remote: remote.into(),
        reference: plan.destination.clone(),
        old: old.map(|id| id.to_string()),
        new: plan.new.map(|id| id.to_string()),
        up_to_date,
        objects,
    })
}