use std::time::Duration;

use anyhow::Result;
use futures::future::BoxFuture;
use russh_sftp::client::SftpSession;
use tokio::time::timeout;
use tracing;

use super::super::super::models::config::SshConfig;
use super::super::super::models::types::{
    ExecRequest, ExecResult, FileEntry, ReadFileRequest, SystemProfile, WriteFileRequest,
};
use super::super::super::pool::connection::{ConnectionPool, Session};
use super::super::super::shell::run_shell;
use super::super::super::shell::{
    detect_remote_shell, shell_for_system_type, SystemProfileOutputs, SystemType,
};
use super::super::super::traits::{ExecService, FileService};
use super::connect::connect_and_auth;
use super::exec::{disconnect_session, exec_collect};

/// Convert an SFTP DirEntry to the internal FileEntry.
fn sftp_entry_to_file_entry(entry: &russh_sftp::client::fs::DirEntry, _parent: &str) -> FileEntry {
    let metadata = entry.metadata();
    let perms = metadata.permissions();
    FileEntry {
        path: entry.path(),
        name: entry.file_name(),
        is_dir: entry.file_type().is_dir(),
        size: metadata.len(),
        permissions: Some(format!("{}", perms)),
        modified_unix_ts: metadata.mtime.map(|t| t as u64),
    }
}

/// SSH tool implementation providing command execution and file operations.
///
/// Supports two modes:
/// - **Pool-less mode** (default): each operation establishes and tears down its own connection. Suitable for one-off operations.
/// - **Pooled mode** (`RusshTool::with_pool`): holds an internal connection pool, preferring to reuse connections. Suitable for batch operations.
#[derive(Clone, Debug)]
pub struct RusshTool {
    pool: Option<ConnectionPool>,
}

impl Default for RusshTool {
    fn default() -> Self {
        Self { pool: None }
    }
}

impl RusshTool {
    /// Create a pool-less tool instance where each operation establishes its own connection.
    pub fn new() -> Self {
        Self::default()
    }

    /// Create a pooled tool instance with an internal connection pool for connection reuse.
    pub fn with_pool(pool: ConnectionPool) -> Self {
        Self { pool: Some(pool) }
    }

