use bytes::Bytes;
use russh::ChannelId;
use russh::server::Session;

use uuid::Uuid;

use crate::share::access::authorize_repo_access;
use crate::share::errors::GitError;
use crate::smart::error::SmartError;
use crate::ssh::error::SshError;
use crate::ssh::git_service::{
    GitService, build_git_advertise_command, parse_git_command, parse_repo_path, sanitize_for_log,
};
use crate::ssh::handler::GitSshHandler;
use crate::ssh::handler::git_op::GitOperation;

impl GitSshHandler {
    pub async fn exec_request_impl(
        &mut self,
        channel: ChannelId,
        data: &[u8],
        session: &mut Session,
    ) -> Result<(), russh::Error> {
        let cmd = match std::str::from_utf8(data) {
            Ok(c) => c,
            Err(_) => {
                invalid_command("invalid command encoding", session, channel).await;
                return Ok(());
            }
        };
        let cmd = cmd.trim();
        tracing::info!(
            "ssh exec from {:?}: {}",
            self.client_addr,
            sanitize_for_log(cmd)
        );

        let Some((service, raw_path)) = parse_git_command(cmd) else {
            invalid_command("invalid git command", session, channel).await;
            return Ok(());
        };
        let Some((space_name, repo_name)) = parse_repo_path(raw_path) else {
            tracing::debug!(path = %sanitize_for_log(raw_path), "ssh exec: invalid repository path");
            invalid_command("invalid repository path", session, channel).await;
            return Ok(());
        };
        tracing::debug!(
            service = service.as_subcommand(),
            space = space_name,
            repo = repo_name,
            "ssh exec: parsed git command"
        );

        let actor_id = self.operator.as_ref().map(|u| u.uid);
        let required = if service.is_write() { "write" } else { "read" };

        if let Err(msg) = self
            .authorize_repo(space_name, repo_name, service.is_write())
            .await
        {
            tracing::warn!("ssh {service:?} {space_name}/{repo_name} denied: {msg}");
            invalid_command(&msg, session, channel).await;
            return Ok(());
        }

        let ctx = match self.service.smart.resolve(space_name, repo_name).await {
            Ok(c) => c,
            Err(e) => {
                tracing::warn!(space = space_name, repo = repo_name, error = %e, "ssh exec: repo resolve failed");
                let msg = format!("fatal: {e}");
                invalid_command(&msg, session, channel).await;
                return Ok(());
            }
        };
        tracing::debug!(
            space = space_name,
            repo = repo_name,
            repo_id = %ctx.model.id,
            "ssh exec: repo access authorized and resolved"
        );
        let space_id = match self.space_id_for(&ctx.model.dir_id).await {
            Ok(id) => id,
            Err(error) => {
                tracing::warn!(%error, "space lookup failed, falling back to dir id");
                ctx.model.dir_id
            }
        };

        let bare = ctx.repo_dir();
        if !bare.is_dir() {
            tracing::warn!(bare = %bare.display(), "ssh exec: bare repository directory missing");
            let msg = format!("fatal: repository '{space_name}/{repo_name}' does not exist");
            invalid_command(&msg, session, channel).await;
            return Ok(());
        }

        // Push is driven from a fully buffered request, so receive-pack only
        // runs after EOF and its advertisement is produced up front here.
        // Upload services are duplex and emit the advertisement themselves.
        let mut adv_len = 0usize;
        if service.is_write() {
            let adv = match run_advertise(
                &bare,
                service,
                self.service.smart.git.state.config.git.git_rpc_timeout_secs,
            )
            .await
            {
                Ok(adv) => adv,
                Err(e) => {
                    tracing::warn!(
                        service = service.as_subcommand(),
                        bare = %bare.display(),
                        error = %e,
                        "ssh exec: ref advertisement failed"
                    );
                    let msg = format!("fatal: unable to read repository advertisement: {e}");
                    invalid_command(&msg, session, channel).await;
                    return Ok(());
                }
            };
            adv_len = adv.len();
            if !adv.is_empty() {
                let _ = session.data(channel, Bytes::from(adv));
            }
        }
        let _ = session.channel_success(channel);
        let _ = session.flush();
        tracing::debug!(
            service = service.as_subcommand(),
            repo_id = %ctx.model.id,
            adv_bytes = adv_len,
            "ssh exec: channel armed"
        );

        let op = match service {
            GitService::UploadPack => GitOperation::UploadPack { bare: bare.clone() },
            GitService::UploadArchive => GitOperation::UploadArchive { bare: bare.clone() },
            GitService::ReceivePack => GitOperation::ReceivePack { bare: bare.clone() },
        };
        // Fetch/clone/archive is duplex: waiting for EOF before starting git,
        // as the push path does, would deadlock against a client that is
        // waiting for the packfile.
        let streaming = !op.is_write();
        {
            let state = self.channels.entry(channel).or_default();
            state.git_op = Some(op);
            state.repo_id = Some(ctx.model.id);
            state.space_id = Some(space_id);
            state.actor_id = actor_id;
        }
        if streaming {
            let state = self.channels.remove(&channel).expect("channel state");
            let mut state = state;
            state.rpc_timeout_secs = self.service.smart.git.state.config.git.git_rpc_timeout_secs;
            let (tx, rx) = tokio::sync::mpsc::unbounded_channel::<Vec<u8>>();
            let abort = crate::ssh::handler::channel::spawn_streaming_git(
                channel,
                state,
                session.handle(),
                rx,
            );
            let state = self.channels.entry(channel).or_default();
            state.dispatched = true;
            state.stdin_tx = Some(tx);
            state.stream_abort = Some(abort);
        }
        let _ = required;
        Ok(())
    }

