use std::{
    collections::{HashMap, HashSet},
    io::{self, Read, Write},
    net::{Shutdown, TcpStream},
    sync::{
        atomic::{AtomicBool, Ordering},
        Arc, Mutex,
    },
    thread,
    time::Duration,
};

use ssh2::Session;
use tauri::{AppHandle, Emitter};
use uuid::Uuid;

use crate::domain::ssh::{
    ConnectionClosedEvent, ExecResponse, ForwardStatus, HostMetricsResponse, ShellClosedEvent,
    ShellDataEvent, SshConnectRequest,
};

use super::tunnel::{connection_fingerprint, TunnelManager};

mod auth;
mod host_metrics;
mod known_hosts;
mod path;
mod proxy;
mod sftp;

pub use known_hosts::{bind_app_handle, respond as respond_host_key, HostKeyDecision};

use auth::authenticate;
use host_metrics::{parse_host_metrics, HOST_METRICS_SCRIPT};
use known_hosts::verify_host_key;
use proxy::connect_transport;

#[derive(Clone)]
struct Connection {
    session: Session,
    config: SshConnectRequest,
}

struct Shell {
    connection_id: String,
    session: Session,
    channel: ssh2::Channel,
    /// When false, reader buffers bytes until the UI opens the stream (avoids lost MOTD).
    output_live: Arc<AtomicBool>,
    pending_output: Arc<Mutex<Vec<u8>>>,
}

/// Dedicated SSH transport + persistent SFTP channel for one workload.
/// Each UI SFTP tab owns a browsing channel and may lazily open a separate
/// transfer channel; neither shares the pooled session used by exec/keepalive.
struct ManagedSftp {
    connection_id: String,
    session: Session,
    sftp: ssh2::Sftp,
}

#[derive(Default)]
pub struct SshService {
    connections: Mutex<HashMap<String, Connection>>,
    shells: Mutex<HashMap<String, Shell>>,
    /// Interactive browsing channel for each SFTP tab.
    sftps: Mutex<HashMap<String, Arc<Mutex<ManagedSftp>>>>,
    /// Lazily-created transfer channel, isolated from browsing/file operations.
    sftp_transfers: Mutex<HashMap<String, Arc<Mutex<ManagedSftp>>>>,
    /// Per-connection cancel flag for in-flight `ssh_exec` (Agent 停止).
    active_execs: Mutex<HashMap<String, Arc<AtomicBool>>>,
    /// Per-SFTP-session cancel for the active upload/download copy loop.
    sftp_transfer_cancels: Mutex<HashMap<String, Arc<AtomicBool>>>,
    tunnels: TunnelManager,
}

impl SshService {
    pub fn connect(&self, config: SshConnectRequest) -> Result<String, String> {
        validate_config(&config)?;
        let session = establish_session(&config)?;
        let id = Uuid::new_v4().to_string();
        let fingerprint = connection_fingerprint(&config);
        let live = {
            let mut connections = self.connections.lock().map_err(lock_error)?;
            connections.insert(id.clone(), Connection { session, config });
            connections.keys().cloned().collect::<HashSet<_>>()
        };
        // Reattach listeners that survived a reconnect (same SSH identity,
        // previous connection id is gone). Runtime is never persisted.
        self.tunnels.adopt_orphans(&id, &fingerprint, &live)?;
        Ok(id)
    }

    pub fn disconnect(
        &self,
        app: &AppHandle,
        connection_id: &str,
        retain_forwards: bool,
        reason: Option<String>,
    ) -> Result<(), String> {
        let reason = reason
            .map(|value| value.trim().to_string())
            .filter(|value| !value.is_empty())
            .unwrap_or_else(|| "connection closed".to_string());
        // Dedicated tunnel sessions keep listening across terminal flaps when
        // retain_forwards is true. Explicit teardown (tab close / panel disconnect)
        // must still close real Listeners.
        if !retain_forwards {
            let _ = self.tunnels.stop_connection(connection_id);
        }
        if let Some(flag) = self
            .active_execs
            .lock()
            .map_err(lock_error)?
            .remove(connection_id)
        {
            flag.store(true, Ordering::Release);
        }
        let shell_ids = {
            let shells = self.shells.lock().map_err(lock_error)?;
            shells
                .iter()
                .filter_map(|(id, shell)| {
                    (shell.connection_id == connection_id).then_some(id.clone())
                })
                .collect::<Vec<_>>()
        };
        for shell_id in shell_ids {
            if let Some(mut shell) = self.shells.lock().map_err(lock_error)?.remove(&shell_id) {
                let _ = shell.channel.close();
                let _ = shell.session.disconnect(None, &reason, None);
            }
        }
        let sftp_ids = {
            let sftps = self.sftps.lock().map_err(lock_error)?;
            sftps
                .iter()
                .filter_map(|(id, handle)| {
                    handle
                        .lock()
                        .ok()
                        .filter(|sftp| sftp.connection_id == connection_id)
                        .map(|_| id.clone())
                })
                .collect::<Vec<_>>()
        };
        for sftp_id in sftp_ids {
            let _ = self.close_sftp(&sftp_id);
        }
        if let Some(connection) = self
            .connections
            .lock()
            .map_err(lock_error)?
            .remove(connection_id)
        {
            let _ = connection.session.disconnect(None, &reason, None);
        }
        let _ = app.emit(
            "ssh:connection-closed",
            ConnectionClosedEvent {
                connection_id: connection_id.to_string(),
                reason,
            },
        );
        Ok(())
    }

