use std::{
    collections::{HashMap, HashSet},
    io,
    net::SocketAddr,
    path::PathBuf,
    process::Stdio,
    sync::Arc,
    time::Duration,
};

use cache::AppCache;
use db::{database::AppDatabase, sqlx};
use model::{
    repos::{RepoModel, RepoProtectModel},
    users::UserModel,
};
use russh::{
    Channel, ChannelId, Disconnect,
    keys::{Certificate, PublicKey},
    server::{Auth, Msg, Session},
};
use tokio::{
    io::AsyncWriteExt,
    process::ChildStdin,
    sync::{Mutex, mpsc::Sender},
    time::sleep,
};

use crate::{
    ssh::{
        SshTokenService,
        authz::SshAuthService,
        branch_protect::check_branch_protection,
        forward::forward,
        git_service::{
            GitService, build_git_command, parse_git_command, parse_repo_path,
        },
        ref_update::RefUpdate,
    },
    sync::{
        ReceiveSyncService, RepoReceiveSyncTask,
        push_queue::{
            PushQueueEvent, PushQueueWaitError, wait_for_push_queue_slot,
        },
    },
};

static SSH_METRICS: std::sync::OnceLock<track::MetricsRegistry> =
    std::sync::OnceLock::new();

/// Call once during server init to wire SSH metrics into the Prometheus registry.
pub fn set_ssh_metrics(registry: track::MetricsRegistry) {
    let _ = SSH_METRICS.set(registry);
}

fn record_ssh(operation: &str, outcome: &str) {
    if let Some(reg) = SSH_METRICS.get() {
        reg.register_counter_vec(
            "git_ssh_operations_total",
            "Total SSH git operations",
            &["operation", "outcome"],
        )
        .map(|cv| cv.with_label_values(&[operation, outcome]).inc())
        .ok();
    }
}

fn record_ssh_auth(method: &str, outcome: &str) {
    if let Some(reg) = SSH_METRICS.get() {
        reg.register_counter_vec(
            "git_ssh_auth_attempts_total",
            "SSH authentication attempts",
            &["method", "outcome"],
        )
        .map(|cv| cv.with_label_values(&[method, outcome]).inc())
        .ok();
    }
}

const PRE_PACK_LIMIT: usize = 1_048_576;
const ZERO_OID: &str = "0000000000000000000000000000000000000000";

pub struct SSHandle {
    pub repo: Option<PathBuf>,
    pub model: Option<RepoModel>,
    pub stdin: HashMap<ChannelId, ChildStdin>,
    pub eof: HashMap<ChannelId, Sender<bool>>,
    pub operator: Option<UserModel>,
    pub db: AppDatabase,
    pub auth: SshAuthService,
    pub buffer: HashMap<ChannelId, Vec<u8>>,
    pub branch: HashMap<ChannelId, Vec<RefUpdate>>,
    pub post_receive_refs: HashMap<ChannelId, Arc<Mutex<Vec<RefUpdate>>>>,
    pub service: Option<GitService>,
    pub cache: AppCache,
    pub sync: ReceiveSyncService,
    pub upload_pack_eof_sent: HashSet<ChannelId>,
    pub token_service: SshTokenService,
    pub client_addr: Option<SocketAddr>,
}

impl SSHandle {
    pub fn new(
        db: AppDatabase,
        cache: AppCache,
        sync: ReceiveSyncService,
        token_service: SshTokenService,
        client_addr: Option<SocketAddr>,
    ) -> Self {
        let auth = SshAuthService::new(db.clone());
        let addr_str = client_addr
            .map(|addr| format!("{}", addr))
            .unwrap_or_else(|| "unknown".to_string());
        tracing::info!("SSH handler created client={}", addr_str);
        Self {
            repo: None,
            model: None,
            stdin: HashMap::new(),
            eof: HashMap::new(),
            operator: None,
            db,
            auth,
            buffer: HashMap::new(),
            branch: HashMap::new(),
            post_receive_refs: HashMap::new(),
            service: None,
            cache,
            sync,
            upload_pack_eof_sent: HashSet::new(),
            token_service,
            client_addr,
        }
    }

