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::{ChannelState, GitSshHandler};
use crate::sync::post_receive::enqueue_post_receive_push_with_ci_logged;

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

fn hex_preview(buf: &[u8]) -> String {
    let n = buf.len().min(96);
    let mut s = String::with_capacity(n * 2);
    for b in &buf[..n] {
        s.push_str(&format!("{b:02x}"));
    }
    s
}

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);
        if state.git_op.is_some() && state.buffer.len() == data.len() {
            tracing::debug!(
                ?channel,
                first_hex = %hex_preview(&state.buffer),
                "ssh: first data chunk on armed channel"
            );
        }
        let is_receive = matches!(
            state.git_op,
            Some(crate::ssh::handler::git_op::GitOperation::ReceivePack { .. })
        );
        if !is_receive {
            return Ok(());
        }
        if !state.nak_sent && Self::wants_section_complete(&state.buffer) {
            // Over SSH the request and response share one stream: send-pack
            // stops after the wants section and waits for NAK before it sends
            // the pack. We only spawn receive-pack after channel EOF, so
            // synthesize the negotiation reply now; the real process' own NAK
            // is stripped from stdout later (see run_git).
            tracing::debug!(?channel, "ssh push: synthesized NAK for wants section");
            state.nak_sent = true;
            let _ = session.data(channel, Bytes::from_static(b"0008NAK\n0000"));
            let _ = session.flush();
        }
        Ok(())
    }

    /// True when the buffer head is a complete `want` negotiation section
    /// terminated by a flush packet.
    fn wants_section_complete(buf: &[u8]) -> bool {
        let mut i = 0usize;
        let mut saw_want = false;
        while i + 4 <= buf.len() {
            let Ok(len) =
                usize::from_str_radix(std::str::from_utf8(&buf[i..i + 4]).unwrap_or_default(), 16)
            else {
                return false;
            };
            if len == 0 {
                return saw_want;
            }
            if len < 4 || i + len > buf.len() {
                return false;
            }
            if !buf[i + 4..i + len].starts_with(b"want ") {
                return false;
            }
            saw_want = true;
            i += len;
        }
        false
    }

    pub async fn channel_close_impl(&mut self, channel: ChannelId) {
        tracing::debug!(?channel, "ssh channel closed");
        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.remove(&channel) else {
            let _ = session.close(channel);
            let _ = session.flush();
            return Ok(());
        };
        self.dispatch_state(channel, state, session.handle(), "eof");
        Ok(())
    }

    /// Detach buffered channel state and run its git operation in the
    /// background. `reason` is only used for tracing.
    fn dispatch_state(
        &mut self,
        channel: ChannelId,
        state: ChannelState,
        handle: russh::server::Handle,
        reason: &str,
    ) {
        let mut state = state;
        let overflowed = state.buffer_overflow;
        let nak_sent = state.nak_sent;
        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;
        tracing::debug!(
            ?channel,
            op = op.as_ref().map(|o| o.service_arg()),
            buffered_bytes = buffer.len(),
            repo_id = ?repo_id,
            actor_id = ?actor_id,
            overflowed,
            %reason,
            "ssh dispatching git operation"
        );
        let push_coordinator = self.service.push_coordinator();
        let queue = self.service.smart.queue.clone();
        let sync_settings = self.service.sync_settings.clone();
        let ci_topic = self.service.smart.ci_event_topic().map(|t| t.to_string());
        let ai_ctx = self
            .service
            .smart
            .ai_config()
            .map(|cfg| crate::sync::post_receive::AiPushCtx {
                db: self.service.smart.git.state.database.clone(),
                redis: self.service.smart.git.state.redis.clone(),
                config: cfg.clone(),
            });
        let ssh_service = self.service.clone();

        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;
                        }
                        tracing::debug!(repo_id = %repo_id, "ssh pre-receive validation passed");
                    }
                    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());
                tracing::debug!(push_key = %push_key, repo_id = ?repo_id, "ssh push: acquiring push lock");

                let run = async {
                    run_git(
                        op.as_ref(),
                        &buffer,
                        &handle,
                        channel,
                        rpc_timeout,
                        nak_sent,
                    )
                    .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 let Err(e) = &result {
                    tracing::warn!(repo_id = ?repo_id, error = %e, "ssh push failed");
                }
                if result.is_ok()
                    && let (Some(repo_id), Some(space_id), Some(queue)) =
                        (repo_id, space_id, &queue)
                {
                    enqueue_post_receive_push_with_ci_logged(
                        queue,
                        &sync_settings,
                        ci_topic.as_deref(),
                        ai_ctx,
                        space_id,
                        repo_id,
                        actor_id,
                        &bare,
                    )
                    .await;
                }
                let code = if result.is_ok() { 0 } else { 1 };
                tracing::debug!(
                    ?channel,
                    code,
                    nak_sent,
                    "ssh: sending exit-status for push"
                );
                let r1 = handle.exit_status_request(channel, code).await;
                // Windows OpenSSH can drop buffered channel data if CLOSE
                // races ahead of the client's read loop; give the client time
                // to drain the report before tearing the channel down.
                tokio::time::sleep(std::time::Duration::from_millis(250)).await;
                let r2 = handle.eof(channel).await;
                let r3 = handle.close(channel).await;
                tracing::debug!(exit_status = ?r1, eof = ?r2, close = ?r3, "ssh: channel shutdown results");
            } else {
                let result = run_git(
                    op.as_ref(),
                    &buffer,
                    &handle,
                    channel,
                    rpc_timeout,
                    nak_sent,
                )
                .await;
                let code = if result.is_ok() { 0 } else { 1 };
                _ = handle.exit_status_request(channel, code);
                tokio::time::sleep(std::time::Duration::from_millis(250)).await;
                _ = handle.eof(channel);
                _ = handle.close(channel);
            }
        });
    }
}

