//! Git LFS routes: the batch API, object GET and PUT, and verify.
//!
//! Each route accepts either a `RemoteAuth <token>` action token (issued by the batch API or by
//! the SSH `git-lfs-authenticate` command, scoped to one repo) or HTTP Basic with a personal access
//! token, or no credentials for public reads. The `{repo}` segment may carry the `.git` suffix,
//! which `find_repo` strips, so both the `.git` and the bare paths reach these handlers.

use std::net::IpAddr;
use std::path::Path;
use std::sync::atomic::{AtomicU64, Ordering};
use std::time::{SystemTime, UNIX_EPOCH};

use forge_core::Action;
use forge_core::lfs_token::{self, LfsOp};
use futures_util::{StreamExt, TryStreamExt};
use http::header::{AUTHORIZATION, CONTENT_LENGTH, CONTENT_TYPE, HeaderValue};
use http_body::Frame;
use http_body_util::StreamBody;
use serde::Deserialize;
use serde_json::{Value, json};
use sha2::{Digest, Sha256};
use tokio::io::AsyncWriteExt;
use tokio_util::io::ReaderStream;
use topcoat::context::{Cx, app_context};
use topcoat::router::request::headers;
use topcoat::router::response::Response;
use topcoat::router::{Body, HeaderMap, StatusCode, path_param, route, to_bytes};

use super::{is_valid_oid, object_path, tmp_dir};
use crate::access::{AccessError, find_repo, resolve_repo_for};
use crate::client_ip::client_ip;
use crate::git_http::{access_error, authenticate};
use crate::state::AppState;

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

/// Media type of LFS JSON bodies.
const LFS_JSON: &str = "application/vnd.git-lfs+json";
/// Lifetime of action tokens, in seconds.
const TOKEN_TTL_SECS: i64 = 3600;
/// Largest batch request body accepted.
const MAX_BATCH_BYTES: usize = 4 * 1024 * 1024;
/// Most objects one batch may name.
const MAX_BATCH_OBJECTS: usize = 1000;
/// Largest verify request body accepted.
const MAX_VERIFY_BYTES: usize = 4096;

/// Distinguishes the temp files of concurrent uploads in one process.
static TEMP_SEQ: AtomicU64 = AtomicU64::new(0);

#[derive(Deserialize)]
struct BatchRequest {
    operation: String,
    #[serde(default)]
    transfers: Vec<String>,
    #[serde(default)]
    hash_algo: Option<String>,
    objects: Vec<BatchObject>,
}

#[derive(Deserialize)]
struct BatchObject {
    oid: String,
    size: u64,
}

#[derive(Deserialize)]
struct VerifyRequest {
    oid: String,
    size: u64,
}

#[route(POST "/{owner}/{repo}/info/lfs/objects/batch")]
async fn batch(cx: &Cx, body: Body) -> topcoat::Result<Response> {
    let state = app_context::<AppState>(cx);
    let headers = headers(cx);
    let owner = path_param::<Owner>(cx);
    let name = path_param::<Repo>(cx);

    let bytes = to_bytes(body, MAX_BATCH_BYTES).await?;
    let BatchPlan { request, action } = match parse_batch_operation(&bytes) {
        Ok(plan) => plan,
        Err(response) => return Ok(*response),
    };
    let max = state.config.lfs.max_object_bytes();
    for object in &request.objects {
        if !is_valid_oid(&object.oid) {
            return Ok(message(StatusCode::UNPROCESSABLE_ENTITY, "invalid oid"));
        }
        if object.size > max {
            return Ok(message(
                StatusCode::UNPROCESSABLE_ENTITY,
                "an object exceeds the size limit",
            ));
        }
    }

    let repo = match authorize(state, client_ip(cx), headers, owner, name, action).await {
        Ok(repo) => repo,
        Err(response) => return Ok(*response),
    };
    let writing = matches!(action, Action::Write);
    let writable = if writing {
        ensure_writable(state, &repo).await
    } else {
        Ok(())
    };
    if let Err(response) = writable {
        return Ok(*response);
    }

    let token_op = if writing {
        LfsOp::Upload
    } else {
        LfsOp::Download
    };
    let base = batch_base(state, &repo);
    let token = lfs_token::issue(&state.key, &repo.id, token_op, now_unix() + TOKEN_TTL_SECS);

    let mut objects = Vec::with_capacity(request.objects.len());
    for object in &request.objects {
        let stored = match state.store.lfs_object_size(&repo.id, &object.oid).await {
            Ok(stored) => stored,
            Err(e) => return Ok(server_error(e)),
        };
        let href = format!("{base}/{}", object.oid);
        let entry = match (writing, stored) {
            (true, Some(_)) => json!({ "oid": object.oid, "size": object.size }),
            (true, None) => json!({
                "oid": object.oid,
                "size": object.size,
                "actions": {
                    "upload": action_json(&href, &token),
                    "verify": action_json(&format!("{base}/verify"), &token),
                },
            }),
            (false, Some(size)) => json!({
                "oid": object.oid,
                "size": size,
                "actions": { "download": action_json(&href, &token) },
            }),
            (false, None) => json!({
                "oid": object.oid,
                "size": object.size,
                "error": { "code": 404, "message": "Object does not exist" },
            }),
        };
        objects.push(entry);
    }
    Ok(lfs_json(
        StatusCode::OK,
        json!({ "transfer": "basic", "objects": objects }),
    ))
}

