// This Source Code Form is subject to the terms of the Mozilla Public
// License, v. 2.0. If a copy of the MPL was not distributed with this
// file, You can obtain one at https://mozilla.org/MPL/2.0/.

//! Git-LFS: the [Batch API] and basic transfer (upload / download) over HTTP.
//!
//! Objects are content-addressed by their sha256 `oid` and stored on disk under
//! `storage.lfs_dir`, sharded `{oid[0..2]}/{oid[2..4]}/{oid}`; the `lfs_objects`
//! table records which repository holds which object (for access control and the
//! batch "do you have this?" answer). Uploads stream to a temp file, are verified
//! against the claimed oid, and only then moved into place — a corrupt or
//! mislabelled upload never becomes a servable object.
//!
//! Authorization runs before any bytes move: download needs [`Access::Read`],
//! upload needs [`Access::Write`]. HTTPS clients authenticate with Basic
//! credentials; SSH clients present the bearer token minted by
//! `git-lfs-authenticate` (see [`crate::git_http::authenticate`]).
//!
//! [Batch API]: https://github.com/git-lfs/git-lfs/blob/main/docs/api/batch.md

use std::collections::HashMap;
use std::sync::atomic::{AtomicU64, Ordering};

use axum::body::Body;
use axum::extract::{Path, State};
use axum::http::{HeaderMap, StatusCode, header};
use axum::response::{IntoResponse, Response};
use axum_extra::extract::cookie::CookieJar;
use blob::Namespace;
use futures_util::TryStreamExt as _;
use serde::{Deserialize, Serialize};
use sha2::{Digest, Sha256};
use tokio::io::{AsyncReadExt as _, AsyncWriteExt as _};
use tokio_util::io::{ReaderStream, StreamReader};

use crate::AppState;
use crate::git_http::authorize;

/// The media type every LFS JSON request and response uses.
const LFS_JSON: &str = "application/vnd.git-lfs+json";

/// Per-process counter for unique upload temp-file names.
static TMP_SEQ: AtomicU64 = AtomicU64::new(0);

/// Split a repo-relative path on `/info/lfs/`, returning `(repo_path, lfs_tail)`
/// with any trailing `.git` stripped from the repo path. `lfs_tail` is e.g.
/// `objects/batch` or `objects/<oid>`.
#[must_use]
pub(crate) fn split_lfs(rest: &str) -> Option<(&str, &str)> {
    let idx = rest.find("/info/lfs/")?;
    let repo = rest[..idx].strip_suffix(".git").unwrap_or(&rest[..idx]);
    Some((repo, &rest[idx + "/info/lfs/".len()..]))
}

/// A 64-char lowercase-hex sha256 is the only shape we treat as an oid; anything
/// else is rejected before it can touch the filesystem.
fn valid_oid(oid: &str) -> bool {
    oid.len() == 64
        && oid
            .bytes()
            .all(|b| b.is_ascii_digit() || (b'a'..=b'f').contains(&b))
}

/// The blob key for an object, sharded by the first two byte-pairs of the oid.
/// Callers pass only validated 64-char oids, so the slices are in bounds.
fn object_key(oid: &str) -> String {
    format!("{}/{}/{}", &oid[0..2], &oid[2..4], oid)
}

/// Hex-encode a byte slice (lowercase), for comparing a computed digest to an oid.
fn hex(bytes: &[u8]) -> String {
    let mut s = String::with_capacity(bytes.len() * 2);
    for b in bytes {
        s.push(char::from_digit(u32::from(b >> 4), 16).unwrap_or('0'));
        s.push(char::from_digit(u32::from(b & 0xf), 16).unwrap_or('0'));
    }
    s
}

/// `PUT /{owner}/{*rest}` — the object-upload route. Recognizes only the LFS
/// `…/info/lfs/objects/{oid}` shape and rejects everything else.
pub(crate) async fn put(
    State(state): State<AppState>,
    jar: CookieJar,
    headers: HeaderMap,
    Path((owner, rest)): Path<(String, String)>,
    body: Body,
) -> Response {
    let Some((repo_path, tail)) = split_lfs(&rest) else {
        return StatusCode::NOT_FOUND.into_response();
    };
    let Some(oid) = tail.strip_prefix("objects/") else {
        return StatusCode::NOT_FOUND.into_response();
    };
    if oid.contains('/') {
        return StatusCode::NOT_FOUND.into_response();
    }
    upload(
        state,
        jar,
        headers,
        owner,
        repo_path.to_string(),
        oid.to_string(),
        body,
    )
    .await
}