async fn run_git(
    op: Option<&crate::ssh::handler::git_op::GitOperation>,
    input: &[u8],
    handle: &russh::server::Handle,
    channel: russh::ChannelId,
    timeout_secs: u64,
    strip_nak: bool,
) -> 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");
        }
        // Same as the advertise path: receive-pack requires an explicit
        // <git-dir> argument, GIT_DIR alone is not enough.
        cmd.arg(op.bare());
        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)?;
        tracing::debug!(
            service = op.service_arg(),
            bare = %op.bare().display(),
            input_bytes = input.len(),
            timeout_secs,
            "spawned git process for ssh operation"
        );
        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)?;
        let mut stdout_buf = stdout_buf;
        if strip_nak {
            // We already answered the wants section with a synthetic NAK, so
            // remove receive-pack's own NAK+flush to keep the client's
            // negotiation read aligned with the report that follows.
            const NAK: &[u8] = b"0008NAK\n";
            if stdout_buf.starts_with(NAK) {
                let mut off = NAK.len();
                if stdout_buf.get(off..off + 4) == Some(b"0000") {
                    off += 4;
                }
                stdout_buf.drain(..off);
            }
        }
        tracing::debug!(
            service = op.service_arg(),
            exit = ?status.code(),
            stdout_bytes = stdout_buf.len(),
            stderr_bytes = stderr_buf.len(),
            "git process for ssh operation finished"
        );
        if !stdout_buf.is_empty() {
            if let Err(e) = handle.data(channel, Bytes::from(stdout_buf.clone())).await {
                tracing::warn!(
                    ?e,
                    bytes = stdout_buf.len(),
                    "ssh: failed to send git stdout to client"
                );
            }
        }
        if !stderr_buf.is_empty() && !status.success() {
            if let Err(e) = handle
                .extended_data(channel, 1, Bytes::from(stderr_buf.clone()))
                .await
            {
                tracing::warn!(?e, "ssh: failed to send git stderr to client");
            }
        }
        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))
}