    pub fn exec_with_status(
        &self,
        connection_id: &str,
        command: &str,
    ) -> Result<(String, i32), String> {
        let cancel = Arc::new(AtomicBool::new(false));
        {
            let mut active = self.active_execs.lock().map_err(lock_error)?;
            active.insert(connection_id.to_string(), cancel.clone());
        }
        let result = self.exec_cancellable(connection_id, command, &cancel);
        let _ = self
            .active_execs
            .lock()
            .map(|mut active| active.remove(connection_id));
        result
    }

    /// 请求终止该连接上正在执行的 `ssh_exec`（关闭远端命令通道）。
    pub fn cancel_exec(&self, connection_id: &str) -> Result<(), String> {
        if let Some(flag) = self
            .active_execs
            .lock()
            .map_err(lock_error)?
            .get(connection_id)
        {
            flag.store(true, Ordering::Release);
        }
        Ok(())
    }

    fn exec_cancellable(
        &self,
        connection_id: &str,
        command: &str,
        cancel: &AtomicBool,
    ) -> Result<(String, i32), String> {
        let connection = self.connection(connection_id)?;
        // Short poll interval so Agent「停止」能尽快关闭通道；执行结束后恢复。
        let poll_ms = 250_u32;
        let restore_ms = connection
            .config
            .timeout_seconds
            .unwrap_or(15)
            .clamp(3, 120)
            .saturating_mul(1_000) as u32;
        connection.session.set_timeout(poll_ms);

        let mut channel = connection
            .session
            .channel_session()
            .map_err(|error| format!("创建命令通道失败: {error}"))?;
        if let Err(error) = channel.exec(command) {
            connection.session.set_timeout(restore_ms);
            return Err(format!("执行命令失败: {error}"));
        }

        let mut output = Vec::new();
        let mut buffer = [0_u8; 8192];
        let cancelled = loop {
            if cancel.load(Ordering::Acquire) {
                break true;
            }
            match channel.read(&mut buffer) {
                Ok(0) if channel.eof() => break false,
                Ok(0) => thread::sleep(Duration::from_millis(20)),
                Ok(length) => output.extend_from_slice(&buffer[..length]),
                Err(error) if error.kind() == io::ErrorKind::WouldBlock => {
                    thread::sleep(Duration::from_millis(20));
                }
                Err(error) if error.kind() == io::ErrorKind::TimedOut => continue,
                Err(error) => {
                    connection.session.set_timeout(restore_ms);
                    return Err(format!("读取命令输出失败: {error}"));
                }
            }
        };

        if cancelled {
            let _ = channel.close();
            let _ = channel.wait_close();
            connection.session.set_timeout(restore_ms);
            return Err("命令已终止".to_string());
        }

        let mut stderr = Vec::new();
        loop {
            if cancel.load(Ordering::Acquire) {
                let _ = channel.close();
                let _ = channel.wait_close();
                connection.session.set_timeout(restore_ms);
                return Err("命令已终止".to_string());
            }
            match channel.stderr().read(&mut buffer) {
                Ok(0) => break,
                Ok(length) => stderr.extend_from_slice(&buffer[..length]),
                Err(error) if error.kind() == io::ErrorKind::WouldBlock => break,
                Err(error) if error.kind() == io::ErrorKind::TimedOut => break,
                Err(_) => break,
            }
        }

        let _ = channel.wait_close();
        let exit_code = channel.exit_status().unwrap_or(-1);
        connection.session.set_timeout(restore_ms);

        let mut text = String::from_utf8_lossy(&output).into_owned();
        if !stderr.is_empty() {
            text.push_str(&String::from_utf8_lossy(&stderr));
        }
        Ok((text, exit_code))
    }