// ---- Batch API ----

/// One `{oid, size}` pair in a batch request.
#[derive(Debug, Deserialize)]
struct ObjectId {
    oid: String,
    #[serde(default)]
    size: i64,
}

/// A `POST objects/batch` request body.
#[derive(Debug, Deserialize)]
struct BatchRequest {
    operation: String,
    objects: Vec<ObjectId>,
}

/// A single transfer action (an href the client PUTs to or GETs from).
#[derive(Debug, Serialize)]
struct Action {
    href: String,
    #[serde(skip_serializing_if = "Option::is_none")]
    header: Option<HashMap<String, String>>,
}

/// The actions available for one object (only the relevant ones are set).
#[derive(Debug, Default, Serialize)]
struct Actions {
    #[serde(skip_serializing_if = "Option::is_none")]
    download: Option<Action>,
    #[serde(skip_serializing_if = "Option::is_none")]
    upload: Option<Action>,
    #[serde(skip_serializing_if = "Option::is_none")]
    verify: Option<Action>,
}

/// A per-object error in the batch response.
#[derive(Debug, Serialize)]
struct ObjectError {
    code: u16,
    message: String,
}

/// One object in the batch response.
#[derive(Debug, Serialize)]
struct BatchObject {
    oid: String,
    size: i64,
    #[serde(skip_serializing_if = "Option::is_none")]
    actions: Option<Actions>,
    #[serde(skip_serializing_if = "Option::is_none")]
    error: Option<ObjectError>,
}

/// The `POST objects/batch` response body.
#[derive(Debug, Serialize)]
struct BatchResponse {
    transfer: String,
    objects: Vec<BatchObject>,
}

/// `POST .../info/lfs/objects/batch` — negotiate which objects the client may
/// upload or download and where.
pub(crate) async fn batch(
    state: AppState,
    jar: CookieJar,
    headers: HeaderMap,
    owner: String,
    repo_path: String,
    body: Body,
) -> Response {
    let Ok(raw) = axum::body::to_bytes(body, 1024 * 1024).await else {
        return lfs_error(StatusCode::BAD_REQUEST, "request too large");
    };
    let Ok(req) = serde_json::from_slice::<BatchRequest>(&raw) else {
        return lfs_error(StatusCode::BAD_REQUEST, "malformed batch request");
    };
    let uploading = req.operation == "upload";
    let required = if uploading {
        auth::Access::Write
    } else {
        auth::Access::Read
    };
    let repo = match authorize(&state, &jar, &headers, &owner, &repo_path, required).await {
        Ok(repo) => repo,
        Err(resp) => return resp,
    };

    let base = format!(
        "{}/{}/{}.git/info/lfs/objects",
        state.config.instance.url.trim_end_matches('/'),
        owner,
        repo_path
    );
    let mut objects = Vec::with_capacity(req.objects.len());
    for obj in &req.objects {
        objects.push(if uploading {
            batch_upload(&state, &repo.id, &base, obj).await
        } else {
            batch_download(&state, &repo.id, &base, obj).await
        });
    }
    let payload = BatchResponse {
        transfer: "basic".to_string(),
        objects,
    };
    lfs_json(StatusCode::OK, &payload)
}

/// Resolve one object for a download batch: offer a download action if we hold it.
async fn batch_download(
    state: &AppState,
    repo_id: &str,
    base: &str,
    obj: &ObjectId,
) -> BatchObject {
    if valid_oid(&obj.oid)
        && state
            .store
            .lfs_object_exists(repo_id, &obj.oid)
            .await
            .unwrap_or(false)
    {
        BatchObject {
            oid: obj.oid.clone(),
            size: obj.size,
            actions: Some(Actions {
                download: Some(Action {
                    href: format!("{base}/{}", obj.oid),
                    header: None,
                }),
                ..Actions::default()
            }),
            error: None,
        }
    } else {
        BatchObject {
            oid: obj.oid.clone(),
            size: obj.size,
            actions: None,
            error: Some(ObjectError {
                code: 404,
                message: "object does not exist".to_string(),
            }),
        }
    }
}

