use std::path::Path;
use std::process::Stdio;
use std::sync::Arc;
use std::time::Duration;

use bytes::Bytes;
use futures::StreamExt;
use tokio::io::{AsyncReadExt, AsyncWriteExt};
use tokio::process::Command;
use tokio::sync::{Mutex, oneshot};
use tokio_util::io::ReaderStream;

use crate::smart::error::{SmartError, SmartResult};

pub type RpcStream = futures::stream::BoxStream<'static, Result<Bytes, std::io::Error>>;

const MAX_STDERR_BYTES: usize = 64 * 1024 * 1024;

#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum GitService {
    UploadPack,
    ReceivePack,
}

impl GitService {
    pub fn from_name(name: &str) -> SmartResult<Self> {
        match name {
            "git-upload-pack" => Ok(Self::UploadPack),
            "git-receive-pack" => Ok(Self::ReceivePack),
            _ => Err(SmartError::BadRequest(format!(
                "unsupported git service: {name}"
            ))),
        }
    }

    pub fn name(self) -> &'static str {
        match self {
            Self::UploadPack => "upload-pack",
            Self::ReceivePack => "receive-pack",
        }
    }

    pub fn dashed_name(self) -> &'static str {
        match self {
            Self::UploadPack => "git-upload-pack",
            Self::ReceivePack => "git-receive-pack",
        }
    }

    pub fn advertisement_content_type(self) -> &'static str {
        match self {
            Self::UploadPack => "application/x-git-upload-pack-advertisement",
            Self::ReceivePack => "application/x-git-receive-pack-advertisement",
        }
    }

    pub fn result_content_type(self) -> &'static str {
        match self {
            Self::UploadPack => "application/x-git-upload-pack-result",
            Self::ReceivePack => "application/x-git-receive-pack-result",
        }
    }
}

fn build_command(bare_path: &Path, service: GitService, advertise: bool) -> Command {
    let mut cmd = Command::new("git");
    cmd.arg(service.name()).arg("--stateless-rpc");
    if advertise {
        cmd.arg("--advertise-refs");
    }
    cmd.arg(bare_path)
        .env("GIT_DIR", bare_path)
        .env("GIT_PROTOCOL", "version=2")
        .env("GIT_TERMINAL_PROMPT", "0")
        .env("GIT_CONFIG_NOSYSTEM", "1")
        .env("GIT_NO_REPLACE_OBJECTS", "1")
        .stdin(Stdio::piped())
        .stdout(Stdio::piped())
        .stderr(Stdio::piped());
    cmd
}

#[allow(clippy::type_complexity)]
fn spawn_with_supervisor(
    bare_path: &Path,
    service: GitService,
    advertise: bool,
    input: Bytes,
    timeout_secs: u64,
) -> SmartResult<(
    RpcStream,
    Option<tokio::sync::oneshot::Receiver<SmartResult<()>>>,
)> {
    let mut cmd = build_command(bare_path, service, advertise);
    let mut child = cmd.spawn().map_err(|e| {
        SmartError::Internal(format!("failed to spawn git {}: {e}", service.name()))
    })?;
    tracing::debug!(
        svc = service.name(),
        advertise,
        bare = %bare_path.display(),
        input_bytes = input.len(),
        timeout_secs,
        pid = ?child.id(),
        "spawned git rpc process"
    );
    let mut stdin = child
        .stdin
        .take()
        .ok_or_else(|| SmartError::Internal("stdin not piped".to_string()))?;
    let stdout = child
        .stdout
        .take()
        .ok_or_else(|| SmartError::Internal("stdout not piped".to_string()))?;
    let mut stderr = child
        .stderr
        .take()
        .ok_or_else(|| SmartError::Internal("stderr not piped".to_string()))?;

    let child_arc = Arc::new(Mutex::new(Some(child)));

    {
        let child_arc = child_arc.clone();
        tokio::spawn(async move {
            if stdin.write_all(&input).await.is_err() {
                tracing::warn!(
                    svc = service.name(),
                    "writing rpc input to git stdin failed; killing process"
                );
                if let Some(mut ch) = child_arc.lock().await.take() {
                    let _ = ch.start_kill();
                    let _ = ch.wait().await;
                }
            }
            let _ = stdin.shutdown().await;
        });
    }

    let (tx, rx) = oneshot::channel();
    {
        let child_arc = child_arc.clone();
        tokio::spawn(async move {
            let status = {
                let mut guard = child_arc.lock().await;
                match guard.as_mut() {
                    None => None,
                    Some(ch) => {
                        match tokio::time::timeout(Duration::from_secs(timeout_secs), ch.wait())
                            .await
                        {
                            Ok(Ok(s)) => Some(s),
                            _ => {
                                tracing::warn!(
                                    svc = service.name(),
                                    timeout_secs,
                                    "git rpc process timed out; killing"
                                );
                                let _ = ch.start_kill();
                                let _ = ch.wait().await;
                                None
                            }
                        }
                    }
                }
            };
            let mut buf = vec![0u8; 8192];
            let mut total = 0usize;
            let mut text = Vec::new();
            loop {
                match stderr.read(&mut buf).await {
                    Ok(0) | Err(_) => break,
                    Ok(n) => {
                        total += n;
                        if total <= MAX_STDERR_BYTES {
                            text.extend_from_slice(&buf[..n]);
                        }
                    }
                }
            }
            let verdict = match status {
                Some(s) if s.success() => {
                    tracing::debug!(
                        svc = service.name(),
                        stderr_bytes = total,
                        "git rpc process exited successfully"
                    );
                    Ok(())
                }
                Some(s) => {
                    tracing::warn!(
                        svc = service.name(),
                        exit = %s,
                        stderr = %String::from_utf8_lossy(text.as_slice()).trim(),
                        "git rpc process exited with error"
                    );
                    Err(SmartError::Internal(format!(
                        "git {} exited with {s}: {}",
                        service.name(),
                        String::from_utf8_lossy(text.as_slice()).trim()
                    )))
                }
                None => Err(SmartError::Internal(format!(
                    "git {} timed out after {timeout_secs}s",
                    service.name()
                ))),
            };
            let _ = tx.send(verdict);
        });
    }

    let stream: RpcStream = Box::pin(ReaderStream::new(stdout));
    Ok((stream, Some(rx)))
}

pub async fn run_rpc_stream(
    bare_path: &Path,
    service: GitService,
    advertise: bool,
    input: Bytes,
    timeout_secs: u64,
) -> SmartResult<RpcStream> {
    let (stream, _rx) = spawn_with_supervisor(bare_path, service, advertise, input, timeout_secs)?;
    Ok(stream)
}

pub async fn run_rpc_bytes(
    bare_path: &Path,
    service: GitService,
    advertise: bool,
    input: Bytes,
    timeout_secs: u64,
) -> SmartResult<Bytes> {
    let (mut stream, rx) =
        spawn_with_supervisor(bare_path, service, advertise, input, timeout_secs)?;
    let mut out: Vec<u8> = Vec::new();
    while let Some(chunk) = stream.next().await {
        out.extend_from_slice(&chunk.map_err(|e| {
            SmartError::Internal(format!("git {} stdout read failed: {e}", service.name()))
        })?);
    }
    if let Some(rx) = rx {
        let _ = rx.await;
    }
    tracing::debug!(
        svc = service.name(),
        output_bytes = out.len(),
        "git rpc output collected"
    );
    Ok(Bytes::from(out))
}