//! Smart-HTTP routes: info/refs, git-upload-pack and git-receive-pack.
//! Streams bodies to and from `git <svc> --stateless-rpc`; gzip request bodies are decoded.
//!
//! Compression must be off for these responses. Topcoat's compression is configured only on the
//! router builder (`Router::builder().compression(Compression::off())`), so boot.rs has to set it.

use std::net::IpAddr;
use std::time::Instant;

use async_compression::tokio::bufread::GzipDecoder;
use forge_core::User;
use forge_git::{GitService, pktline};
use futures_util::TryStreamExt;
use http::header::{
    AUTHORIZATION, CACHE_CONTROL, CONTENT_ENCODING, CONTENT_TYPE, HeaderValue, RETRY_AFTER,
    WWW_AUTHENTICATE,
};
use http_body::Frame;
use http_body_util::StreamBody;
use tokio::io::AsyncReadExt;
use tokio::process::{Child, ChildStdin};
use tokio_util::io::{ReaderStream, StreamReader};
use topcoat::context::{Cx, app_context};
use topcoat::router::request::headers;
use topcoat::router::response::Response;
use topcoat::router::{Body, HeaderMap, StatusCode, path_param, query_params, route};

use crate::access::{AccessError, Transport, find_repo, git_process, resolve_repo};
use crate::client_ip::client_ip;
use crate::git_auth::{parse_basic_auth, user_from_basic};
use crate::state::AppState;
use crate::throttle::{self, GIT_IP_LIMIT};

path_param!(owner);
path_param!(repo);

#[query_params(error = bad_request)]
struct InfoRefsQuery {
    service: Option<String>,
}

#[route(GET "/{owner}/{repo}/info/refs")]
async fn info_refs(cx: &Cx) -> topcoat::Result<Response> {
    let query = query_params::<InfoRefsQuery>(cx)?;
    let Some(service) = query.service.as_deref().and_then(GitService::from_name) else {
        return Ok(text(StatusCode::FORBIDDEN, "dumb protocol not supported"));
    };

    let state = app_context::<AppState>(cx);
    let headers = headers(cx);
    let owner = path_param::<Owner>(cx);
    let repo_name = path_param::<Repo>(cx);

    let (actor, repo, path) =
        match caller_repo(state, client_ip(cx), headers, owner, repo_name, service).await {
            Ok(found) => found,
            Err(response) => return Ok(*response),
        };

    let protocol = git_protocol(headers);
    let advertisement = git_process(
        state,
        &repo,
        &path,
        actor.as_ref(),
        service,
        Transport::Http,
        true,
        protocol,
    )
    .output()
    .await;
    let refs = match advertisement {
        Ok(refs) => refs,
        Err(e) => {
            tracing::error!(error = %e, service = service.name(), "info/refs failed");
            return Ok(internal_error());
        }
    };

    let mut payload = Vec::with_capacity(refs.len() + 64);
    if needs_service_header(service, wants_v2(protocol)) {
        payload.extend(pktline::service_advertisement_header(service));
    }
    payload.extend(refs);

    Ok(respond(
        StatusCode::OK,
        advertisement_content_type(service),
        Body::from(payload),
    ))
}

#[route(POST "/{owner}/{repo}/git-upload-pack")]
async fn upload_pack(cx: &Cx, body: Body) -> topcoat::Result<Response> {
    Ok(serve_rpc(cx, GitService::UploadPack, body).await)
}

#[route(POST "/{owner}/{repo}/git-receive-pack")]
async fn receive_pack(cx: &Cx, body: Body) -> topcoat::Result<Response> {
    Ok(serve_rpc(cx, GitService::ReceivePack, body).await)
}