    /// Active-tab status bar snapshot. Skips when Agent/`ssh_exec` holds the connection.
    pub fn host_metrics(&self, connection_id: &str) -> HostMetricsResponse {
        let busy = match self.active_execs.lock() {
            Ok(map) => map.contains_key(connection_id),
            Err(_) => true,
        };
        if busy {
            return HostMetricsResponse::skipped();
        }

        let command = format!("sh -c {}", shell_single_quote(HOST_METRICS_SCRIPT));
        match self.exec_with_status(connection_id, &command) {
            Ok((raw, _)) => HostMetricsResponse::ok(parse_host_metrics(&raw)),
            Err(error) => HostMetricsResponse::error(error),
        }
    }

    pub fn keepalive(&self, connection_id: &str) -> Result<(), String> {
        self.connection(connection_id)?
            .session
            .keepalive_send()
            .map(|_| ())
            .map_err(|error| format!("SSH 保活失败: {error}"))?;

        // Interactive panes / SFTP use dedicated SSH sessions so a busy
        // SFTP/exec channel cannot stall terminal IO. Keep those alive too.
        let shell_sessions = self
            .shells
            .lock()
            .map_err(lock_error)?
            .values()
            .filter(|shell| shell.connection_id == connection_id)
            .map(|shell| shell.session.clone())
            .collect::<Vec<_>>();
        for session in shell_sessions {
            session
                .keepalive_send()
                .map_err(|error| format!("Shell SSH 保活失败: {error}"))?;
        }
        let sftp_sessions = {
            let sftps = self.sftps.lock().map_err(lock_error)?;
            sftps
                .values()
                .filter_map(|handle| {
                    let guard = handle.lock().ok()?;
                    (guard.connection_id == connection_id).then(|| guard.session.clone())
                })
                .collect::<Vec<_>>()
        };
        for session in sftp_sessions {
            session
                .keepalive_send()
                .map_err(|error| format!("SFTP SSH 保活失败: {error}"))?;
        }
        Ok(())
    }

    pub fn create_shell(
        self: &std::sync::Arc<Self>,
        app: AppHandle,
        connection_id: &str,
        cols: u32,
        rows: u32,
        preferred_id: Option<String>,
    ) -> Result<String, String> {
        let connection = self.connection(connection_id)?;
        // A dedicated libssh2 session keeps the interactive non-blocking reader
        // from serializing SFTP and exec operations on the pooled connection.
        let session = establish_session(&connection.config)?;
        let mut channel = session
            .channel_session()
            .map_err(|error| format!("创建 Shell 通道失败: {error}"))?;
        // Best-effort: many servers only AcceptEnv LANG/LC_*. Ignore failures.
        let _ = channel.setenv("COLORTERM", "truecolor");
        let _ = channel.setenv("TERM", "xterm-256color");
        let _ = channel.setenv("LANG", &utf8_lang_for_ssh());
        channel
            .request_pty("xterm-256color", None, Some((cols, rows, 0, 0)))
            .map_err(|error| format!("申请远程 PTY 失败: {error}"))?;
        channel
            .shell()
            .map_err(|error| format!("启动远程 Shell 失败: {error}"))?;
        session.set_blocking(false);

        let shell_id = preferred_id.unwrap_or_else(|| format!("shell-{}", Uuid::new_v4()));
        let output_live = Arc::new(AtomicBool::new(false));
        let pending_output = Arc::new(Mutex::new(Vec::new()));
        let reader = channel.clone();
        self.shells.lock().map_err(lock_error)?.insert(
            shell_id.clone(),
            Shell {
                connection_id: connection_id.to_string(),
                session,
                channel,
                output_live: Arc::clone(&output_live),
                pending_output: Arc::clone(&pending_output),
            },
        );
        spawn_shell_reader(
            std::sync::Arc::clone(self),
            app,
            connection_id.to_string(),
            shell_id.clone(),
            reader,
            output_live,
            pending_output,
        );
        Ok(shell_id)
    }

    /// Flush MOTD/banner buffered before the UI subscribed to `ssh:shell-data`, then go live.
    pub fn shell_open_output(&self, shell_id: &str) -> Result<Vec<u8>, String> {
        let (output_live, pending_output) = {
            let shells = self.shells.lock().map_err(lock_error)?;
            let shell = shells
                .get(shell_id)
                .ok_or_else(|| format!("远程 Shell 不存在: {shell_id}"))?;
            (
                Arc::clone(&shell.output_live),
                Arc::clone(&shell.pending_output),
            )
        };

        // Give the server a short window to push MOTD / lastlog before we go live.
        let deadline = std::time::Instant::now() + Duration::from_millis(300);
        while std::time::Instant::now() < deadline {
            let has_data = pending_output
                .lock()
                .map(|guard| !guard.is_empty())
                .unwrap_or(false);
            if has_data {
                break;
            }
            thread::sleep(Duration::from_millis(15));
        }

        let mut pending = pending_output.lock().map_err(lock_error)?;
        output_live.store(true, Ordering::Release);
        Ok(std::mem::take(&mut *pending))
    }

