use std::path::PathBuf;
use std::sync::Arc;

use axum::body::Body;
use axum::http::{HeaderMap, StatusCode, header};
use axum::response::{IntoResponse, Response};
use bytes::Bytes;
use flate2::read::GzDecoder;
use futures::StreamExt as _;
use std::io::Read;
use uuid::Uuid;

use crate::context::facade::GitContext;
use crate::share::access::{BasicAuth, allow_repo_read, require_writer};
use crate::smart::error::{SmartError, SmartResult};
use crate::smart::info_refs as info_refs_module;
use crate::smart::pre_receive::{self, parse_receive_pack_refs};
use crate::smart::rpc::{self, GitService};
use crate::smart::service::SmartService;
use crate::sync::post_receive::enqueue_post_receive_push;

const MAX_RPC_BODY: usize = 512 * 1024 * 1024;

pub async fn info_refs(
    service: &SmartService,
    space: &str,
    repo: &str,
    _headers: &HeaderMap,
    auth: &BasicAuth,
    service_kind: Option<&str>,
) -> SmartResult<Response> {
    let (ctx, _actor_id) = match service_kind {
        Some(name) => {
            let git_service = GitService::from_name(name)?;
            let want_write = matches!(git_service, GitService::ReceivePack);
            let ctx = service.resolve(space, repo).await?;
            let actor_id = authorize(service, &ctx, auth, want_write).await?;
            (ctx, actor_id)
        }
        None => {
            let ctx = service.resolve(space, repo).await?;
            allow_repo_read(&service.git, auth, &ctx).await?;
            (ctx, None)
        }
    };
    let bare = service.bare_path(&ctx);
    match service_kind {
        Some(name) => {
            let git_service = GitService::from_name(name)?;
            let timeout = service.git.state.config.git.git_rpc_timeout_secs;
            info_refs_module::smart_advertisement(&bare, git_service, timeout).await
        }
        None => info_refs_module::dumb_info_refs(&bare).await,
    }
}

#[allow(clippy::too_many_arguments)]
pub async fn rpc(
    service: &SmartService,
    space: &str,
    repo: &str,
    headers: &HeaderMap,
    body: Body,
    auth: &BasicAuth,
    git_service: GitService,
) -> SmartResult<Response> {
    let want_write = matches!(git_service, GitService::ReceivePack);
    let ctx = service.resolve(space, repo).await?;
    let actor_id = authorize(service, &ctx, auth, want_write).await?;
    let bare = service.bare_path(&ctx);
    let space_id = service.owning_space_id(&ctx).await?;

    let run = async {
        let input = decode_request_body(headers, body, MAX_RPC_BODY).await?;
        let is_receive = git_service == GitService::ReceivePack;
        if is_receive && actor_id.is_some() {
            let updates = parse_receive_pack_refs(&input)?;
            if !updates.is_empty()
                && let Some(actor_uid) = actor_id {
                    let baby = service.git.baby(&ctx);
                    let verdict = pre_receive::validate_pre_receive(
                        service.git.db_read(),
                        &baby,
                        ctx.model.id,
                        actor_uid,
                        &updates,
                    )
                    .await?;
                    verdict.accepted_only()?;
                }
            if let Err(e) = crate::sync::post_receive::prepare_post_receive(&bare) {
                tracing::warn!(?e, "post-receive hook setup failed");
            }
        }
        let timeout = service.git.state.config.git.git_rpc_timeout_secs;
        let stream = rpc::run_rpc_stream(&bare, git_service, false, input, timeout).await?;
        let post = (
            is_receive,
            service.queue.clone(),
            service.sync_settings.clone(),
            space_id,
            ctx.model.id,
            actor_id,
            bare.clone(),
        );
        let stream = stream.map(|chunk| chunk.map_err(|e| axum::Error::new(e.to_string())));
        let stream = body_stream_with_post_receive(stream, post);
        Ok::<_, SmartError>(
            (
                StatusCode::OK,
                [
                    (header::CONTENT_TYPE, git_service.result_content_type()),
                    (header::CACHE_CONTROL, "no-cache"),
                ],
                Body::from_stream(stream),
            )
                .into_response(),
        )
    };

    match git_service {
        GitService::ReceivePack => {
            let key = format!("push:{}:{}", space_id, ctx.model.id);
            service.push_coordinator().run(&key, run).await
        }
        GitService::UploadPack => run.await,
    }
}

async fn authorize(
    service: &SmartService,
    ctx: &crate::context::facade::GitContext,
    auth: &BasicAuth,
    want_write: bool,
) -> SmartResult<Option<Uuid>> {
    let svc = &service.git;
    if want_write {
        let uid = require_writer(svc, auth, ctx).await?;
        Ok(Some(uid))
    } else {
        let uid = allow_repo_read(svc, auth, ctx).await?;
        Ok(uid)
    }
}

fn body_stream_with_post_receive(
    stream: impl futures::Stream<Item = Result<Bytes, axum::Error>> + Send + 'static,
    post: (
        bool,
        Option<Arc<crate::sync::queue::SyncQueue>>,
        config::git::SyncSettings,
        Uuid,
        Uuid,
        Option<Uuid>,
        std::path::PathBuf,
    ),
) -> impl futures::Stream<Item = Result<Bytes, axum::Error>> + Send + 'static {
    use futures::{StreamExt, stream};
    stream.chain(stream::once(async move {
        let (is_receive, queue, settings, space_id, repo_id, actor_id, bare) = post;
        if is_receive
            && let Some(queue) = queue
        {
            match enqueue_post_receive_push(
                &queue,
                &settings,
                space_id,
                repo_id,
                actor_id,
                &bare,
            )
            .await
            {
                Ok(n) if n > 0 => {
                    tracing::debug!(count = n, "enqueued post-receive ref updates")
                }
                Ok(_) => tracing::debug!("no refs to enqueue"),
                Err(e) => tracing::warn!(?e, "post-receive enqueue failed"),
            }
        }
        Ok::<_, axum::Error>(Bytes::new())
    }))
}

pub async fn decode_request_body(
    headers: &HeaderMap,
    body: Body,
    limit: usize,
) -> SmartResult<Bytes> {
    let bytes = axum::body::to_bytes(body, limit)
        .await
        .map_err(|e| SmartError::BadRequest(format!("failed to read request body: {e}")))?;
    let encoding = headers
        .get(header::CONTENT_ENCODING)
        .and_then(|v| v.to_str().ok())
        .map(|v| v.to_ascii_lowercase());
    if encoding.as_deref() == Some("gzip") {
        let decoder = GzDecoder::new(bytes.as_ref());
        let mut out = Vec::new();
        let cap = limit as u64 + 1;
        decoder
            .take(cap)
            .read_to_end(&mut out)
            .map_err(|e| SmartError::BadRequest(format!("failed to decompress gzip body: {e}")))?;
        if out.len() as u64 > limit as u64 {
            return Err(SmartError::BadRequest(format!(
                "decompressed body exceeds limit of {limit} bytes"
            )));
        }
        Ok(Bytes::from(out))
    } else {
        Ok(bytes)
    }
}

#[allow(dead_code)]
fn _unused(_ctx: &GitContext, _p: &PathBuf) {}