feat(cli,fuse,core): copy files and folders on Proton Drive

`pdfs copy PATH... NEW_PARENT` sends a new control request, Copy, which
copies a file with its content or a folder with its subtree on Proton
Drive through proton-drive-rs's copy_node. Nothing is downloaded or
uploaded. A name already taken in the destination gets the next free
" (n)" suffix through get_available_name.

A copy cannot wait in the queue, because Drive copies what it holds.
So it is refused offline, and while the mount still owes Drive part of
the node: a local placeholder, a queued op, unsaved writes, or, for a
folder, any queued op at all. Local-first does not change this.

DriveApi gains copy_node and get_available_name, and the fake drive
copies subtrees, so a FUSE sim test covers the copy, the suffix, the
folder subtree and both refusals. The activity feed logs a new Copy
kind, translated in every catalog.

Needs the proton-drive-rs release that adds copy_node; until then, build
with a [patch.crates-io] pointing at proton-sdk-rs.

crates/pdfs-fuse/src/control.rs

@@ -1063,6 +1063,31 @@ fn handle_control_conn(core: &Core, username: &str, mountpoint: &Path, stream: U
                 (Err(e), _) | (_, Err(e)) => CtlResponse::error(e),
             }
         }
+        Ok(CtlRequest::Copy { path, new_parent }) => {
+            match (
+                rel_to_mount(mountpoint, &path),
+                rel_to_mount(mountpoint, &new_parent),
+            ) {
+                (Ok(rel), Ok(parent_rel)) => match core.copy_to(&rel, &parent_rel) {
+                    Ok(name) => {
+                        let dest = if new_parent.is_empty() {
+                            "My files".to_string()
+                        } else {
+                            new_parent.clone()
+                        };
+                        core.log_activity(ActivityKind::Copy, &name, format!("to {dest}"), true);
+                        CtlResponse::Ok {
+                            message: format!("copied to {name}"),
+                        }
+                    }
+                    Err(e) => {
+                        core.log_activity(ActivityKind::Copy, &path, &e, false);
+                        CtlResponse::error(e)
+                    }
+                },
+                (Err(e), _) | (_, Err(e)) => CtlResponse::error(e),
+            }
+        }
         Ok(CtlRequest::Delete { path }) => match route_node(core, mountpoint, &path) {
             Ok((core, rel)) => match core.delete(&rel) {
                 Ok(name) => {

crates/pdfs-fuse/src/drive.rs

@@ -28,8 +28,8 @@ use proton_drive_rs::proton_sdk::account::Quota;
 use proton_drive_rs::proton_sdk::error::Result;
 use proton_drive_rs::proton_sdk::ids::{DriveEventId, NodeUid};
 use proton_drive_rs::{
-    Device, DeviceType, DriveEvent, DriveEventScopeId, Node, NodeMoveItem, ProtonDriveClient,
-    RevisionReader, Thumbnail,
+    AvailableNameTarget, Device, DeviceType, DriveEvent, DriveEventScopeId, Node, NodeMoveItem,
+    ProtonDriveClient, RevisionReader, Thumbnail,
 };
 
 /// One outcome per node of a batch request; see [`ProtonDriveClient::trash_nodes`].
@@ -138,6 +138,22 @@ pub(crate) trait DriveApi: Send + Sync {
 
     async fn move_node(&self, uid: &NodeUid, new_parent: &NodeUid) -> Result<()>;
 
+    /// See [`ProtonDriveClient::copy_node`].
+    async fn copy_node(
+        &self,
+        uid: &NodeUid,
+        new_parent: &NodeUid,
+        name: Option<&str>,
+    ) -> Result<NodeUid>;
+
+    /// See [`ProtonDriveClient::get_available_name`].
+    async fn get_available_name(
+        &self,
+        parent_uid: &NodeUid,
+        name: &str,
+        target: AvailableNameTarget,
+    ) -> Result<String>;
+
     fn move_nodes_streaming<'a>(
         &'a self,
         items: Vec<NodeMoveItem>,
@@ -448,6 +464,24 @@ impl DriveApi for ProtonDriveClient {
         ProtonDriveClient::move_node(self, uid, new_parent).await
     }
 
+    async fn copy_node(
+        &self,
+        uid: &NodeUid,
+        new_parent: &NodeUid,
+        name: Option<&str>,
+    ) -> Result<NodeUid> {
+        ProtonDriveClient::copy_node(self, uid, new_parent, name).await
+    }
+
+    async fn get_available_name(
+        &self,
+        parent_uid: &NodeUid,
+        name: &str,
+        target: AvailableNameTarget,
+    ) -> Result<String> {
+        ProtonDriveClient::get_available_name(self, parent_uid, name, target).await
+    }
+
     fn move_nodes_streaming<'a>(
         &'a self,
         items: Vec<NodeMoveItem>,

crates/pdfs-fuse/src/lib.rs

@@ -74,8 +74,8 @@ use proton_drive_rs::proton_sdk::api::ResponseCode;
 use proton_drive_rs::proton_sdk::error::ProtonError;
 use proton_drive_rs::proton_sdk::ids::{DriveEventId, LinkId, NodeUid, ShareId, VolumeId};
 use proton_drive_rs::{
-    DriveEvent, DriveEventScopeId, MemberRole, Node, NodeKind, NodeMoveItem, ProtonDriveClient,
-    ProtonPhotosClient, SharedWithMeItem, ThumbnailType,
+    AvailableNameTarget, DriveEvent, DriveEventScopeId, MemberRole, Node, NodeKind, NodeMoveItem,
+    ProtonDriveClient, ProtonPhotosClient, SharedWithMeItem, ThumbnailType,
 };
 
 mod albums;
@@ -5206,6 +5206,88 @@ impl Core {
         Ok(name)
     }
 
+    /// Copy a file or folder into `new_parent_rel` on Proton Drive, which copies
+    /// the content there: nothing is downloaded or uploaded. Both paths are
+    /// mountpoint-relative. A name already taken in the folder gets the next
+    /// free ` (n)` suffix. Returns the copy's name.
+    ///
+    /// Drive copies what it holds, so the copy is refused offline and while the
+    /// mount still owes Drive part of it: a node or folder only queued locally,
+    /// a file with a queued upload or unsaved writes, or — for a folder, whose
+    /// queued changes could sit anywhere below it — any queued op at all.
+    fn copy_to(&self, rel: &Path, new_parent_rel: &Path) -> CoreResult<String> {
+        let (ino, uid) = self.resolve(rel)?;
+        let (pino, new_parent_uid) = self
+            .resolve_path(new_parent_rel)
+            .map_err(|e| self.errno_error(e, "resolve new parent"))?;
+        self.ensure_children(pino)
+            .map_err(|e| self.errno_error(e, "enumerate"))?;
+        self.require_uid_writable(&new_parent_uid)
+            .map_err(|error| self.errno_error(error, "copy access"))?;
+        let (name, is_folder, unsaved) = {
+            let st = self.state();
+            let entry = st
+                .entries
+                .get(&ino)
+                .ok_or_else(|| CoreError::not_found(format!("{} is gone", rel.display())))?;
+            let unsaved = st.active_writes.get(&ino).is_some_and(|w| w.dirty);
+            (entry.node.name.clone(), entry.node.is_folder(), unsaved)
+        };
+        // Unlike a move, a copy cannot wait in the queue: it needs the node as
+        // Drive has it, so it goes now or not at all, local-first or not.
+        if !self.is_online() {
+            return Err(CoreError::offline());
+        }
+        let owed = unsaved
+            || is_local_uid(&uid)
+            || is_local_uid(&new_parent_uid)
+            || self.db.has_any_op(&uid.to_string()).unwrap_or(true)
+            || (is_folder
+                && self
+                    .db
+                    .pending_op_counts()
+                    .map(|counts| counts.uploads + counts.changes > 0)
+                    .unwrap_or(true));
+        if owed {
+            return Err(CoreError::conflict(format!(
+                "{name} has changes that have not reached Proton Drive yet; \
+                 copy it once they have synced"
+            )));
+        }
+        let target = if is_folder {
+            AvailableNameTarget::Folder
+        } else {
+            AvailableNameTarget::File
+        };
+        let copy_name = self
+            .rt
+            .block_on(
+                self.drive
+                    .get_available_name(&new_parent_uid, &name, target),
+            )
+            .map_err(|e| CoreError::from_api(&e, "copy"))?;
+        let new_uid = self
+            .rt
+            .block_on(
+                self.drive
+                    .copy_node(&uid, &new_parent_uid, Some(&copy_name)),
+            )
+            .map_err(|e| CoreError::from_api(&e, "copy"))?;
+        let node = self
+            .fetch_minted_node(&new_uid, &copy_name)
+            .ok_or_else(|| minted_but_unlisted(&copy_name))?;
+        let mut st = self.state();
+        let copy_ino = st.intern(pino, node);
+        if let Some(kids) = st.children.get_mut(&pino)
+            && !kids.contains(&copy_ino)
+        {
+            kids.push(copy_ino);
+        }
+        drop(st);
+        self.flush_access_changes();
+        Ok(copy_name)
+    }
+
     /// Trash a file or folder. `rel` is mountpoint-relative. Forgets the node,
     /// evicts any cached content, and invalidates the parent listing.
     fn delete(&self, rel: &Path) -> CoreResult<String> {

crates/pdfs-fuse/src/sim/daemon.rs

@@ -2214,6 +2214,75 @@ mod tests {
         let _ = std::fs::remove_dir_all(&dir);
     }
 
+    #[test]
+    #[ignore = "mounts FUSE: run with `cargo test -p pdfs-fuse sim:: -- --ignored`"]
+    fn a_pdfs_copy_copies_on_drive_and_waits_for_unsynced_changes() {
+        let drive = FakeDrive::new();
+        let dir = scratch("pdfs-copy");
+        let daemon = Daemon::start(&dir, drive.client(1, Faults::lan())).unwrap();
+
+        let mount = &daemon.mountpoint;
+        std::fs::create_dir_all(mount.join("src/inner")).unwrap();
+        std::fs::create_dir_all(mount.join("dest")).unwrap();
+        std::fs::write(mount.join("src/a.txt"), b"alpha").unwrap();
+        std::fs::write(mount.join("src/inner/b.txt"), b"beta").unwrap();
+        let file = |bytes: &[u8]| Some(Entry::File(Arc::new(bytes.to_vec())));
+        assert!(wait_until(Duration::from_secs(30), || {
+            drive.tree().get("src/inner/b.txt").cloned() == file(b"beta")
+                && daemon.pending().is_ok_and(|ops| ops.is_empty())
+        }));
+
+        let path = |rel: &str| mount.join(rel).to_string_lossy().into_owned();
+        let copy = |from: &str, to: &str| {
+            daemon.request(&Request::Copy {
+                path: path(from),
+                new_parent: path(to),
+            })
+        };
+        let uploads = drive.tree().len();
+        copy("src/a.txt", "dest").unwrap();
+        // A name already taken gets a suffix rather than an error.
+        copy("src/a.txt", "dest").unwrap();
+        copy("src", "dest").unwrap();
+        assert_eq!(std::fs::read(mount.join("dest/a.txt")).unwrap(), b"alpha");
+        assert_eq!(
+            std::fs::read(mount.join("dest/a (1).txt")).unwrap(),
+            b"alpha"
+        );
+        assert_eq!(
+            std::fs::read(mount.join("dest/src/inner/b.txt")).unwrap(),
+            b"beta"
+        );
+        let tree = drive.tree();
+        assert_eq!(tree.get("dest/src/a.txt").cloned(), file(b"alpha"));
+        assert_eq!(
+            tree.len(),
+            uploads + 6,
+            "{:?}",
+            tree.keys().collect::<Vec<_>>()
+        );
+
+        // Drive would copy what it has, not the bytes still queued.
+        let paused = |paused| Request::SetSyncPaused {
+            paused,
+            until: None,
+        };
+        daemon.request(&paused(true)).unwrap();
+        std::fs::write(mount.join("src/a.txt"), b"alpha, edited").unwrap();
+        match copy("src/a.txt", "dest") {
+            Err(e) => assert!(e.to_string().contains("not reached Proton Drive"), "{e}"),
+            Ok(answer) => panic!("the copy went ahead: {answer:?}"),
+        }
+        match copy("src", "dest") {
+            Err(e) => assert!(e.to_string().contains("not reached Proton Drive"), "{e}"),
+            Ok(answer) => panic!("the folder copy went ahead: {answer:?}"),
+        }
+        daemon.request(&paused(false)).unwrap();
+
+        assert!(matches!(daemon.stop(), Ok(MountOutcome::Shutdown)));
+        let _ = std::fs::remove_dir_all(&dir);
+    }
+
     #[test]
     #[ignore = "mounts FUSE: run with `cargo test -p pdfs-fuse sim:: -- --ignored`"]
     fn a_pdfs_rename_after_a_queued_rename_is_not_undone() {

crates/pdfs-fuse/src/sim/fake_drive.rs

@@ -61,8 +61,8 @@ use proton_drive_rs::proton_sdk::ids::{
     DeviceUid, DriveEventId, LinkId, NodeUid, ShareId, VolumeId,
 };
 use proton_drive_rs::{
-    Device as Computer, DeviceType, DriveEvent, DriveEventScopeId, Node, NodeKind, NodeMoveItem,
-    RevisionState, Thumbnail,
+    AvailableNameTarget, Device as Computer, DeviceType, DriveEvent, DriveEventScopeId, Node,
+    NodeKind, NodeMoveItem, RevisionState, Thumbnail,
 };
 use sha1::{Digest as _, Sha1};
 
@@ -185,6 +185,39 @@ impl Server {
         }
     }
 
+    /// Copy `uid` and, for a folder, everything live below it under `parent` as
+    /// `name`, the way Drive's copy endpoint does. Returns the copy's uid.
+    fn copy_subtree(&mut self, uid: &NodeUid, parent: &NodeUid, name: String) -> NodeUid {
+        let source = self
+            .current(uid)
+            .cloned()
+            .expect("the caller checked the source");
+        let copy = self.mint_uid();
+        let now = now_secs();
+        self.put(
+            &copy,
+            Some(Remote {
+                parent: Some(parent.clone()),
+                name,
+                created: now,
+                ..source
+            }),
+        );
+        let children: Vec<(NodeUid, String)> = self
+            .nodes
+            .keys()
+            .filter_map(|child| {
+                let remote = self.current(child)?;
+                (remote.parent.as_ref() == Some(uid) && !remote.trashed)
+                    .then(|| (child.clone(), remote.name.clone()))
+            })
+            .collect();
+        for (child, child_name) in children {
+            self.copy_subtree(&child, &copy, child_name);
+        }
+        copy
+    }
+
     fn mint_uid(&mut self) -> NodeUid {
         self.next_link += 1;
         uid(&format!("L{}", self.next_link))
@@ -1490,6 +1523,61 @@ impl DriveApi for FakeClient {
         .await
     }
 
+    async fn copy_node(
+        &self,
+        uid: &NodeUid,
+        new_parent: &NodeUid,
+        name: Option<&str>,
+    ) -> Result<NodeUid> {
+        self.request(true, |server| {
+            if !server.is_folder(new_parent) {
+                return Err(api(ResponseCode::DoesNotExist, "parent folder not found"));
+            }
+            let source = server
+                .current(uid)
+                .filter(|remote| !remote.trashed)
+                .cloned()
+                .ok_or_else(|| api(ResponseCode::DoesNotExist, "node not found"))?;
+            if server.is_below(new_parent, uid) {
+                return Err(api(
+                    ResponseCode::InvalidRequirements,
+                    "cannot copy a folder into itself",
+                ));
+            }
+            let name = name.map_or(source.name.clone(), str::to_owned);
+            if server.live_child_named(new_parent, &name).is_some() {
+                return Err(api(
+                    ResponseCode::AlreadyExists,
+                    "a file or folder with that name already exists",
+                ));
+            }
+            Ok(server.copy_subtree(uid, new_parent, name))
+        })
+        .await
+    }
+
+    async fn get_available_name(
+        &self,
+        parent_uid: &NodeUid,
+        name: &str,
+        target: AvailableNameTarget,
+    ) -> Result<String> {
+        self.request(false, |server| {
+            let (stem, extension) = match (target, name.rsplit_once('.')) {
+                (AvailableNameTarget::File, Some((stem, ext))) if !stem.is_empty() => {
+                    (stem, format!(".{ext}"))
+                }
+                _ => (name, String::new()),
+            };
+            let free = std::iter::once(name.to_owned())
+                .chain((1..).map(|n| format!("{stem} ({n}){extension}")))
+                .find(|candidate| server.live_child_named(parent_uid, candidate).is_none())
+                .expect("an unbounded sequence has a free name");
+            Ok(free)
+        })
+        .await
+    }
+
     fn move_nodes_streaming<'a>(
         &'a self,
         items: Vec<NodeMoveItem>,