    fn cleanup_channel(&mut self, channel_id: ChannelId) {
        if let Some(stdin) = self.stdin.remove(&channel_id) {
            let channel_id_for_task = channel_id;
            tokio::spawn(async move {
                let _ = tokio::time::timeout(Duration::from_secs(5), async {
                    let mut stdin = stdin;
                    if let Err(e) = stdin.flush().await {
                        tracing::warn!(error = %e, "ssh_cleanup_flush_failed channel={:?}", channel_id_for_task);
                    }
                    let _ = stdin.shutdown().await;
                })
                .await;
            });
        }
        self.eof.remove(&channel_id);
        self.post_receive_refs.remove(&channel_id);
        self.upload_pack_eof_sent.remove(&channel_id);
    }

    fn format_post_receive_hints(
        namespace: &str,
        repo: &RepoModel,
        refs: &[RefUpdate],
        queue: Option<(usize, usize)>,
    ) -> String {
        let mut lines = Vec::new();
        for r#ref in refs {
            if r#ref.old_oid == ZERO_OID
                && r#ref.name.starts_with("refs/heads/")
            {
                let branch = r#ref.name.trim_start_matches("refs/heads/");
                lines.push(format!(
                    "remote: new branch '{}' pushed. Create a PR: /{}/repo/{}/pulls/new?head={}\r\n",
                    branch,
                    namespace,
                    repo.name,
                    branch
                ));
            }
        }
        if let Some((position, total)) = queue {
            lines.push(format!(
                "remote: repository sync queued ({}/{}). Metadata, webhooks and search indexes will update shortly.\r\n",
                position, total
            ));
        }
        lines.concat()
    }
}

impl Drop for SSHandle {
    fn drop(&mut self) {
        let addr_str = self
            .client_addr
            .map(|addr| format!("{}", addr))
            .unwrap_or_else(|| "unknown".to_string());
        tracing::info!("ssh_handler_dropped client={}", addr_str);

        let channel_ids: Vec<_> = self.stdin.keys().copied().collect();
        for channel_id in channel_ids {
            self.cleanup_channel(channel_id);
        }
    }
}

impl russh::server::Handler for SSHandle {
    type Error = russh::Error;

    async fn auth_none(&mut self, user: &str) -> Result<Auth, Self::Error> {
        let client_info = self
            .client_addr
            .map(|addr| format!("{}", addr))
            .unwrap_or_else(|| "unknown".to_string());
        tracing::info!(
            "auth_none_received user={} client={}",
            user,
            client_info
        );
        Ok(Auth::UnsupportedMethod)
    }

    async fn auth_password(
        &mut self,
        _user: &str,
        token: &str,
    ) -> Result<Auth, Self::Error> {
        let client_info = self
            .client_addr
            .map(|addr| format!("{}", addr))
            .unwrap_or_else(|| "unknown".to_string());

        if token.is_empty() {
            tracing::warn!("auth_rejected_empty_token client={}", client_info);
            return Err(russh::Error::NotAuthenticated);
        }

        tracing::info!("auth_token_attempt client={}", client_info);

        let user_model =
            match self.token_service.find_user_by_token(token).await {
                Ok(Some(model)) => model,
                Ok(None) => {
                    tracing::warn!(
                        "auth_rejected_token_not_found client={}",
                        client_info
                    );
                    return Err(russh::Error::NotAuthenticated);
                }
                Err(e) => {
                    tracing::error!("auth_token_error error={}", e);
                    return Err(russh::Error::NotAuthenticated);
                }
            };

        tracing::info!(
            "auth_token_success user={} client={}",
            user_model.username,
            client_info
        );
        self.operator = Some(user_model);
        Ok(Auth::Accept)
    }

