use std::sync::Arc;

use bytes::Bytes;
use cache::CacheClient;
use russh::ChannelId;
use russh::server::Session;
use sea_orm::EntityTrait;
use tokio::io::{AsyncReadExt, AsyncWriteExt};

use crate::context::facade::GitContext;
use crate::smart::error::SmartError;
use crate::smart::pre_receive::parse_receive_pack_refs;
use crate::ssh::error::SshError;
use crate::ssh::git_service::apply_hardened_env;
use crate::ssh::handler::GitSshHandler;
use crate::sync::post_receive::enqueue_post_receive_push;

pub const MAX_SSH_CHANNEL_BUF: usize = 64 * 1024 * 1024;
pub const MAX_GIT_OUTPUT_BUF: usize = 256 * 1024 * 1024;

impl GitSshHandler {
    pub async fn data_impl(
        &mut self,
        channel: ChannelId,
        data: &[u8],
        session: &mut Session,
    ) -> Result<(), russh::Error> {
        let state = self.channels.entry(channel).or_default();
        if state.buffer_overflow {
            return Ok(());
        }
        let projected = state.buffer.len().saturating_add(data.len());
        if projected > MAX_SSH_CHANNEL_BUF {
            tracing::warn!(
                channel = ?channel,
                buffered = state.buffer.len(),
                incoming = data.len(),
                limit = MAX_SSH_CHANNEL_BUF,
                "ssh channel buffer overflow; closing channel"
            );
            state.buffer_overflow = true;
            state.buffer.clear();
            let _ = session.close(channel);
            let _ = session.flush();
            return Ok(());
        }
        state.buffer.extend_from_slice(data);
        Ok(())
    }

    pub async fn channel_close_impl(&mut self, channel: ChannelId) {
        self.channels.remove(&channel);
    }

