use std::collections::HashMap;
use std::pin::Pin;
use std::task::Poll;

use export::storage::AppStorage;
use models::git::git_lfs;
use models::repo::repos;
use sea_orm::sea_query::Expr;
use sea_orm::{ColumnTrait, EntityTrait, ExprTrait, QueryFilter, Set};
use sha2::{Digest, Sha256};
use time::OffsetDateTime;
use uuid::Uuid;

use crate::lfs::types::{
    LfsAction, LfsBatchObject, LfsBatchRequest, LfsBatchResponse, LfsError, LfsOperation,
    LfsVerifyRequest,
};
use crate::lfs::validate::{canonical_oid, validate_size};
use crate::service::GitService;
use crate::share::errors::GitError;

pub const UPLOAD_TTL_SECS: u32 = 3600;
pub const DOWNLOAD_TTL_SECS: u32 = 3600;
pub const MAX_BATCH_OBJECTS: usize = 200;

pub async fn batch(
    svc: &GitService,
    space_name: String,
    repo_name: String,
    req: LfsBatchRequest,
) -> Result<LfsBatchResponse, GitError> {
    if req.objects.len() > MAX_BATCH_OBJECTS {
        return Err(GitError::InvalidArgument(format!(
            "batch request exceeds {MAX_BATCH_OBJECTS} objects"
        )));
    }
    let ctx = svc.context(space_name.clone(), repo_name.clone()).await?;
    if !ctx.model.enable_lfs {
        return Err(GitError::NotFound);
    }
    let storage: &AppStorage = &svc.state.storage;
    let cap = svc.state.config.git.max_blob_size;
    let mut out: Vec<LfsBatchObject> = Vec::with_capacity(req.objects.len());
    for obj in req.objects {
        let oid = match canonical_oid(&obj.oid) {
            Ok(oid) => oid,
            Err(e) => {
                out.push(object_error(obj.oid, obj.size, e));
                continue;
            }
        };
        if let Err(e) = validate_size(obj.size, cap) {
            out.push(object_error(oid, obj.size, e));
            continue;
        }
        let present = storage
            .client()
            .exists(&storage_key(&ctx.model.id, &oid))
            .await
            .map_err(|e| GitError::Internal(format!("storage exists: {e}")))?;
        let actions = build_actions(
            svc,
            &ctx.model,
            &space_name,
            &repo_name,
            &oid,
            req.operation,
            present,
        )?;
        let mut resp = LfsBatchObject {
            oid,
            size: obj.size,
            actions,
            error: None,
        };
        if req.operation == LfsOperation::Download && !present {
            resp.error = Some(LfsError {
                code: 404,
                message: "object not found".to_string(),
            });
        }
        out.push(resp);
    }
    Ok(LfsBatchResponse {
        transfer: "basic".to_string(),
        hash_algo: "sha256".to_string(),
        objects: out,
    })
}

fn build_actions(
    svc: &GitService,
    repo: &repos::Model,
    space_name: &str,
    repo_name: &str,
    oid: &str,
    operation: LfsOperation,
    present: bool,
) -> Result<HashMap<String, LfsAction>, GitError> {
    let storage: &AppStorage = &svc.state.storage;
    let mut actions: HashMap<String, LfsAction> = HashMap::new();
    match (operation, present) {
        (LfsOperation::Download, true) | (LfsOperation::Verify, true) => {
            actions.insert(
                "download".to_string(),
                LfsAction {
                    href: storage
                        .client()
                        .presign_get(&storage_key(&repo.id, oid), DOWNLOAD_TTL_SECS)
                        .map_err(|e| GitError::Internal(format!("presign_get: {e}")))?,
                    header: None,
                    expires_at: Some(now_plus_secs(DOWNLOAD_TTL_SECS as i64)),
                },
            );
        }
        (LfsOperation::Upload, _) => {
            let href = storage
                .client()
                .presign_put(&storage_key(&repo.id, oid), UPLOAD_TTL_SECS)
                .map_err(|e| GitError::Internal(format!("presign_put: {e}")))?;
            actions.insert(
                "upload".to_string(),
                LfsAction {
                    href: href.clone(),
                    header: None,
                    expires_at: Some(now_plus_secs(UPLOAD_TTL_SECS as i64)),
                },
            );
            if !present {
                actions.insert(
                    "verify".to_string(),
                    LfsAction {
                        href: verify_href(svc, space_name, repo_name),
                        header: None,
                        expires_at: Some(now_plus_secs(UPLOAD_TTL_SECS as i64)),
                    },
                );
            }
        }
        _ => {}
    }
    Ok(actions)
}

