use std::collections::HashMap;
use std::net::SocketAddr;
use std::sync::Arc;

use russh::server::{Auth, Handler, Msg, Session};
use russh::{Channel, ChannelId};
use tokio::sync::OwnedSemaphorePermit;
use uuid::Uuid;

use models::users::user;

use crate::ssh::handler::git_op::GitOperation;
use crate::ssh::service::GitSshService;

pub mod git_op;

mod auth;
pub(crate) mod channel;
mod command;

#[derive(Default)]
pub struct ChannelState {
    pub buffer: Vec<u8>,
    pub git_op: Option<GitOperation>,
    pub repo_id: Option<Uuid>,
    pub space_id: Option<Uuid>,
    pub actor_id: Option<Uuid>,
    pub buffer_overflow: bool,
    pub nak_sent: bool,
    /// Set when a duplex git session (upload-pack / upload-archive) is running
    /// for this channel; client data is then relayed to its stdin instead of
    /// being buffered.
    pub dispatched: bool,
    pub stdin_tx: Option<tokio::sync::mpsc::UnboundedSender<Vec<u8>>>,
    /// Abort handle for the task that owns the streaming git child process.
    pub stream_abort: Option<tokio::task::AbortHandle>,
    /// Wall-clock budget for the duplex git process, captured at exec time.
    pub rpc_timeout_secs: u64,
}

pub struct GitSshHandler {
    pub service: Arc<GitSshService>,
    pub client_addr: Option<SocketAddr>,
    pub session_permit: Option<OwnedSemaphorePermit>,
    pub operator: Option<user::Model>,
    pub channels: HashMap<ChannelId, ChannelState>,
    pub over_capacity: bool,
}

impl GitSshHandler {
    pub fn new(
        service: Arc<GitSshService>,
        client_addr: Option<SocketAddr>,
        session_permit: Option<OwnedSemaphorePermit>,
    ) -> Self {
        let over_capacity = session_permit.is_none();
        Self {
            service,
            client_addr,
            session_permit,
            operator: None,
            channels: HashMap::new(),
            over_capacity,
        }
    }

    pub fn config(&self) -> &config::git::GitSshConfig {
        &self.service.config
    }
}

impl Handler for GitSshHandler {
    type Error = russh::Error;

    async fn auth_none(&mut self, user: &str) -> Result<Auth, Self::Error> {
        tracing::debug!(user, client = ?self.client_addr, "ssh auth_none rejected");
        Ok(Auth::reject())
    }

    async fn auth_password(&mut self, user: &str, _password: &str) -> Result<Auth, Self::Error> {
        tracing::debug!(user, client = ?self.client_addr, "ssh password auth rejected (publickey only)");
        Ok(Auth::reject())
    }

    async fn auth_publickey_offered(
        &mut self,
        user: &str,
        public_key: &russh::keys::PublicKey,
    ) -> Result<Auth, Self::Error> {
        self.auth_publickey_offered_impl(user, public_key).await
    }

    async fn auth_publickey(
        &mut self,
        user: &str,
        public_key: &russh::keys::PublicKey,
    ) -> Result<Auth, Self::Error> {
        self.auth_publickey_impl(user, public_key).await
    }

    async fn auth_openssh_certificate(
        &mut self,
        user: &str,
        _certificate: &russh::keys::Certificate,
    ) -> Result<Auth, Self::Error> {
        tracing::debug!(user, "ssh certificate auth rejected");
        Ok(Auth::reject())
    }

    async fn authentication_banner(&mut self) -> Result<Option<String>, Self::Error> {
        Ok(self.config().banner.clone())
    }

    async fn channel_open_session(
        &mut self,
        _channel: Channel<Msg>,
        reply: russh::server::ChannelOpenHandle,
        _session: &mut Session,
    ) -> Result<(), Self::Error> {
        reply.accept().await;
        Ok(())
    }

    async fn channel_close(
        &mut self,
        channel: ChannelId,
        _session: &mut Session,
    ) -> Result<(), Self::Error> {
        self.channel_close_impl(channel).await;
        Ok(())
    }

    async fn channel_eof(
        &mut self,
        channel: ChannelId,
        session: &mut Session,
    ) -> Result<(), Self::Error> {
        self.channel_eof_impl(channel, session).await
    }

    async fn data(
        &mut self,
        channel: ChannelId,
        data: &[u8],
        session: &mut Session,
    ) -> Result<(), Self::Error> {
        self.data_impl(channel, data, session).await
    }

    async fn exec_request(
        &mut self,
        channel: ChannelId,
        data: &[u8],
        session: &mut Session,
    ) -> Result<(), Self::Error> {
        self.exec_request_impl(channel, data, session).await
    }

    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> {
        let _ = session.flush();
        Ok(())
    }

    async fn shell_request(
        &mut self,
        channel: ChannelId,
        session: &mut Session,
    ) -> Result<(), Self::Error> {
        self.shell_request_impl(channel, session).await
    }

    async fn subsystem_request(
        &mut self,
        _channel: ChannelId,
        _name: &str,
        session: &mut Session,
    ) -> Result<(), Self::Error> {
        let _ = session.flush();
        Ok(())
    }

    async fn env_request(
        &mut self,
        _channel: ChannelId,
        _variable_name: &str,
        _variable_value: &str,
        session: &mut Session,
    ) -> Result<(), Self::Error> {
        let _ = session.flush();
        Ok(())
    }

    async fn window_change_request(
        &mut self,
        _channel: ChannelId,
        _col_width: u32,
        _row_height: u32,
        _pix_width: u32,
        _pix_height: u32,
        session: &mut Session,
    ) -> Result<(), Self::Error> {
        let _ = session.flush();
        Ok(())
    }
}