    pub fn shell_write(&self, shell_id: &str, data: &[u8]) -> Result<(), String> {
        let mut shells = self.shells.lock().map_err(lock_error)?;
        let shell = shells
            .get_mut(shell_id)
            .ok_or_else(|| format!("远程 Shell 不存在: {shell_id}"))?;
        write_nonblocking(&mut shell.channel, data)
            .map_err(|error| format!("写入远程 Shell 失败: {error}"))
    }

    pub fn shell_resize(&self, shell_id: &str, cols: u32, rows: u32) -> Result<(), String> {
        let mut shells = self.shells.lock().map_err(lock_error)?;
        let shell = shells
            .get_mut(shell_id)
            .ok_or_else(|| format!("远程 Shell 不存在: {shell_id}"))?;
        retry_would_block(|| shell.channel.request_pty_size(cols, rows, None, None))
            .map_err(|error| format!("调整远程终端尺寸失败: {error}"))
    }

    pub fn close_shell(&self, shell_id: &str) -> Result<(), String> {
        let mut shell = self
            .shells
            .lock()
            .map_err(lock_error)?
            .remove(shell_id)
            .ok_or_else(|| format!("远程 Shell 不存在: {shell_id}"))?;
        let _ = shell.channel.send_eof();
        let _ = shell.channel.close();
        let _ = shell.session.disconnect(None, "shell closed by user", None);
        Ok(())
    }

    pub fn forward_local(
        &self,
        connection_id: &str,
        forward_id: Option<String>,
        local_bind_host: String,
        local_port: u16,
        remote_host: String,
        remote_port: u16,
    ) -> Result<ForwardStatus, String> {
        let config = self.connection(connection_id)?.config;
        self.tunnels.start_local(
            forward_id,
            connection_id,
            config,
            local_bind_host,
            local_port,
            remote_host,
            remote_port,
        )
    }

    pub fn forward_dynamic(
        &self,
        connection_id: &str,
        forward_id: Option<String>,
        local_port: u16,
    ) -> Result<ForwardStatus, String> {
        let config = self.connection(connection_id)?.config;
        self.tunnels
            .start_dynamic(forward_id, connection_id, config, local_port)
    }

    pub fn forward_remote(
        &self,
        connection_id: &str,
        forward_id: Option<String>,
        remote_bind_host: String,
        remote_port: u16,
        local_host: String,
        local_port: u16,
    ) -> Result<ForwardStatus, String> {
        let config = self.connection(connection_id)?.config;
        self.tunnels.start_remote(
            forward_id,
            connection_id,
            config,
            remote_bind_host,
            remote_port,
            local_host,
            local_port,
        )
    }

    pub fn forward_stop(&self, forward_id: &str) -> Result<(), String> {
        self.tunnels.stop(forward_id)
    }

    pub fn forward_list(&self, connection_id: Option<&str>) -> Result<Vec<ForwardStatus>, String> {
        self.tunnels.list(connection_id)
    }

    fn connection(&self, connection_id: &str) -> Result<Connection, String> {
        self.connections
            .lock()
            .map_err(lock_error)?
            .get(connection_id)
            .cloned()
            .ok_or_else(|| format!("SSH 连接不存在: {connection_id}"))
    }
}

fn validate_config(config: &SshConnectRequest) -> Result<(), String> {
    if config.host.trim().is_empty() || config.username.trim().is_empty() {
        return Err("主机和用户名不能为空".to_string());
    }
    if config.port == 0 {
        return Err("SSH 端口必须在 1-65535 范围内".to_string());
    }
    Ok(())
}

