remote_fs/

ftp.rs

use std::io::Cursor;
use std::path::{Path, PathBuf};
use std::time::SystemTime;

use async_ftp::FtpStream;
use async_trait::async_trait;
use tokio::io::AsyncReadExt;

use crate::error::{Error, Result};
use crate::traits::RemoteFileSystem;
use crate::types::{FileEntry, FileMeta, FtpConfig};

/// FTP remote file system client (backed by `async-ftp`).
pub struct FtpClient {
    config: FtpConfig,
    stream: Option<FtpStream>,
}

impl FtpClient {
    pub fn new(config: FtpConfig) -> Self {
        Self {
            config,
            stream: None,
        }
    }

    fn stream(&mut self) -> Result<&mut FtpStream> {
        self.stream.as_mut().ok_or(Error::NotConnected)
    }

    fn addr(&self) -> String {
        format!("{}:{}", self.config.host, self.config.port)
    }
}

/// Parse a single FTP LIST line (Unix or DOS-ish).
fn parse_list_line(base: &str, line: &str) -> Option<FileEntry> {
    let line = line.trim();
    if line.is_empty() {
        return None;
    }

    // Unix: drwxr-xr-x 2 user group 4096 Jan 1 12:00 name
    if line.starts_with('d')
        || line.starts_with('-')
        || line.starts_with('l')
        || line.starts_with('b')
        || line.starts_with('c')
    {
        let parts: Vec<&str> = line.split_whitespace().collect();
        if parts.len() < 9 {
            return None;
        }
        let name = parts[8..].join(" ");
        if name == "." || name == ".." {
            return None;
        }
        let is_dir = parts[0].starts_with('d');
        let is_symlink = parts[0].starts_with('l');
        let size = parts[4].parse().unwrap_or(0);
        let path = join_ftp(base, &name);
        return Some(FileEntry {
            path,
            name,
            is_dir,
            is_file: !is_dir && !is_symlink,
            is_symlink,
            size,
            modified: None,
        });
    }

    // DOS: 01-01-26  12:00AM       <DIR>          name
    //      01-01-26  12:00AM                  123 name
    let parts: Vec<&str> = line.split_whitespace().collect();
    if parts.len() >= 4 {
        let is_dir = parts[2].eq_ignore_ascii_case("<DIR>");
        let (size, name_idx) = if is_dir {
            (0u64, 3)
        } else {
            (parts[2].parse().unwrap_or(0), 3)
        };
        if name_idx >= parts.len() {
            return None;
        }
        let name = parts[name_idx..].join(" ");
        if name == "." || name == ".." {
            return None;
        }
        let path = join_ftp(base, &name);
        return Some(FileEntry {
            path,
            name,
            is_dir,
            is_file: !is_dir,
            is_symlink: false,
            size,
            modified: None,
        });
    }

    None
}

fn join_ftp(base: &str, name: &str) -> PathBuf {
    if base.is_empty() || base == "/" {
        PathBuf::from(format!("/{name}"))
    } else {
        PathBuf::from(base.trim_end_matches('/')).join(name)
    }
}

#[async_trait]
impl RemoteFileSystem for FtpClient {
    async fn connect(&mut self) -> Result<()> {
        let mut stream = FtpStream::connect(self.addr().as_str()).await?;
        stream
            .login(&self.config.username, &self.config.password)
            .await?;
        let _ = self.config.passive; // async-ftp uses passive by default
        self.stream = Some(stream);
        Ok(())
    }

    async fn disconnect(&mut self) -> Result<()> {
        if let Some(mut stream) = self.stream.take() {
            let _ = stream.quit().await;
        }
        Ok(())
    }

    fn is_connected(&self) -> bool {
        self.stream.is_some()
    }

    async fn list(&mut self, path: &str) -> Result<Vec<FileEntry>> {
        let stream = self.stream()?;
        let pathname = if path.is_empty() || path == "/" {
            None
        } else {
            Some(path)
        };
        let lines = stream.list(pathname).await?;
        Ok(lines
            .iter()
            .filter_map(|line| parse_list_line(path, line))
            .collect())
    }

    async fn metadata(&mut self, path: &str) -> Result<FileMeta> {
        let stream = self.stream()?;
        let size = stream.size(path).await?;
        let modified = stream
            .mdtm(path)
            .await?
            .map(|dt| SystemTime::from(dt));

        if let Some(size) = size {
            return Ok(FileMeta {
                path: PathBuf::from(path),
                is_dir: false,
                is_file: true,
                is_symlink: false,
                size: size as u64,
                modified,
            });
        }

        // Might be a directory — try listing it.
        match stream.list(Some(path)).await {
            Ok(_) => Ok(FileMeta {
                path: PathBuf::from(path),
                is_dir: true,
                is_file: false,
                is_symlink: false,
                size: 0,
                modified,
            }),
            Err(err) => Err(err.into()),
        }
    }

    async fn exists(&mut self, path: &str) -> Result<bool> {
        match self.metadata(path).await {
            Ok(_) => Ok(true),
            Err(_) => Ok(false),
        }
    }

    async fn mkdir(&mut self, path: &str) -> Result<()> {
        self.stream()?.mkdir(path).await?;
        Ok(())
    }

    async fn mkdir_all(&mut self, path: &str) -> Result<()> {
        let path = path.trim_matches('/');
        if path.is_empty() {
            return Ok(());
        }
        let mut current = String::new();
        for part in path.split('/').filter(|p| !p.is_empty()) {
            current = if current.is_empty() {
                format!("/{part}")
            } else {
                format!("{current}/{part}")
            };
            if !self.exists(&current).await? {
                self.mkdir(&current).await?;
            }
        }
        Ok(())
    }

    async fn rmdir(&mut self, path: &str) -> Result<()> {
        self.stream()?.rmdir(path).await?;
        Ok(())
    }

    async fn remove_file(&mut self, path: &str) -> Result<()> {
        self.stream()?.rm(path).await?;
        Ok(())
    }

    async fn remove(&mut self, path: &str) -> Result<()> {
        let meta = self.metadata(path).await?;
        if meta.is_dir {
            let children = self.list(path).await?;
            for child in children {
                let child_path = child.path.to_string_lossy().to_string();
                Box::pin(self.remove(&child_path)).await?;
            }
            self.rmdir(path).await
        } else {
            self.remove_file(path).await
        }
    }

    async fn rename(&mut self, from: &str, to: &str) -> Result<()> {
        self.stream()?.rename(from, to).await?;
        Ok(())
    }

    async fn read(&mut self, path: &str) -> Result<Vec<u8>> {
        let cursor = self.stream()?.simple_retr(path).await?;
        Ok(cursor.into_inner())
    }

    async fn write(&mut self, path: &str, data: &[u8]) -> Result<()> {
        let mut reader = Cursor::new(data.to_vec());
        self.stream()?.put(path, &mut reader).await?;
        Ok(())
    }

    async fn download(&mut self, remote: &str, local: &Path) -> Result<()> {
        let data = self.read(remote).await?;
        tokio::fs::write(local, data).await?;
        Ok(())
    }

    async fn upload(&mut self, local: &Path, remote: &str) -> Result<()> {
        let data = tokio::fs::read(local).await?;
        self.write(remote, &data).await
    }
}

#[allow(dead_code)]
async fn read_data_stream<R: AsyncReadExt + Unpin>(mut reader: R) -> Result<Vec<u8>> {
    let mut buf = Vec::new();
    reader.read_to_end(&mut buf).await?;
    Ok(buf)
}