/// Shared body of the two POST routes: authorize, spawn `git <svc> --stateless-rpc`, pipe the
/// request body into its stdin and stream its stdout back.
async fn serve_rpc(cx: &Cx, service: GitService, body: Body) -> Response {
    let state = app_context::<AppState>(cx);
    let headers = headers(cx);
    let owner = path_param::<Owner>(cx);
    let repo_name = path_param::<Repo>(cx);

    let (actor, repo, path) =
        match caller_repo(state, client_ip(cx), headers, owner, repo_name, service).await {
            Ok(found) => found,
            Err(response) => return *response,
        };

    let gzip = match body_encoding(
        headers
            .get(CONTENT_ENCODING)
            .map(|v| v.to_str().unwrap_or("")),
    ) {
        Some(gzip) => gzip,
        None => {
            return text(
                StatusCode::UNSUPPORTED_MEDIA_TYPE,
                "unsupported content encoding",
            );
        }
    };

    let mut command = git_process(
        state,
        &repo,
        &path,
        actor.as_ref(),
        service,
        Transport::Http,
        false,
        git_protocol(headers),
    )
    .into_tokio();
    command
        .stdin(std::process::Stdio::piped())
        .stdout(std::process::Stdio::piped())
        .stderr(std::process::Stdio::piped());

    let mut child = match command.spawn() {
        Ok(child) => child,
        Err(e) => {
            tracing::error!(error = %e, service = service.name(), "failed to spawn git");
            return internal_error();
        }
    };
    let (Some(stdin), Some(stdout)) = (child.stdin.take(), child.stdout.take()) else {
        tracing::error!(service = service.name(), "git child is missing piped stdio");
        return internal_error();
    };

    tokio::spawn(pump_request(body, gzip, stdin));
    tokio::spawn(reap(child, service));

    let chunks = ReaderStream::new(stdout).map_ok(Frame::data);
    respond(
        StatusCode::OK,
        result_content_type(service),
        Body::new(StreamBody::new(chunks)),
    )
}

/// Copies the request body (decoding gzip when `gzip` is set) into git's stdin, then closes it.
async fn pump_request(body: Body, gzip: bool, mut stdin: ChildStdin) {
    let reader = StreamReader::new(body.into_data_stream().map_err(std::io::Error::other));
    let copied = if gzip {
        tokio::io::copy(&mut GzipDecoder::new(reader), &mut stdin).await
    } else {
        let mut reader = reader;
        tokio::io::copy(&mut reader, &mut stdin).await
    };
    if let Err(e) = copied {
        tracing::debug!(error = %e, "request body ended before git consumed it");
    }
    // Closing stdin tells git the request is complete.
    let _ = tokio::io::AsyncWriteExt::shutdown(&mut stdin).await;
}

/// Drains stderr into the log and waits for the child, so no zombie is left behind.
async fn reap(mut child: Child, service: GitService) {
    let mut stderr_buf = Vec::new();
    if let Some(mut stderr) = child.stderr.take() {
        let _ = stderr.read_to_end(&mut stderr_buf).await;
    }
    let stderr_text = String::from_utf8_lossy(&stderr_buf);
    match child.wait().await {
        Ok(status) if status.success() => {
            tracing::debug!(service = service.name(), "git finished");
        }
        Ok(status) => {
            tracing::warn!(
                service = service.name(),
                %status,
                stderr = %stderr_text.trim(),
                "git exited with failure"
            );
        }
        Err(e) => {
            tracing::warn!(service = service.name(), error = %e, "failed to wait for git");
        }
    }
}

/// Extracts a `Git-Protocol` header value only if it matches `^[a-z0-9=:,._-]{1,64}$`.
fn git_protocol(headers: &HeaderMap) -> Option<&str> {
    let value = headers.get("git-protocol")?.to_str().ok()?;
    valid_git_protocol(value).then_some(value)
}

fn valid_git_protocol(value: &str) -> bool {
    (1..=64).contains(&value.len())
        && value.bytes().all(|b| {
            b.is_ascii_lowercase()
                || b.is_ascii_digit()
                || matches!(b, b'=' | b':' | b',' | b'.' | b'_' | b'-')
        })
}

/// True when the protocol string asks for version 2.
fn wants_v2(protocol: Option<&str>) -> bool {
    protocol.is_some_and(|p| p.split(':').any(|kv| kv == "version=2"))
}

/// The `# service=` preamble is required for v0 advertisements. With v2 the upload-pack server
/// writes its own version line instead. Receive-pack has no v2, so it always gets the preamble.
fn needs_service_header(service: GitService, v2: bool) -> bool {
    !(service == GitService::UploadPack && v2)
}

/// Decodes the request body encoding. `Some(true)` for gzip, `Some(false)` for none or identity,
/// `None` for anything else.
fn body_encoding(value: Option<&str>) -> Option<bool> {
    match value.map(|v| v.trim().to_ascii_lowercase()) {
        None => Some(false),
        Some(v) if v == "gzip" => Some(true),
        Some(v) if v == "identity" => Some(false),
        Some(_) => None,
    }
}

fn advertisement_content_type(service: GitService) -> &'static str {
    match service {
        GitService::UploadPack => "application/x-git-upload-pack-advertisement",
        GitService::ReceivePack => "application/x-git-receive-pack-advertisement",
    }
}

