//! SFTP channels over the shared SSH connection (replaces ssh2 `client.sftp()`).
//!
//! Built on `russh_sftp::client::RawSftpSession` so positioned reads/writes can be pipelined
//! (the Transfer Manager keeps up to 8 requests in flight, like 0.1.0). Errors keep the SFTP
//! status code so callers can map them exactly like the 0.1.0 `sftpError`/`errorCode` helpers.

use crate::core::error::{CoreError, CoreResult};
use crate::core::ssh::session::{not_connected, SessionManager, SessionRecord};
use russh_sftp::client::error::Error as RawError;
use russh_sftp::client::RawSftpSession;
use russh_sftp::protocol::{FileAttributes, OpenFlags, StatusCode};
use std::sync::Arc;
use std::time::Duration;

/// SFTP request timeout (a dead link never answers; the transfer stall detector uses 60 s too).
const REQUEST_TIMEOUT_SECS: u64 = 60;

#[derive(Debug, Clone)]
pub enum SftpError {
    /// SSH_FX_* status code and server message.
    Status(u32, String),
    /// Channel closed / connection lost / timeout.
    Connection(String),
    Other(String),
}

impl SftpError {
    pub fn status(&self) -> Option<u32> {
        match self {
            SftpError::Status(code, _) => Some(*code),
            _ => None,
        }
    }

    pub fn message(&self) -> String {
        match self {
            SftpError::Status(_, m) | SftpError::Connection(m) | SftpError::Other(m) => m.clone(),
        }
    }

    /// 0.1.0 `sftpError`: FILE_* codes for the File Manager.
    pub fn file_error(&self, message: &str) -> CoreError {
        let code = match self {
            SftpError::Status(2, _) => "FILE_NOT_FOUND",
            SftpError::Status(3, _) => "FILE_PERMISSION_DENIED",
            SftpError::Status(6 | 7, _) | SftpError::Connection(_) => "SESSION_NOT_CONNECTED",
            SftpError::Status(8, _) => "FILE_OPERATION_UNSUPPORTED",
            _ => "FILE_REMOTE_IO_FAILED",
        };
        let text = if message.is_empty() { self.message() } else { message.to_string() };
        CoreError::new(code, format!("{code}: {text}"))
    }

    /// 0.1.0 Transfer Manager `errorCode`.
    pub fn transfer_code(&self) -> &'static str {
        match self {
            SftpError::Status(6 | 7, _) | SftpError::Connection(_) => "SESSION_NOT_CONNECTED",
            SftpError::Status(3, _) => "FILE_PERMISSION_DENIED",
            SftpError::Status(2, _) => "FILE_NOT_FOUND",
            _ => "TRANSFER_FAILED",
        }
    }
}

impl std::fmt::Display for SftpError {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        f.write_str(&self.message())
    }
}

fn convert(error: RawError) -> SftpError {
    match error {
        RawError::Status(status) => SftpError::Status(status_number(status.status_code), status.error_message),
        RawError::Timeout => SftpError::Connection("SFTP request timed out.".into()),
        RawError::UnexpectedBehavior(m) if m.contains("closed") => SftpError::Connection(m),
        RawError::IO(m) => SftpError::Connection(m),
        other => SftpError::Other(other.to_string()),
    }
}

fn status_number(code: StatusCode) -> u32 {
    code as u32
}

pub type SftpResult<T> = Result<T, SftpError>;

/// Normalized attributes (ssh2 `Stats` equivalent).
#[derive(Debug, Clone, Default)]
pub struct Stat {
    pub size: u64,
    pub mode: u32,
    pub mtime: u32,
    pub uid: Option<u32>,
    pub gid: Option<u32>,
}

const S_IFMT: u32 = 0o170000;

impl Stat {
    fn from(attrs: &FileAttributes) -> Stat {
        Stat { size: attrs.size.unwrap_or(0), mode: attrs.permissions.unwrap_or(0), mtime: attrs.mtime.unwrap_or(0), uid: attrs.uid, gid: attrs.gid }
    }
    pub fn is_dir(&self) -> bool {
        self.mode & S_IFMT == 0o040000
    }
    pub fn is_file(&self) -> bool {
        self.mode & S_IFMT == 0o100000
    }
    pub fn is_symlink(&self) -> bool {
        self.mode & S_IFMT == 0o120000
    }
}

#[derive(Debug, Clone)]
pub struct DirEntry {
    pub name: String,
    pub stat: Stat,
}