/// Stream an object upload: the reader is consumed chunk-by-chunk into
/// storage while its bytes are hashed with SHA-256 on the fly. The hash is
/// compared against the canonical oid only after the upload finishes; a
/// mismatch (or declared-size mismatch) deletes the stored object again.
/// Uploads are idempotent (already-present objects succeed without a second
/// write). `repos.lfs_size` is incremented atomically.
#[allow(clippy::too_many_arguments)]
pub async fn upload_stream<R>(
    svc: &GitService,
    space_name: String,
    repo_name: String,
    caller: i64,
    oid: String,
    size: i64,
    content_type: Option<String>,
    reader: R,
) -> Result<bool, GitError>
where
    R: tokio::io::AsyncRead + Unpin + Send,
{
    let _ = caller;
    let ctx = svc.context(space_name, repo_name).await?;
    if !ctx.model.enable_lfs {
        return Err(GitError::NotFound);
    }
    let oid = canonical_oid(&oid)?;
    validate_size(size, svc.state.config.git.max_blob_size)?;
    let storage: &AppStorage = &svc.state.storage;
    let key = storage_key(&ctx.model.id, &oid);
    let existed = storage
        .client()
        .exists(&key)
        .await
        .map_err(|e| GitError::Internal(format!("storage exists: {e}")))?;

    if !existed {
        let mut reader = HashingReader::new(tokio::io::AsyncReadExt::take(reader, size as u64 + 1));
        let sent = storage
            .client()
            .put_stream(
                &key,
                &mut reader,
                content_type
                    .as_deref()
                    .unwrap_or("application/octet-stream"),
            )
            .await
            .map_err(|e| GitError::Internal(format!("lfs put_stream: {e}")))?;
        let (hex, _count) = reader.finish();
        if sent as i64 != size {
            let _ = storage.client().delete(&key).await;
            return Err(GitError::InvalidArgument(format!(
                "size mismatch: declared {size}, received {sent}"
            )));
        }
        let actual = format!("sha256:{hex}");
        if actual != oid {
            let _ = storage.client().delete(&key).await;
            return Err(GitError::InvalidArgument(format!(
                "oid mismatch: expected {oid}, got {actual}"
            )));
        }
    }
    match git_lfs::Entity::insert(git_lfs::ActiveModel {
        id: Set(Uuid::now_v7()),
        repo_id: Set(ctx.model.id),
        oid: Set(oid),
        size: Set(size),
        content_type: Set(content_type),
        created_at: Set(OffsetDateTime::now_utc()),
    })
    .exec(svc.db_write())
    .await
    {
        Ok(_) => {}
        Err(db) => match GitError::from(db) {
            GitError::Conflict(_) => {}
            other => return Err(other),
        },
    }

    if !existed {
        let update = repos::Entity::update_many()
            .col_expr(
                repos::Column::LfsSize,
                Expr::col(repos::Column::LfsSize).add(size),
            )
            .col_expr(
                repos::Column::UpdatedAt,
                Expr::val(OffsetDateTime::now_utc()),
            )
            .filter(repos::Column::Id.eq(ctx.model.id));
        let res = if let Some(cap) = ctx.model.storage_limit_size {
            let limit = cap.saturating_sub(size);
            update
                .filter(Expr::col(repos::Column::LfsSize).lte(limit))
                .exec(svc.db_write())
                .await?
        } else {
            update.exec(svc.db_write()).await?
        };
        if res.rows_affected == 0 {
            let _ = storage.client().delete(&key).await;
            return Err(GitError::InvalidArgument(
                "lfs storage limit exceeded".to_string(),
            ));
        }
    }
    Ok(true)
}