    async fn auth_publickey_offered(
        &mut self,
        user: &str,
        public_key: &PublicKey,
    ) -> Result<Auth, Self::Error> {
        let client_info = self
            .client_addr
            .map(|addr| format!("{}", addr))
            .unwrap_or_else(|| "unknown".to_string());

        if user != "git" {
            tracing::warn!(
                "auth_publickey_offer_rejected_invalid_username user={} client={}",
                user,
                client_info
            );
            return Err(russh::Error::NotAuthenticated);
        }

        let public_key_str = public_key.to_string();
        if public_key_str.len() < 32 {
            tracing::warn!(
                "auth_publickey_offer_rejected_invalid_key_length key_length={}",
                public_key_str.len()
            );
            return Err(russh::Error::NotAuthenticated);
        }

        tracing::info!("auth_publickey_offer client={}", client_info);
        match self.auth.find_user_by_public_key(&public_key_str).await {
            Ok(Some(key_user)) => {
                tracing::info!(
                    "auth_publickey_offer_accepted user={} key={} client={}",
                    key_user.user.username,
                    key_user.key_title,
                    client_info
                );
                Ok(Auth::Accept)
            }
            Ok(None) => {
                tracing::warn!(
                    "auth_publickey_offer_rejected_key_not_found client={}",
                    client_info
                );
                Err(russh::Error::NotAuthenticated)
            }
            Err(e) => {
                tracing::error!("auth_publickey_offer_error error={}", e);
                Err(russh::Error::NotAuthenticated)
            }
        }
    }

    async fn auth_publickey(
        &mut self,
        user: &str,
        public_key: &PublicKey,
    ) -> Result<Auth, Self::Error> {
        let client_info = self
            .client_addr
            .map(|addr| format!("{}", addr))
            .unwrap_or_else(|| "unknown".to_string());

        if user != "git" {
            tracing::warn!(
                "auth_rejected_invalid_username user={} client={}",
                user,
                client_info
            );
            return Err(russh::Error::NotAuthenticated);
        }
        let public_key_str = public_key.to_string();
        if public_key_str.len() < 32 {
            tracing::warn!(
                "auth_rejected_invalid_key_length key_length={}",
                public_key_str.len()
            );
            return Err(russh::Error::NotAuthenticated);
        }

        tracing::info!("auth_publickey_attempt client={}", client_info);
        let key_user =
            match self.auth.find_user_by_public_key(&public_key_str).await {
                Ok(Some(key_user)) => key_user,
                Ok(None) => {
                    tracing::warn!(
                        "auth_rejected_key_not_found client={}",
                        client_info
                    );
                    return Err(russh::Error::NotAuthenticated);
                }
                Err(e) => {
                    tracing::error!("auth_publickey_error error={}", e);
                    return Err(russh::Error::NotAuthenticated);
                }
            };

        tracing::info!(
            "auth_publickey_success user={} client={}",
            key_user.user.username,
            client_info
        );
        self.auth.update_key_last_used_async(key_user.key_id);
        self.operator = Some(key_user.user);
        Ok(Auth::Accept)
    }

    async fn auth_openssh_certificate(
        &mut self,
        user: &str,
        certificate: &Certificate,
    ) -> Result<Auth, Self::Error> {
        let client_info = self
            .client_addr
            .map(|addr| format!("{}", addr))
            .unwrap_or_else(|| "unknown".to_string());

        if user != "git" {
            tracing::warn!(
                "auth_rejected_invalid_username user={} client={}",
                user,
                client_info
            );
            return Err(russh::Error::NotAuthenticated);
        }
        let public_key_str = certificate.to_string();
        if public_key_str.len() < 32 {
            tracing::warn!(
                "auth_rejected_invalid_key_length key_length={}",
                public_key_str.len()
            );
            return Err(russh::Error::NotAuthenticated);
        }

        tracing::info!("auth_publickey_attempt client={}", client_info);
        let key_user =
            match self.auth.find_user_by_public_key(&public_key_str).await {
                Ok(Some(key_user)) => key_user,
                Ok(None) => {
                    tracing::warn!(
                        "auth_rejected_key_not_found client={}",
                        client_info
                    );
                    return Err(russh::Error::NotAuthenticated);
                }
                Err(e) => {
                    tracing::error!("auth_publickey_error error={}", e);
                    return Err(russh::Error::NotAuthenticated);
                }
            };

        tracing::info!(
            "auth_publickey_success user={} client={}",
            key_user.user.username,
            client_info
        );
        self.auth.update_key_last_used_async(key_user.key_id);
        self.operator = Some(key_user.user);
        Ok(Auth::Accept)
    }