fn result_content_type(service: GitService) -> &'static str {
    match service {
        GitService::UploadPack => "application/x-git-upload-pack-result",
        GitService::ReceivePack => "application/x-git-receive-pack-result",
    }
}

fn respond(status: StatusCode, content_type: &'static str, body: Body) -> Response {
    let mut res = Response::new(body);
    *res.status_mut() = status;
    res.headers_mut()
        .insert(CONTENT_TYPE, HeaderValue::from_static(content_type));
    res.headers_mut()
        .insert(CACHE_CONTROL, HeaderValue::from_static("no-cache"));
    res
}

fn text(status: StatusCode, message: &'static str) -> Response {
    let mut res = Response::new(Body::from(message));
    *res.status_mut() = status;
    res.headers_mut().insert(
        CONTENT_TYPE,
        HeaderValue::from_static("text/plain; charset=utf-8"),
    );
    res
}

fn internal_error() -> Response {
    text(StatusCode::INTERNAL_SERVER_ERROR, "internal server error")
}

/// Authenticates the request's credentials under the per-IP git failure limit. Only rejected
/// credentials count, so anonymous requests and successful logins never do. Returns the response
/// to send when the request must stop: 429 while the IP is throttled, or the access error.
pub(crate) async fn authenticate(
    state: &AppState,
    ip: IpAddr,
    headers: &HeaderMap,
) -> Result<Option<User>, Box<Response>> {
    let key = throttle::ip_key("git-ip", ip);
    let now = Instant::now();
    if let Some(retry) = state.throttle.check(&key, GIT_IP_LIMIT, now) {
        tracing::warn!(
            ip = %ip,
            retry_after_secs = retry.as_secs(),
            "git authentication throttled"
        );
        let mut res = text(
            StatusCode::TOO_MANY_REQUESTS,
            "too many failed attempts, try again later",
        );
        res.headers_mut()
            .insert(RETRY_AFTER, HeaderValue::from(retry.as_secs().max(1)));
        return Err(Box::new(res));
    }
    match user_from_basic(state, headers).await {
        Ok(actor) => Ok(actor),
        Err(AccessError::Unauthorized) if headers.contains_key(AUTHORIZATION) => {
            state.throttle.record_failure(&key, now);
            Err(Box::new(access_error(AccessError::Unauthorized)))
        }
        Err(error) => Err(Box::new(access_error(error))),
    }
}

/// Authorizes a smart-HTTP request. A CI job token (see [`job_token_read`]) is tried first;
/// otherwise the caller is a user or anonymous, and the normal rules apply. Returns the actor
/// (`None` for a job), the repo and its path.
async fn caller_repo(
    state: &AppState,
    ip: IpAddr,
    headers: &HeaderMap,
    owner: &str,
    name: &str,
    service: GitService,
) -> Result<(Option<User>, forge_core::Repo, std::path::PathBuf), Box<Response>> {
    if let Some(found) = job_token_read(state, ip, headers, owner, name, service).await {
        return found.map(|(repo, path)| (None, repo, path));
    }
    let actor = authenticate(state, ip, headers).await?;
    let (repo, path) = resolve_repo(state, actor.as_ref(), owner, name, service)
        .await
        .map_err(|e| Box::new(access_error(e)))?;
    Ok((actor, repo, path))
}

/// Username of a CI job's git credential. The password is `<job id>:<job token>`.
pub(crate) const JOB_GIT_USERNAME: &str = "ci-job";