pub(super) fn establish_session(config: &SshConnectRequest) -> Result<Session, String> {
    let timeout_seconds = config.timeout_seconds.unwrap_or(15).clamp(3, 120);
    let address = format!("{}:{}", config.host.trim(), config.port);
    let tcp = connect_transport(config, timeout_seconds)?;
    tcp.set_read_timeout(Some(Duration::from_secs(timeout_seconds)))
        .ok();
    tcp.set_write_timeout(Some(Duration::from_secs(timeout_seconds)))
        .ok();

    let mut session = Session::new().map_err(|error| format!("初始化 SSH 失败: {error}"))?;
    session.set_tcp_stream(tcp);
    session.set_timeout(timeout_seconds.saturating_mul(1_000) as u32);
    session
        .handshake()
        .map_err(|error| format!("SSH 握手失败 ({address}): {error}"))?;
    session.set_keepalive(config.keep_alive.unwrap_or(true), 15);
    verify_host_key(&session, &config.host, config.port)?;
    authenticate(&session, config)?;
    Ok(session)
}

fn spawn_shell_reader(
    service: std::sync::Arc<SshService>,
    app: AppHandle,
    connection_id: String,
    shell_id: String,
    mut channel: ssh2::Channel,
    output_live: Arc<AtomicBool>,
    pending_output: Arc<Mutex<Vec<u8>>>,
) {
    thread::spawn(move || {
        let mut buffer = [0_u8; 8192];
        let mut output_tail = Vec::with_capacity(SHELL_OUTPUT_TAIL_LIMIT);
        let mut read_error = None;
        loop {
            match channel.read(&mut buffer) {
                Ok(0) if channel.eof() => break,
                Ok(0) => thread::sleep(Duration::from_millis(8)),
                Ok(length) => {
                    let chunk = &buffer[..length];
                    append_output_tail(&mut output_tail, chunk);
                    let should_emit = {
                        let mut pending = match pending_output.lock() {
                            Ok(guard) => guard,
                            Err(_) => break,
                        };
                        if output_live.load(Ordering::Acquire) {
                            true
                        } else {
                            // Cap bootstrap buffer so a stuck UI cannot grow forever.
                            const MAX_PENDING: usize = 256 * 1024;
                            if pending.len() < MAX_PENDING {
                                let room = MAX_PENDING - pending.len();
                                pending.extend_from_slice(&chunk[..length.min(room)]);
                            }
                            false
                        }
                    };
                    if should_emit {
                        let _ = app.emit(
                            "ssh:shell-data",
                            ShellDataEvent {
                                connection_id: connection_id.clone(),
                                shell_id: shell_id.clone(),
                                data: chunk.to_vec(),
                            },
                        );
                    }
                }
                Err(error) if error.kind() == io::ErrorKind::WouldBlock => {
                    thread::sleep(Duration::from_millis(8));
                }
                Err(error) => {
                    read_error = Some(error.to_string());
                    break;
                }
            }
        }

        // disconnect() removes the shell before closing its channel. In that
        // case the connection-closed event is authoritative and a second
        // shell-closed event would briefly show a misleading recovery prompt.
        let was_active = service
            .shells
            .lock()
            .map(|mut shells| shells.remove(&shell_id).is_some())
            .unwrap_or(false);
        if !was_active {
            return;
        }

        // Interactive shells use a dedicated SSH session. EOF, Ctrl+D, logout
        // and server-side TMOUT close only this shell; the pooled SSH
        // connection remains available to SFTP, tunnels and sibling panes.
        let (reason, idle_timeout) = classify_shell_close(&output_tail, read_error.as_deref());
        let _ = app.emit(
            "ssh:shell-closed",
            ShellClosedEvent {
                connection_id,
                shell_id,
                reason,
                idle_timeout,
            },
        );
    });
}

const SHELL_OUTPUT_TAIL_LIMIT: usize = 4096;

fn append_output_tail(tail: &mut Vec<u8>, chunk: &[u8]) {
    if chunk.len() >= SHELL_OUTPUT_TAIL_LIMIT {
        tail.clear();
        tail.extend_from_slice(&chunk[chunk.len() - SHELL_OUTPUT_TAIL_LIMIT..]);
        return;
    }
    let overflow = tail
        .len()
        .saturating_add(chunk.len())
        .saturating_sub(SHELL_OUTPUT_TAIL_LIMIT);
    if overflow > 0 {
        tail.drain(..overflow);
    }
    tail.extend_from_slice(chunk);
}

fn classify_shell_close(output_tail: &[u8], read_error: Option<&str>) -> (String, bool) {
    let output = String::from_utf8_lossy(output_tail).to_ascii_lowercase();
    let idle_timeout = output.contains("timed out waiting for input")
        || output.contains("auto-logout")
        || output.contains("auto logout");
    if idle_timeout {
        return ("shell.idleTimeout".to_string(), true);
    }
    if let Some(error) = read_error {
        return (format!("shell.readFailed: {error}"), false);
    }
    ("shell.exited".to_string(), false)
}