    fn exec_structured(
        &self,
        cfg: SshConfig,
        req: ExecRequest,
        known_system_type: Option<SystemType>,
    ) -> BoxFuture<'static, Result<ExecResult>> {
        let pool = self.pool.clone();
        Box::pin(async move {
            let (session, from_pool) = if let Some(ref pool) = pool {
                let s = pool.get_or_connect(&cfg).await?;
                (s.handle, true)
            } else {
                let handle = connect_and_auth(cfg.clone()).await?;
                (handle, false)
            };
            let shell = match known_system_type {
                Some(st) => shell_for_system_type(st),
                None => detect_remote_shell(&session).await,
            };
            let command = shell.build_exec_command(&req);
            let result = if let Some(timeout_ms) = req.timeout_ms {
                match timeout(
                    Duration::from_millis(timeout_ms),
                    exec_collect(&session, &command),
                )
                .await
                {
                    Ok(exec_result) => {
                        let (exit_code, stdout, stderr) = exec_result?;
                        ExecResult {
                            exit_code,
                            stdout,
                            stderr,
                            timed_out: false,
                        }
                    }
                    Err(_) => {
                        tracing::warn!(command = %command, "exec command timed out");
                        ExecResult {
                            exit_code: 124,
                            stdout: Vec::new(),
                            stderr: b"command timed out".to_vec(),
                            timed_out: true,
                        }
                    }
                }
            } else {
                let (exit_code, stdout, stderr) = exec_collect(&session, &command).await?;
                ExecResult {
                    exit_code,
                    stdout,
                    stderr,
                    timed_out: false,
                }
            };

            if let Some(ref pool) = pool {
                if from_pool {
                    let s = Session::from_handle(cfg, session);
                    pool.return_connection(s).await;
                }
            } else {
                disconnect_session(&session).await?;
            }
            Ok(result)
        })
    }

    pub fn exec_with_system_type(
        &self,
        cfg: SshConfig,
        req: ExecRequest,
        system_type: SystemType,
    ) -> BoxFuture<'static, Result<ExecResult>> {
        self.exec_structured(cfg, req, Some(system_type))
    }

    fn detect_system_profile(
        &self,
        cfg: SshConfig,
        known_system_type: Option<SystemType>,
    ) -> BoxFuture<'static, Result<SystemProfile>> {
        let pool = self.pool.clone();
        Box::pin(async move {
            let (session, from_pool) = if let Some(ref pool) = pool {
                let s = pool.get_or_connect(&cfg).await?;
                (s.handle, true)
            } else {
                let handle = connect_and_auth(cfg.clone()).await?;
                (handle, false)
            };
            let shell = match known_system_type {
                Some(st) => shell_for_system_type(st),
                None => detect_remote_shell(&session).await,
            };
            let commands = shell.build_system_profile_commands();

            let os_info =
                exec_probe_string(&session, &shell.build_exec_command(&commands.os_info)).await;
            let arch = exec_probe_string(&session, &shell.build_exec_command(&commands.arch)).await;
            let shell_output =
                exec_probe_string(&session, &shell.build_exec_command(&commands.shell)).await;
            let package_manager = exec_probe_string(
                &session,
                &shell.build_exec_command(&commands.package_manager),
            )
            .await;

            if let Some(ref pool) = pool {
                if from_pool {
                    let s = Session::from_handle(cfg, session);
                    pool.return_connection(s).await;
                }
            } else {
                disconnect_session(&session).await?;
            }

            Ok(shell.parse_system_profile(&SystemProfileOutputs {
                os_info,
                arch,
                shell: shell_output,
                package_manager,
            }))
        })
    }

    pub fn detect_system_with_type(
        &self,
        cfg: SshConfig,
        system_type: SystemType,
    ) -> BoxFuture<'static, Result<SystemProfile>> {
        self.detect_system_profile(cfg, Some(system_type))
    }

    fn detect_system_type_inner(&self, cfg: SshConfig) -> BoxFuture<'static, Result<SystemType>> {
        let pool = self.pool.clone();
        Box::pin(async move {
            let (session, from_pool) = if let Some(ref pool) = pool {
                let s = pool.get_or_connect(&cfg).await?;
                (s.handle, true)
            } else {
                let handle = connect_and_auth(cfg.clone()).await?;
                (handle, false)
            };
            let shell = detect_remote_shell(&session).await;
            let system_type = shell.system_type();
            if let Some(ref pool) = pool {
                if from_pool {
                    let s = Session::from_handle(cfg, session);
                    pool.return_connection(s).await;
                }
            } else {
                disconnect_session(&session).await?;
            }
            Ok(system_type)
        })
    }

    /// Execute operations using the SFTP subsystem, reusing connections in pooled mode.
    fn with_sftp<F, Fut, R>(&self, cfg: SshConfig, f: F) -> BoxFuture<'static, Result<R>>
    where
        F: FnOnce(SftpSession) -> Fut + Send + 'static,
        Fut: std::future::Future<Output = Result<R>> + Send,
        R: Send + 'static,
    {
        let pool = self.pool.clone();
        Box::pin(async move {
            let (handle, from_pool) = if let Some(ref pool) = pool {
                let session = pool.get_or_connect(&cfg).await?;
                (session.handle, true)
            } else {
                let handle = connect_and_auth(cfg.clone()).await?;
                (handle, false)
            };

            // Open SFTP subsystem
            let sftp_result: Result<SftpSession> = async {
                let channel = handle
                    .channel_open_session()
                    .await
                    .map_err(anyhow::Error::from)?;
                channel
                    .request_subsystem(true, "sftp")
                    .await
                    .map_err(anyhow::Error::from)?;
                SftpSession::new(channel.into_stream())
                    .await
                    .map_err(anyhow::Error::from)
            }
            .await;

            let sftp = match sftp_result {
                Ok(s) => s,
                Err(e) => {
                    if let Some(ref pool) = pool {
                        if from_pool {
                            let session = Session::from_handle(cfg, handle);
                            pool.return_connection(session).await;
                        }
                    } else {
                        let _ = disconnect_session(&handle).await;
                    }
                    return Err(e);
                }
            };

            // Execute user operation
            let result = f(sftp).await;

            // Clean up connection
            if let Some(ref pool) = pool {
                if from_pool {
                    let session = Session::from_handle(cfg, handle);
                    pool.return_connection(session).await;
                }
            } else {
                let _ = disconnect_session(&handle).await;
            }

            result
        })
    }

    fn read_file_bytes(
        &self,
        cfg: SshConfig,
        req: ReadFileRequest,
    ) -> BoxFuture<'static, Result<Vec<u8>>> {
        self.with_sftp(cfg, move |sftp| async move {
            let data = if req.offset.is_some() || req.limit.is_some() {
                use tokio::io::{AsyncReadExt, AsyncSeekExt};
                let mut file = sftp.open(&req.path).await?;
                if let Some(offset) = req.offset {
                    file.seek(std::io::SeekFrom::Start(offset)).await?;
                }
                match req.limit {
                    Some(limit) => {
                        let mut buf = vec![0u8; limit as usize];
                        let n = file.read(&mut buf).await?;
                        buf.truncate(n);
                        buf
                    }
                    None => {
                        let mut buf = Vec::new();
                        file.read_to_end(&mut buf).await?;
                        buf
                    }
                }
            } else {
                sftp.read(&req.path).await?
            };
            Ok(data)
        })
    }

    fn write_file_bytes(
        &self,
        cfg: SshConfig,
        req: WriteFileRequest,
    ) -> BoxFuture<'static, Result<()>> {
        self.with_sftp(cfg, move |sftp| async move {
            if req.create_parent {
                if let Some(parent) = std::path::Path::new(&req.path).parent() {
                    let parent_str = parent.to_string_lossy().to_string();
                    if !parent_str.is_empty() && parent_str != "." {
                        let mut current = std::path::PathBuf::new();
                        for component in parent.components() {
                            current.push(component);
                            let dir = current.to_string_lossy().to_string();
                            if !dir.is_empty() && dir != "." {
                                match sftp.create_dir(&dir).await {
                                    Ok(()) => {}
                                    Err(e) => {
                                        let err_msg = e.to_string();
                                        if !err_msg.contains("already exists")
                                            && !err_msg.contains("File exists")
                                        {
                                            return Err(anyhow::anyhow!(
                                                "failed to create directory '{}': {}",
                                                dir,
                                                e
                                            ));
                                        }
                                    }
                                }
                            }
                        }
                    }
                }
            }

            if !req.overwrite {
                match sftp.try_exists(&req.path).await {
                    Ok(true) => {
                        return Err(anyhow::anyhow!(
                            "file already exists and overwrite is false: {}",
                            req.path
                        ));
                    }
                    Err(e) => {
                        tracing::warn!(path = %req.path, error = %e, "failed to check file existence");
                    }
                    _ => {}
                }
            }

            sftp.write(&req.path, &req.content).await?;
            Ok(())
        })
    }

    fn list_directory_entries(
        &self,
        cfg: SshConfig,
        path: String,
    ) -> BoxFuture<'static, Result<Vec<FileEntry>>> {
        self.with_sftp(cfg, move |sftp| async move {
            let read_dir = sftp.read_dir(&path).await?;
            let entries: Vec<FileEntry> = read_dir
                .map(|entry| sftp_entry_to_file_entry(&entry, &path))
                .collect();
            Ok(entries)
        })
    }

    fn download_file_impl(
        &self,
        cfg: SshConfig,
        remote_path: &str,
        local_path: &str,
    ) -> BoxFuture<'static, Result<()>> {
        let remote_path = remote_path.to_string();
        let local_path = local_path.to_string();
        self.with_sftp(cfg, move |sftp| async move {
            let data = sftp.read(&remote_path).await?;
            tokio::fs::write(&local_path, &data).await?;
            Ok(())
        })
    }

    fn upload_file_impl(
        &self,
        cfg: SshConfig,
        local_path: &str,
        remote_path: &str,
    ) -> BoxFuture<'static, Result<()>> {
        let local_path = local_path.to_string();
        let remote_path = remote_path.to_string();
        self.with_sftp(cfg, move |sftp| async move {
            let data = tokio::fs::read(&local_path).await?;
            sftp.write(&remote_path, &data).await?;
            Ok(())
        })
    }
}