/// A batch request that passed its own checks, with the action it asks for.
struct BatchPlan {
    request: BatchRequest,
    action: Action,
}

/// Parses and checks a batch body. The error is the response to send.
fn parse_batch_operation(bytes: &[u8]) -> Result<BatchPlan, Box<Response>> {
    let request: BatchRequest = serde_json::from_slice(bytes).map_err(|e| {
        Box::new(message(
            StatusCode::UNPROCESSABLE_ENTITY,
            format!("invalid batch request: {e}"),
        ))
    })?;
    let action = match request.operation.as_str() {
        "download" => Action::Read,
        "upload" => Action::Write,
        _ => return Err(unprocessable("unsupported operation")),
    };
    if !request.transfers.is_empty() && !request.transfers.iter().any(|t| t == "basic") {
        return Err(unprocessable("only the basic transfer is supported"));
    }
    if request
        .hash_algo
        .as_deref()
        .is_some_and(|algo| algo != "sha256")
    {
        return Err(unprocessable("unsupported hash algorithm"));
    }
    if request.objects.len() > MAX_BATCH_OBJECTS {
        return Err(unprocessable(format!(
            "a batch may name at most {MAX_BATCH_OBJECTS} objects"
        )));
    }
    Ok(BatchPlan { request, action })
}

#[route(GET "/{owner}/{repo}/info/lfs/objects/{oid}")]
async fn get_object(cx: &Cx) -> topcoat::Result<Response> {
    let state = app_context::<AppState>(cx);
    let headers = headers(cx);
    let owner = path_param::<Owner>(cx);
    let name = path_param::<Repo>(cx);
    let oid = path_param::<Oid>(cx);
    if !is_valid_oid(oid) {
        return Ok(message(StatusCode::UNPROCESSABLE_ENTITY, "invalid oid"));
    }

    let repo = match authorize(state, client_ip(cx), headers, owner, name, Action::Read).await {
        Ok(repo) => repo,
        Err(response) => return Ok(*response),
    };
    let size = match state.store.lfs_object_size(&repo.id, oid).await {
        Ok(Some(size)) => size,
        Ok(None) => return Ok(message(StatusCode::NOT_FOUND, "Object does not exist")),
        Err(e) => return Ok(server_error(e)),
    };
    let file = match tokio::fs::File::open(object_path(&state.config.data_dir, oid)).await {
        Ok(file) => file,
        Err(e) => {
            // The link exists but the bytes do not: the store is damaged, not the request.
            tracing::error!(oid = %oid, error = %e, "lfs object file is missing");
            return Ok(server_error(e));
        }
    };

    let chunks = ReaderStream::new(file).map_ok(Frame::data);
    let mut res = Response::new(Body::new(StreamBody::new(chunks)));
    res.headers_mut().insert(
        CONTENT_TYPE,
        HeaderValue::from_static("application/octet-stream"),
    );
    res.headers_mut()
        .insert(CONTENT_LENGTH, HeaderValue::from(size));
    Ok(res)
}

#[route(PUT "/{owner}/{repo}/info/lfs/objects/{oid}")]
async fn put_object(cx: &Cx, body: Body) -> topcoat::Result<Response> {
    let state = app_context::<AppState>(cx);
    let headers = headers(cx);
    let owner = path_param::<Owner>(cx);
    let name = path_param::<Repo>(cx);
    let oid = path_param::<Oid>(cx);
    if !is_valid_oid(oid) {
        return Ok(message(StatusCode::UNPROCESSABLE_ENTITY, "invalid oid"));
    }

    let repo = match authorize(state, client_ip(cx), headers, owner, name, Action::Write).await {
        Ok(repo) => repo,
        Err(response) => return Ok(*response),
    };
    if let Err(response) = ensure_writable(state, &repo).await {
        return Ok(*response);
    }

    let max = state.config.lfs.max_object_bytes();
    let declared = headers
        .get(CONTENT_LENGTH)
        .and_then(|v| v.to_str().ok())
        .and_then(|v| v.parse::<u64>().ok());
    if declared.is_some_and(|len| len > max) {
        return Ok(message(
            StatusCode::PAYLOAD_TOO_LARGE,
            "the object exceeds the size limit",
        ));
    }

    let size = match receive_object(&state.config.data_dir, body, oid, max).await {
        Ok(size) => size,
        Err(response) => return Ok(*response),
    };
    if let Err(e) = state.store.insert_lfs_object(&repo.id, oid, size).await {
        return Ok(server_error(e));
    }
    Ok(Response::new(Body::empty()))
}