fn write_nonblocking(channel: &mut ssh2::Channel, data: &[u8]) -> io::Result<()> {
    let mut written = 0;
    while written < data.len() {
        match channel.write(&data[written..]) {
            Ok(0) => thread::sleep(Duration::from_millis(5)),
            Ok(length) => written += length,
            Err(error) if error.kind() == io::ErrorKind::WouldBlock => {
                thread::sleep(Duration::from_millis(5));
            }
            Err(error) => return Err(error),
        }
    }
    Ok(())
}

fn retry_would_block<T>(
    mut operation: impl FnMut() -> Result<T, ssh2::Error>,
) -> Result<T, ssh2::Error> {
    loop {
        match operation() {
            Err(error) if error.code() == ssh2::ErrorCode::Session(-37) => {
                thread::sleep(Duration::from_millis(5));
            }
            result => return result,
        }
    }
}

fn lock_error<T>(_: std::sync::PoisonError<T>) -> String {
    "SSH 服务状态锁异常".to_string()
}

fn shell_single_quote(value: &str) -> String {
    format!("'{}'", value.replace('\'', "'\\''"))
}

/// Best-effort UTF-8 LANG for remote AcceptEnv (mirrors local_terminal).
fn utf8_lang_for_ssh() -> String {
    if let Ok(lang) = std::env::var("LANG") {
        let upper = lang.to_ascii_uppercase();
        if upper.contains("UTF-8") || upper.contains("UTF8") {
            return lang;
        }
        if let Some(base) = lang.split('.').next() {
            if !base.is_empty() {
                return format!("{base}.UTF-8");
            }
        }
    }
    "en_US.UTF-8".to_string()
}

fn relay_channel(session: Session, mut channel: ssh2::Channel, mut socket: TcpStream) {
    session.set_blocking(false);
    let _ = socket.set_nonblocking(true);
    let mut network_buffer = [0_u8; 32 * 1024];
    let mut channel_buffer = [0_u8; 32 * 1024];
    let mut network_eof = false;
    loop {
        let mut progressed = false;
        if !network_eof {
            match socket.read(&mut network_buffer) {
                Ok(0) => {
                    network_eof = true;
                    let _ = channel.send_eof();
                }
                Ok(length) => {
                    if write_nonblocking(&mut channel, &network_buffer[..length]).is_err() {
                        break;
                    }
                    progressed = true;
                }
                Err(error) if error.kind() == io::ErrorKind::WouldBlock => {}
                Err(_) => break,
            }
        }
        match channel.read(&mut channel_buffer) {
            Ok(0) if channel.eof() => break,
            Ok(0) => {}
            Ok(length) => {
                if write_socket_nonblocking(&mut socket, &channel_buffer[..length]).is_err() {
                    break;
                }
                progressed = true;
            }
            Err(error) if error.kind() == io::ErrorKind::WouldBlock => {}
            Err(_) => break,
        }
        if !progressed {
            thread::sleep(Duration::from_millis(5));
        }
    }
    let _ = socket.shutdown(Shutdown::Both);
    let _ = channel.close();
}

fn write_socket_nonblocking(socket: &mut TcpStream, data: &[u8]) -> io::Result<()> {
    let mut written = 0;
    while written < data.len() {
        match socket.write(&data[written..]) {
            Ok(0) => thread::sleep(Duration::from_millis(5)),
            Ok(length) => written += length,
            Err(error) if error.kind() == io::ErrorKind::WouldBlock => {
                thread::sleep(Duration::from_millis(5))
            }
            Err(error) => return Err(error),
        }
    }
    Ok(())
}

#[allow(dead_code)]
fn _exec_response(result: Result<String, String>) -> ExecResponse {
    match result {
        Ok(output) => ExecResponse {
            success: true,
            result: Some(output),
            exit_code: None,
            error: None,
        },
        Err(error) => ExecResponse {
            success: false,
            result: None,
            exit_code: None,
            error: Some(error),
        },
    }
}

#[cfg(test)]
mod tests {
    use std::{
        fs,
        io::{Seek, SeekFrom},
        path::Path,
    };

    use super::sftp::{copy_with_progress, download_file, local_partial_path, upload_file};
    use super::*;

    #[test]
    fn detects_remote_shell_idle_timeout_from_recent_output() {
        let (reason, idle_timeout) =
            classify_shell_close(b"timed out waiting for input: auto-logout\r\n", None);

        assert!(idle_timeout);
        assert_eq!(reason, "shell.idleTimeout");
    }

