use bytes::Bytes;
use gitsmart::pre_receive::{self, parse_receive_pack_refs};
use russh::ChannelId;
use russh::server::Session;

use crate::handler::GitSshHandler;

pub const MAX_SSH_CHANNEL_BUF: usize = 1 << 30;

impl GitSshHandler {
    pub async fn data_impl(&mut self, channel: ChannelId, data: &[u8]) -> Result<(), russh::Error> {
        let state = self.channels.entry(channel).or_default();
        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; dropping channel"
            );
            state.buffer_overflow = true;
            return Err(russh::Error::Disconnect);
        }
        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(state) = self.channels.get_mut(&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 channel_id = state.channel_id;
        let actor_id = state.actor_id;
        let push_coordinator = self.service.push_coordinator();
        let queue = self.service.queue.clone();
        let sync_settings = self.service.sync_settings.clone();

        let handle = session.handle();
        let db = self.service.auth.db.clone();
        tokio::spawn(async move {
            if overflowed {
                tracing::warn!(
                    channel = ?channel,
                    "refusing to spawn git for ssh channel that exceeded buffer cap"
                );
                let _ = handle.exit_status_request(channel, 1).await;
                let _ = handle.eof(channel).await;
                let _ = handle.close(channel).await;
                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::handler::git_op::GitOperation::ReceivePack { .. })
            );

            if is_receive && let (Some(repo_id), Some(actor_uid)) = (repo_id, actor_id) {
                let updates = parse_receive_pack_refs(&buffer);
                if !updates.is_empty() {
                    let db_conn = db.reader();
                    match pre_receive::validate_pre_receive(&db_conn, repo_id, actor_uid, &updates)
                        .await
                    {
                        Ok(verdict) => {
                            if let Err(e) = verdict.accepted_only() {
                                let msg = format!("remote: {}\n", e);
                                let _ = handle.data(channel, Bytes::from(msg.into_bytes())).await;
                                let _ = handle.exit_status_request(channel, 1).await;
                                let _ = handle.eof(channel).await;
                                let _ = handle.close(channel).await;
                                return;
                            }
                        }
                        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;
                            let _ = handle.exit_status_request(channel, 1).await;
                            let _ = handle.eof(channel).await;
                            let _ = handle.close(channel).await;
                            return;
                        }
                    }
                }
            }

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

            let run = async { run_git(op.as_ref(), &buffer, &handle, channel).await };
            let result = if is_write {
                push_coordinator
                    .run(&push_key, async {
                        run.await
                            .map_err(|e| gitsmart::SmartError::Internal(e.to_string()))
                    })
                    .await
                    .map_err(crate::error::SshError::from)
            } else {
                run.await
            };
            if result.is_ok()
                && is_write
                && let (Some(repo_id), Some(channel_id), Some(bare)) = (
                    repo_id,
                    channel_id,
                    op.as_ref().map(|o| o.bare().to_path_buf()),
                )
                && let Some(queue) = &queue
            {
                match gitsync::enqueue_post_receive_push(
                    queue,
                    &sync_settings,
                    channel_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"),
                }
            }
            let code = match &result {
                Ok(_) => 0,
                Err(_) => 1,
            };
            let _ = handle.exit_status_request(channel, code).await;
            tokio::time::sleep(std::time::Duration::from_millis(20)).await;
            let _ = handle.eof(channel).await;
            let _ = handle.close(channel).await;
        });

        Ok(())
    }

    pub async fn shell_request_impl(
        &mut self,
        channel: ChannelId,
        session: &mut Session,
    ) -> Result<(), russh::Error> {
        let msg = Bytes::from_static(
            b"Welcome to GitSsh. This is a git-only server; shell access is disabled.\r\n",
        );
        let _ = session.data(channel, msg);
        let _ = session.exit_status_request(channel, 0);
        let _ = session.eof(channel);
        let _ = session.close(channel);
        let _ = session.flush();
        Ok(())
    }
}

async fn run_git(
    op: Option<&crate::handler::git_op::GitOperation>,
    input: &[u8],
    handle: &russh::server::Handle,
    channel: russh::ChannelId,
) -> Result<(), crate::error::SshError> {
    let Some(op) = op else {
        return Ok(());
    };
    let _timer = crate::metrics::SubprocessTimer::start();
    let result: Result<(), crate::error::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());

        let mut child = cmd.spawn().map_err(crate::error::SshError::Io)?;
        use tokio::io::AsyncWriteExt;
        if let Some(mut stdin) = child.stdin.take() {
            stdin
                .write_all(input)
                .await
                .map_err(crate::error::SshError::Io)?;
            let _ = stdin.shutdown().await;
        }
        let output = child
            .wait_with_output()
            .await
            .map_err(crate::error::SshError::Io)?;
        if !output.stdout.is_empty() {
            let _ = handle.data(channel, Bytes::from(output.stdout)).await;
        }
        if !output.stderr.is_empty() && !output.status.success() {
            let _ = handle
                .extended_data(channel, 1, Bytes::from(output.stderr))
                .await;
        }
        if output.status.success() {
            Ok(())
        } else {
            Err(crate::error::SshError::Internal(format!(
                "git {} failed: exit {:?}",
                op.service_arg(),
                output.status.code()
            )))
        }
    }
    .await;
    crate::metrics::record_subprocess(op.service_arg(), result.is_ok());
    result
}