impl ExecService for RusshTool {
    fn validate(&self, cfg: SshConfig) -> BoxFuture<'static, Result<()>> {
        Box::pin(async move {
            let session = connect_and_auth(cfg).await?;
            disconnect_session(&session).await?;
            Ok(())
        })
    }

    fn exec(&self, cfg: SshConfig, req: ExecRequest) -> BoxFuture<'static, Result<ExecResult>> {
        self.exec_structured(cfg, req, None)
    }

    fn detect_system_type(&self, cfg: SshConfig) -> BoxFuture<'static, Result<SystemType>> {
        self.detect_system_type_inner(cfg)
    }

    fn detect_system(&self, cfg: SshConfig) -> BoxFuture<'static, Result<SystemProfile>> {
        self.detect_system_profile(cfg, None)
    }

    fn open_shell(&self, cfg: SshConfig) -> BoxFuture<'static, Result<()>> {
        Box::pin(async move {
            let session = connect_and_auth(cfg).await?;
            run_shell(&session).await?;
            disconnect_session(&session).await?;
            Ok(())
        })
    }
}

impl FileService for RusshTool {
    fn list_dir(&self, cfg: SshConfig, path: String) -> BoxFuture<'static, Result<Vec<FileEntry>>> {
        self.list_directory_entries(cfg, path)
    }

    fn read_file(
        &self,
        cfg: SshConfig,
        req: ReadFileRequest,
    ) -> BoxFuture<'static, Result<Vec<u8>>> {
        self.read_file_bytes(cfg, req)
    }

    fn write_file(&self, cfg: SshConfig, req: WriteFileRequest) -> BoxFuture<'static, Result<()>> {
        self.write_file_bytes(cfg, req)
    }

    fn download_file(
        &self,
        cfg: SshConfig,
        remote_path: &str,
        local_path: &str,
    ) -> BoxFuture<'static, Result<()>> {
        self.download_file_impl(cfg, remote_path, local_path)
    }

    fn upload_file(
        &self,
        cfg: SshConfig,
        local_path: &str,
        remote_path: &str,
    ) -> BoxFuture<'static, Result<()>> {
        self.upload_file_impl(cfg, local_path, remote_path)
    }
}

async fn exec_probe_string(
    session: &russh::client::Handle<super::connect::ClientHandler>,
    command: &str,
) -> String {
    match exec_collect(session, command).await {
        Ok((0, stdout, _)) => String::from_utf8_lossy(&stdout).to_string(),
        _ => String::new(),
    }
}