#[route(POST "/{owner}/{repo}/info/lfs/objects/verify")]
async fn verify(cx: &Cx, body: Body) -> topcoat::Result<Response> {
    let state = app_context::<AppState>(cx);
    let headers = headers(cx);
    let owner = path_param::<Owner>(cx);
    let name = path_param::<Repo>(cx);

    let repo = match authorize(state, client_ip(cx), headers, owner, name, Action::Write).await {
        Ok(repo) => repo,
        Err(response) => return Ok(*response),
    };
    if let Err(response) = ensure_writable(state, &repo).await {
        return Ok(*response);
    }

    let bytes = to_bytes(body, MAX_VERIFY_BYTES).await?;
    let request: VerifyRequest = match serde_json::from_slice(&bytes) {
        Ok(request) => request,
        Err(e) => {
            return Ok(message(
                StatusCode::UNPROCESSABLE_ENTITY,
                format!("invalid verify request: {e}"),
            ));
        }
    };
    if !is_valid_oid(&request.oid) {
        return Ok(message(StatusCode::UNPROCESSABLE_ENTITY, "invalid oid"));
    }
    Ok(
        match state.store.lfs_object_size(&repo.id, &request.oid).await {
            Ok(Some(size)) if size == request.size => Response::new(Body::empty()),
            Ok(Some(_)) => message(StatusCode::UNPROCESSABLE_ENTITY, "the size does not match"),
            Ok(None) => message(StatusCode::NOT_FOUND, "Object does not exist"),
            Err(e) => server_error(e),
        },
    )
}

/// Who may act on a repo for `action`. A `RemoteAuth` token is checked against the repo and the
/// operation it grants, and skips the user check; otherwise HTTP Basic (a PAT) or anonymous access
/// goes through the usual repo authorization. The error is the response to send.
async fn authorize(
    state: &AppState,
    ip: IpAddr,
    headers: &HeaderMap,
    owner: &str,
    name: &str,
    action: Action,
) -> Result<forge_core::Repo, Box<Response>> {
    if let Some(token) = remote_auth_token(headers) {
        let repo = match find_repo(state, owner, name).await {
            Ok(Some(repo)) => repo,
            Ok(None) => return Err(Box::new(access_failure(AccessError::NotFound))),
            Err(e) => return Err(Box::new(access_failure(e))),
        };
        let wanted = match action {
            Action::Write => LfsOp::Upload,
            _ => LfsOp::Download,
        };
        if !lfs_token::verify(&state.key, token, &repo.id, wanted, now_unix()) {
            return Err(Box::new(access_failure(AccessError::Unauthorized)));
        }
        return Ok(repo);
    }

    let actor = authenticate(state, ip, headers).await?;
    resolve_repo_for(state, actor.as_ref(), owner, name, action)
        .await
        .map_err(|e| Box::new(access_failure(e)))
}

/// The token of a `Authorization: RemoteAuth <token>` header, if there is one.
fn remote_auth_token(headers: &HeaderMap) -> Option<&str> {
    let value = headers.get(AUTHORIZATION)?.to_str().ok()?;
    let (scheme, token) = value.split_once(' ')?;
    let token = token.trim();
    (scheme.eq_ignore_ascii_case("remoteauth") && !token.is_empty()).then_some(token)
}

/// Refuses writes to an archived repo or to a pull mirror, which is read-only by design.
async fn ensure_writable(state: &AppState, repo: &forge_core::Repo) -> Result<(), Box<Response>> {
    if repo.archived {
        return Err(Box::new(message(
            StatusCode::FORBIDDEN,
            "the repository is archived",
        )));
    }
    match state.store.pull_mirror(&repo.id).await {
        Ok(Some(_)) => Err(Box::new(message(
            StatusCode::FORBIDDEN,
            "the repository is a pull mirror",
        ))),
        Ok(None) => Ok(()),
        Err(e) => Err(Box::new(server_error(e))),
    }
}

