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;
use crate::ssh::handler::GitSshHandler;
use crate::ssh::handler::git_op::GitOperation;
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(());
        }
        if state.dispatched {
            if let Some(tx) = &state.stdin_tx {
                let _ = tx.send(data.to_vec());
            }
            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"
            );
        }
        // Push-only: receive-pack runs after EOF, so its negotiation reply has
        // to be synthesized here. Upload services must not receive it, their
        // own process emits the real NAK/ACK.
        let is_receive = matches!(state.git_op, Some(GitOperation::ReceivePack { .. }));
        if is_receive && !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");
        if let Some(state) = self.channels.remove(&channel)
            && let Some(abort) = state.stream_abort
        {
            abort.abort();
        }
    }

    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(());
        };
        if state.dispatched {
            // Half-close only: the git process may still have data to write
            // back (upload-pack answers with the packfile after `done`), so the
            // session is torn down when the pump task finishes, not here.
            if let Some(tx) = state.stdin_tx.take() {
                drop(tx);
            }
            tracing::debug!(?channel, "ssh: client half-closed duplex channel");
            return Ok(());
        }
        let state = self.channels.remove(&channel).expect("channel state");
        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,
                        &crate::sync::PushTarget {
                            space_id,
                            repo_id,
                            actor_id,
                            bare: &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);
            }
        });
    }
}

/// Start a duplex git session for `upload-pack` / `upload-archive`: stdin is fed
/// from `rx` and stdout is streamed back as it is produced. The client keeps the
/// channel open while it waits for the packfile, so the response cannot be
/// deferred until EOF the way the push path does.
pub(crate) fn spawn_streaming_git(
    channel: russh::ChannelId,
    state: ChannelState,
    handle: russh::server::Handle,
    rx: tokio::sync::mpsc::UnboundedReceiver<Vec<u8>>,
) -> tokio::task::AbortHandle {
    let op = state.git_op.clone();
    let service = state.git_op.as_ref().map(|o| o.service_arg());
    let timeout_secs = state.rpc_timeout_secs;
    let task = tokio::spawn(async move {
        let Some(op) = op else {
            return;
        };
        let result = streaming_git(&op, &handle, channel, rx, timeout_secs).await;
        export::metrics::git_op(
            op.service_arg(),
            if result.is_ok() { "ok" } else { "error" },
        );
        match &result {
            Ok(()) => tracing::debug!(?channel, "ssh streaming git finished"),
            Err(e) => tracing::warn!(?channel, error = %e, "ssh streaming git failed"),
        }
        let code = if result.is_ok() { 0 } else { 1 };
        _ = handle.exit_status_request(channel, code).await;
        // Give the client a moment to drain the final bytes before the
        // channel is torn down (the same race the push path guards against).
        tokio::time::sleep(std::time::Duration::from_millis(250)).await;
        _ = handle.eof(channel).await;
        _ = handle.close(channel).await;
    });
    let abort = task.abort_handle();
    if let Some(name) = service {
        tracing::debug!(?channel, service = %name, "ssh streaming git started");
    }
    abort
}

/// Drive one duplex git process: pump `rx` into stdin, forward stdout to the
/// SSH channel, and collect stderr for diagnostics.
#[tracing::instrument(name = "git.ssh", skip_all, fields(service = op.service_arg()))]
async fn streaming_git(
    op: &crate::ssh::handler::git_op::GitOperation,
    handle: &russh::server::Handle,
    channel: russh::ChannelId,
    mut rx: tokio::sync::mpsc::UnboundedReceiver<Vec<u8>>,
    timeout_secs: u64,
) -> Result<(), SshError> {
    // Stateful, like a real sshd: advertise refs, then answer the request
    // arriving on stdin. Must not be gated on stdin EOF.
    let mut cmd = tokio::process::Command::new("git");
    cmd.current_dir(op.bare()).arg(op.service_arg());
    cmd.arg(op.bare())
        .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);

    cmd.kill_on_drop(true);
    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 mut 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()))?;

    tracing::debug!(
        service = op.service_arg(),
        bare = %op.bare().display(),
        "spawned git process for duplex ssh operation"
    );

    let writer = tokio::spawn(async move {
        while let Some(chunk) = rx.recv().await {
            if stdin.write_all(&chunk).await.is_err() {
                break;
            }
        }
        let _ = stdin.shutdown().await;
    });

    let reader = {
        let handle = handle.clone();
        tokio::spawn(async move {
            let mut buf = vec![0u8; 64 * 1024];
            loop {
                match stdout.read(&mut buf).await {
                    Ok(0) => break,
                    Ok(n) => {
                        if handle
                            .data(channel, Bytes::copy_from_slice(&buf[..n]))
                            .await
                            .is_err()
                        {
                            break;
                        }
                    }
                    Err(e) => {
                        tracing::warn!(?channel, error = %e, "ssh: git stdout read failed");
                        break;
                    }
                }
            }
        })
    };

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

    let status = match tokio::time::timeout(
        std::time::Duration::from_secs(timeout_secs),
        child.wait(),
    )
    .await
    {
        Ok(Ok(s)) => s,
        _ => {
            let _ = child.start_kill();
            let _ = child.wait().await;
            return Err(SshError::Internal(format!(
                "git {} timed out",
                op.service_arg()
            )));
        }
    };

    // The writer only stops on client half-close, so awaiting it here would
    // hang once the process has already finished.
    let _ = reader.await;
    writer.abort();
    let _ = writer.await;
    let stderr_bytes = stderr_task.await.unwrap_or_default();
    tracing::debug!(
        service = op.service_arg(),
        exit = ?status.code(),
        stderr_bytes = stderr_bytes.len(),
        "git process for duplex ssh operation finished"
    );
    if !stderr_bytes.is_empty() && !status.success() {
        let _ = handle
            .extended_data(channel, 1, Bytes::from(stderr_bytes.clone()))
            .await;
    }
    if status.success() {
        Ok(())
    } else {
        Err(SshError::Internal(format!(
            "git {} failed: exit {:?}",
            op.service_arg(),
            status.code()
        )))
    }
}

#[tracing::instrument(
    name = "git.ssh",
    skip_all,
    fields(service = op.map(|o| o.service_arg()).unwrap_or("none"))
)]
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);

        cmd.kill_on_drop(true);
        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()
                    && 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()
            && 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()
            && 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;
    export::metrics::git_op(
        op.service_arg(),
        if result.is_ok() { "ok" } else { "error" },
    );
    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 = Arc::new(CacheClient::new(service.smart.git.state.redis.clone()));
    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))
}