/// Read-only git access for a running CI job. Returns `None` unless the request uses the
/// `ci-job` username, so other callers take the normal path. The token must belong to a job that
/// is Running (see `job_by_token`), the repo must be that job's repo, and receive-pack is refused.
/// Failed attempts count against the git failure limit.
pub(crate) async fn job_token_read(
    state: &AppState,
    ip: IpAddr,
    headers: &HeaderMap,
    owner: &str,
    name: &str,
    service: GitService,
) -> Option<Result<(forge_core::Repo, std::path::PathBuf), Box<Response>>> {
    let (username, password) = parse_basic_auth(headers)?;
    if !username.eq_ignore_ascii_case(JOB_GIT_USERNAME) {
        return None;
    }
    let key = throttle::ip_key("git-ip", ip);
    let now = Instant::now();
    if let Some(retry) = state.throttle.check(&key, GIT_IP_LIMIT, now) {
        let mut res = text(
            StatusCode::TOO_MANY_REQUESTS,
            "too many failed attempts, try again later",
        );
        res.headers_mut()
            .insert(RETRY_AFTER, HeaderValue::from(retry.as_secs().max(1)));
        return Some(Err(Box::new(res)));
    }
    if service == GitService::ReceivePack {
        return Some(Err(Box::new(text(
            StatusCode::FORBIDDEN,
            "CI job tokens are read-only",
        ))));
    }
    let job_token = match password.split_once(':') {
        Some((job_id, token)) => state
            .store
            .job_by_token(job_id, token)
            .await
            .map_err(|e| internal_error_for(&e)),
        None => Ok(None),
    };
    let run_repo = match job_token {
        Ok(Some(job)) => state
            .store
            .run_by_id(&job.run_id)
            .await
            .map_err(|e| internal_error_for(&e))
            .map(|run| run.map(|run| run.repo_id)),
        Ok(None) => Ok(None),
        Err(response) => Err(response),
    };
    let run_repo = match run_repo {
        Ok(run_repo) => run_repo,
        Err(response) => return Some(Err(Box::new(response))),
    };
    let Some(job_repo_id) = run_repo else {
        state.throttle.record_failure(&key, now);
        return Some(Err(Box::new(access_error(AccessError::Unauthorized))));
    };
    // `<repo>.wiki` names the job's repo's wiki. It is hidden while the wiki is off, as for users.
    let (repo_name, wiki) = match forge_core::names::wiki_base(name) {
        Some(base) => (base, true),
        None => (name, false),
    };
    let repo = match find_repo(state, owner, repo_name).await {
        Ok(Some(repo)) if repo.id == job_repo_id => repo,
        Ok(_) => return Some(Err(Box::new(access_error(AccessError::Unauthorized)))),
        Err(e) => return Some(Err(Box::new(access_error(e)))),
    };
    if wiki {
        match state.store.wiki_enabled(&repo.id).await {
            Ok(true) => {}
            Ok(false) => return Some(Err(Box::new(access_error(AccessError::Unauthorized)))),
            Err(e) => return Some(Err(Box::new(internal_error_for(&e)))),
        }
        return match forge_git::repo::wiki_path(&state.config.data_dir, &repo.id) {
            Ok(path) => Some(Ok((repo, path))),
            Err(_) => Some(Err(Box::new(internal_error()))),
        };
    }
    match forge_git::repo::repo_path(&state.config.data_dir, &repo.id) {
        Ok(path) => Some(Ok((repo, path))),
        Err(_) => Some(Err(Box::new(internal_error()))),
    }
}

fn internal_error_for(error: &forge_core::Error) -> Response {
    tracing::error!(error = %error, "ci job token check failed");
    internal_error()
}