    async fn channel_close(
        &mut self,
        channel: ChannelId,
        _: &mut Session,
    ) -> Result<(), Self::Error> {
        tracing::info!(
            "channel_close channel={:?} client={:?}",
            channel,
            self.client_addr
        );
        self.cleanup_channel(channel);
        Ok(())
    }

    async fn channel_eof(
        &mut self,
        channel: ChannelId,
        _: &mut Session,
    ) -> Result<(), Self::Error> {
        tracing::info!(
            "channel_eof channel={:?} client={:?}",
            channel,
            self.client_addr
        );

        if let Some(eof) = self.eof.get(&channel) {
            let _ = eof.send(true).await;
        }

        if let Some(mut stdin) = self.stdin.remove(&channel) {
            tracing::info!(
                "Closing stdin channel={:?} client={:?}",
                channel,
                self.client_addr
            );
            let _ = tokio::time::timeout(Duration::from_secs(5), async {
                if let Err(e) = stdin.flush().await {
                    tracing::warn!(error = %e, "ssh_eof_flush_failed channel={:?}", channel);
                }
                let _ = stdin.shutdown().await;
            })
            .await;
        }

        Ok(())
    }

    #[tracing::instrument(skip(self, channel, session))]
    async fn channel_open_session(
        &mut self,
        channel: Channel<Msg>,
        session: &mut Session,
    ) -> Result<bool, Self::Error> {
        let client_info = self
            .client_addr
            .map(|addr| format!("{}", addr))
            .unwrap_or_else(|| "unknown".to_string());
        tracing::info!(
            "channel_open_session channel={:?} client={}",
            channel,
            client_info
        );
        if let Err(e) = session.flush() {
            tracing::warn!(error = %e, "ssh_session_flush_failed");
        }
        Ok(true)
    }

    async fn pty_request(
        &mut self,
        channel: ChannelId,
        term: &str,
        col_width: u32,
        row_height: u32,
        _pix_width: u32,
        _pix_height: u32,
        _modes: &[(russh::Pty, u32)],
        session: &mut Session,
    ) -> Result<(), Self::Error> {
        tracing::warn!(
            "pty_request not supported channel={:?} term={} cols={} rows={}",
            channel,
            term,
            col_width,
            row_height
        );
        if let Err(e) = session.flush() {
            tracing::warn!(error = %e, "ssh_session_flush_failed");
        }
        Ok(())
    }

    async fn subsystem_request(
        &mut self,
        channel: ChannelId,
        name: &str,
        session: &mut Session,
    ) -> Result<(), Self::Error> {
        tracing::info!(
            "subsystem_request channel={:?} subsystem={}",
            channel,
            name
        );
        if let Err(e) = session.flush() {
            tracing::warn!(error = %e, "ssh_session_flush_failed");
        }
        Ok(())
    }