    pub async fn channel_eof_impl(
        &mut self,
        channel: ChannelId,
        session: &mut Session,
    ) -> Result<(), russh::Error> {
        let Some(mut state) = self.channels.remove(&channel) else {
            let _ = session.close(channel);
            let _ = session.flush();
            return Ok(());
        };
        let overflowed = state.buffer_overflow;
        let op = state.git_op.take();
        let buffer = std::mem::take(&mut state.buffer);
        let repo_id = state.repo_id;
        let space_id = state.space_id;
        let actor_id = state.actor_id;
        let push_coordinator = self.service.push_coordinator();
        let queue = self.service.smart.queue.clone();
        let sync_settings = self.service.sync_settings.clone();
        let ssh_service = self.service.clone();

        let handle = session.handle();
        let db = self.service.smart.git.state.database.clone();
        let rpc_timeout = self.service.smart.git.state.config.git.git_rpc_timeout_secs;
        tokio::spawn(async move {
            if overflowed {
                _ = handle.exit_status_request(channel, 1);
                _ = handle.eof(channel);
                _ = handle.close(channel);
                let _ = handle
                    .data(
                        channel,
                        Bytes::from_static(
                            b"remote: request rejected: ssh channel buffer cap exceeded\n",
                        ),
                    )
                    .await;
                return;
            }
            let is_write = op.as_ref().is_some_and(|o| o.is_write());
            let is_receive = matches!(
                op.as_ref(),
                Some(crate::ssh::handler::git_op::GitOperation::ReceivePack { .. })
            );

            if is_receive && let (Some(repo_id), Some(actor_uid)) = (repo_id, actor_id) {
                match parse_receive_pack_refs(&buffer) {
                    Ok(updates) if !updates.is_empty() => {
                        let baby = match build_baby_for_pre_receive(&ssh_service, repo_id).await {
                            Some(b) => b,
                            None => {
                                tracing::error!(
                                    repo_id = %repo_id,
                                    "cannot build baby for pre-receive; refusing push"
                                );
                                let _ = handle
                                    .data(
                                        channel,
                                        Bytes::from_static(
                                            b"remote: internal error during pre-receive setup\n",
                                        ),
                                    )
                                    .await;
                                _ = handle.exit_status_request(channel, 1);
                                _ = handle.eof(channel);
                                _ = handle.close(channel);
                                return;
                            }
                        };
                        let verdict = match crate::smart::pre_receive::validate_pre_receive(
                            db.read(),
                            &baby,
                            repo_id,
                            actor_uid,
                            &updates,
                        )
                        .await
                        {
                            Ok(v) => v,
                            Err(e) => {
                                tracing::error!(
                                    ?e, repo_id = %repo_id,
                                    "pre-receive validation failed; refusing push"
                                );
                                let _ = handle
                                    .data(
                                        channel,
                                        Bytes::from_static(
                                            b"remote: internal error during pre-receive validation\n",
                                        ),
                                    )
                                    .await;
                                _ = handle.exit_status_request(channel, 1);
                                _ = handle.eof(channel);
                                _ = handle.close(channel);
                                return;
                            }
                        };
                        if let Err(e) = verdict.accepted_only() {
                            let msg = format!("remote: {e}\n");
                            _ = handle.data(channel, Bytes::from(msg.into_bytes())).await;
                            _ = handle.exit_status_request(channel, 1);
                            _ = handle.eof(channel);
                            _ = handle.close(channel);
                            return;
                        }
                    }
                    Ok(_) => {}
                    Err(e) => {
                        tracing::error!(
                            ?e, repo_id = %repo_id,
                            "receive-pack ref parse failed; refusing push"
                        );
                        let _ = handle
                            .data(
                                channel,
                                Bytes::from_static(
                                    b"remote: invalid receive-pack ref update\n",
                                ),
                            )
                            .await;
                        _ = handle.exit_status_request(channel, 1);
                        _ = handle.eof(channel);
                        _ = handle.close(channel);
                        return;
                    }
                }
            }

            if is_write && let Some(bare) = op.as_ref().map(|o| o.bare().clone()) {
                if let Err(e) = crate::sync::post_receive::prepare_post_receive(&bare) {
                    tracing::warn!(?e, "post-receive hook setup failed");
                }
                let push_key = format!("git_op:{}", bare.display());

                let run = async {
                    run_git(op.as_ref(), &buffer, &handle, channel, rpc_timeout).await
                };
                let result = push_coordinator
                    .run(&push_key, async {
                        run.await.map_err(|e| SmartError::Internal(e.to_string()))
                    })
                    .await
                    .map_err(|e: SmartError| SshError::from(e));
                if result.is_ok()
                    && let (Some(repo_id), Some(space_id), Some(queue)) = (repo_id, space_id, &queue)
                {
                        match enqueue_post_receive_push(
                            queue,
                            &sync_settings,
                            space_id,
                            repo_id,
                            actor_id,
                            &bare,
                        )
                        .await
                        {
                            Ok(n) if n > 0 => {
                                tracing::debug!(
                                    count = n,
                                    space_id = %space_id,
                                    repo_id = %repo_id,
                                    "enqueued post-receive ref updates"
                                )
                            }
                            Ok(_) => tracing::debug!(
                                space_id = %space_id,
                                repo_id = %repo_id,
                                "no refs to enqueue"
                            ),
                            Err(e) => tracing::warn!(
                                error = %e,
                                space_id = %space_id,
                                repo_id = %repo_id,
                                "post-receive enqueue failed"
                            ),
                        }
                }
                let code = if result.is_ok() { 0 } else { 1 };
                _ = handle.exit_status_request(channel, code);
                tokio::time::sleep(std::time::Duration::from_millis(20)).await;
                _ = handle.eof(channel);
                _ = handle.close(channel);
            } else {
                let result = run_git(op.as_ref(), &buffer, &handle, channel, rpc_timeout).await;
                let code = if result.is_ok() { 0 } else { 1 };
                _ = handle.exit_status_request(channel, code);
                tokio::time::sleep(std::time::Duration::from_millis(20)).await;
                _ = handle.eof(channel);
                _ = handle.close(channel);
            }
        });

        Ok(())
    }
}

