use axum::body::Body;
use axum::http::{HeaderMap, Method, StatusCode, header};
use axum::response::{IntoResponse, Response};
use futures::StreamExt as _;
use serde_json::json;
use tokio_util::io::StreamReader;

use crate::lfs::types::{
    LfsBatchRequest, LfsLockCreateRequest, LfsOperation, LfsVerifyLocksRequest,
};
use crate::lfs::{lock, object};
use crate::share::access::{BasicAuth, allow_repo_read, require_writer};
use crate::smart::error::{SmartError, SmartResult};
use crate::smart::service::SmartService;

const MAX_LFS_BODY: usize = 64 * 1024 * 1024;

#[allow(clippy::too_many_arguments)]
pub async fn dispatch(
    service: &SmartService,
    space: &str,
    repo: &str,
    headers: &HeaderMap,
    path: &str,
    method: Method,
    body: Body,
    auth: &BasicAuth,
) -> SmartResult<Response> {
    match path {
        "info/lfs/objects/batch" if method == Method::POST => {
            let bytes = read_body(headers, body, MAX_LFS_BODY).await?;
            let req: LfsBatchRequest = serde_json::from_slice(&bytes)
                .map_err(|e| SmartError::BadRequest(format!("invalid batch request: {e}")))?;
            batch(service, space, repo, req, auth).await
        }
        "info/lfs/verify" if method == Method::POST => {
            let bytes = read_body(headers, body, MAX_LFS_BODY).await?;
            verify(service, space, repo, bytes, auth).await
        }
        p if p.starts_with("info/lfs/objects/") => {
            let oid = &p["info/lfs/objects/".len()..];
            match method {
                Method::PUT => {
                    upload(service, space, repo, oid.to_string(), headers, body, auth).await
                }
                Method::GET => download(service, space, repo, oid, auth).await,
                _ => Ok(SmartError::BadRequest("method not allowed".into()).into_response()),
            }
        }
        "info/lfs/locks" if method == Method::GET => list_locks(service, space, repo, auth).await,
        "info/lfs/locks" if method == Method::POST => {
            create_lock(service, space, repo, body, auth).await
        }
        "info/lfs/locks/verify" if method == Method::POST => {
            verify_locks(service, space, repo, body, auth).await
        }
        p if p.starts_with("info/lfs/locks/")
            && p.ends_with("/unlock")
            && method == Method::POST =>
        {
            let id = &p["info/lfs/locks/".len()..p.len() - "/unlock".len()];
            unlock(service, space, repo, id.to_string(), auth).await
        }
        _ => Ok(SmartError::NotFound("lfs endpoint not found".into()).into_response()),
    }
}

async fn batch(
    service: &SmartService,
    space: &str,
    repo: &str,
    req: LfsBatchRequest,
    auth: &BasicAuth,
) -> SmartResult<Response> {
    let svc = &service.git;
    let ctx = service.resolve(space, repo).await?;
    match req.operation {
        LfsOperation::Upload | LfsOperation::Verify => {
            require_writer(svc, auth, &ctx).await?;
        }
        LfsOperation::Download => {
            allow_repo_read(svc, auth, &ctx).await?;
        }
    }
    let resp = object::batch(svc, space.to_string(), repo.to_string(), req).await?;
    Ok((
        StatusCode::OK,
        [(header::CONTENT_TYPE, "application/vnd.git-lfs+json")],
        serde_json::to_vec(&resp).map_err(|e| SmartError::Internal(format!("serialize: {e}")))?,
    )
        .into_response())
}

async fn verify(
    service: &SmartService,
    space: &str,
    repo: &str,
    bytes: bytes::Bytes,
    auth: &BasicAuth,
) -> SmartResult<Response> {
    let svc = &service.git;
    let req: crate::lfs::types::LfsVerifyRequest = serde_json::from_slice(&bytes)
        .map_err(|e| SmartError::BadRequest(format!("invalid verify request: {e}")))?;
    let ctx = service.resolve(space, repo).await?;
    require_writer(svc, auth, &ctx).await?;
    object::verify(svc, space.to_string(), repo.to_string(), req).await?;
    Ok(StatusCode::OK.into_response())
}

#[allow(clippy::too_many_arguments)]
async fn upload(
    service: &SmartService,
    space: &str,
    repo: &str,
    oid: String,
    headers: &HeaderMap,
    body: Body,
    auth: &BasicAuth,
) -> SmartResult<Response> {
    let svc = &service.git;
    let ctx = service.resolve(space, repo).await?;
    let uid = require_writer(svc, auth, &ctx).await?;
    let canonical = crate::lfs::canonical_oid(&oid)?;
    let size = headers
        .get(header::CONTENT_LENGTH)
        .and_then(|v| v.to_str().ok())
        .and_then(|s| s.parse::<i64>().ok())
        .unwrap_or(0);
    let content_type = headers
        .get(header::CONTENT_TYPE)
        .and_then(|v| v.to_str().ok())
        .map(str::to_string);
    let stream = body
        .into_data_stream()
        .map(|chunk| chunk.map_err(|e| std::io::Error::other(e.to_string())));
    let reader = StreamReader::new(stream);
    object::upload_stream(
        svc,
        space.to_string(),
        repo.to_string(),
        uid,
        canonical,
        size,
        content_type,
        reader,
    )
    .await?;
    Ok(StatusCode::OK.into_response())
}