    async fn data(
        &mut self,
        channel: ChannelId,
        data: &[u8],
        session: &mut Session,
    ) -> Result<(), Self::Error> {
        if matches!(self.service, Some(GitService::ReceivePack)) {
            if !self.branch.contains_key(&channel) {
                let bf = self.buffer.entry(channel).or_default();

                if bf.len() + data.len() > PRE_PACK_LIMIT {
                    tracing::warn!(
                        "ssh_pre_pack_too_large channel={:?}",
                        channel
                    );
                    let msg = "remote: Ref negotiation exceeds size limit\r\n";
                    let _ = session.extended_data(
                        channel,
                        1,
                        msg.as_bytes().to_vec(),
                    );
                    let _ = session.exit_status_request(channel, 1);
                    let _ = session.eof(channel);
                    let _ = session.close(channel);
                    self.cleanup_channel(channel);
                    return Ok(());
                }

                bf.extend_from_slice(data);

                if !bf.windows(4).any(|w| w == b"0000") {
                    return Ok(());
                }

                let buffered = self.buffer.remove(&channel).unwrap_or_default();

                match RefUpdate::parse_ref_updates(&buffered) {
                    Ok(refs) => {
                        if let Some(model) = &self.model {
                            let branch_protect_roles = sqlx::query_as::<_, RepoProtectModel>(
                                "SELECT id, repo, pattern, require_pull_request, required_approvals, require_status_checks, required_status_contexts, enforce_admins, allow_force_pushes, allow_deletions, created_at, updated_at FROM repo_protect WHERE repo = $1",
                            )
                                .bind(model.id)
                                .fetch_all(self.db.reader())
                                .await
                                .map_err(|e| {
                                    russh::Error::IO(io::Error::new(io::ErrorKind::Other, e))
                                })?;

                            for r#ref in &refs {
                                if let Some(msg) = check_branch_protection(
                                    &branch_protect_roles,
                                    r#ref,
                                ) {
                                    let full_msg =
                                        format!("remote: {}\r\n", msg);
                                    let _ = session.extended_data(
                                        channel,
                                        1,
                                        full_msg.into_bytes(),
                                    );
                                    let _ =
                                        session.exit_status_request(channel, 1);
                                    let _ = session.eof(channel);
                                    let _ = session.close(channel);
                                    self.cleanup_channel(channel);
                                    return Ok(());
                                }
                            }
                        }
                        if let Some(refs_for_hints) =
                            self.post_receive_refs.get(&channel)
                        {
                            *refs_for_hints.lock().await = refs.clone();
                        }
                        self.branch.insert(channel, refs);
                    }
                    Err(e) => {
                        tracing::warn!("ref_update_parse_error error={:?}", e);
                        if let Some(refs_for_hints) =
                            self.post_receive_refs.get(&channel)
                        {
                            refs_for_hints.lock().await.clear();
                        }
                        self.branch.insert(channel, vec![]);
                    }
                }

                if let Some(stdin) = self.stdin.get_mut(&channel) {
                    stdin.write_all(&buffered).await?;
                    stdin.flush().await?;
                }
                return Ok(());
            }

            if let Some(stdin) = self.stdin.get_mut(&channel) {
                stdin.write_all(data).await?;
                stdin.flush().await?;
            }
            return Ok(());
        }