    async fn authorize_repo(
        &self,
        space: &str,
        repo: &str,
        want_write: bool,
    ) -> Result<(), String> {
        let svc = &self.service.smart.git;
        let ctx = svc
            .context(space.to_string(), repo.to_string())
            .await
            .map_err(|e| format!("{e}"))?;
        let Some(operator) = &self.operator else {
            return Err("authentication required".into());
        };
        authorize_repo_access(svc, operator.uid, &ctx, want_write, false)
            .await
            .map_err(|e| match e {
                GitError::Forbidden => "permission denied".to_string(),
                other => format!("{other}"),
            })
    }

    async fn space_id_for(&self, dir_id: &Uuid) -> Result<Uuid, SmartError> {
        crate::share::access::space_for_dir(self.service.smart.git.db_read(), *dir_id)
            .await
            .map_err(SmartError::from)?
            .map(|s| s.id)
            .ok_or_else(|| SmartError::NotFound("space".into()))
    }

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

async fn invalid_command(msg: &str, session: &mut Session, channel: ChannelId) {
    let _ = session.data(channel, Bytes::from(format!("{msg}\n").into_bytes()));
    let _ = session.channel_failure(channel);
    let _ = session.exit_status_request(channel, 1);
    let _ = session.eof(channel);
    let _ = session.close(channel);
    let _ = session.flush();
}

async fn run_advertise(
    bare: &std::path::Path,
    service: GitService,
    timeout_secs: u64,
) -> Result<Vec<u8>, SshError> {
    let mut cmd = build_git_advertise_command(service, bare);
    cmd.stdout(std::process::Stdio::piped())
        .stderr(std::process::Stdio::piped());
    let mut child = cmd.spawn().map_err(SshError::Io)?;
    let stderr_task = child.stderr.take().map(|stderr| {
        tokio::spawn(async move {
            use tokio::io::AsyncReadExt;
            let s = stderr;
            let mut buf = Vec::new();
            let _ = s
                .take((64 * 1024 * 1024) as u64 + 1)
                .read_to_end(&mut buf)
                .await;
            buf
        })
    });
    let stdout = child
        .stdout
        .take()
        .ok_or_else(|| SshError::Internal("git stdout missing".to_string()))?;
    let mut out = Vec::new();
    use tokio::io::AsyncReadExt;
    stdout
        .take((64 * 1024 * 1024) as u64 + 1)
        .read_to_end(&mut out)
        .await
        .map_err(SshError::Io)?;
    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 {} --advertise-refs timed out",
                service.as_subcommand()
            )));
        }
    };
    let stderr_bytes = match stderr_task {
        Some(t) => t
            .await
            .map_err(|e| SshError::Internal(format!("stderr task join failed: {e}")))?,
        None => Vec::new(),
    };
    if !status.success() {
        return Err(SshError::Internal(format!(
            "git {} --advertise-refs failed: {}",
            service.as_subcommand(),
            String::from_utf8_lossy(&stderr_bytes)
        )));
    }
    Ok(out)
}