async fn download(
    service: &SmartService,
    space: &str,
    repo: &str,
    oid: &str,
    auth: &BasicAuth,
) -> SmartResult<Response> {
    let svc = &service.git;
    let ctx = service.resolve(space, repo).await?;
    allow_repo_read(svc, auth, &ctx).await?;
    let canonical = crate::lfs::canonical_oid(oid)?;
    let key = format!("lfs/{}/{}", ctx.model.id.simple(), canonical);
    let stream = svc
        .state
        .storage
        .client()
        .get_stream(&key)
        .await
        .map_err(|e| SmartError::Internal(format!("lfs get: {e}")))?;
    let body = Body::from_stream(
        stream.map(|chunk| chunk.map_err(|e| -> axum::Error { axum::Error::new(e.to_string()) })),
    );
    Ok((
        StatusCode::OK,
        [(header::CONTENT_TYPE, "application/octet-stream")],
        body,
    )
        .into_response())
}

async fn list_locks(
    service: &SmartService,
    space: &str,
    repo: &str,
    auth: &BasicAuth,
) -> SmartResult<Response> {
    let svc = &service.git;
    let ctx = service.resolve(space, repo).await?;
    allow_repo_read(svc, auth, &ctx).await?;
    let resp = lock::list(svc, space.to_string(), repo.to_string(), None, 100).await?;
    Ok((
        StatusCode::OK,
        [(header::CONTENT_TYPE, "application/vnd.git-lfs+json")],
        serde_json::to_vec(&resp).map_err(|e| SmartError::Internal(format!("serialize: {e}")))?,
    )
        .into_response())
}

async fn create_lock(
    service: &SmartService,
    space: &str,
    repo: &str,
    body: Body,
    auth: &BasicAuth,
) -> SmartResult<Response> {
    let svc = &service.git;
    let ctx = service.resolve(space, repo).await?;
    let uid = require_writer(svc, auth, &ctx).await?;
    let bytes = axum::body::to_bytes(body, MAX_LFS_BODY)
        .await
        .map_err(|e| SmartError::BadRequest(format!("read body: {e}")))?;
    let req: LfsLockCreateRequest = serde_json::from_slice(&bytes)
        .map_err(|e| SmartError::BadRequest(format!("invalid lock request: {e}")))?;
    let lock = crate::lfs::create_lock(
        svc,
        space.to_string(),
        repo.to_string(),
        uid,
        req.path,
        None,
    )
    .await?;
    Ok((
        StatusCode::CREATED,
        [(header::CONTENT_TYPE, "application/vnd.git-lfs+json")],
        serde_json::to_vec(&json!({ "lock": lock }))
            .map_err(|e| SmartError::Internal(format!("serialize: {e}")))?,
    )
        .into_response())
}

async fn verify_locks(
    service: &SmartService,
    space: &str,
    repo: &str,
    body: Body,
    auth: &BasicAuth,
) -> SmartResult<Response> {
    let svc = &service.git;
    let ctx = service.resolve(space, repo).await?;
    let uid = allow_repo_read(svc, auth, &ctx).await?;
    let uid = uid.ok_or(SmartError::Unauthorized)?;
    let bytes = axum::body::to_bytes(body, MAX_LFS_BODY)
        .await
        .map_err(|e| SmartError::BadRequest(format!("read body: {e}")))?;
    let req: LfsVerifyLocksRequest = serde_json::from_slice(&bytes)
        .map_err(|e| SmartError::BadRequest(format!("invalid request: {e}")))?;
    let resp = lock::verify_locks(
        svc,
        space.to_string(),
        repo.to_string(),
        uid,
        req.limit.unwrap_or(100) as u64,
    )
    .await?;
    Ok((
        StatusCode::OK,
        [(header::CONTENT_TYPE, "application/vnd.git-lfs+json")],
        serde_json::to_vec(&resp).map_err(|e| SmartError::Internal(format!("serialize: {e}")))?,
    )
        .into_response())
}

async fn unlock(
    service: &SmartService,
    space: &str,
    repo: &str,
    lock_id: String,
    auth: &BasicAuth,
) -> SmartResult<Response> {
    let svc = &service.git;
    let ctx = service.resolve(space, repo).await?;
    let uid = require_writer(svc, auth, &ctx).await?;
    crate::lfs::unlock_by_id(
        svc,
        space.to_string(),
        repo.to_string(),
        uid,
        lock_id,
        false,
    )
    .await?;
    Ok((
        StatusCode::OK,
        [(header::CONTENT_TYPE, "application/vnd.git-lfs+json")],
        serde_json::to_vec(&json!({ "message": "unlock successful" }))
            .map_err(|e| SmartError::Internal(format!("serialize: {e}")))?,
    )
        .into_response())
}

async fn read_body(_headers: &HeaderMap, body: Body, limit: usize) -> SmartResult<bytes::Bytes> {
    axum::body::to_bytes(body, limit)
        .await
        .map_err(|e| SmartError::BadRequest(format!("failed to read body: {e}")))
}