pub(crate) fn access_error(error: AccessError) -> Response {
    match error {
        AccessError::Unauthorized => {
            let mut res = text(StatusCode::UNAUTHORIZED, "unauthorized");
            res.headers_mut().insert(
                WWW_AUTHENTICATE,
                HeaderValue::from_static("Basic realm=\"forge\""),
            );
            res
        }
        AccessError::NotFound => text(StatusCode::NOT_FOUND, "not found"),
        AccessError::Forbidden => text(StatusCode::FORBIDDEN, "forbidden"),
        AccessError::Internal(message) => {
            tracing::error!(error = %message, "smart-http request failed");
            internal_error()
        }
    }
}

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

    #[test]
    fn git_protocol_accepts_documented_charset() {
        assert!(valid_git_protocol("version=2"));
        assert!(valid_git_protocol("version=2:object-format=sha1"));
        assert!(valid_git_protocol("a.b,c_d-e:f"));
    }

    #[test]
    fn git_protocol_rejects_bad_values() {
        assert!(!valid_git_protocol(""));
        assert!(!valid_git_protocol("Version=2"));
        assert!(!valid_git_protocol("version 2"));
        assert!(!valid_git_protocol("version=2;x"));
        assert!(!valid_git_protocol("version=2\n"));
        assert!(!valid_git_protocol(&"a".repeat(65)));
        assert!(valid_git_protocol(&"a".repeat(64)));
    }

    #[test]
    fn git_protocol_header_is_dropped_when_invalid() {
        let mut headers = HeaderMap::new();
        headers.insert("git-protocol", HeaderValue::from_static("version=2"));
        assert_eq!(git_protocol(&headers), Some("version=2"));
        headers.insert("git-protocol", HeaderValue::from_static("BAD VALUE"));
        assert_eq!(git_protocol(&headers), None);
        assert_eq!(git_protocol(&HeaderMap::new()), None);
    }

    #[test]
    fn wants_v2_detects_version_pair() {
        assert!(wants_v2(Some("version=2")));
        assert!(wants_v2(Some("object-format=sha1:version=2")));
        assert!(!wants_v2(Some("version=1")));
        assert!(!wants_v2(None));
    }

    #[test]
    fn service_header_rules() {
        assert!(needs_service_header(GitService::UploadPack, false));
        assert!(!needs_service_header(GitService::UploadPack, true));
        assert!(needs_service_header(GitService::ReceivePack, false));
        assert!(needs_service_header(GitService::ReceivePack, true));
    }

    #[test]
    fn body_encoding_rules() {
        assert_eq!(body_encoding(None), Some(false));
        assert_eq!(body_encoding(Some("gzip")), Some(true));
        assert_eq!(body_encoding(Some("GZIP")), Some(true));
        assert_eq!(body_encoding(Some(" identity ")), Some(false));
        assert_eq!(body_encoding(Some("br")), None);
        assert_eq!(body_encoding(Some("gzip, gzip")), None);
        assert_eq!(body_encoding(Some("")), None);
    }

    #[tokio::test]
    async fn gzip_request_body_decodes() {
        use async_compression::tokio::write::GzipEncoder;
        use tokio::io::{AsyncWriteExt, BufReader};

        let payload = b"0032want 0123456789abcdef0123456789abcdef01234567\n0000";
        let mut encoder = GzipEncoder::new(Vec::new());
        encoder.write_all(payload).await.unwrap();
        encoder.shutdown().await.unwrap();
        let compressed = encoder.into_inner();

        let mut decoded = Vec::new();
        let mut decoder = GzipDecoder::new(BufReader::new(&compressed[..]));
        tokio::io::copy(&mut decoder, &mut decoded).await.unwrap();
        assert_eq!(decoded, payload);
    }

    #[test]
    fn content_types_match_service() {
        assert_eq!(
            advertisement_content_type(GitService::UploadPack),
            "application/x-git-upload-pack-advertisement"
        );
        assert_eq!(
            result_content_type(GitService::ReceivePack),
            "application/x-git-receive-pack-result"
        );
    }

    #[test]
    fn access_errors_map_to_status() {
        assert_eq!(
            access_error(AccessError::Unauthorized).status(),
            StatusCode::UNAUTHORIZED
        );
        assert_eq!(
            access_error(AccessError::NotFound).status(),
            StatusCode::NOT_FOUND
        );
        assert_eq!(
            access_error(AccessError::Forbidden).status(),
            StatusCode::FORBIDDEN
        );
        assert_eq!(
            access_error(AccessError::Internal("x".into())).status(),
            StatusCode::INTERNAL_SERVER_ERROR
        );
        let res = access_error(AccessError::Unauthorized);
        assert_eq!(
            res.headers()
                .get(WWW_AUTHENTICATE)
                .and_then(|v| v.to_str().ok()),
            Some("Basic realm=\"forge\"")
        );
    }

    fn basic_headers(plain: &str) -> HeaderMap {
        let mut headers = HeaderMap::new();
        use base64::Engine;

        let value = format!("Basic {}", base64::prelude::BASE64_STANDARD.encode(plain));
        headers.insert(AUTHORIZATION, HeaderValue::from_str(&value).unwrap());
        headers
    }

    #[tokio::test]
    async fn thirty_first_bad_token_is_throttled() {
        let tmp = tempfile::tempdir().expect("tempdir");
        let dir = tmp.path();
        let state = crate::test_support::test_state(dir).await;
        let ip: IpAddr = "192.0.2.50".parse().unwrap();
        let bad = basic_headers("alice:forge_not_a_real_token");

        for _ in 0..30 {
            let res = authenticate(&state, ip, &bad).await.unwrap_err();
            assert_eq!(res.status(), StatusCode::UNAUTHORIZED);
        }
        let res = authenticate(&state, ip, &bad).await.unwrap_err();
        assert_eq!(res.status(), StatusCode::TOO_MANY_REQUESTS);
        assert!(res.headers().get(RETRY_AFTER).is_some());

        // Anonymous requests from the same IP are stopped too, and other IPs are unaffected.
        let other: IpAddr = "192.0.2.51".parse().unwrap();
        assert!(
            authenticate(&state, other, &HeaderMap::new())
                .await
                .unwrap()
                .is_none()
        );
        let res = authenticate(&state, ip, &HeaderMap::new())
            .await
            .unwrap_err();
        assert_eq!(res.status(), StatusCode::TOO_MANY_REQUESTS);
    }
}