/// Resolve one object for an upload batch: offer upload+verify unless we have it.
async fn batch_upload(state: &AppState, repo_id: &str, base: &str, obj: &ObjectId) -> BatchObject {
    if !valid_oid(&obj.oid) {
        return BatchObject {
            oid: obj.oid.clone(),
            size: obj.size,
            actions: None,
            error: Some(ObjectError {
                code: 422,
                message: "invalid oid".to_string(),
            }),
        };
    }
    // Already stored (in this repo): no action needed, the client skips it.
    if state
        .store
        .lfs_object_exists(repo_id, &obj.oid)
        .await
        .unwrap_or(false)
    {
        return BatchObject {
            oid: obj.oid.clone(),
            size: obj.size,
            actions: None,
            error: None,
        };
    }
    BatchObject {
        oid: obj.oid.clone(),
        size: obj.size,
        actions: Some(Actions {
            upload: Some(Action {
                href: format!("{base}/{}", obj.oid),
                header: None,
            }),
            verify: Some(Action {
                href: format!("{base}/{}/verify", obj.oid),
                header: None,
            }),
            ..Actions::default()
        }),
        error: None,
    }
}

// ---- Basic transfer ----

/// `GET .../info/lfs/objects/{oid}` — stream a stored object.
pub(crate) async fn download(
    state: AppState,
    jar: CookieJar,
    headers: HeaderMap,
    owner: String,
    repo_path: String,
    oid: String,
) -> Response {
    let repo = match authorize(
        &state,
        &jar,
        &headers,
        &owner,
        &repo_path,
        auth::Access::Read,
    )
    .await
    {
        Ok(repo) => repo,
        Err(resp) => return resp,
    };
    if !valid_oid(&oid)
        || !state
            .store
            .lfs_object_exists(&repo.id, &oid)
            .await
            .unwrap_or(false)
    {
        return StatusCode::NOT_FOUND.into_response();
    }
    let Ok(Some(read)) = state.blobs.open(Namespace::Lfs, &object_key(&oid)).await else {
        return StatusCode::NOT_FOUND.into_response();
    };
    let body = Body::from_stream(ReaderStream::new(read.reader));
    (
        [
            (header::CONTENT_TYPE, "application/octet-stream".to_string()),
            (header::CONTENT_LENGTH, read.size.to_string()),
        ],
        body,
    )
        .into_response()
}

/// `PUT .../info/lfs/objects/{oid}` — receive an object, verifying its sha256
/// matches the oid before it becomes servable.
pub(crate) async fn upload(
    state: AppState,
    jar: CookieJar,
    headers: HeaderMap,
    owner: String,
    repo_path: String,
    oid: String,
    body: Body,
) -> Response {
    let repo = match authorize(
        &state,
        &jar,
        &headers,
        &owner,
        &repo_path,
        auth::Access::Write,
    )
    .await
    {
        Ok(repo) => repo,
        Err(resp) => return resp,
    };
    if !valid_oid(&oid) {
        return lfs_error(StatusCode::UNPROCESSABLE_ENTITY, "invalid oid");
    }
    // Idempotent: if we already hold it, drain the body and succeed.
    if state
        .store
        .lfs_object_exists(&repo.id, &oid)
        .await
        .unwrap_or(false)
    {
        return StatusCode::OK.into_response();
    }

    // Stream to a local scratch file to hash and size it before it becomes
    // servable; `data_dir` is always local, whatever the blob backend is.
    let tmp_dir = state.config.storage.data_dir.join("tmp");
    if tokio::fs::create_dir_all(&tmp_dir).await.is_err() {
        return lfs_error(StatusCode::INTERNAL_SERVER_ERROR, "storage error");
    }
    let tmp = tmp_dir.join(format!(
        "lfs-{}.{}.tmp",
        std::process::id(),
        TMP_SEQ.fetch_add(1, Ordering::Relaxed)
    ));

    let Ok((digest, size)) = stream_to_file(body, &tmp).await else {
        let _ = tokio::fs::remove_file(&tmp).await;
        return lfs_error(StatusCode::INTERNAL_SERVER_ERROR, "write failed");
    };
    if digest != oid {
        let _ = tokio::fs::remove_file(&tmp).await;
        return lfs_error(StatusCode::UNPROCESSABLE_ENTITY, "oid mismatch");
    }

    // Move the verified object into the blob store (local rename or S3 upload).
    if state
        .blobs
        .put_file(
            Namespace::Lfs,
            &object_key(&oid),
            &tmp,
            "application/octet-stream",
        )
        .await
        .is_err()
    {
        let _ = tokio::fs::remove_file(&tmp).await;
        return lfs_error(StatusCode::INTERNAL_SERVER_ERROR, "storage error");
    }
    if state
        .store
        .lfs_object_add(&repo.id, &oid, size)
        .await
        .is_err()
    {
        return lfs_error(StatusCode::INTERNAL_SERVER_ERROR, "storage error");
    }
    StatusCode::OK.into_response()
}