async fn run_git(
    op: Option<&crate::ssh::handler::git_op::GitOperation>,
    input: &[u8],
    handle: &russh::server::Handle,
    channel: russh::ChannelId,
    timeout_secs: u64,
) -> Result<(), SshError> {
    let Some(op) = op else {
        return Ok(());
    };
    let result: Result<(), SshError> = async {
        let mut cmd = tokio::process::Command::new("git");
        cmd.current_dir(op.bare()).arg(op.service_arg());
        if !op.is_upload_archive() {
            cmd.arg("--stateless-rpc");
        }
        cmd.env("GIT_DIR", op.bare())
            .env("GIT_TERMINAL_PROMPT", "0")
            .stdin(std::process::Stdio::piped())
            .stdout(std::process::Stdio::piped())
            .stderr(std::process::Stdio::piped());
        apply_hardened_env(&mut cmd);

        let mut child = cmd.spawn().map_err(SshError::Io)?;
        let mut stdin = child
            .stdin
            .take()
            .ok_or_else(|| SshError::Internal("git stdin missing".to_string()))?;
        let stdout = child
            .stdout
            .take()
            .ok_or_else(|| SshError::Internal("git stdout missing".to_string()))?;
        let stderr = child
            .stderr
            .take()
            .ok_or_else(|| SshError::Internal("git stderr missing".to_string()))?;
        let child_arc = Arc::new(tokio::sync::Mutex::new(Some(child)));

        {
            let child_arc = child_arc.clone();
            let input = input.to_vec();
            tokio::spawn(async move {
                if stdin.write_all(&input).await.is_err() {
                    if let Some(mut ch) = child_arc.lock().await.take() {
                        let _ = ch.start_kill();
                        let _ = ch.wait().await;
                    }
                }
                let _ = stdin.shutdown().await;
            });
        }

        let stdout_task = tokio::spawn(async move {
            let cap = MAX_GIT_OUTPUT_BUF as u64 + 1;
            let mut buf = Vec::new();
            let res = stdout.take(cap).read_to_end(&mut buf).await;
            (buf, res)
        });
        let stderr_task = tokio::spawn(async move {
            let cap = MAX_GIT_OUTPUT_BUF as u64 + 1;
            let mut buf = Vec::new();
            let res = stderr.take(cap).read_to_end(&mut buf).await;
            (buf, res)
        });

        let wait_task = {
            let child_arc = child_arc.clone();
            async move {
                let mut guard = child_arc.lock().await;
                match guard.as_mut() {
                    None => None,
                    Some(ch) => {
                        match tokio::time::timeout(
                            std::time::Duration::from_secs(timeout_secs),
                            ch.wait(),
                        )
                        .await
                        {
                            Ok(Ok(s)) => Some(s),
                            _ => {
                                let _ = ch.start_kill();
                                let _ = ch.wait().await;
                                None
                            }
                        }
                    }
                }
            }
        };

        let (wait_out, stdout_res, stderr_res) =
            tokio::join!(wait_task, stdout_task, stderr_task);

        let (stdout_buf, stdout_read) = stdout_res
            .map_err(|e| SshError::Internal(format!("stdout task: {e}")))?;
        let (stderr_buf, _stderr_read) = stderr_res
            .map_err(|e| SshError::Internal(format!("stderr task: {e}")))?;
        let status = wait_out.ok_or_else(|| {
            SshError::Internal(format!("git {} timed out", op.service_arg()))
        })?;

        if stdout_buf.len() > MAX_GIT_OUTPUT_BUF || stderr_buf.len() > MAX_GIT_OUTPUT_BUF {
            return Err(SshError::Internal(format!(
                "git {} output exceeded {} bytes",
                op.service_arg(),
                MAX_GIT_OUTPUT_BUF
            )));
        }
        stdout_read.map_err(SshError::Io)?;
        if !stdout_buf.is_empty() {
            let _ = handle.data(channel, Bytes::from(stdout_buf)).await;
        }
        if !stderr_buf.is_empty() && !status.success() {
            let _ = handle
                .extended_data(channel, 1, Bytes::from(stderr_buf))
                .await;
        }
        if status.success() {
            Ok(())
        } else {
            Err(SshError::Internal(format!(
                "git {} failed: exit {:?}",
                op.service_arg(),
                status.code()
            )))
        }
    }
    .await;
    result
}

/// Build a GitBaby for `validate_pre_receive`. The SSH path does not have
/// `(space, repo)` strings available — only `repo_id` — so we resolve the
/// repo model directly and assemble a `GitContext` from `AppState`. This
/// mirrors `service::context::context` minus the space-name lookup, which
/// is unnecessary here because `validate_pre_receive` only needs the bare
/// repo metadata + a configured repo_root/objects_root.
async fn build_baby_for_pre_receive(
    service: &Arc<crate::ssh::service::GitSshService>,
    repo_id: uuid::Uuid,
) -> Option<gitbaby::GitBaby> {
    use models::repo::repos;
    let db = service.smart.git.state.database.clone();
    let repo = repos::Entity::find_by_id(repo_id).one(db.read()).await.ok()??;
    let cache = match CacheClient::new(&service.smart.git.state.config.cache).await {
        Ok(c) => Arc::new(c),
        Err(error) => {
            tracing::warn!(%error, "cache client init failed, skipping pre-receive check");
            return None;
        }
    };
    let git_cfg = &service.smart.git.state.config.git;
    let ctx = GitContext {
        model: repo,
        cache,
        database: db,
        repo_root: std::path::PathBuf::from(&git_cfg.repo_root),
        objects_root: std::path::PathBuf::from(&git_cfg.objects_root),
    };
    Some(service.smart.git.baby(&ctx))
}