    #[test]
    fn shell_output_tail_is_bounded_and_keeps_latest_marker() {
        let mut tail = vec![b'x'; SHELL_OUTPUT_TAIL_LIMIT];
        append_output_tail(&mut tail, b"auto-logout");

        assert_eq!(tail.len(), SHELL_OUTPUT_TAIL_LIMIT);
        assert!(tail.ends_with(b"auto-logout"));
    }

    #[test]
    fn reports_normal_shell_eof_without_marking_connection_dead() {
        let (reason, idle_timeout) = classify_shell_close(b"logout\r\n", None);

        assert!(!idle_timeout);
        assert_eq!(reason, "shell.exited");
    }

    #[test]
    fn resumes_copy_from_existing_offset() {
        let mut source = io::Cursor::new(b"abcdef".to_vec());
        source.seek(SeekFrom::Start(3)).unwrap();
        let mut target = io::Cursor::new(b"abc".to_vec());
        target.seek(SeekFrom::Start(3)).unwrap();
        let mut progress = Vec::new();

        copy_with_progress(&mut source, &mut target, 3, 6, None, |value| {
            progress.push(value)
        })
        .unwrap();

        assert_eq!(target.into_inner(), b"abcdef");
        assert_eq!(progress.first(), Some(&3));
        assert_eq!(progress.last(), Some(&6));
    }

    #[test]
    #[ignore = "requires an explicitly configured SSH endpoint"]
    fn real_ssh_exec_and_sftp_probe() {
        let host = std::env::var("OPSX_TEST_SSH_HOST").expect("OPSX_TEST_SSH_HOST");
        let username = std::env::var("OPSX_TEST_SSH_USER").unwrap_or_else(|_| "root".to_string());
        let port = std::env::var("OPSX_TEST_SSH_PORT")
            .ok()
            .and_then(|value| value.parse().ok())
            .unwrap_or(22);
        let private_key = std::env::var("OPSX_TEST_SSH_KEY").ok();
        let service = SshService::default();
        let connection_id = service
            .connect(SshConnectRequest {
                host,
                port,
                username,
                password: None,
                private_key,
                passphrase: None,
                proxy: None,
                jump_host: None,
                jump_hosts: None,
                timeout_seconds: Some(15),
                keep_alive: Some(true),
            })
            .expect("SSH connect");
        let (output, exit_code) = service
            .exec_with_status(&connection_id, "printf OPSX_TAURI_OK")
            .unwrap();
        assert_eq!(output, "OPSX_TAURI_OK");
        assert_eq!(exit_code, 0);
        service
            .keepalive(&connection_id)
            .expect("SSH protocol keepalive");
        let sftp_id = service.create_sftp(&connection_id).expect("SFTP connect");
        assert!(!service
            .sftp_list(&sftp_id, "/")
            .expect("SFTP list")
            .is_empty());
        let probe_id = Uuid::new_v4().to_string();
        let local_source = std::env::temp_dir().join(format!("opsxshell-upload-{probe_id}.txt"));
        let local_download =
            std::env::temp_dir().join(format!("opsxshell-download-{probe_id}.txt"));
        let remote_path = format!("/tmp/opsxshell-sftp-{probe_id}.txt");
        let payload = format!("OPSX_SFTP_ROUNDTRIP_{probe_id}\n").into_bytes();
        fs::write(&local_source, &payload).expect("write local SFTP fixture");
        let transfer_result = service.with_sftp(&sftp_id, |sftp| {
            upload_file(sftp, &local_source, &remote_path, None, |_, _| {})?;
            download_file(sftp, &remote_path, &local_download, None, |_, _| {})?;
            let downloaded = fs::read(&local_download)
                .map_err(|error| format!("read downloaded probe: {error}"))?;
            if downloaded != payload {
                return Err("SFTP round-trip content mismatch".to_string());
            }
            let _ = sftp.unlink(Path::new(&remote_path));
            let _ = sftp.unlink(Path::new(&format!("{remote_path}.opsx-part")));
            Ok(())
        });
        let _ = fs::remove_file(&local_source);
        let _ = fs::remove_file(&local_download);
        let _ = fs::remove_file(local_partial_path(&local_download));
        transfer_result.expect("SFTP upload/download round trip");
        service.close_sftp(&sftp_id).expect("close SFTP");
        let connection = service.connection(&connection_id).unwrap();
        let shell_session = establish_session(&connection.config).expect("shell SSH session");
        let mut shell = shell_session.channel_session().expect("shell channel");
        shell
            .request_pty("xterm-256color", None, Some((80, 24, 0, 0)))
            .expect("request PTY");
        shell.shell().expect("start shell");
        shell
            .write_all(b"printf OPSX_SHELL_OK; exit\n")
            .expect("write shell");
        let mut output = String::new();
        shell.read_to_string(&mut output).expect("read shell");
        assert!(
            output.contains("OPSX_SHELL_OK"),
            "unexpected shell output: {output}"
        );
    }