        if let Some(stdin) = self.stdin.get_mut(&channel) {
            stdin.write_all(data).await?;
            if matches!(self.service, Some(GitService::UploadPack))
                && !self.upload_pack_eof_sent.contains(&channel)
            {
                let has_flush_pkt = data.windows(4).any(|w| w == b"0000");
                if has_flush_pkt {
                    stdin.flush().await?;
                    let _ = stdin.shutdown().await;
                    self.upload_pack_eof_sent.insert(channel);
                }
            }
        }
        Ok(())
    }

    async fn shell_request(
        &mut self,
        channel_id: ChannelId,
        session: &mut Session,
    ) -> Result<(), Self::Error> {
        if let Some(user) = &self.operator {
            let welcome_msg = format!(
                "Hi {}! You've successfully authenticated, but interactive shell access is not provided.\r\n",
                user.username
            );

            tracing::info!("shell_request user={}", user.username);
            let _ = session.data(channel_id, welcome_msg.into_bytes());
            let _ = session.exit_status_request(channel_id, 0);
            let _ = session.eof(channel_id);
            let _ = session.close(channel_id);
            let _ = session.flush();
        } else {
            tracing::warn!(
                "shell_request_unauthenticated channel={:?}",
                channel_id
            );
            let msg = "Authentication required\r\n";
            let _ = session.data(channel_id, msg.as_bytes().to_vec());
            let _ = session.exit_status_request(channel_id, 1);
            let _ = session.eof(channel_id);
            let _ = session.close(channel_id);
            let _ = session.flush();
        }
        Ok(())
    }

    #[tracing::instrument(skip(self, session), fields(cmd = ?String::from_utf8_lossy(data)))]
    async fn exec_request(
        &mut self,
        channel_id: ChannelId,
        data: &[u8],
        session: &mut Session,
    ) -> Result<(), Self::Error> {
        let client_info = self
            .client_addr
            .map(|addr| format!("{}", addr))
            .unwrap_or_else(|| "unknown".to_string());

        tracing::info!(
            "exec_request received channel={:?} client={}",
            channel_id,
            client_info
        );

        let git_shell_cmd = match std::str::from_utf8(data) {
            Ok(cmd) => cmd.trim(),
            Err(e) => {
                tracing::error!("invalid_command_encoding error={}", e);
                let _ = session.disconnect(
                    Disconnect::ServiceNotAvailable,
                    "Invalid command encoding",
                    "",
                );
                return Err(russh::Error::Disconnect);
            }
        };
        let (service, path) = match parse_git_command(git_shell_cmd) {
            Some((s, p)) => (s, p),
            None => {
                tracing::error!(
                    "invalid_git_command command={}",
                    git_shell_cmd
                );
                let msg = format!("Invalid git command: {}", git_shell_cmd);
                let _ = session.disconnect(
                    Disconnect::ServiceNotAvailable,
                    &msg,
                    "",
                );
                return Err(russh::Error::Disconnect);
            }
        };
        self.service = Some(service);
        let (owner, repo) = match parse_repo_path(path) {
            Some(pair) => pair,
            None => {
                let msg = format!("Invalid repository path: {}", path);
                tracing::error!("invalid_repo_path path={}", path);
                let _ = session.disconnect(
                    Disconnect::ServiceNotAvailable,
                    &msg,
                    "",
                );
                return Err(russh::Error::Disconnect);
            }
        };
        let namespace = owner.to_string();
        let repo = repo.strip_suffix(".git").unwrap_or(repo).to_string();

        let repo = match self.auth.find_repo(owner, &repo).await {
            Ok(repo) => repo,
            Err(e) => {
                tracing::error!("repo_fetch_error error={}", e);
                let _ = session.disconnect(
                    Disconnect::ServiceNotAvailable,
                    "Repository not found",
                    "",
                );
                return Err(russh::Error::Disconnect);
            }
        };

        self.model = Some(repo.clone());
        let operator = match &self.operator {
            Some(user) => user,
            None => {
                let msg = "Authentication error: no authenticated user";
                tracing::error!(
                    "exec_no_authenticated_user channel={:?}",
                    channel_id
                );
                let _ = session.disconnect(Disconnect::ByApplication, msg, "");
                return Err(russh::Error::Disconnect);
            }
        };

        let is_write = service == GitService::ReceivePack;
        let has_permission = self
            .auth
            .check_repo_permission(operator, &repo, is_write)
            .await;

        if !has_permission {
            let msg = format!(
                "Access denied: user '{}' does not have {} permission for repository {}",
                operator.username,
                if is_write { "write" } else { "read" },
                repo.name
            );
            tracing::error!(
                "access_denied user={} repo={} is_write={}",
                operator.username,
                repo.name,
                is_write
            );
            let _ = session.disconnect(Disconnect::ByApplication, &msg, "");
            return Err(russh::Error::Disconnect);
        }

        tracing::info!(
            "access_granted user={} repo={} is_write={}",
            operator.username,
            repo.name,
            is_write
        );

        let mut push_queue_lease = if is_write {
            let repo_id = repo.id;
            let queue_result =
                wait_for_push_queue_slot(self.sync.clone(), repo_id, |event, request_id| {
                    let request_id = request_id.to_string();
                    match event {
                        PushQueueEvent::Waiting(position) => {
                            let msg = format!(
                                "remote: another push is running for this repository. Queued {}/{}.\r\n",
                                position.position, position.total
                            );
                            let _ = session.extended_data(channel_id, 1, msg.as_bytes().to_vec());
                            let _ = session.flush();
                            tracing::info!(
                                repo_id = %repo_id,
                                request_id = %request_id,
                                position = position.position,
                                total = position.total,
                                "push_queue_waiting"
                            );
                        }
                        PushQueueEvent::Acquired => {
                            let msg = "remote: push queue slot acquired. Processing now.\r\n";
                            let _ = session.extended_data(channel_id, 1, msg.as_bytes().to_vec());
                            let _ = session.flush();
                            tracing::info!(
                                repo_id = %repo_id,
                                request_id = %request_id,
                                "push_queue_acquired"
                            );
                        }
                    }
                })
                .await;

            match queue_result {
                Ok(lease) => Some(lease),
                Err(error) => {
                    match &error {
                        PushQueueWaitError::Join(e) => {
                            tracing::error!(error = %e, repo = %repo.name, "push_queue_join_failed");
                            let msg = "remote: push queue is temporarily unavailable. Please retry later.\r\n";
                            let _ = session.extended_data(
                                channel_id,
                                1,
                                msg.as_bytes().to_vec(),
                            );
                        }
                        PushQueueWaitError::Lock(e) => {
                            tracing::error!(error = %e, repo_id = %repo.id, "push_queue_lock_failed");
                            let msg = "remote: push queue lock failed. Please retry later.\r\n";
                            let _ = session.extended_data(
                                channel_id,
                                1,
                                msg.as_bytes().to_vec(),
                            );
                        }
                        PushQueueWaitError::Timeout => {
                            tracing::warn!(repo_id = %repo.id, "push_queue_timeout");
                            let msg = "remote: push queue timed out. Please retry in a moment.\r\n";
                            let _ = session.extended_data(
                                channel_id,
                                1,
                                msg.as_bytes().to_vec(),
                            );
                        }
                    }
                    let _ = session.channel_failure(channel_id);
                    let _ = session.close(channel_id);
                    self.cleanup_channel(channel_id);
                    return if matches!(error, PushQueueWaitError::Timeout) {
                        Ok(())
                    } else {
                        Err(russh::Error::IO(io::Error::new(
                            io::ErrorKind::Other,
                            error.to_string(),
                        )))
                    };
                }
            }
        } else {
            None
        };

        let repo = match &self.model {
            Some(m) => m,
            None => {
                let msg = "Repository model not available";
                tracing::error!("repo_model_missing");
                let _ = session.disconnect(Disconnect::ByApplication, msg, "");
                return Err(russh::Error::Disconnect);
            }
        };
        let repo_path = PathBuf::from(&repo.storage_path);
        if !repo_path.exists() {
            tracing::error!("repo_path_not_found path={}", repo_path.display());
        }
        tracing::info!(
            "spawn_git_process service={:?} path={}",
            service,
            repo_path.display()
        );
        let mut cmd = build_git_command(service, repo_path);

        let mut shell = match cmd
            .stdin(Stdio::piped())
            .stdout(Stdio::piped())
            .stderr(Stdio::piped())
            .spawn()
        {
            Ok(shell) => {
                let _ = session.channel_success(channel_id);
                shell
            }
            Err(e) => {
                tracing::error!("process_spawn_failed error={}", e);
                if let Some(lease) = &mut push_queue_lease {
                    lease.release().await;
                }
                let _ = session.channel_failure(channel_id);
                self.cleanup_channel(channel_id);
                return Err(russh::Error::IO(e));
            }
        };
        let session_handle = session.handle();
        let stdin = match shell.stdin.take() {
            Some(s) => s,
            None => {
                tracing::error!(
                    "stdin pipe unavailable for channel={:?}",
                    channel_id
                );
                if let Some(lease) = &mut push_queue_lease {
                    lease.release().await;
                }
                let _ = session_handle.channel_failure(channel_id).await;
                return Err(russh::Error::IO(io::Error::new(
                    io::ErrorKind::Other,
                    "stdin unavailable",
                )));
            }
        };
        self.stdin.insert(channel_id, stdin);
        let mut shell_stdout = match shell.stdout.take() {
            Some(s) => s,
            None => {
                tracing::error!(
                    "stdout pipe unavailable for channel={:?}",
                    channel_id
                );
                if let Some(lease) = &mut push_queue_lease {
                    lease.release().await;
                }
                return Err(russh::Error::IO(io::Error::new(
                    io::ErrorKind::Other,
                    "stdout unavailable",
                )));
            }
        };
        let mut shell_stderr = match shell.stderr.take() {
            Some(s) => s,
            None => {
                tracing::error!(
                    "stderr pipe unavailable for channel={:?}",
                    channel_id
                );
                if let Some(lease) = &mut push_queue_lease {
                    lease.release().await;
                }
                return Err(russh::Error::IO(io::Error::new(
                    io::ErrorKind::Other,
                    "stderr unavailable",
                )));
            }
        };

        let (eof_tx, mut eof_rx) = tokio::sync::mpsc::channel::<bool>(10);
        self.eof.insert(channel_id, eof_tx);
        let refs_for_hints = Arc::new(Mutex::new(Vec::new()));
        self.post_receive_refs
            .insert(channel_id, refs_for_hints.clone());
        let repo_uid = repo.id;
        let repo_for_hints = repo.clone();
        let namespace_for_hints = namespace.clone();
        let should_sync = service == GitService::ReceivePack;
        let sync = self.sync.clone();
        let mut push_queue_lease = push_queue_lease;

        let fut = async move {
            tracing::info!(channel = ?channel_id, "git_task_started");

            let mut stdout_done = false;
            let mut stderr_done = false;

            let stdout_fut = forward(
                &session_handle,
                channel_id,
                &mut shell_stdout,
                |handle, chan, data| async move { handle.data(chan, data).await },
            );
            tokio::pin!(stdout_fut);

            let stderr_fut = forward(
                &session_handle,
                channel_id,
                &mut shell_stderr,
                |handle, chan, data| async move {
                    handle.extended_data(chan, 1, data).await
                },
            );
            tokio::pin!(stderr_fut);

            loop {
                tokio::select! {
                    result = shell.wait() => {
                        let status = match result {
                            Ok(status) => status,
                            Err(e) => {
                                if let Some(lease) = &mut push_queue_lease {
                                    lease.release().await;
                                }
                                return Err(russh::Error::IO(e));
                            }
                        };
                        let status_code = status.code().unwrap_or(128) as u32;

                        tracing::info!("git_process_exited channel={:?} status={}", channel_id, status_code);

                        if let Some(lease) = &mut push_queue_lease {
                            lease.release().await;
                        }

                        if !stdout_done || !stderr_done {
                            let _ = tokio::time::timeout(Duration::from_millis(100), async {
                                tokio::join!(
                                    async {
                                        if !stdout_done {
                                            let _ = (&mut stdout_fut).await;
                                        }
                                    },
                                    async {
                                        if !stderr_done {
                                            let _ = (&mut stderr_fut).await;
                                        }
                                    }
                                );
                            }).await;
                        }

                        if should_sync && status_code == 0 {
                            let queue = sync.send(RepoReceiveSyncTask { repo_uid }).await;
                            let refs_for_hints = refs_for_hints.lock().await.clone();
                            let msg = SSHandle::format_post_receive_hints(
                                &namespace_for_hints,
                                &repo_for_hints,
                                &refs_for_hints,
                                queue,
                            );
                            if !msg.is_empty() {
                                let _ = session_handle
                                    .extended_data(channel_id, 1, msg.into_bytes())
                                    .await;
                            }
                        }

                        let _ = session_handle.exit_status_request(channel_id, status_code).await;
                        sleep(Duration::from_millis(50)).await;
                        let _ = session_handle.eof(channel_id).await;
                        let _ = session_handle.close(channel_id).await;
                        tracing::info!(channel = ?channel_id, "channel_closed");
                        break;
                    }
                    result = &mut stdout_fut, if !stdout_done => {
                        tracing::info!("stdout completed");
                        stdout_done = true;
                        if let Err(e) = result {
                            tracing::warn!(error = ?e, "stdout_forward_error");
                        }
                    }
                    result = &mut stderr_fut, if !stderr_done => {
                        tracing::info!("stderr completed");
                        stderr_done = true;
                        if let Err(e) = result {
                            tracing::warn!(error = ?e, "stderr_forward_error");
                        }
                    }
                }
            }

            Ok::<(), russh::Error>(())
        };

        tokio::spawn(async move {
            if let Err(e) = fut.await {
                tracing::error!("git_ssh_channel_task_error error={}", e);
            }
            while eof_rx.recv().await.is_some() {}
        });
        Ok(())
    }
}