/// Streams an upload into a temp file, checks its SHA-256 against `oid` and its size against
/// `max`, then renames it into the object store. Returns the object size. A failed upload leaves
/// no file behind.
async fn receive_object(
    data_dir: &Path,
    body: Body,
    oid: &str,
    max: u64,
) -> Result<u64, Box<Response>> {
    let dir = tmp_dir(data_dir);
    tokio::fs::create_dir_all(&dir)
        .await
        .map_err(|e| Box::new(server_error(e)))?;
    let seq = TEMP_SEQ.fetch_add(1, Ordering::Relaxed);
    let temp = dir.join(format!("{oid}.{}.{seq}.part", std::process::id()));

    let (size, digest) = match write_and_hash(&temp, body, max).await {
        Ok(written) => written,
        Err(response) => {
            let _ = tokio::fs::remove_file(&temp).await;
            return Err(response);
        }
    };
    if digest != oid {
        let _ = tokio::fs::remove_file(&temp).await;
        return Err(unprocessable("the content does not match the oid"));
    }

    let target = object_path(data_dir, oid);
    if let Some(parent) = target.parent() {
        tokio::fs::create_dir_all(parent)
            .await
            .map_err(|e| Box::new(server_error(e)))?;
    }
    if let Err(e) = tokio::fs::rename(&temp, &target).await {
        let _ = tokio::fs::remove_file(&temp).await;
        return Err(Box::new(server_error(e)));
    }
    Ok(size)
}

/// Writes the request body to `temp` and returns its size and hex SHA-256.
async fn write_and_hash(temp: &Path, body: Body, max: u64) -> Result<(u64, String), Box<Response>> {
    let mut file = tokio::fs::OpenOptions::new()
        .write(true)
        .create_new(true)
        .open(temp)
        .await
        .map_err(|e| Box::new(server_error(e)))?;
    let mut hasher = Sha256::new();
    let mut size: u64 = 0;
    let mut stream = std::pin::pin!(body.into_data_stream());
    while let Some(chunk) = stream.next().await {
        let Ok(chunk) = chunk else {
            return Err(Box::new(message(
                StatusCode::BAD_REQUEST,
                "the request body could not be read",
            )));
        };
        size = size.saturating_add(u64::try_from(chunk.len()).unwrap_or(u64::MAX));
        if size > max {
            return Err(Box::new(message(
                StatusCode::PAYLOAD_TOO_LARGE,
                "the object exceeds the size limit",
            )));
        }
        hasher.update(&chunk);
        file.write_all(&chunk)
            .await
            .map_err(|e| Box::new(server_error(e)))?;
    }
    file.sync_all()
        .await
        .map_err(|e| Box::new(server_error(e)))?;
    Ok((size, hex::encode(hasher.finalize())))
}

/// The `href` prefix of LFS actions for `repo`, always in the `.git` form.
fn batch_base(state: &AppState, repo: &forge_core::Repo) -> String {
    format!(
        "{}/{}/{}.git/info/lfs/objects",
        state.config.base_url.trim_end_matches('/'),
        repo.owner_name,
        repo.name
    )
}

/// One action of a batch response.
fn action_json(href: &str, token: &str) -> Value {
    json!({
        "href": href,
        "header": { "Authorization": format!("RemoteAuth {token}") },
        "expires_in": TOKEN_TTL_SECS,
    })
}

/// Maps an access failure to a response. A 401 carries the LFS challenge, so git-lfs asks for
/// credentials.
fn access_failure(error: AccessError) -> Response {
    let unauthorized = error == AccessError::Unauthorized;
    let mut res = access_error(error);
    if unauthorized {
        res.headers_mut().insert(
            "LFS-Authenticate",
            HeaderValue::from_static("Basic realm=\"forge\""),
        );
    }
    res
}

/// A 422 response, boxed so it can be the error of a `Result`.
fn unprocessable(text: impl Into<String>) -> Box<Response> {
    Box::new(message(StatusCode::UNPROCESSABLE_ENTITY, text))
}

/// Logs an internal failure and answers 500.
fn server_error(error: impl std::fmt::Display) -> Response {
    tracing::error!(error = %error, "lfs request failed");
    message(StatusCode::INTERNAL_SERVER_ERROR, "internal server error")
}

/// A JSON answer in the LFS media type.
fn lfs_json(status: StatusCode, value: Value) -> Response {
    let mut res = Response::new(Body::from(value.to_string()));
    *res.status_mut() = status;
    res.headers_mut()
        .insert(CONTENT_TYPE, HeaderValue::from_static(LFS_JSON));
    res
}

/// An LFS JSON error with a `message` field.
fn message(status: StatusCode, text: impl Into<String>) -> Response {
    lfs_json(status, json!({ "message": text.into() }))
}

fn now_unix() -> i64 {
    SystemTime::now()
        .duration_since(UNIX_EPOCH)
        .map_or(0, |d| i64::try_from(d.as_secs()).unwrap_or(i64::MAX))
}