/// One SFTP subsystem channel. Cheap to share (`Arc`); requests may run concurrently.
pub struct Sftp {
    raw: RawSftpSession,
    /// Largest data payload per READ/WRITE request (ssh2 rule: OpenSSH 256 KiB packets,
    /// otherwise 34000 bytes; `limits@openssh.com` wins when offered).
    max_read: usize,
    max_write: usize,
}

/// Open file handle (closed explicitly; dropping without close leaks only the server handle
/// until the channel closes).
#[derive(Debug, Clone)]
pub struct FileHandle(pub String);

pub fn mode_attrs(mode: u32) -> FileAttributes {
    let mut attrs = FileAttributes::empty();
    attrs.permissions = Some(mode);
    attrs
}

impl Sftp {
    /// `openSftp`: new SFTP channel on the connected session.
    pub async fn open(manager: &SessionManager, session_id: &str) -> CoreResult<Arc<Sftp>> {
        let record = manager.require_connected(session_id)?;
        Self::open_record(&record).await
    }

    pub async fn open_record(record: &Arc<SessionRecord>) -> CoreResult<Arc<Sftp>> {
        let handle = record.handle()?;
        let failed = |e: String| if record.is_connected() { CoreError::new("SFTP_OPEN_FAILED", e) } else { not_connected() };
        let opened = tokio::time::timeout(Duration::from_secs(30), async {
            let channel = handle.channel_open_session().await.map_err(|e| e.to_string())?;
            channel.request_subsystem(true, "sftp").await.map_err(|e| e.to_string())?;
            let raw = RawSftpSession::new(channel.into_stream());
            raw.set_timeout(REQUEST_TIMEOUT_SECS);
            let version = raw.init().await.map_err(|e| e.to_string())?;
            // Large requests only when the server announces its limits; Win32-OpenSSH 8.1
            // (Windows Server 2022) fails 255 KiB writes. Small pieces are sent concurrently.
            let base = 34_000 - 1024;
            let (mut max_read, mut max_write) = (base, base);
            if version.extensions.contains_key("limits@openssh.com") {
                if let Ok(limits) = raw.limits().await {
                    let large = 256 * 1024 - 1024;
                    max_read = large;
                    max_write = large;
                    if limits.max_read_len > 0 {
                        max_read = max_read.min(limits.max_read_len as usize);
                    }
                    if limits.max_write_len > 0 {
                        max_write = max_write.min(limits.max_write_len as usize);
                    }
                }
            }
            Ok::<_, String>((raw, max_read, max_write))
        })
        .await
        .map_err(|_| failed("SFTP subsystem did not answer.".into()))?
        .map_err(failed)?;
        let (raw, max_read, max_write) = opened;
        Ok(Arc::new(Sftp { raw, max_read, max_write }))
    }

    pub fn close(&self) {
        let _ = self.raw.close_session();
    }

    pub async fn stat(&self, path: &str) -> SftpResult<Stat> {
        self.raw.stat(path).await.map(|a| Stat::from(&a.attrs)).map_err(convert)
    }

    pub async fn lstat(&self, path: &str) -> SftpResult<Stat> {
        self.raw.lstat(path).await.map(|a| Stat::from(&a.attrs)).map_err(convert)
    }

    pub async fn exists(&self, path: &str) -> bool {
        self.lstat(path).await.is_ok()
    }

    pub async fn read_dir(&self, path: &str) -> SftpResult<Vec<DirEntry>> {
        let handle = self.raw.opendir(path).await.map_err(convert)?.handle;
        let mut entries = Vec::new();
        let result = loop {
            match self.raw.readdir(handle.as_str()).await {
                Ok(name) => {
                    entries.extend(name.files.into_iter().map(|f| DirEntry { stat: Stat::from(&f.attrs), name: f.filename }));
                    if entries.len() > 1_000_000 {
                        break Err(SftpError::Other("Directory listing is too large.".into()));
                    }
                }
                Err(RawError::Status(status)) if status.status_code == StatusCode::Eof => break Ok(()),
                Err(error) => break Err(convert(error)),
            }
        };
        let _ = self.raw.close(handle).await;
        result.map(|_| entries)
    }

    pub async fn mkdir(&self, path: &str, mode: u32) -> SftpResult<()> {
        self.raw.mkdir(path, mode_attrs(mode)).await.map(|_| ()).map_err(convert)
    }

    pub async fn rmdir(&self, path: &str) -> SftpResult<()> {
        self.raw.rmdir(path).await.map(|_| ()).map_err(convert)
    }

    pub async fn unlink(&self, path: &str) -> SftpResult<()> {
        self.raw.remove(path).await.map(|_| ()).map_err(convert)
    }

