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::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_with_ci_logged;

const MAX_RPC_BODY: usize = 512 * 1024 * 1024;

#[tracing::instrument(name = "git.info_refs", skip_all, fields(space, repo))]
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);
    tracing::debug!(
        space,
        repo,
        service = service_kind.unwrap_or("dumb"),
        bare = %bare.display(),
        "info/refs authorized, generating advertisement"
    );
    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)]
#[tracing::instrument(name = "git.rpc", skip_all, fields(space, repo, op = ?git_service))]
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?;
    tracing::info!(
        space,
        repo,
        service = git_service.name(),
        actor_id = ?actor_id,
        repo_id = %ctx.model.id,
        "git rpc authorized, processing request"
    );

    let run = async {
        let input = decode_request_body(headers, body, MAX_RPC_BODY).await?;
        let is_receive = git_service == GitService::ReceivePack;
        tracing::debug!(
            service = git_service.name(),
            body_bytes = input.len(),
            "rpc request body decoded"
        );
        if is_receive && actor_id.is_some() {
            let updates = parse_receive_pack_refs(&input)?;
            tracing::debug!(
                repo_id = %ctx.model.id,
                ref_count = updates.len(),
                "parsed receive-pack ref updates"
            );
            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()?;
                tracing::debug!(repo_id = %ctx.model.id, "pre-receive validation passed");
            }
            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 = PostReceiveStream {
            is_receive,
            queue: service.queue.clone(),
            settings: service.sync_settings.clone(),
            ci_topic: service.ci_event_topic().map(|t| t.to_string()),
            space_id,
            repo_id: ctx.model.id,
            actor_id,
            bare: 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<i64>> {
    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)
    }
}

/// Everything the post-receive hook needs once the response body has been
/// streamed through. A named struct rather than a nine-element tuple: the fields
/// are all `Uuid`-shaped, so positional access silently swaps tenant and repo.
pub struct PostReceiveStream {
    pub is_receive: bool,
    pub queue: Option<Arc<crate::sync::queue::SyncQueue>>,
    pub settings: config::git::SyncSettings,
    pub ci_topic: Option<String>,
    pub space_id: Uuid,
    pub repo_id: Uuid,
    pub actor_id: Option<i64>,
    pub bare: std::path::PathBuf,
}

fn body_stream_with_post_receive(
    stream: impl futures::Stream<Item = Result<Bytes, axum::Error>> + Send + 'static,
    post: PostReceiveStream,
) -> impl futures::Stream<Item = Result<Bytes, axum::Error>> + Send + 'static {
    use futures::{StreamExt, stream};
    stream.chain(stream::once(async move {
        let PostReceiveStream {
            is_receive,
            queue,
            settings,
            ci_topic,
            space_id,
            repo_id,
            actor_id,
            bare,
        } = post;
        if is_receive && let Some(queue) = queue {
            enqueue_post_receive_push_with_ci_logged(
                &queue,
                &settings,
                ci_topic.as_deref(),
                &crate::sync::PushTarget {
                    space_id,
                    repo_id,
                    actor_id,
                    bare: &bare,
                },
            )
            .await;
        }
        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"
            )));
        }
        tracing::debug!(
            compressed = bytes.len(),
            decompressed = out.len(),
            "gzip request body decompressed"
        );
        Ok(Bytes::from(out))
    } else {
        Ok(bytes)
    }
}