feat: add sync files\folders

src/main.rs

@@ -2,6 +2,7 @@ mod cli;
 mod config;
 mod local;
 mod ssh;
+mod sync;
 
 use std::process::ExitCode;
 
@@ -93,10 +94,10 @@ fn run(cli: &cli::Cli) -> Result<()> {
             session = Some(ssh::connect(&resolved)?);
         }
 
-        if !project.sync.is_empty() {
-            println!("stage 4 not implemented yet: SFTP file transfer");
-        }
         if let Some(session) = session.as_mut() {
+            for rule in &project.sync {
+                sync::upload(session, &rule.source, &rule.target.to_string_lossy())?;
+            }
             for cmd in remote_cmds {
                 session.exec(cmd)?;
             }

src/ssh.rs

@@ -121,7 +121,7 @@ fn query_ssh_config(alias: &str) -> HostConfig {
     }
 }
 
-struct ClientHandler {
+pub(crate) struct ClientHandler {
     host: String,
     port: u16,
 }
@@ -173,7 +173,13 @@ pub struct Session {
     handle: Handle<ClientHandler>,
 }
 
-fn runtime() -> &'static tokio::runtime::Runtime {
+impl Session {
+    pub(crate) fn handle(&mut self) -> &mut Handle<ClientHandler> {
+        &mut self.handle
+    }
+}
+
+pub(crate) fn runtime() -> &'static tokio::runtime::Runtime {
     static RT: OnceLock<tokio::runtime::Runtime> = OnceLock::new();
     RT.get_or_init(|| {
         tokio::runtime::Builder::new_current_thread()

src/sync.rs

@@ -0,0 +1,178 @@
+use std::path::Path;
+
+use anyhow::{Context, Result, bail};
+use russh_sftp::client::SftpSession;
+use tokio::io::AsyncWriteExt;
+use walkdir::WalkDir;
+
+use crate::ssh::{Session, runtime};
+
+pub struct Stats {
+    pub files: u64,
+    pub bytes: u64,
+}
+
+impl std::fmt::Display for Stats {
+    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
+        write!(f, "{} file(s), {}", self.files, human_size(self.bytes))
+    }
+}
+
+fn human_size(bytes: u64) -> String {
+    const KIB: f64 = 1024.0;
+    const MIB: f64 = 1024.0 * 1024.0;
+    let b = bytes as f64;
+    if b >= MIB {
+        format!("{:.1} MiB", b / MIB)
+    } else if b >= KIB {
+        format!("{:.1} KiB", b / KIB)
+    } else {
+        format!("{bytes} B")
+    }
+}
+
+pub fn upload(session: &mut Session, source: &Path, target: &str) -> Result<Stats> {
+    println!("[SFTP] transferring {} -> {} ...", source.display(), target);
+    let stats = runtime().block_on(async {
+        if source.is_dir() {
+            upload_dir(session, source, target).await
+        } else if source.is_file() {
+            upload_file(session, source, target).await
+        } else {
+            bail!("sync source '{}' does not exist", source.display());
+        }
+    })?;
+    println!("[SFTP] ... Done! ({stats})");
+    Ok(stats)
+}
+
+async fn open_sftp(session: &mut Session) -> Result<SftpSession> {
+    let channel = session
+        .handle()
+        .channel_open_session()
+        .await
+        .context("failed to open SFTP channel")?;
+    channel
+        .request_subsystem(true, "sftp")
+        .await
+        .context("remote host has no SFTP subsystem")?;
+    let sftp = SftpSession::new(channel.into_stream())
+        .await
+        .context("SFTP session init failed")?;
+    Ok(sftp)
+}
+
+async fn mkdir_if_missing(sftp: &SftpSession, path: &str) -> Result<()> {
+    match sftp.metadata(path).await {
+        Ok(meta) if meta.is_dir() => return Ok(()),
+        Ok(_) => bail!("remote path exists but is not a directory: {path}"),
+        Err(_) => {}
+    }
+    sftp.create_dir(path)
+        .await
+        .with_context(|| format!("cannot create remote directory {path}"))
+}
+
+fn join_remote(base: &str, rel: &str) -> String {
+    let rel = rel.replace('\\', "/");
+    if rel.is_empty() {
+        base.trim_end_matches('/').to_string()
+    } else {
+        format!("{}/{}", base.trim_end_matches('/'), rel)
+    }
+}
+
+async fn upload_dir(session: &mut Session, source: &Path, target: &str) -> Result<Stats> {
+    let sftp = open_sftp(session).await?;
+    mkdir_if_missing(&sftp, target).await?;
+
+    let mut stats = Stats { files: 0, bytes: 0 };
+    for entry in WalkDir::new(source) {
+        let entry = entry.with_context(|| format!("cannot read local dir {}", source.display()))?;
+        let rel = entry
+            .path()
+            .strip_prefix(source)
+            .expect("walkdir entry is inside source");
+        let remote_path = join_remote(target, &rel.to_string_lossy());
+
+        if entry.file_type().is_dir() {
+            if !rel.as_os_str().is_empty() {
+                mkdir_if_missing(&sftp, &remote_path).await?;
+            }
+        } else if entry.file_type().is_file() {
+            write_remote_file(&sftp, entry.path(), &remote_path).await?;
+            stats.files += 1;
+            stats.bytes += std::fs::metadata(entry.path())
+                .map(|m| m.len())
+                .unwrap_or(0);
+        }
+    }
+    Ok(stats)
+}
+
+async fn upload_file(session: &mut Session, source: &Path, target: &str) -> Result<Stats> {
+    let sftp = open_sftp(session).await?;
+    let dest_is_dir = sftp
+        .metadata(target)
+        .await
+        .map(|meta| meta.is_dir())
+        .unwrap_or(false);
+    let dest = if dest_is_dir {
+        let name = source
+            .file_name()
+            .with_context(|| format!("cannot determine file name of {}", source.display()))?;
+        join_remote(target, &name.to_string_lossy())
+    } else {
+        target.to_string()
+    };
+    write_remote_file(&sftp, source, &dest).await?;
+    let size = std::fs::metadata(source).map(|m| m.len()).unwrap_or(0);
+    Ok(Stats {
+        files: 1,
+        bytes: size,
+    })
+}
+
+async fn write_remote_file(sftp: &SftpSession, local: &Path, remote_path: &str) -> Result<()> {
+    let data = std::fs::read(local).with_context(|| format!("cannot read {}", local.display()))?;
+    let mut remote_file = sftp
+        .create(remote_path)
+        .await
+        .with_context(|| format!("cannot create remote file {remote_path}"))?;
+    remote_file
+        .write_all(&data)
+        .await
+        .with_context(|| format!("upload interrupted: {remote_path}"))?;
+    remote_file
+        .shutdown()
+        .await
+        .with_context(|| format!("cannot finish writing {remote_path}"))?;
+    Ok(())
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    #[test]
+    fn join_remote_handles_slashes_and_edges() {
+        assert_eq!(join_remote("/var/www/pages", "dist"), "/var/www/pages/dist");
+        assert_eq!(
+            join_remote("/var/www/pages/", "a/b.js"),
+            "/var/www/pages/a/b.js"
+        );
+        assert_eq!(join_remote("/var/www/pages", ""), "/var/www/pages");
+        assert_eq!(
+            join_remote("/var/www/pages", "sub\\file.css"),
+            "/var/www/pages/sub/file.css"
+        );
+    }
+
+    #[test]
+    fn human_size_formats() {
+        assert_eq!(human_size(0), "0 B");
+        assert_eq!(human_size(512), "512 B");
+        assert_eq!(human_size(2048), "2.0 KiB");
+        assert_eq!(human_size(3 * 1024 * 1024), "3.0 MiB");
+    }
+}