    pub async fn rename(&self, from: &str, to: &str) -> SftpResult<()> {
        self.raw.rename(from, to).await.map(|_| ()).map_err(convert)
    }

    /// ssh2 flag strings used by 0.1.0: "r", "r+", "wx".
    pub async fn open_file(&self, path: &str, flags: &str, mode: Option<u32>) -> SftpResult<FileHandle> {
        let flags = match flags {
            "r" => OpenFlags::READ,
            "r+" => OpenFlags::READ | OpenFlags::WRITE,
            "wx" => OpenFlags::WRITE | OpenFlags::CREATE | OpenFlags::EXCLUDE | OpenFlags::TRUNCATE,
            _ => OpenFlags::WRITE | OpenFlags::CREATE | OpenFlags::TRUNCATE,
        };
        let attrs = mode.map(mode_attrs).unwrap_or_else(FileAttributes::empty);
        self.raw.open(path, flags, attrs).await.map(|h| FileHandle(h.handle)).map_err(convert)
    }

    pub async fn close_file(&self, handle: &FileHandle) -> SftpResult<()> {
        self.raw.close(handle.0.as_str()).await.map(|_| ()).map_err(convert)
    }

    pub async fn fstat(&self, handle: &FileHandle) -> SftpResult<Stat> {
        self.raw.fstat(handle.0.as_str()).await.map(|a| Stat::from(&a.attrs)).map_err(convert)
    }

    pub async fn truncate(&self, handle: &FileHandle, size: u64) -> SftpResult<()> {
        let mut attrs = FileAttributes::empty();
        attrs.size = Some(size);
        self.raw.fsetstat(handle.0.as_str(), attrs).await.map(|_| ()).map_err(convert)
    }

    pub async fn fsync(&self, handle: &FileHandle) -> SftpResult<()> {
        self.raw.fsync(handle.0.as_str()).await.map(|_| ()).map_err(convert)
    }

    /// Reads up to `len` bytes at `offset`; an empty result means end of file.
    pub async fn read_at(&self, handle: &FileHandle, offset: u64, len: u32) -> SftpResult<Vec<u8>> {
        match self.raw.read(handle.0.as_str(), offset, len).await {
            Ok(data) => Ok(data.data),
            Err(RawError::Status(status)) if status.status_code == StatusCode::Eof => Ok(Vec::new()),
            Err(error) => Err(convert(error)),
        }
    }

    /// Fills `len` bytes from `offset` (servers may answer short reads), stops at EOF.
    pub async fn read_full(&self, handle: &FileHandle, offset: u64, len: usize) -> SftpResult<Vec<u8>> {
        // All pieces in parallel; a short piece (EOF or a server limit) falls back to sequential reads.
        let pieces: Vec<(u64, usize)> = (0..len).step_by(self.max_read.max(1)).map(|start| (offset + start as u64, (len - start).min(self.max_read))).collect();
        let results = futures::future::join_all(pieces.iter().map(|(at, want)| self.read_at(handle, *at, *want as u32))).await;
        let mut out = Vec::with_capacity(len);
        for ((_, want), result) in pieces.iter().zip(results) {
            let data = result?;
            let full = data.len() == *want;
            out.extend_from_slice(&data);
            if !full {
                break;
            }
        }
        while out.len() < len {
            let want = (len - out.len()).min(self.max_read) as u32;
            let chunk = self.read_at(handle, offset + out.len() as u64, want).await?;
            if chunk.is_empty() {
                break;
            }
            out.extend_from_slice(&chunk);
        }
        Ok(out)
    }

    /// Writes `data` at `offset` in protocol-sized pieces.
    pub async fn write_at(&self, handle: &FileHandle, offset: u64, data: &[u8]) -> SftpResult<()> {
        let writes = data.chunks(self.max_write.max(1)).enumerate().map(|(i, piece)| self.raw.write(handle.0.as_str(), offset + (i * self.max_write) as u64, piece.to_vec()));
        for result in futures::future::join_all(writes).await {
            result.map_err(convert)?;
        }
        Ok(())
    }

    /// Whole-file helpers for small files (editor temp copies, probes).
    pub async fn write_new_file(&self, path: &str, data: &[u8], mode: u32) -> SftpResult<()> {
        let handle = self.open_file(path, "wx", Some(mode)).await?;
        let result = self.write_at(&handle, 0, data).await;
        let closed = self.close_file(&handle).await;
        result.and(closed)
    }
}

impl Drop for Sftp {
    fn drop(&mut self) {
        let _ = self.raw.close_session();
    }
}