/// `POST .../info/lfs/objects/{oid}/verify` — confirm an object landed at the
/// expected size (git-lfs calls this after an upload).
pub(crate) async fn verify(
    state: AppState,
    jar: CookieJar,
    headers: HeaderMap,
    owner: String,
    repo_path: String,
    body: Body,
) -> Response {
    let repo = match authorize(
        &state,
        &jar,
        &headers,
        &owner,
        &repo_path,
        auth::Access::Read,
    )
    .await
    {
        Ok(repo) => repo,
        Err(resp) => return resp,
    };
    let Ok(raw) = axum::body::to_bytes(body, 64 * 1024).await else {
        return lfs_error(StatusCode::BAD_REQUEST, "request too large");
    };
    let Ok(obj) = serde_json::from_slice::<ObjectId>(&raw) else {
        return lfs_error(StatusCode::BAD_REQUEST, "malformed request");
    };
    match state.store.lfs_object_size(&repo.id, &obj.oid).await {
        Ok(Some(size)) if size == obj.size => StatusCode::OK.into_response(),
        _ => lfs_error(StatusCode::NOT_FOUND, "object not found"),
    }
}

/// Stream `body` into `path`, returning `(hex_sha256, size)`.
async fn stream_to_file(body: Body, path: &std::path::Path) -> std::io::Result<(String, i64)> {
    let stream = body.into_data_stream().map_err(std::io::Error::other);
    let reader = StreamReader::new(stream);
    tokio::pin!(reader);
    let mut file = tokio::fs::File::create(path).await?;
    let mut hasher = Sha256::new();
    let mut buf = vec![0u8; 64 * 1024];
    let mut size: i64 = 0;
    loop {
        let n = reader.read(&mut buf).await?;
        if n == 0 {
            break;
        }
        hasher.update(&buf[..n]);
        file.write_all(&buf[..n]).await?;
        size += i64::try_from(n).unwrap_or(0);
    }
    file.flush().await?;
    Ok((hex(&hasher.finalize()), size))
}

/// Render a value as an LFS JSON response with the required content type.
fn lfs_json<T: Serialize>(status: StatusCode, value: &T) -> Response {
    match serde_json::to_vec(value) {
        Ok(bytes) => (status, [(header::CONTENT_TYPE, LFS_JSON)], bytes).into_response(),
        Err(_) => StatusCode::INTERNAL_SERVER_ERROR.into_response(),
    }
}

/// Render an LFS error body (`{"message": "..."}`) with the given status.
fn lfs_error(status: StatusCode, message: &str) -> Response {
    let body = serde_json::json!({ "message": message });
    (status, [(header::CONTENT_TYPE, LFS_JSON)], body.to_string()).into_response()
}

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

    #[test]
    fn split_lfs_finds_the_repo_and_tail() {
        assert_eq!(
            split_lfs("owner-scope/repo.git/info/lfs/objects/batch"),
            Some(("owner-scope/repo", "objects/batch"))
        );
        // The `.git` is optional (git-lfs derives the endpoint from the remote).
        assert_eq!(
            split_lfs("grp/sub/repo/info/lfs/objects/abc"),
            Some(("grp/sub/repo", "objects/abc"))
        );
        assert_eq!(split_lfs("owner/repo.git/info/refs"), None);
    }

    #[test]
    fn valid_oid_accepts_only_lowercase_hex_sha256() {
        assert!(valid_oid(&"a".repeat(64)));
        assert!(valid_oid(&"0123456789abcdef".repeat(4)));
        assert!(!valid_oid(&"a".repeat(63)), "too short");
        assert!(!valid_oid(&"A".repeat(64)), "uppercase rejected");
        assert!(!valid_oid(&"g".repeat(64)), "non-hex rejected");
    }

    #[test]
    fn hex_matches_a_known_sha256() {
        // sha256("") = e3b0c442...
        let digest = Sha256::digest(b"");
        assert_eq!(
            hex(&digest),
            "e3b0c44298fc1c149afbf4c8996fb92427ae41e4649b934ca495991b7852b855"
        );
    }
}