/// In-memory convenience wrapper over [`upload_stream`] for small bodies.
#[allow(clippy::too_many_arguments)]
pub async fn upload(
    svc: &GitService,
    space_name: String,
    repo_name: String,
    caller: i64,
    oid: String,
    size: i64,
    content_type: Option<String>,
    body: Vec<u8>,
) -> Result<bool, GitError> {
    upload_stream(
        svc,
        space_name,
        repo_name,
        caller,
        oid,
        size,
        content_type,
        std::io::Cursor::new(body),
    )
    .await
}

/// `POST objects/verify` — confirm a stored object matches the recorded
/// `(oid, size)`; fails with `NotFound` when the object or its DB row is
/// missing.
pub async fn verify(
    svc: &GitService,
    space_name: String,
    repo_name: String,
    req: LfsVerifyRequest,
) -> Result<(), GitError> {
    let oid = canonical_oid(&req.oid)?;
    let ctx = svc.context(space_name, repo_name).await?;
    let row = git_lfs::Entity::find()
        .filter(git_lfs::Column::RepoId.eq(ctx.model.id))
        .filter(git_lfs::Column::Oid.eq(&oid))
        .one(svc.db_read())
        .await?
        .ok_or(GitError::NotFound)?;
    if row.size != req.size {
        return Err(GitError::InvalidArgument("size mismatch".to_string()));
    }
    let storage: &AppStorage = &svc.state.storage;
    let present = storage
        .client()
        .exists(&storage_key(&ctx.model.id, &oid))
        .await
        .map_err(|e| GitError::Internal(format!("storage exists: {e}")))?;
    if !present {
        return Err(GitError::NotFound);
    }
    Ok(())
}

/// Stream an object into `writer` (bounded memory, no whole-object buffering).
/// Returns the number of bytes written.
pub async fn download_to<W>(
    svc: &GitService,
    space_name: String,
    repo_name: String,
    oid: String,
    writer: &mut W,
) -> Result<u64, GitError>
where
    W: tokio::io::AsyncWrite + Send + Unpin,
{
    let oid = canonical_oid(&oid)?;
    let ctx = svc.context(space_name, repo_name).await?;
    let storage: &AppStorage = &svc.state.storage;
    storage
        .client()
        .get_to_writer(&storage_key(&ctx.model.id, &oid), writer)
        .await
        .map_err(|e| GitError::Internal(format!("lfs get: {e}")))
}

/// Whole-object convenience wrapper over [`download_to`] (use only for small
/// objects; prefer the streaming variant in hot paths).
pub async fn download(
    svc: &GitService,
    space_name: String,
    repo_name: String,
    oid: String,
) -> Result<Vec<u8>, GitError> {
    let mut buf = Vec::new();
    download_to(svc, space_name, repo_name, oid, &mut buf).await?;
    Ok(buf)
}