    #[test]
    #[ignore = "requires an explicitly configured SSH endpoint"]
    fn real_dynamic_forward_probe() {
        let _ = tracing_subscriber::fmt().with_test_writer().try_init();
        let host = std::env::var("OPSX_TEST_SSH_HOST").expect("OPSX_TEST_SSH_HOST");
        let username = std::env::var("OPSX_TEST_SSH_USER").unwrap_or_else(|_| "root".to_string());
        let port = std::env::var("OPSX_TEST_SSH_PORT")
            .ok()
            .and_then(|value| value.parse().ok())
            .unwrap_or(22);
        let private_key = std::env::var("OPSX_TEST_SSH_KEY").ok();
        let service = SshService::default();
        let connection_id = service
            .connect(SshConnectRequest {
                host: host.clone(),
                port,
                username,
                password: None,
                private_key,
                passphrase: None,
                proxy: None,
                jump_host: None,
                jump_hosts: None,
                timeout_seconds: Some(20),
                keep_alive: Some(true),
            })
            .expect("SSH connect");
        let forward_id = format!("real-socks-{}", Uuid::new_v4());
        let status = service
            .forward_dynamic(&connection_id, Some(forward_id.clone()), 0)
            .expect("start dynamic forward");

        let mut client = TcpStream::connect(("127.0.0.1", status.bound_port))
            .expect("connect local SOCKS5 listener");
        client
            .set_read_timeout(Some(Duration::from_secs(25)))
            .unwrap();
        client.write_all(&[5, 1, 0]).unwrap();
        let mut greeting = [0_u8; 2];
        client.read_exact(&mut greeting).unwrap();
        assert_eq!(greeting, [5, 0]);

        let target = host.as_bytes();
        assert!(target.len() <= 255);
        let mut request = vec![5, 1, 0, 3, target.len() as u8];
        request.extend_from_slice(target);
        request.extend_from_slice(&port.to_be_bytes());
        client.write_all(&request).unwrap();
        let mut reply = [0_u8; 10];
        client.read_exact(&mut reply).unwrap();
        assert_eq!(&reply[..2], &[5, 0]);

        let mut banner = [0_u8; 4];
        client.read_exact(&mut banner).unwrap();
        assert_eq!(&banner, b"SSH-");
        service.forward_stop(&forward_id).expect("stop forward");
        assert!(service
            .forward_list(Some(&connection_id))
            .unwrap()
            .is_empty());
    }

    #[test]
    #[ignore = "requires an explicitly configured SSH endpoint reachable through a jump host"]
    fn real_proxy_jump_probe() {
        let host = std::env::var("OPSX_TEST_SSH_HOST").expect("OPSX_TEST_SSH_HOST");
        let jump_host = std::env::var("OPSX_TEST_SSH_JUMP_HOST").expect("OPSX_TEST_SSH_JUMP_HOST");
        let username = std::env::var("OPSX_TEST_SSH_USER").unwrap_or_else(|_| "root".to_string());
        let port = std::env::var("OPSX_TEST_SSH_PORT")
            .ok()
            .and_then(|value| value.parse().ok())
            .unwrap_or(22);
        let jump_port = std::env::var("OPSX_TEST_SSH_JUMP_PORT")
            .ok()
            .and_then(|value| value.parse().ok())
            .unwrap_or(22);
        let jump = serde_json::json!({
            "host": jump_host,
            "port": jump_port,
            "username": username,
        });
        let jump_count = std::env::var("OPSX_TEST_SSH_JUMP_COUNT")
            .ok()
            .and_then(|value| value.parse::<usize>().ok())
            .unwrap_or(1)
            .clamp(1, 8);
        let service = SshService::default();
        let connection_id = service
            .connect(SshConnectRequest {
                host,
                port,
                username,
                password: None,
                private_key: None,
                passphrase: None,
                proxy: None,
                jump_host: (jump_count == 1).then(|| jump.clone()),
                jump_hosts: (jump_count > 1).then(|| vec![jump; jump_count]),
                timeout_seconds: Some(20),
                keep_alive: Some(true),
            })
            .expect("SSH connect through jump host");
        let (output, exit_code) = service
            .exec_with_status(&connection_id, "printf OPSX_PROXY_JUMP_OK")
            .unwrap();
        assert_eq!(output, "OPSX_PROXY_JUMP_OK");
        assert_eq!(exit_code, 0);
    }
}