/// Consistency repair + unreferenced-object sweep for one repo:
///
/// 1. drop `git_lfs` rows whose storage object has vanished;
/// 2. delete storage keys under `lfs/{repo}/` that have no DB row.
///
/// Returns the number of removed entries.
pub async fn cleanup_unreferenced(
    svc: &GitService,
    space_name: String,
    repo_name: String,
) -> Result<i64, GitError> {
    let ctx = svc.context(space_name, repo_name).await?;
    let repo_id = ctx.model.id;
    let storage: &AppStorage = &svc.state.storage;
    let mut removed: i64 = 0;
    let db = svc.db_write();

    let rows = git_lfs::Entity::find()
        .filter(git_lfs::Column::RepoId.eq(repo_id))
        .all(db)
        .await?;
    for row in rows {
        let key = storage_key(&repo_id, &row.oid);
        let exists = storage
            .client()
            .exists(&key)
            .await
            .map_err(|e| GitError::Internal(format!("storage exists: {e}")))?;
        if !exists {
            git_lfs::Entity::delete_by_id(row.id).exec(db).await?;
            removed += 1;
        }
    }

    let prefix = format!("lfs/{}/", repo_id.simple());
    let live: std::collections::HashSet<String> = git_lfs::Entity::find()
        .filter(git_lfs::Column::RepoId.eq(repo_id))
        .all(svc.db_read())
        .await?
        .into_iter()
        .map(|r| r.oid)
        .collect();
    let mut token: Option<String> = None;
    loop {
        let page = storage
            .client()
            .list_page(&prefix, token.as_deref(), Some(200))
            .await
            .map_err(|e| GitError::Internal(format!("storage list: {e}")))?;
        for obj in page.objects {
            let Some(oid) = obj.key.strip_prefix(&prefix) else {
                continue;
            };
            if !live.contains(oid) {
                storage
                    .client()
                    .delete(&obj.key)
                    .await
                    .map_err(|e| GitError::Internal(format!("storage delete: {e}")))?;
                removed += 1;
            }
        }
        match page.next_continuation_token {
            Some(next) if !next.is_empty() => token = Some(next),
            _ => break,
        }
    }
    Ok(removed)
}

fn object_error(oid: String, size: i64, err: GitError) -> LfsBatchObject {
    LfsBatchObject {
        oid,
        size,
        actions: HashMap::new(),
        error: Some(LfsError {
            code: crate::lfs::error::http_status(&err) as i32,
            message: err.to_string(),
        }),
    }
}

fn storage_key(repo_id: &Uuid, oid: &str) -> String {
    format!("lfs/{}/{}", repo_id.simple(), oid)
}

/// Tee between the upload reader and storage: every byte read is fed into a
/// running SHA-256 and counted, so the upload can be verified without
/// buffering the whole body.
struct HashingReader<R> {
    inner: R,
    hasher: Sha256,
    count: u64,
}

impl<R> HashingReader<R> {
    fn new(inner: R) -> Self {
        Self {
            inner,
            hasher: Sha256::new(),
            count: 0,
        }
    }

    fn finish(self) -> (String, u64) {
        let hex = self
            .hasher
            .finalize()
            .iter()
            .map(|b| format!("{b:02x}"))
            .collect();
        (hex, self.count)
    }
}

impl<R: tokio::io::AsyncRead + Unpin> tokio::io::AsyncRead for HashingReader<R> {
    fn poll_read(
        mut self: Pin<&mut Self>,
        cx: &mut std::task::Context<'_>,
        buf: &mut tokio::io::ReadBuf<'_>,
    ) -> Poll<std::io::Result<()>> {
        let before = buf.filled().len();
        let poll = Pin::new(&mut self.inner).poll_read(cx, buf);
        if poll.is_ready() {
            let written = &buf.filled()[before..];
            if !(written.is_empty() && buf.remaining() > 0) {
                self.hasher.update(written);
                self.count += written.len() as u64;
            }
        }
        poll
    }
}

fn now_plus_secs(secs: i64) -> String {
    OffsetDateTime::now_utc()
        .saturating_add(time::Duration::seconds(secs))
        .format(&time::format_description::well_known::Rfc3339)
        .unwrap_or_default()
}

/// The batch `verify` action always points back at this server (the initial
/// upload itself goes directly to the presigned S3 URL). The href follows the
/// Gitea-style `/{space}/{repo}.git/info/lfs` layout, prefixed with
/// `web.web_base_url` when configured.
fn verify_href(svc: &GitService, space_name: &str, repo_name: &str) -> String {
    let base = svc.state.config.web.web_base_url.trim_end_matches('/');
    if base.is_empty() {
        format!("/{}/{}.git/info/lfs/objects/verify", space_name, repo_name)
    } else {
        format!(
            "{base}/{}/{}.git/info/lfs/objects/verify",
            space_name, repo_name
        )
    }
}