Merge pull request #1 from narrrl/feature/sharing-endpoints

Feature/sharing + devices
This commit is contained in:
nvrl authored and GitHub committed 2026-07-17 11:12:13 +02:00
commit fb39594453
14 files changed
+7774 -199

No files matched your search

Generated
+226 -28
View File
@@ -19,6 +19,16 @@ dependencies = [
"generic-array",
]
[[package]]
name = "aead"
version = "0.6.1"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "1973cfbc1a2daf9cf550e74e1f088c28e7f7d8c1e1418fb6c9dc5184b7e84c99"
dependencies = [
"crypto-common 0.2.2",
"inout 0.2.2",
]
[[package]]
name = "aes"
version = "0.8.4"
@@ -30,17 +40,42 @@ dependencies = [
"cpufeatures 0.2.17",
]
[[package]]
name = "aes"
version = "0.9.1"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "f1fc76eaeac4c9164506c466d4ffdd8ec9d0c5bf57ee97177c4d8eceb3a0e138"
dependencies = [
"cipher 0.5.2",
"cpubits",
"cpufeatures 0.3.0",
]
[[package]]
name = "aes-gcm"
version = "0.10.3"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "831010a0f742e1209b3bcea8fab6a8e149051ba6099432c8cb2cc117dec3ead1"
dependencies = [
"aead",
"aes",
"aead 0.5.2",
"aes 0.8.4",
"cipher 0.4.4",
"ctr",
"ghash",
"ctr 0.9.2",
"ghash 0.5.1",
"subtle",
]
[[package]]
name = "aes-gcm"
version = "0.11.0"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "fdf011db2e21ce0d575593d749db5554b47fed37aff429e4dc50bc91ac93a028"
dependencies = [
"aead 0.6.1",
"aes 0.9.1",
"cipher 0.5.2",
"ctr 0.10.1",
"ghash 0.6.0",
"subtle",
]
@@ -50,7 +85,7 @@ version = "0.2.1"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "69fa2b352dcefb5f7f3a5fb840e02665d311d878955380515e4fd50095dd3d8c"
dependencies = [
"aes",
"aes 0.8.4",
]
[[package]]
@@ -740,6 +775,17 @@ version = "0.2.1"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "613afe47fcd5fac7ccf1db93babcb082c5994d996f20b8b159f2ad1658eb5724"
[[package]]
name = "chacha20"
version = "0.10.1"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "d524456ba66e72eb8b115ff89e01e497f8e6d11d78b70b1aa13c0fbd97540a81"
dependencies = [
"cfg-if",
"cpufeatures 0.3.0",
"rand_core 0.10.1",
]
[[package]]
name = "chrono"
version = "0.4.45"
@@ -769,6 +815,7 @@ version = "0.5.2"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "e8cf2a2c93cd704877c0858356ed03480ff301ee950b43f1cbe4573b088bfa6c"
dependencies = [
"block-buffer 0.12.1",
"crypto-common 0.2.2",
"inout 0.2.2",
]
@@ -922,6 +969,12 @@ version = "0.8.7"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "773648b94d0e5d620f64f280777445740e61fe701025087ec8b57f45c791888b"
[[package]]
name = "cpubits"
version = "0.1.1"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "15b85f9c39137c3a891689859392b1bd49812121d0d61c9caf00d46ed5ce06ae"
[[package]]
name = "cpufeatures"
version = "0.2.17"
@@ -1015,7 +1068,9 @@ version = "0.2.2"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "ce6e4c961d6cd6c9a86db418387425e8bdeaf05b3c8bc1411e6dca4c252f1453"
dependencies = [
"getrandom 0.4.3",
"hybrid-array",
"rand_core 0.10.1",
]
[[package]]
@@ -1027,6 +1082,15 @@ dependencies = [
"cipher 0.4.4",
]
[[package]]
name = "ctr"
version = "0.10.1"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "baaca1c4b237092596f64d571e9db6ce4109c4ef9742e27590f1709594461f21"
dependencies = [
"cipher 0.5.2",
]
[[package]]
name = "ctutils"
version = "0.4.2"
@@ -1152,7 +1216,7 @@ version = "4.1.0"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "708b509edf7889e53d7efb0ffadd994cc6c2345ccb62f55cfd6b0682165e4fa6"
dependencies = [
"aes",
"aes 0.8.4",
"block-padding",
"cbc",
"dbus",
@@ -1331,10 +1395,10 @@ version = "0.5.0"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "9954fabd903b82b9d7a68f65f97dc96dd9ad368e40ccc907a7c19d53e6bfac28"
dependencies = [
"aead",
"aead 0.5.2",
"cipher 0.4.4",
"cmac",
"ctr",
"ctr 0.9.2",
"subtle",
]
@@ -1607,6 +1671,15 @@ version = "1.3.0"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "42703706b716c37f96a77aea830392ad231f44c9e9a67872fa5548707e11b11c"
[[package]]
name = "fsevent-sys"
version = "4.1.0"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "76ee7a02da4d231650c7cea31349b889be2f45ddb3ef3032d2ec8185f6313fd2"
dependencies = [
"libc",
]
[[package]]
name = "funty"
version = "2.0.0"
@@ -1838,6 +1911,7 @@ dependencies = [
"cfg-if",
"libc",
"r-efi 6.0.0",
"rand_core 0.10.1",
]
[[package]]
@@ -1847,7 +1921,16 @@ source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "f0d8a4362ccb29cb0b265253fb0a2728f592895ee6854fd9bc13f2ffda266ff1"
dependencies = [
"opaque-debug",
"polyval",
"polyval 0.6.2",
]
[[package]]
name = "ghash"
version = "0.6.0"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "2eecf2d5dc9b66b732b97707a0210906b1d30523eb773193ab777c0c84b3e8d5"
dependencies = [
"polyval 0.7.2",
]
[[package]]
@@ -2485,6 +2568,26 @@ dependencies = [
"hashbrown 0.17.1",
]
[[package]]
name = "inotify"
version = "0.11.4"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "153be1941a183ec9ccd095ddbe17a8b8d435ef6c76e9e02451b933c3999af2c8"
dependencies = [
"bitflags 2.13.0",
"inotify-sys",
"libc",
]
[[package]]
name = "inotify-sys"
version = "0.1.8"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "c033f80b2c113cdf91ab7a33faa9cbc014726dcad99880c8609af2a370edf37d"
dependencies = [
"libc",
]
[[package]]
name = "inout"
version = "0.1.4"
@@ -2651,6 +2754,26 @@ dependencies = [
"zeroize",
]
[[package]]
name = "kqueue"
version = "1.2.0"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "273c0752728918e0ac4976f2b275b6fefb9ecd400585dec929419f3844cd87b5"
dependencies = [
"kqueue-sys",
"libc",
]
[[package]]
name = "kqueue-sys"
version = "1.1.2"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "07293a4e297ac234359b510362495713f75ea345d5307140414f20c69ffeb087"
dependencies = [
"bitflags 2.13.0",
"libc",
]
[[package]]
name = "ksni"
version = "0.2.2"
@@ -2885,6 +3008,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "02bd0af71c67b473010cbbc60715ee815645a4dc942899111f494b4b737d6fda"
dependencies = [
"libc",
"log",
"wasi",
"windows-sys 0.61.2",
]
@@ -2961,6 +3085,33 @@ version = "0.3.0"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "0676bb32a98c1a483ce53e500a81ad9c3d5b3f7c920c28c24e9cb0980d0b5bc8"
[[package]]
name = "notify"
version = "8.2.0"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "4d3d07927151ff8575b7087f245456e549fea62edf0ec4e565a5ee50c8402bc3"
dependencies = [
"bitflags 2.13.0",
"fsevent-sys",
"inotify",
"kqueue",
"libc",
"log",
"mio",
"notify-types",
"walkdir",
"windows-sys 0.60.2",
]
[[package]]
name = "notify-types"
version = "2.1.0"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "42b8cfee0e339a0337359f3c88165702ac6e600dc01c0cc9579a92d62b08477a"
dependencies = [
"bitflags 2.13.0",
]
[[package]]
name = "nu-ansi-term"
version = "0.50.3"
@@ -2976,7 +3127,7 @@ version = "0.4.3"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "35bd024e8b2ff75562e5f34e7f4905839deb4b22955ef5e73d2fea1b9813cb23"
dependencies = [
"num-bigint",
"num-bigint 0.4.6",
"num-complex",
"num-integer",
"num-iter",
@@ -2994,6 +3145,16 @@ dependencies = [
"num-traits",
]
[[package]]
name = "num-bigint"
version = "0.5.1"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "93e7820bc0a80a0238e650327316f929ba18d5be054b647490a3a6a339f3e7c0"
dependencies = [
"num-integer",
"num-traits",
]
[[package]]
name = "num-bigint-dig"
version = "0.8.6"
@@ -3058,7 +3219,7 @@ version = "0.4.2"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "f83d14da390562dca69fc84082e73e548e1ad308d24accdedd2720017cb37824"
dependencies = [
"num-bigint",
"num-bigint 0.4.6",
"num-integer",
"num-traits",
]
@@ -3101,9 +3262,9 @@ version = "0.1.0"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "c196e0276c471c843dd5777e7543a36a298a4be942a2a688d8111cd43390dedb"
dependencies = [
"aead",
"aead 0.5.2",
"cipher 0.4.4",
"ctr",
"ctr 0.9.2",
"subtle",
]
@@ -3311,6 +3472,7 @@ dependencies = [
"fuser",
"image",
"libc",
"notify",
"pdfs-core",
"proton-drive-rs",
"serde_json",
@@ -3356,9 +3518,9 @@ version = "0.20.0"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "1cfa4743b28656065ff4c0ba09e46b357a65e8c00fc2341e89084b82f87cbdf1"
dependencies = [
"aead",
"aes",
"aes-gcm",
"aead 0.5.2",
"aes 0.8.4",
"aes-gcm 0.10.3",
"aes-kw",
"argon2",
"base64",
@@ -3500,7 +3662,18 @@ dependencies = [
"cfg-if",
"cpufeatures 0.2.17",
"opaque-debug",
"universal-hash",
"universal-hash 0.5.1",
]
[[package]]
name = "polyval"
version = "0.7.2"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "b20f20e954175de5f463f67781b35583397d916b1d148738923711b2ad16bee8"
dependencies = [
"cpubits",
"cpufeatures 0.3.0",
"universal-hash 0.6.1",
]
[[package]]
@@ -3569,14 +3742,13 @@ dependencies = [
[[package]]
name = "proton-drive-rs"
version = "0.1.5"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "d6862368afe7da79766f4d86845ba0ad7cdfc619ef6d962aeb45ac444612dc42"
version = "0.1.7"
dependencies = [
"async-trait",
"base64",
"bytes",
"futures",
"getrandom 0.4.3",
"hex",
"hmac 0.13.0",
"pgp",
@@ -3594,24 +3766,23 @@ dependencies = [
[[package]]
name = "proton-sdk"
version = "0.1.5"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "7fa00fcf9c1ad4866ab8bc145295efe12e518232ab7cebb6cf2bf0caa4f7e3da"
version = "0.1.7"
dependencies = [
"aes-gcm",
"aes-gcm 0.11.0",
"async-trait",
"base64",
"bcrypt",
"bytes",
"chrono",
"futures",
"getrandom 0.2.17",
"getrandom 0.4.3",
"hex",
"hkdf 0.13.0",
"hmac 0.13.0",
"num-bigint",
"num-bigint 0.5.1",
"num-traits",
"pgp",
"rand 0.10.2",
"rand 0.8.6",
"reqwest",
"serde",
@@ -3711,7 +3882,7 @@ dependencies = [
"once_cell",
"socket2",
"tracing",
"windows-sys 0.60.2",
"windows-sys 0.59.0",
]
[[package]]
@@ -3762,6 +3933,17 @@ dependencies = [
"rand_core 0.9.5",
]
[[package]]
name = "rand"
version = "0.10.2"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "c7f5fa3a058cd35567ef9bfa5e75732bee0f9e4c55fa90477bef2dfcdbc4be80"
dependencies = [
"chacha20",
"getrandom 0.4.3",
"rand_core 0.10.1",
]
[[package]]
name = "rand_chacha"
version = "0.3.1"
@@ -3800,6 +3982,12 @@ dependencies = [
"getrandom 0.3.4",
]
[[package]]
name = "rand_core"
version = "0.10.1"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "63b8176103e19a2643978565ca18b50549f6101881c443590420e4dc998a3c69"
[[package]]
name = "rav1e"
version = "0.8.1"
@@ -4238,7 +4426,7 @@ version = "4.0.0"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "e4d35ad99a181be0a60ffcbe85d680d98f87bdc4d7644ade319b87076b9dbfd4"
dependencies = [
"aes",
"aes 0.8.4",
"cbc",
"futures-util",
"generic-array",
@@ -5012,6 +5200,16 @@ dependencies = [
"subtle",
]
[[package]]
name = "universal-hash"
version = "0.6.1"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "f4987bdc12753382e0bec4a65c50738ffaabc998b9cdd1f952fb5f39b0048a96"
dependencies = [
"crypto-common 0.2.2",
"ctutils",
]
[[package]]
name = "untrusted"
version = "0.9.0"
+8 -2
View File
@@ -10,8 +10,8 @@ authors = ["Nils Pukropp <contact@narl.io>"]
rust-version = "1.96"
[workspace.dependencies]
proton-sdk = "0.1.5"
proton-drive-rs = "0.1.5"
proton-sdk = "0.1.6"
proton-drive-rs = "0.1.6"
# Internal crates.
pdfs-core = { path = "crates/pdfs-core" }
@@ -37,3 +37,9 @@ clap = { version = "4", features = ["derive"] }
rpassword = "7"
fuser = "0.17"
libc = "0.2"
# Development override: build against the local SDK checkout until 0.1.6 is
# published to crates.io. Remove this block to consume the released crate.
[patch.crates-io]
proton-sdk = { path = "../proton-sdk-rs/crates/proton-sdk" }
proton-drive-rs = { path = "../proton-sdk-rs/crates/proton-drive-rs" }
+619 -18
View File
@@ -10,7 +10,9 @@ use clap::{Parser, Subcommand};
use pdfs_core::auth;
use pdfs_core::cache::ContentCache;
use pdfs_core::config::AppDirs;
use pdfs_core::control::{Request as CtlRequest, Response as CtlResponse};
use pdfs_core::control::{
Request as CtlRequest, Response as CtlResponse, ShareEntryKind, SyncPhase,
};
use pdfs_core::db::Db;
#[derive(Parser)]
@@ -113,11 +115,15 @@ enum Command {
/// New folder name (a single path component).
name: String,
},
/// Upload a local file into a Drive folder via the running daemon.
/// Upload local files and/or folders into a Drive folder via the running
/// daemon. Folders are uploaded recursively. The daemon uploads in the
/// background; watch progress with `pdfs transfers`.
Upload {
/// Local file to upload.
file: PathBuf,
/// Destination folder path, inside the mountpoint or relative to it.
/// Local files and/or folders to upload.
#[arg(required = true)]
sources: Vec<PathBuf>,
/// Destination folder, inside the mountpoint or relative to it.
#[arg(short = 't', long = "to", default_value = ".")]
parent: PathBuf,
},
/// Show the daemon's in-flight transfers (active uploads/downloads).
@@ -139,6 +145,200 @@ enum Command {
},
/// Permanently delete everything in the trash. This cannot be undone.
EmptyTrash,
/// Manage the account's registered devices.
Devices {
#[command(subcommand)]
action: DeviceCmd,
},
/// Sync local folders to this machine's Proton Drive device.
Sync {
#[command(subcommand)]
action: SyncCmd,
},
/// Share a file or folder with Proton and/or external email addresses.
Share {
/// File/folder path, inside the mountpoint or relative to it.
path: PathBuf,
/// One or more invitee emails (Proton or external, auto-detected).
#[arg(required = true)]
emails: Vec<String>,
/// Role to grant: viewer, editor or admin.
#[arg(long, default_value = "viewer")]
role: String,
/// Optional message included in the invitation email.
#[arg(long)]
message: Option<String>,
},
/// List a node's members, pending invitations and public link.
Members {
/// File/folder path, inside the mountpoint or relative to it.
path: PathBuf,
},
/// Change the role of a member or pending invitation on a node's share.
ShareRole {
/// File/folder path, inside the mountpoint or relative to it.
path: PathBuf,
/// Entry id (membership or invitation id), from `pdfs members`.
id: String,
/// Entry kind: member, proton or external.
kind: ShareKindArg,
/// New role: viewer, editor or admin.
role: String,
},
/// Remove a member or pending invitation from a node's share.
Unshare {
/// File/folder path, inside the mountpoint or relative to it.
path: PathBuf,
/// Entry id (membership or invitation id), from `pdfs members`.
id: String,
/// Entry kind: member, proton or external.
kind: ShareKindArg,
},
/// Manage a node's public link.
PublicLink {
#[command(subcommand)]
action: PublicLinkCmd,
},
/// List the items I have shared with others (members, invites or a link).
Shared,
/// List nodes shared with me that I have accepted.
SharedWithMe,
/// Leave a shared node by its uid (from `pdfs shared-with-me`).
Leave {
/// Node uid in `volume~link` form.
uid: String,
},
/// Manage invitations addressed to me.
Invitations {
#[command(subcommand)]
action: InvitationCmd,
},
/// Manage saved public-link bookmarks.
Bookmarks {
#[command(subcommand)]
action: BookmarkCmd,
},
/// Show the daemon's recent activity, newest first.
Activity {
/// Maximum entries to show.
#[arg(short, long, default_value_t = 50)]
limit: usize,
},
}
/// Which collection a share entry lives in, for `share-role` / `unshare`.
#[derive(Clone, Copy, clap::ValueEnum)]
enum ShareKindArg {
/// An accepted member.
Member,
/// A pending invitation to a Proton user.
Proton,
/// A pending invitation to a non-Proton email.
External,
}
impl ShareKindArg {
fn to_kind(self) -> ShareEntryKind {
match self {
ShareKindArg::Member => ShareEntryKind::Member,
ShareKindArg::Proton => ShareEntryKind::ProtonInvite,
ShareKindArg::External => ShareEntryKind::ExternalInvite,
}
}
}
#[derive(Subcommand)]
enum DeviceCmd {
/// List registered devices.
List,
/// Rename a device by uid.
Rename { uid: String, name: String },
/// Delete (deregister) a device by uid.
Rm { uid: String },
}
#[derive(Subcommand)]
enum SyncCmd {
/// Add a local folder to this machine's device and upload its contents.
Add {
/// Local directory to sync.
path: PathBuf,
},
/// List this machine's synced folders.
List,
/// Remove a synced folder by id.
Rm {
/// Synced-folder id (from `sync list`).
id: i64,
/// Also delete the folder's copy in Proton Drive.
#[arg(long)]
delete_remote: bool,
},
/// Force a sync pass now (all folders, or one by id).
Now {
/// Optional synced-folder id; omit to reconcile every folder.
id: Option<i64>,
},
/// Switch a folder between full-copy sync and on-demand (FUSE, no local storage).
Mode {
/// Synced-folder id (from `sync list`).
id: i64,
/// `mirror` (full local copy) or `ondemand` (FUSE mount, reclaims disk).
#[arg(value_parser = ["mirror", "ondemand"])]
mode: String,
},
}
#[derive(Subcommand)]
enum PublicLinkCmd {
/// Create a public link on a node.
Create {
/// File/folder path, inside the mountpoint or relative to it.
path: PathBuf,
/// Role: viewer or editor.
#[arg(long, default_value = "viewer")]
role: String,
/// Optional custom password protecting the link.
#[arg(long)]
password: Option<String>,
/// Optional expiry, Unix epoch seconds.
#[arg(long)]
expires: Option<i64>,
},
/// Remove a node's public link.
Remove {
/// File/folder path, inside the mountpoint or relative to it.
path: PathBuf,
/// Public-link id, from `pdfs members`.
id: String,
},
}
#[derive(Subcommand)]
enum InvitationCmd {
/// List invitations addressed to me.
List,
/// Accept an invitation by id.
Accept { id: String },
/// Reject an invitation by id.
Reject { id: String },
}
#[derive(Subcommand)]
enum BookmarkCmd {
/// List saved public-link bookmarks.
List,
/// Save a public link (optionally password-protected) as a bookmark.
Add {
/// Public link URL, including the `#password` fragment.
url: String,
/// Optional custom password for the link.
#[arg(long)]
password: Option<String>,
},
/// Remove a saved bookmark by its token.
Rm { token: String },
}
fn main() -> Result<()> {
@@ -170,15 +370,395 @@ fn main() -> Result<()> {
Command::Move { path, new_parent } => cmd_move(path, new_parent),
Command::Rm { path } => cmd_rm(path),
Command::Mkdir { parent, name } => cmd_mkdir(parent, name),
Command::Upload { file, parent } => cmd_upload(file, parent),
Command::Upload { sources, parent } => cmd_upload(sources, parent),
Command::Transfers => cmd_transfers(),
Command::Trash => cmd_trash(),
Command::Restore { uids } => cmd_restore(uids),
Command::DeleteForever { uids } => cmd_delete_forever(uids),
Command::EmptyTrash => cmd_empty_trash(),
Command::Devices { action } => cmd_devices(action),
Command::Sync { action } => cmd_sync(action),
Command::Share {
path,
emails,
role,
message,
} => cmd_share(path, emails, role, message),
Command::Members { path } => cmd_members(path),
Command::ShareRole {
path,
id,
kind,
role,
} => cmd_share_role(path, id, kind, role),
Command::Unshare { path, id, kind } => cmd_unshare(path, id, kind),
Command::PublicLink { action } => cmd_public_link(action),
Command::Shared => cmd_shared(),
Command::SharedWithMe => cmd_shared_with_me(),
Command::Leave { uid } => cmd_leave(uid),
Command::Invitations { action } => cmd_invitations(action),
Command::Bookmarks { action } => cmd_bookmarks(action),
Command::Activity { limit } => cmd_activity(limit),
}
}
fn cmd_devices(action: DeviceCmd) -> Result<()> {
match action {
DeviceCmd::List => match control_request(CtlRequest::ListDevices)? {
CtlResponse::Devices { items } if items.is_empty() => println!("No devices."),
CtlResponse::Devices { items } => {
for d in items {
let sync = d
.last_sync
.map(|t| t.to_string())
.unwrap_or_else(|| "never".to_string());
println!(
"{} {} (synced: {sync}) [{}]",
d.device_type, d.name, d.uid
);
}
}
CtlResponse::Error { message } => bail!("{message}"),
other => bail!("unexpected response: {other:?}"),
},
DeviceCmd::Rename { uid, name } => {
ok_or_bail(control_request(CtlRequest::RenameDevice { uid, name })?)?
}
DeviceCmd::Rm { uid } => ok_or_bail(control_request(CtlRequest::DeleteDevice { uid })?)?,
}
Ok(())
}
fn cmd_sync(action: SyncCmd) -> Result<()> {
match action {
SyncCmd::Add { path } => {
let abs = std::fs::canonicalize(&path)
.with_context(|| format!("resolve {}", path.display()))?;
let local_path = abs
.to_str()
.ok_or_else(|| anyhow!("path is not valid UTF-8"))?
.to_string();
ok_or_bail(control_request(CtlRequest::AddSyncFolder { local_path })?)?
}
SyncCmd::List => match control_request(CtlRequest::ListSyncFolders)? {
CtlResponse::SyncFolders { items } if items.is_empty() => {
println!("No synced folders.")
}
CtlResponse::SyncFolders { items } => {
for f in items {
let sync = if f.last_sync == 0 {
"never".to_string()
} else {
f.last_sync.to_string()
};
println!(
"[{}] {} ({}, {}, synced: {sync})",
f.id, f.local_path, f.mode, f.state
);
// A pass in flight says what it is doing, indented under it.
if let Some(p) = &f.progress {
match p.phase {
SyncPhase::Scanning => println!(" scanning…"),
SyncPhase::Applying => println!(
" {} of {} {}",
p.done + 1,
p.total.max(p.done + 1),
p.current
),
}
}
}
}
CtlResponse::Error { message } => bail!("{message}"),
other => bail!("unexpected response: {other:?}"),
},
SyncCmd::Rm { id, delete_remote } => {
ok_or_bail(control_request(CtlRequest::RemoveSyncFolder {
id,
delete_remote,
})?)?
}
SyncCmd::Now { id } => ok_or_bail(control_request(CtlRequest::SyncNow { id })?)?,
SyncCmd::Mode { id, mode } => {
ok_or_bail(control_request(CtlRequest::SetSyncFolderMode { id, mode })?)?
}
}
Ok(())
}
fn cmd_share(
path: PathBuf,
emails: Vec<String>,
role: String,
message: Option<String>,
) -> Result<()> {
ok_or_bail(control_request(CtlRequest::ShareNode {
path: path_arg(&path)?,
emails,
role,
message,
})?)
}
fn cmd_members(path: PathBuf) -> Result<()> {
match control_request(CtlRequest::ListShare {
path: path_arg(&path)?,
})? {
CtlResponse::Share { entries, link } => {
if entries.is_empty() {
println!("No members or pending invitations.");
}
for e in entries {
let kind = match e.kind {
ShareEntryKind::Member => "member",
ShareEntryKind::ProtonInvite => "invited (proton)",
ShareEntryKind::ExternalInvite => "invited (external)",
};
println!("{:<18} {:<8} {} [{}]", kind, e.role, e.email, e.id);
}
match link {
Some(l) => {
let url = l.url.as_deref().unwrap_or("(url hidden)");
let pw = if l.has_password { " +password" } else { "" };
println!("public link ({}{pw}): {url} [{}]", l.role, l.id);
}
None => println!("public link: none"),
}
}
CtlResponse::Error { message } => bail!("{message}"),
other => bail!("unexpected response: {other:?}"),
}
Ok(())
}
fn cmd_share_role(path: PathBuf, id: String, kind: ShareKindArg, role: String) -> Result<()> {
ok_or_bail(control_request(CtlRequest::UpdateShareRole {
path: path_arg(&path)?,
id,
kind: kind.to_kind(),
role,
})?)
}
fn cmd_unshare(path: PathBuf, id: String, kind: ShareKindArg) -> Result<()> {
ok_or_bail(control_request(CtlRequest::RemoveShareEntry {
path: path_arg(&path)?,
id,
kind: kind.to_kind(),
})?)
}
fn cmd_public_link(action: PublicLinkCmd) -> Result<()> {
match action {
PublicLinkCmd::Create {
path,
role,
password,
expires,
} => match control_request(CtlRequest::CreatePublicLink {
path: path_arg(&path)?,
role,
password,
expires,
})? {
CtlResponse::PublicLink { link } => {
println!("{}", link.url.as_deref().unwrap_or("(no url returned)"));
}
CtlResponse::Error { message } => bail!("{message}"),
other => bail!("unexpected response: {other:?}"),
},
PublicLinkCmd::Remove { path, id } => {
ok_or_bail(control_request(CtlRequest::RemovePublicLink {
path: path_arg(&path)?,
id,
})?)?
}
}
Ok(())
}
fn cmd_shared() -> Result<()> {
match control_request(CtlRequest::ListSharedByMe)? {
CtlResponse::SharedByMe { items } if items.is_empty() => {
println!("You haven't shared anything.")
}
CtlResponse::SharedByMe { items } => {
for it in items {
let kind = if it.is_dir { "d" } else { "-" };
let mut tags = Vec::new();
if it.member_count > 0 {
tags.push(format!("{} member(s)", it.member_count));
}
if it.invite_count > 0 {
tags.push(format!("{} pending", it.invite_count));
}
if let Some(link) = &it.link {
tags.push(match &link.url {
Some(url) => format!("link: {url}"),
None => "link".to_string(),
});
}
let tags = if tags.is_empty() {
String::new()
} else {
format!(" ({})", tags.join(", "))
};
println!("{kind} {}{tags} [{}]", it.name, it.uid);
}
}
CtlResponse::Error { message } => bail!("{message}"),
other => bail!("unexpected response: {other:?}"),
}
Ok(())
}
fn cmd_shared_with_me() -> Result<()> {
match control_request(CtlRequest::ListSharedWithMe)? {
CtlResponse::Entries { entries } if entries.is_empty() => {
println!("Nothing shared with you.")
}
CtlResponse::Entries { entries } => {
for e in entries {
let kind = if e.is_dir { "d" } else { "-" };
println!("{kind} {:>12} {} [{}]", e.size, e.name, e.uid);
}
}
CtlResponse::Error { message } => bail!("{message}"),
other => bail!("unexpected response: {other:?}"),
}
Ok(())
}
fn cmd_activity(limit: usize) -> Result<()> {
match control_request(CtlRequest::ListActivity { limit })? {
CtlResponse::Activity { items } if items.is_empty() => println!("No recent activity."),
CtlResponse::Activity { items } => {
for a in items {
let when = format_epoch(a.time);
let verb = activity_verb(a.kind);
let mark = if a.ok { " " } else { "!" };
let detail = if a.detail.is_empty() {
String::new()
} else {
format!(" {}", a.detail)
};
println!("{mark} {when} {verb:<8} {}{detail}", a.target);
}
}
CtlResponse::Error { message } => bail!("{message}"),
other => bail!("unexpected response: {other:?}"),
}
Ok(())
}
/// A short verb for an activity kind, for the CLI's one-line-per-event feed.
fn activity_verb(kind: pdfs_core::control::ActivityKind) -> &'static str {
use pdfs_core::control::ActivityKind::*;
match kind {
Upload => "upload",
Download => "download",
Sync => "sync",
Rename => "rename",
Move => "move",
CreateFolder => "mkdir",
Trash => "trash",
Restore => "restore",
DeleteForever => "delete",
EmptyTrash => "empty",
Share => "share",
PublicLink => "link",
Unshare => "unshare",
}
}
/// Format an epoch-seconds timestamp as a compact local-ish `MM-DD HH:MM`.
/// Deliberately dependency-free: derived directly from the Unix time via UTC.
fn format_epoch(secs: i64) -> String {
// Minimal civil-time conversion (UTC) — good enough for an activity feed.
let days = secs.div_euclid(86_400);
let tod = secs.rem_euclid(86_400);
let (h, mi) = (tod / 3600, (tod % 3600) / 60);
// Days since 1970-01-01 → y/m/d (Howard Hinnant's civil_from_days).
let z = days + 719_468;
let era = z.div_euclid(146_097);
let doe = z.rem_euclid(146_097);
let yoe = (doe - doe / 1460 + doe / 36_524 - doe / 146_096) / 365;
let y = yoe + era * 400;
let doy = doe - (365 * yoe + yoe / 4 - yoe / 100);
let mp = (5 * doy + 2) / 153;
let d = doy - (153 * mp + 2) / 5 + 1;
let m = if mp < 10 { mp + 3 } else { mp - 9 };
let _ = y;
format!("{m:02}-{d:02} {h:02}:{mi:02}")
}
fn cmd_leave(uid: String) -> Result<()> {
ok_or_bail(control_request(CtlRequest::LeaveShared { uid })?)
}
fn cmd_invitations(action: InvitationCmd) -> Result<()> {
match action {
InvitationCmd::List => match control_request(CtlRequest::ListInvitations)? {
CtlResponse::Invitations { items } if items.is_empty() => {
println!("No pending invitations.")
}
CtlResponse::Invitations { items } => {
for i in items {
let kind = if i.is_dir { "folder" } else { "file" };
let name = i.name.as_deref().unwrap_or("(name hidden)");
println!("{} shared {kind} \"{name}\" [{}]", i.inviter_email, i.id);
}
}
CtlResponse::Error { message } => bail!("{message}"),
other => bail!("unexpected response: {other:?}"),
},
InvitationCmd::Accept { id } => {
ok_or_bail(control_request(CtlRequest::AcceptInvitation { id })?)?
}
InvitationCmd::Reject { id } => {
ok_or_bail(control_request(CtlRequest::RejectInvitation { id })?)?
}
}
Ok(())
}
fn cmd_bookmarks(action: BookmarkCmd) -> Result<()> {
match action {
BookmarkCmd::List => match control_request(CtlRequest::ListBookmarks)? {
CtlResponse::Bookmarks { items } if items.is_empty() => println!("No bookmarks."),
CtlResponse::Bookmarks { items } => {
for b in items {
let name = b.name.as_deref().unwrap_or("(name hidden)");
println!("{name} {} [{}]", b.url, b.token);
}
}
CtlResponse::Error { message } => bail!("{message}"),
other => bail!("unexpected response: {other:?}"),
},
BookmarkCmd::Add { url, password } => {
ok_or_bail(control_request(CtlRequest::CreateBookmark {
url,
password,
})?)?
}
BookmarkCmd::Rm { token } => {
ok_or_bail(control_request(CtlRequest::DeleteBookmark { token })?)?
}
}
Ok(())
}
/// Print an `Ok` message or bail on an `Error`, for the many commands whose only
/// reply is one of those two.
fn ok_or_bail(resp: CtlResponse) -> Result<()> {
match resp {
CtlResponse::Ok { message } => println!("{message}"),
CtlResponse::Error { message } => bail!("{message}"),
other => bail!("unexpected response: {other:?}"),
}
Ok(())
}
/// Read a line from stdin with a prompt (echoed).
fn prompt_line(label: &str) -> Result<String> {
print!("{label}: ");
@@ -366,8 +946,25 @@ fn cmd_unpin(path: PathBuf) -> Result<()> {
fn cmd_transfers() -> Result<()> {
use pdfs_core::control::TransferDirection;
match control_request(CtlRequest::GetQueueStatus)? {
CtlResponse::Transfers { items } if items.is_empty() => println!("No active transfers."),
CtlResponse::Transfers { items } => {
CtlResponse::Transfers { items, jobs } if items.is_empty() && jobs.is_empty() => {
println!("Nothing in progress.")
}
CtlResponse::Transfers { items, jobs } => {
// Jobs first: they are the context for the transfers under them ("of
// 40 files", "still scanning") rather than a separate list.
for j in jobs {
let count = if j.total > 0 {
format!("{} of {}", j.done, j.total)
} else {
"—".to_string()
};
let detail = if j.detail.is_empty() {
String::new()
} else {
format!(" — {}", j.detail)
};
println!("… {:>9} {}{detail}", count, j.title);
}
for t in items {
let arrow = match t.direction {
TransferDirection::Download => "↓",
@@ -595,17 +1192,21 @@ fn cmd_mkdir(parent: PathBuf, name: String) -> Result<()> {
Ok(())
}
fn cmd_upload(file: PathBuf, parent: PathBuf) -> Result<()> {
let bytes = std::fs::read(&file).with_context(|| format!("read {}", file.display()))?;
let name = file
.file_name()
.and_then(|n| n.to_str())
.ok_or_else(|| anyhow!("source file has no valid name"))?
.to_owned();
match control_request(CtlRequest::UploadFile {
fn cmd_upload(sources: Vec<PathBuf>, parent: PathBuf) -> Result<()> {
// The daemon reads the sources off its own (local) filesystem, so send it
// absolute, canonicalized paths.
let mut abs = Vec::with_capacity(sources.len());
for src in &sources {
let p = std::fs::canonicalize(src).with_context(|| format!("resolve {}", src.display()))?;
abs.push(
p.to_str()
.ok_or_else(|| anyhow!("path is not valid UTF-8: {}", p.display()))?
.to_owned(),
);
}
match control_request(CtlRequest::UploadPaths {
parent: path_arg(&parent)?,
name,
bytes,
sources: abs,
})? {
CtlResponse::Ok { message } => println!("{message}"),
CtlResponse::Error { message } => bail!("{message}"),
+454 -5
View File
@@ -92,6 +92,18 @@ pub enum Request {
name: String,
bytes: Vec<u8>,
},
/// Bulk-upload local files and/or directory trees into the mountpoint-relative
/// `parent` folder. `sources` are absolute paths on the daemon's own
/// filesystem (the daemon is local): each file is uploaded, each directory is
/// recreated remotely and walked recursively. The daemon acks immediately with
/// [`Response::Ok`] and does the work in the background — a big tree far
/// outlasts the socket read timeout — so progress is observed through
/// [`Request::GetQueueStatus`] and completion through the transfer count
/// falling to zero.
UploadPaths {
parent: String,
sources: Vec<String>,
},
/// Delete all unpinned cached blobs and on-demand blocks, keeping pinned
/// files intact. Replies with [`Response::Ok`] reporting the bytes freed.
PurgeCache,
@@ -99,9 +111,11 @@ pub enum Request {
/// persist it to config so the next mount keeps it. Replies with
/// [`Response::Ok`].
SetCacheBudget { bytes: u64 },
/// Snapshot the daemon's in-flight transfers (active uploads/downloads).
/// Replies with [`Response::Transfers`]. Cheap to poll: the daemon keeps the
/// registry in memory, so a front-end can render a live progress widget.
/// Snapshot what the daemon is working on: in-flight transfers (active
/// uploads/downloads) and the longer jobs around them (scans, folder
/// skeletons, the local index, sync passes). Replies with
/// [`Response::Transfers`]. Cheap to poll: the daemon keeps the registry in
/// memory, so a front-end can render a live progress widget.
GetQueueStatus,
/// List what is in the account's trash. Replies with [`Response::Entries`];
/// a trashed node has no path inside the mount, so each entry carries only
@@ -117,6 +131,311 @@ pub enum Request {
/// Permanently delete everything in the trash. Irreversible.
/// Replies with [`Response::Ok`].
EmptyTrash,
// ---- devices ----------------------------------------------------------
/// List the account's registered devices. Replies with [`Response::Devices`].
ListDevices,
/// Rename a device by its uid. Replies with [`Response::Ok`].
RenameDevice { uid: String, name: String },
/// Delete (deregister) a device by its uid. Replies with [`Response::Ok`].
DeleteDevice { uid: String },
// ---- device folder sync (devices.md) ----------------------------------
/// Add a local folder to this machine's device, uploading its tree and
/// registering the device on first use. Replies with [`Response::Ok`].
AddSyncFolder { local_path: String },
/// List this device's synced folders. Replies with [`Response::SyncFolders`].
ListSyncFolders,
/// Remove a synced folder by id; `delete_remote` also trashes its cloud copy.
/// Replies with [`Response::Ok`].
RemoveSyncFolder { id: i64, delete_remote: bool },
/// Switch a synced folder between `mirror` and `ondemand` (Phase 3). Replies
/// with [`Response::Ok`].
SetSyncFolderMode { id: i64, mode: String },
/// Force a reconcile pass: one folder by id, or all when `id` is `None`.
/// Replies with [`Response::Ok`].
SyncNow { id: Option<i64> },
// ---- sharing a node ---------------------------------------------------
/// Invite `emails` (Proton and/or external addresses, auto-detected) to the
/// node at mountpoint-relative `path` at `role` ("viewer"|"editor"|"admin"),
/// with an optional email `message`. Replies with [`Response::Ok`].
ShareNode {
path: String,
emails: Vec<String>,
role: String,
message: Option<String>,
},
/// List the members, pending invitations and public link of the node at
/// mountpoint-relative `path`. Replies with [`Response::Share`].
ListShare { path: String },
/// Change the role of a share entry (member or pending invitation) on the node
/// at `path`. `id` and `kind` identify the entry (from [`Response::Share`]).
/// Replies with [`Response::Ok`].
UpdateShareRole {
path: String,
id: String,
kind: ShareEntryKind,
role: String,
},
/// Remove a share entry (member, pending Proton invite, or external invite)
/// from the node at `path`. Replies with [`Response::Ok`].
RemoveShareEntry {
path: String,
id: String,
kind: ShareEntryKind,
},
/// Create a public link on the node at `path`. `role` is "viewer" or "editor";
/// `password` optionally adds a custom password; `expires` is an optional Unix
/// expiry. Replies with [`Response::PublicLink`] (carrying the shareable URL).
CreatePublicLink {
path: String,
role: String,
password: Option<String>,
expires: Option<i64>,
},
/// Remove the public link `id` from the node at `path`. Replies with
/// [`Response::Ok`].
RemovePublicLink { path: String, id: String },
// ---- shared by me -----------------------------------------------------
/// List the nodes I have shared with others — collaborative shares that still
/// have members, pending invitations or a public link. Replies with
/// [`Response::SharedByMe`]. Each item carries the share's state so the
/// front-end can render it without a follow-up per node.
ListSharedByMe,
// ---- shared with me ---------------------------------------------------
/// List nodes shared with me that I have accepted. Replies with
/// [`Response::Entries`] (each entry carries its `uid`; `path` is empty since
/// the item lives outside the mount tree).
ListSharedWithMe,
/// Leave a shared node by its `uid`, giving up my access. Replies with
/// [`Response::Ok`].
LeaveShared { uid: String },
// ---- incoming invitations ---------------------------------------------
/// List invitations addressed to me, pending accept or reject. Replies with
/// [`Response::Invitations`].
ListInvitations,
/// Accept the invitation `id`, gaining access to the shared node. Replies with
/// [`Response::Ok`].
AcceptInvitation { id: String },
/// Reject the invitation `id`, declining access. Replies with [`Response::Ok`].
RejectInvitation { id: String },
// ---- bookmarks --------------------------------------------------------
/// List public links saved to my account. Replies with [`Response::Bookmarks`].
ListBookmarks,
/// Save a public link `url` (optionally password-protected) as a bookmark.
/// Replies with [`Response::Ok`].
CreateBookmark {
url: String,
password: Option<String>,
},
/// Remove a saved bookmark by its `token`. Replies with [`Response::Ok`].
DeleteBookmark { token: String },
// ---- activity ---------------------------------------------------------
/// Fetch the daemon's recent activity log, newest first, capped at `limit`
/// entries. Replies with [`Response::Activity`]. The log is persisted, so it
/// survives a daemon restart: it records the mutations and transfers the
/// daemon performs (uploads, downloads, deletes, renames, shares, sync
/// passes, …), so a front-end can show a running "what happened" feed
/// without re-deriving it from anywhere.
ListActivity { limit: usize },
}
/// Which kind of share entry a [`ShareEntry`] is, and which collection an
/// [`Request::UpdateShareRole`]/[`Request::RemoveShareEntry`] targets.
#[derive(Serialize, Deserialize, Debug, Clone, Copy, PartialEq, Eq)]
pub enum ShareEntryKind {
/// An accepted member (identified by its membership id).
Member,
/// A pending invitation to a Proton user (identified by its invitation id).
ProtonInvite,
/// A pending invitation to a non-Proton email (identified by its invitation id).
ExternalInvite,
}
/// A registered device in a [`Response::Devices`] listing.
#[derive(Serialize, Deserialize, Debug, Clone)]
pub struct DeviceInfo {
/// Device uid — the handle for [`Request::RenameDevice`]/[`Request::DeleteDevice`].
pub uid: String,
/// Decrypted device name, or a placeholder when it could not be decrypted.
pub name: String,
/// Platform: "Windows", "MacOs" or "Linux".
pub device_type: String,
/// Last sync time, epoch seconds; `None` if it never synced.
pub last_sync: Option<i64>,
}
/// One synced local folder on this machine's device (in [`Response::SyncFolders`]).
#[derive(Serialize, Deserialize, Debug, Clone)]
pub struct SyncFolderInfo {
/// Row id — the handle for [`Request::RemoveSyncFolder`]/[`Request::SetSyncFolderMode`].
pub id: i64,
/// Absolute local folder path.
pub local_path: String,
/// The uid of the folder's remote root under the device root.
pub remote_uid: String,
/// `mirror` (full local copy, two-way synced) or `ondemand` (FUSE mount).
pub mode: String,
/// `idle` | `syncing` | `error` | `conflict`.
pub state: String,
/// Last successful sync, epoch seconds; `0` if never.
pub last_sync: i64,
/// What the folder's sync pass is doing right now, or `None` when no pass is
/// running. Live daemon state, not a stored column.
#[serde(default)]
pub progress: Option<SyncProgress>,
}
/// Which stage a running sync pass is in, in a [`SyncProgress`].
#[derive(Serialize, Deserialize, Debug, Clone, Copy, PartialEq, Eq)]
pub enum SyncPhase {
/// Walking the local tree, the remote tree and the stored baseline to work
/// out what changed. No per-item counts exist yet.
Scanning,
/// Applying the diff: creating folders, uploading, downloading, deleting.
Applying,
}
/// A snapshot of a sync pass in flight (in [`SyncFolderInfo::progress`]), so a
/// front-end can say what the daemon is doing rather than just "syncing".
#[derive(Serialize, Deserialize, Debug, Clone)]
pub struct SyncProgress {
pub phase: SyncPhase,
/// Items applied so far this pass.
pub done: usize,
/// Items known to need applying. This *grows* during a pass: paths are
/// classified depth by depth (a folder must exist remotely before its
/// children can be queued), so the total is what has been discovered so far,
/// not a figure fixed up front.
pub total: usize,
/// The name of an item currently being applied, or empty between items.
/// Several run at once; this is just the most recently started.
pub current: String,
}
/// One member or pending invitation on a node's share (in [`Response::Share`]).
#[derive(Serialize, Deserialize, Debug, Clone)]
pub struct ShareEntry {
/// Membership id (members) or invitation id (invitations) — the handle for
/// [`Request::UpdateShareRole`]/[`Request::RemoveShareEntry`].
pub id: String,
/// The member's / invitee's email address.
pub email: String,
/// Role: "viewer", "editor", "admin" or "inherited".
pub role: String,
/// Whether this is a member, a pending Proton invite, or an external invite.
pub kind: ShareEntryKind,
}
/// A node's public link (in [`Response::Share`] / [`Response::PublicLink`]).
#[derive(Serialize, Deserialize, Debug, Clone)]
pub struct PublicLinkInfo {
/// Public-link id — the handle for [`Request::RemovePublicLink`].
pub id: String,
/// The shareable URL including the `#password` fragment, when known (always on
/// creation; absent when only listed).
pub url: Option<String>,
/// Role granted to anyone with the link ("viewer" or "editor").
pub role: String,
/// Expiry, epoch seconds, if set.
pub expires: Option<i64>,
/// Whether a custom password additionally protects the link.
pub has_password: bool,
}
/// One invitation addressed to me (in [`Response::Invitations`]).
#[derive(Serialize, Deserialize, Debug, Clone)]
pub struct InvitationInfo {
/// Invitation id — the handle for [`Request::AcceptInvitation`]/[`Request::RejectInvitation`].
pub id: String,
/// The email of the user who shared the item.
pub inviter_email: String,
/// The shared item's decrypted name, when available.
pub name: Option<String>,
/// Whether the shared item is a folder.
pub is_dir: bool,
}
/// One saved public link (in [`Response::Bookmarks`]).
#[derive(Serialize, Deserialize, Debug, Clone)]
pub struct BookmarkInfo {
/// Bookmark token — the handle for [`Request::DeleteBookmark`].
pub token: String,
/// The full public URL including the `#password` fragment.
pub url: String,
/// The bookmarked item's decrypted name, when available.
pub name: Option<String>,
/// Whether the bookmarked item is a folder.
pub is_dir: bool,
}
/// One node I have shared with others (in [`Response::SharedByMe`]). Summarizes
/// the share's state so the "Shared" view renders in one pass: how many people
/// have access, how many invitations are still pending, and the public link if
/// the node has one.
#[derive(Serialize, Deserialize, Debug, Clone)]
pub struct SharedItem {
/// Node uid in `volume~link` form — the handle for opening the node's share
/// (via its mount path) or its details.
pub uid: String,
/// The shared node's decrypted name.
pub name: String,
pub is_dir: bool,
/// Mountpoint-relative path, when the daemon can resolve it (the node lives in
/// my own tree). Empty when the path is unknown.
#[serde(default)]
pub path: String,
/// Accepted members (people who already have access), excluding me.
pub member_count: usize,
/// Pending invitations (Proton + external) not yet accepted.
pub invite_count: usize,
/// The node's public link, if it has one.
pub link: Option<PublicLinkInfo>,
}
/// What happened, in an [`ActivityEntry`]. Kept coarse: a front-end maps each to
/// an icon and a verb, and the human detail lives in the entry's fields.
#[derive(Serialize, Deserialize, Debug, Clone, Copy, PartialEq, Eq)]
pub enum ActivityKind {
Upload,
Download,
/// A whole sync pass over one folder, summarising what it moved.
Sync,
Rename,
Move,
CreateFolder,
Trash,
Restore,
DeleteForever,
EmptyTrash,
Share,
PublicLink,
Unshare,
}
/// One line in the daemon's activity log (in [`Response::Activity`]). Newest
/// first. Records a mutation or transfer the daemon performed, with enough
/// context to read as a sentence: "Uploaded report.pdf to /docs".
#[derive(Serialize, Deserialize, Debug, Clone)]
pub struct ActivityEntry {
/// When it happened, epoch seconds.
pub time: i64,
pub kind: ActivityKind,
/// The primary item's name (a file/folder name, or a count like "3 items").
pub target: String,
/// Extra context: a destination path, a new name, an invitee, an error
/// message. Empty when there is nothing to add.
#[serde(default)]
pub detail: String,
/// Whether the operation succeeded. A failed entry still appears, so the log
/// shows what was attempted.
pub ok: bool,
}
/// Which way an active transfer is moving, in a [`TransferItem`].
@@ -126,6 +445,23 @@ pub enum TransferDirection {
Upload,
}
/// One long-running daemon job in a [`Response::Transfers`] snapshot: work that
/// takes long enough to need reporting but doesn't move bytes over the wire —
/// walking a local tree, building a remote folder skeleton, indexing `$HOME`.
/// Byte-moving work is a [`TransferItem`] instead.
#[derive(Serialize, Deserialize, Debug, Clone)]
pub struct JobItem {
/// What the job is, as a front-end would title it ("Uploading files").
pub title: String,
/// What it is doing right now ("Scanning Photos/2024"), or empty.
pub detail: String,
/// Steps finished so far.
pub done: u64,
/// Steps known to need doing, or `0` when unknown (indeterminate progress).
/// May *grow* mid-job as more work is discovered.
pub total: u64,
}
/// One in-flight transfer in a [`Response::Transfers`] snapshot.
#[derive(Serialize, Deserialize, Debug, Clone)]
pub struct TransferItem {
@@ -283,8 +619,37 @@ pub enum Response {
/// is true while a scan of the machine is still running, so a front-end can
/// say "still indexing" instead of "no matches" on a cold first launch.
LocalResults { hits: Vec<LocalHit>, indexing: bool },
/// A snapshot of in-flight transfers (reply to [`Request::GetQueueStatus`]).
Transfers { items: Vec<TransferItem> },
/// A snapshot of what the daemon is working on (reply to
/// [`Request::GetQueueStatus`]): `items` are byte-moving transfers, `jobs`
/// the longer non-transfer work around them (scans, folder skeletons, the
/// local index, sync passes). Both empty means the daemon is idle.
Transfers {
items: Vec<TransferItem>,
#[serde(default)]
jobs: Vec<JobItem>,
},
/// The account's devices (reply to [`Request::ListDevices`]).
Devices { items: Vec<DeviceInfo> },
/// This device's synced folders (reply to [`Request::ListSyncFolders`]).
SyncFolders { items: Vec<SyncFolderInfo> },
/// A node's share: members + pending invitations, and its public link if any
/// (reply to [`Request::ListShare`]).
Share {
entries: Vec<ShareEntry>,
link: Option<PublicLinkInfo>,
},
/// A freshly created public link, carrying the shareable URL (reply to
/// [`Request::CreatePublicLink`]).
PublicLink { link: PublicLinkInfo },
/// Invitations addressed to me (reply to [`Request::ListInvitations`]).
Invitations { items: Vec<InvitationInfo> },
/// Saved public links (reply to [`Request::ListBookmarks`]).
Bookmarks { items: Vec<BookmarkInfo> },
/// Nodes I have shared with others (reply to [`Request::ListSharedByMe`]).
SharedByMe { items: Vec<SharedItem> },
/// The daemon's recent activity, newest first (reply to
/// [`Request::ListActivity`]).
Activity { items: Vec<ActivityEntry> },
/// The request failed.
Error { message: String },
}
@@ -336,6 +701,10 @@ mod tests {
name: "f.bin".into(),
bytes: vec![0, 1, 2, 255],
},
Request::UploadPaths {
parent: "a".into(),
sources: vec!["/home/u/x.txt".into(), "/home/u/pics".into()],
},
];
for req in reqs {
let line = serde_json::to_string(&req).unwrap();
@@ -368,4 +737,84 @@ mod tests {
assert_eq!(line, serde_json::to_string(&back).unwrap());
}
}
/// The sharing and devices requests must survive the same line-delimited JSON
/// round-trip: a mangled id or role would touch the wrong share or member.
#[test]
fn sharing_requests_roundtrip() {
let reqs = [
Request::ListDevices,
Request::RenameDevice {
uid: "dev-1".into(),
name: "laptop".into(),
},
Request::DeleteDevice {
uid: "dev-1".into(),
},
Request::AddSyncFolder {
local_path: "/home/me/Docs".into(),
},
Request::ListSyncFolders,
Request::RemoveSyncFolder {
id: 3,
delete_remote: true,
},
Request::SetSyncFolderMode {
id: 3,
mode: "ondemand".into(),
},
Request::SyncNow { id: Some(3) },
Request::ShareNode {
path: "a/b".into(),
emails: vec!["x@proton.me".into(), "y@example.com".into()],
role: "editor".into(),
message: Some("hi".into()),
},
Request::ListShare { path: "a/b".into() },
Request::UpdateShareRole {
path: "a/b".into(),
id: "mid-1".into(),
kind: ShareEntryKind::Member,
role: "admin".into(),
},
Request::RemoveShareEntry {
path: "a/b".into(),
id: "iid-1".into(),
kind: ShareEntryKind::ExternalInvite,
},
Request::CreatePublicLink {
path: "a/b".into(),
role: "viewer".into(),
password: Some("pw".into()),
expires: Some(1_700_000_000),
},
Request::RemovePublicLink {
path: "a/b".into(),
id: "url-1".into(),
},
Request::ListSharedByMe,
Request::ListActivity { limit: 100 },
Request::ListSharedWithMe,
Request::LeaveShared {
uid: "vol~link".into(),
},
Request::ListInvitations,
Request::AcceptInvitation { id: "inv-1".into() },
Request::RejectInvitation { id: "inv-1".into() },
Request::ListBookmarks,
Request::CreateBookmark {
url: "https://drive.proton.me/urls/tok#pw".into(),
password: None,
},
Request::DeleteBookmark {
token: "tok".into(),
},
];
for req in reqs {
let line = serde_json::to_string(&req).unwrap();
assert!(!line.contains('\n'), "wire form must be a single line");
let back: Request = serde_json::from_str(&line).unwrap();
assert_eq!(line, serde_json::to_string(&back).unwrap());
}
}
}
+456 -6
View File
@@ -19,10 +19,20 @@ use proton_drive_rs::{Node, NodeKind};
use rusqlite::{Connection, OptionalExtension, params};
use crate::Result;
use crate::control::{ActivityEntry, ActivityKind};
use crate::localindex::LocalEntry;
/// Current schema version. Bump on every forward migration added below.
const SCHEMA_VERSION: i64 = 7;
const SCHEMA_VERSION: i64 = 9;
/// How many activity rows to keep. Older rows are pruned on insert, so the feed
/// stays a bounded "recent history" rather than growing without limit.
const ACTIVITY_KEEP: i64 = 2000;
/// Size the WAL is truncated back to after a checkpoint. Comfortably above the
/// steady-state working set (a few MB), so the truncation only claws back the
/// outliers rather than fighting the normal write path for disk.
const WAL_SIZE_LIMIT: i64 = 64 * 1024 * 1024;
/// A photo whose thumbnail state is not known yet: it has never been asked for.
pub const THUMB_UNKNOWN: i64 = 0;
@@ -77,6 +87,46 @@ pub struct StoredTrash {
pub mtime: i64,
}
/// The one Proton Drive Device this machine is registered as (devices.md).
/// Cached so restarts reuse the same device rather than creating a new one.
#[derive(Clone, Debug, PartialEq)]
pub struct StoredDevice {
pub uid: String,
pub share_id: String,
pub root_uid: String,
pub name: String,
pub created: i64,
}
/// One local folder the user added to this device's sync set.
#[derive(Clone, Debug, PartialEq)]
pub struct StoredSyncFolder {
pub id: i64,
pub local_path: String,
pub remote_uid: String,
pub remote_share_id: String,
/// `mirror` (full local copy, two-way synced) or `ondemand` (FUSE mount).
pub mode: String,
/// `idle` | `syncing` | `error` | `conflict`.
pub state: String,
pub last_sync: i64,
}
/// One per-file sync baseline row: what a path looked like on both sides at the
/// last successful sync, so the next reconcile can tell which side changed
/// (devices.md Phase 2). `remote_rev`/`remote_hash` hold the remote signature —
/// its modification time and size as strings — since no cheap content hash is
/// exposed; change detection is `(mtime, size)` on each side.
#[derive(Clone, Debug, PartialEq)]
pub struct StoredSyncEntry {
pub rel_path: String,
pub remote_uid: Option<String>,
pub local_mtime: i64,
pub local_size: i64,
pub remote_rev: Option<String>,
pub remote_hash: Option<String>,
}
/// Handle to the unified metadata database.
///
/// Cheap to wrap in an `Arc`; clone the `Arc`, not this. All access goes through
@@ -95,6 +145,13 @@ impl Db {
conn.pragma_update(None, "journal_mode", "WAL")?;
conn.pragma_update(None, "synchronous", "NORMAL")?;
conn.pragma_update(None, "foreign_keys", "ON")?;
// Hand the WAL's disk back after a checkpoint. Without a limit SQLite
// reuses the file in place but never shrinks it, so a single large
// transaction (a full-index FTS rebuild in `local_finish_scan` is the
// one that reaches this size) leaves the WAL at its high-water mark
// forever — a multi-GB file next to a database two orders of magnitude
// smaller. Checkpointing is unaffected; only the file is truncated back.
conn.pragma_update(None, "journal_size_limit", WAL_SIZE_LIMIT)?;
let db = Self {
conn: Mutex::new(conn),
@@ -161,6 +218,12 @@ impl Db {
if current < 7 {
tx.execute_batch(MIGRATION_V7)?;
}
if current < 8 {
tx.execute_batch(MIGRATION_V8)?;
}
if current < 9 {
tx.execute_batch(MIGRATION_V9)?;
}
tx.execute(
"INSERT INTO sync_state (key, value) VALUES ('schema_version', ?1)
ON CONFLICT(key) DO UPDATE SET value = excluded.value",
@@ -381,9 +444,8 @@ impl Db {
let tx = conn.transaction()?;
let learned: HashMap<String, (Option<f64>, i64)> = {
let mut stmt = tx.prepare("SELECT uid, ratio, thumb_state FROM photos")?;
let rows = stmt.query_map([], |r| {
Ok((r.get::<_, String>(0)?, (r.get(1)?, r.get(2)?)))
})?;
let rows =
stmt.query_map([], |r| Ok((r.get::<_, String>(0)?, (r.get(1)?, r.get(2)?))))?;
rows.collect::<rusqlite::Result<_>>()?
};
@@ -394,8 +456,16 @@ impl Db {
VALUES (?1, ?2, ?3, ?4, ?5, ?6)",
)?;
for (seq, (uid, capture_time, name)) in items.iter().enumerate() {
let (ratio, thumb_state) = learned.get(uid).copied().unwrap_or((None, THUMB_UNKNOWN));
stmt.execute(params![uid, capture_time, name, ratio, thumb_state, seq as i64])?;
let (ratio, thumb_state) =
learned.get(uid).copied().unwrap_or((None, THUMB_UNKNOWN));
stmt.execute(params![
uid,
capture_time,
name,
ratio,
thumb_state,
seq as i64
])?;
}
}
tx.commit()?;
@@ -520,6 +590,266 @@ impl Db {
Ok(items)
}
// ---- device sync (devices.md) -----------------------------------------
/// The registered device for this machine, if one has been created/cached.
pub fn device_get(&self) -> Result<Option<StoredDevice>> {
let conn = self.conn.lock().unwrap();
conn.query_row(
"SELECT uid, share_id, root_uid, name, created FROM device LIMIT 1",
[],
|r| {
Ok(StoredDevice {
uid: r.get(0)?,
share_id: r.get(1)?,
root_uid: r.get(2)?,
name: r.get(3)?,
created: r.get(4)?,
})
},
)
.optional()
.map_err(Into::into)
}
/// Persist (or replace) this machine's device. The table holds a single row.
pub fn device_set(&self, dev: &StoredDevice) -> Result<()> {
let conn = self.conn.lock().unwrap();
conn.execute("DELETE FROM device", [])?;
conn.execute(
"INSERT INTO device (uid, share_id, root_uid, name, created)
VALUES (?1, ?2, ?3, ?4, ?5)",
params![dev.uid, dev.share_id, dev.root_uid, dev.name, dev.created],
)?;
Ok(())
}
/// Add a synced folder, returning its new row id.
pub fn sync_folder_add(
&self,
local_path: &str,
remote_uid: &str,
remote_share_id: &str,
) -> Result<i64> {
let conn = self.conn.lock().unwrap();
conn.execute(
"INSERT INTO sync_folder (local_path, remote_uid, remote_share_id)
VALUES (?1, ?2, ?3)",
params![local_path, remote_uid, remote_share_id],
)?;
Ok(conn.last_insert_rowid())
}
/// Every synced folder, oldest first.
pub fn sync_folder_list(&self) -> Result<Vec<StoredSyncFolder>> {
let conn = self.conn.lock().unwrap();
let mut stmt = conn.prepare(
"SELECT id, local_path, remote_uid, remote_share_id, mode, state, last_sync
FROM sync_folder ORDER BY id",
)?;
let rows = stmt.query_map([], |r| {
Ok(StoredSyncFolder {
id: r.get(0)?,
local_path: r.get(1)?,
remote_uid: r.get(2)?,
remote_share_id: r.get(3)?,
mode: r.get(4)?,
state: r.get(5)?,
last_sync: r.get(6)?,
})
})?;
let mut items = Vec::new();
for row in rows {
items.push(row?);
}
Ok(items)
}
/// Look up one synced folder by id.
pub fn sync_folder_get(&self, id: i64) -> Result<Option<StoredSyncFolder>> {
let conn = self.conn.lock().unwrap();
conn.query_row(
"SELECT id, local_path, remote_uid, remote_share_id, mode, state, last_sync
FROM sync_folder WHERE id = ?1",
params![id],
|r| {
Ok(StoredSyncFolder {
id: r.get(0)?,
local_path: r.get(1)?,
remote_uid: r.get(2)?,
remote_share_id: r.get(3)?,
mode: r.get(4)?,
state: r.get(5)?,
last_sync: r.get(6)?,
})
},
)
.optional()
.map_err(Into::into)
}
/// Remove a synced folder and its per-file baseline.
pub fn sync_folder_remove(&self, id: i64) -> Result<bool> {
let mut conn = self.conn.lock().unwrap();
let tx = conn.transaction()?;
tx.execute("DELETE FROM sync_entry WHERE folder_id = ?1", params![id])?;
let n = tx.execute("DELETE FROM sync_folder WHERE id = ?1", params![id])?;
tx.commit()?;
Ok(n > 0)
}
/// Update a synced folder's mode (`mirror`/`ondemand`).
pub fn sync_folder_set_mode(&self, id: i64, mode: &str) -> Result<()> {
let conn = self.conn.lock().unwrap();
conn.execute(
"UPDATE sync_folder SET mode = ?2 WHERE id = ?1",
params![id, mode],
)?;
Ok(())
}
/// Update a synced folder's state and stamp `last_sync` to now.
pub fn sync_folder_set_state(&self, id: i64, state: &str, last_sync: i64) -> Result<()> {
let conn = self.conn.lock().unwrap();
conn.execute(
"UPDATE sync_folder SET state = ?2, last_sync = ?3 WHERE id = ?1",
params![id, state, last_sync],
)?;
Ok(())
}
/// The whole per-file sync baseline for a folder, keyed by relative path.
pub fn sync_entries(&self, folder_id: i64) -> Result<HashMap<String, StoredSyncEntry>> {
let conn = self.conn.lock().unwrap();
let mut stmt = conn.prepare(
"SELECT rel_path, remote_uid, local_mtime, local_size, remote_rev, remote_hash
FROM sync_entry WHERE folder_id = ?1",
)?;
let rows = stmt.query_map(params![folder_id], |r| {
Ok(StoredSyncEntry {
rel_path: r.get(0)?,
remote_uid: r.get(1)?,
local_mtime: r.get(2)?,
local_size: r.get(3)?,
remote_rev: r.get(4)?,
remote_hash: r.get(5)?,
})
})?;
let mut map = HashMap::new();
for row in rows {
let e = row?;
map.insert(e.rel_path.clone(), e);
}
Ok(map)
}
/// Insert or replace one baseline row.
pub fn sync_entry_upsert(&self, folder_id: i64, e: &StoredSyncEntry) -> Result<()> {
let conn = self.conn.lock().unwrap();
conn.execute(
"INSERT INTO sync_entry
(folder_id, rel_path, remote_uid, local_mtime, local_size, remote_rev, remote_hash)
VALUES (?1, ?2, ?3, ?4, ?5, ?6, ?7)
ON CONFLICT(folder_id, rel_path) DO UPDATE SET
remote_uid = excluded.remote_uid,
local_mtime = excluded.local_mtime,
local_size = excluded.local_size,
remote_rev = excluded.remote_rev,
remote_hash = excluded.remote_hash",
params![
folder_id,
e.rel_path,
e.remote_uid,
e.local_mtime,
e.local_size,
e.remote_rev,
e.remote_hash,
],
)?;
Ok(())
}
/// Drop one baseline row (a path that left the sync set on both sides).
pub fn sync_entry_remove(&self, folder_id: i64, rel_path: &str) -> Result<()> {
let conn = self.conn.lock().unwrap();
conn.execute(
"DELETE FROM sync_entry WHERE folder_id = ?1 AND rel_path = ?2",
params![folder_id, rel_path],
)?;
Ok(())
}
// ---- activity ---------------------------------------------------------
/// Append one entry to the activity log, pruning back to [`ACTIVITY_KEEP`]
/// rows. `kind` round-trips through serde rather than a hand-written string
/// table, so adding an [`ActivityKind`] variant needs no change here.
pub fn activity_add(&self, entry: &ActivityEntry) -> Result<()> {
let kind = serde_json::to_string(&entry.kind)?;
let mut conn = self.conn.lock().unwrap();
let tx = conn.transaction()?;
tx.execute(
"INSERT INTO activity (time, kind, target, detail, ok) VALUES (?1, ?2, ?3, ?4, ?5)",
params![entry.time, kind, entry.target, entry.detail, entry.ok],
)?;
tx.execute(
"DELETE FROM activity WHERE id <= (
SELECT id FROM activity ORDER BY id DESC LIMIT 1 OFFSET ?1
)",
params![ACTIVITY_KEEP],
)?;
tx.commit()?;
Ok(())
}
/// The most recent activity, newest first, capped at `limit` entries. Rows
/// whose stored `kind` no longer parses (written by an older build) are
/// skipped rather than failing the whole read.
pub fn activity_list(&self, limit: usize) -> Result<Vec<ActivityEntry>> {
let conn = self.conn.lock().unwrap();
let mut stmt = conn.prepare(
"SELECT time, kind, target, detail, ok FROM activity
ORDER BY id DESC LIMIT ?1",
)?;
let rows = stmt.query_map(params![limit as i64], |r| {
Ok((
r.get::<_, i64>(0)?,
r.get::<_, String>(1)?,
r.get::<_, String>(2)?,
r.get::<_, String>(3)?,
r.get::<_, bool>(4)?,
))
})?;
let mut out = Vec::new();
for row in rows {
let (time, kind, target, detail, ok) = row?;
let Ok(kind) = serde_json::from_str::<ActivityKind>(&kind) else {
continue;
};
out.push(ActivityEntry {
time,
kind,
target,
detail,
ok,
});
}
Ok(out)
}
/// Drop the entire baseline for a folder. Used when flipping ondemand→mirror:
/// the local tree was evicted, so the old baseline is stale and would make the
/// next reconcile mistake "locally deleted" for "must re-download". Clearing it
/// leaves an empty baseline + full remote = pure download (devices.md P3).
pub fn sync_entries_clear(&self, folder_id: i64) -> Result<()> {
let conn = self.conn.lock().unwrap();
conn.execute(
"DELETE FROM sync_entry WHERE folder_id = ?1",
params![folder_id],
)?;
Ok(())
}
/// Children of `parent`, for the offline `ensure_children` fast path when a
/// listed folder's `children` map entry was trimmed mid-run. Returns `None`
/// when the folder is not marked `listed` (listing unknown, must re-fetch).
@@ -1099,9 +1429,86 @@ CREATE TABLE trash (
);
";
/// Schema v8: device sync (devices.md). This machine registers as one Proton Drive
/// **Device** (a share + root folder on the main volume); `device` is a singleton
/// row cached so we reuse the same device across restarts instead of creating a
/// new one each run. `sync_folder` is one row per local folder the user added,
/// each mapped to a folder under the device root; `mode` is `mirror` (full local
/// copy, two-way synced) or `ondemand` (FUSE mount at `local_path`, no local
/// storage). `sync_entry` is the per-file baseline for three-way merge — added in
/// this migration so Phase 2 has the table, though Phase 1 leaves it empty.
const MIGRATION_V8: &str = "
CREATE TABLE device (
uid TEXT PRIMARY KEY,
share_id TEXT NOT NULL,
root_uid TEXT NOT NULL,
name TEXT NOT NULL,
created INTEGER NOT NULL DEFAULT 0
);
CREATE TABLE sync_folder (
id INTEGER PRIMARY KEY,
local_path TEXT NOT NULL UNIQUE,
remote_uid TEXT NOT NULL,
remote_share_id TEXT NOT NULL,
mode TEXT NOT NULL DEFAULT 'mirror',
state TEXT NOT NULL DEFAULT 'idle',
last_sync INTEGER NOT NULL DEFAULT 0
);
CREATE TABLE sync_entry (
folder_id INTEGER NOT NULL,
rel_path TEXT NOT NULL,
remote_uid TEXT,
local_mtime INTEGER NOT NULL DEFAULT 0,
local_size INTEGER NOT NULL DEFAULT 0,
remote_hash TEXT,
remote_rev TEXT,
PRIMARY KEY (folder_id, rel_path)
);
";
const MIGRATION_V9: &str = "
CREATE TABLE activity (
id INTEGER PRIMARY KEY,
time INTEGER NOT NULL,
kind TEXT NOT NULL,
target TEXT NOT NULL,
detail TEXT NOT NULL DEFAULT '',
ok INTEGER NOT NULL DEFAULT 1
);
CREATE INDEX activity_time ON activity(time DESC);
";
#[cfg(test)]
mod tests {
use super::*;
/// `Db::open` applies the pragmas that `open_in_memory` skips, so the WAL
/// settings can only be checked against a real file.
#[test]
fn open_bounds_the_wal_size() {
let path = std::env::temp_dir().join(format!("pdfs-db-wal-{}.db", std::process::id()));
let _ = std::fs::remove_file(&path);
let db = Db::open(&path).unwrap();
let conn = db.conn.lock().unwrap();
let mode: String = conn
.query_row("PRAGMA journal_mode", [], |r| r.get(0))
.unwrap();
assert_eq!(mode, "wal");
// Without this the WAL is reused in place but never shrinks, so one
// oversized transaction strands its high-water mark on disk forever.
let limit: i64 = conn
.query_row("PRAGMA journal_size_limit", [], |r| r.get(0))
.unwrap();
assert_eq!(limit, WAL_SIZE_LIMIT);
drop(conn);
drop(db);
let _ = std::fs::remove_file(&path);
}
use proton_drive_rs::proton_sdk::ids::{LinkId, VolumeId};
fn uid(link: &str) -> NodeUid {
@@ -1483,6 +1890,49 @@ mod tests {
assert_eq!(db.search("unique", 10).unwrap().len(), 0);
}
fn activity(target: &str, kind: ActivityKind, ok: bool) -> ActivityEntry {
ActivityEntry {
time: 1700,
kind,
target: target.into(),
detail: "detail".into(),
ok,
}
}
#[test]
fn activity_reads_back_newest_first() {
let db = Db::open_in_memory().unwrap();
db.activity_add(&activity("a.txt", ActivityKind::Upload, true))
.unwrap();
db.activity_add(&activity("b.txt", ActivityKind::Download, false))
.unwrap();
let items = db.activity_list(10).unwrap();
assert_eq!(items.len(), 2);
assert_eq!(items[0].target, "b.txt");
assert_eq!(items[0].kind, ActivityKind::Download);
assert!(!items[0].ok);
assert_eq!(items[0].detail, "detail");
assert_eq!(items[0].time, 1700);
assert_eq!(items[1].target, "a.txt");
assert_eq!(db.activity_list(1).unwrap().len(), 1);
}
#[test]
fn activity_prunes_to_the_keep_limit() {
let db = Db::open_in_memory().unwrap();
for i in 0..(ACTIVITY_KEEP + 10) {
db.activity_add(&activity(&format!("f{i}"), ActivityKind::Upload, true))
.unwrap();
}
let items = db.activity_list(ACTIVITY_KEEP as usize * 2).unwrap();
assert_eq!(items.len(), ACTIVITY_KEEP as usize);
// The newest survive; the oldest are the ones dropped.
assert_eq!(items[0].target, format!("f{}", ACTIVITY_KEEP + 9));
}
#[test]
fn opens_and_migrates() {
let db = Db::open_in_memory().unwrap();
+1
View File
@@ -16,3 +16,4 @@ fuser.workspace = true
libc.workspace = true
serde_json.workspace = true
image.workspace = true
notify = "8"
File diff suppressed because it is too large. Load diff
File diff suppressed because it is too large. Load diff
+135 -12
View File
@@ -1,12 +1,17 @@
//! In-memory registry of in-flight transfers, behind [`Request::GetQueueStatus`].
//! In-memory registry of in-flight work, behind [`Request::GetQueueStatus`].
//!
//! The daemon downloads and uploads whole files through the SDK's streaming
//! [`Read`]/[`Write`] variants. Wrapping that reader/writer in a
//! Two kinds of work live here. **Transfers** move bytes: the daemon downloads
//! and uploads whole files through the SDK's streaming [`Read`]/[`Write`]
//! variants, and wrapping that reader/writer in a
//! [`CountingReader`]/[`CountingWriter`] lets each block of bytes tick a
//! per-transfer counter without the SDK knowing anything about progress. A
//! [`TransferGuard`] registers a transfer on creation and deregisters it on
//! drop, so the registry always reflects exactly what is in flight — even if a
//! transfer fails partway and unwinds.
//! per-transfer counter without the SDK knowing anything about progress.
//! **Jobs** ([`JobGuard`]) cover the long stretches around them that move no
//! bytes — walking a local tree, creating the remote folder skeleton, indexing
//! `$HOME` — which otherwise look to a front-end exactly like an idle daemon.
//!
//! Both register on creation and deregister on guard drop, so the registry
//! always reflects exactly what is in flight — even if the work fails partway
//! and unwinds.
//!
//! [`Request::GetQueueStatus`]: pdfs_core::control::Request::GetQueueStatus
@@ -16,7 +21,7 @@ use std::sync::atomic::{AtomicU64, Ordering};
use std::sync::{Arc, Mutex};
use std::time::Instant;
use pdfs_core::control::{TransferDirection, TransferItem};
use pdfs_core::control::{JobItem, TransferDirection, TransferItem};
/// One registered transfer. `done` is bumped from the I/O wrapper without the
/// registry lock; the rest is immutable for the transfer's lifetime.
@@ -29,11 +34,22 @@ struct Entry {
started: Instant,
}
/// The set of transfers currently in flight. Cloned `Arc`-style across the FUSE
/// session and the control-socket task (both share one registry via [`Core`]).
/// One registered job. `detail`/`done`/`total` all move as the job runs; `total`
/// may grow when more work is discovered mid-job.
struct JobEntry {
title: String,
detail: Mutex<String>,
done: AtomicU64,
total: AtomicU64,
}
/// The set of transfers and jobs currently in flight. Cloned `Arc`-style across
/// the FUSE session and the control-socket task (both share one registry via
/// [`Core`]).
#[derive(Default)]
pub struct TransferRegistry {
inner: Mutex<HashMap<u64, Arc<Entry>>>,
jobs: Mutex<HashMap<u64, Arc<JobEntry>>>,
next: AtomicU64,
}
@@ -68,14 +84,58 @@ impl TransferRegistry {
}
}
/// Snapshot every in-flight transfer for [`Response::Transfers`]. Speed is the
/// Register a long-running non-transfer job titled `title`, returning a guard
/// that carries its progress and deregisters the job when dropped. The job
/// starts indeterminate; give it a `total` once one is known.
pub fn begin_job(self: &Arc<Self>, title: impl Into<String>) -> JobGuard {
let id = self.next.fetch_add(1, Ordering::Relaxed);
let entry = Arc::new(JobEntry {
title: title.into(),
detail: Mutex::new(String::new()),
done: AtomicU64::new(0),
total: AtomicU64::new(0),
});
self.jobs.lock().unwrap().insert(id, entry.clone());
JobGuard {
reg: self.clone(),
id,
entry,
}
}
/// Snapshot every in-flight job for [`Response::Transfers`], oldest first, so
/// a front-end's rows keep their order across polls.
///
/// [`Response::Transfers`]: pdfs_core::control::Response::Transfers
pub fn jobs_snapshot(&self) -> Vec<JobItem> {
let map = self.jobs.lock().unwrap();
let mut ids: Vec<_> = map.keys().copied().collect();
ids.sort_unstable();
ids.iter()
.map(|id| {
let e = &map[id];
JobItem {
title: e.title.clone(),
detail: e.detail.lock().unwrap().clone(),
done: e.done.load(Ordering::Relaxed),
total: e.total.load(Ordering::Relaxed),
}
})
.collect()
}
/// Snapshot every in-flight transfer for [`Response::Transfers`], oldest
/// first so a front-end's rows keep their order across polls. Speed is the
/// running average since the transfer began — simple, and stable enough for a
/// progress widget without per-tick sampling state.
///
/// [`Response::Transfers`]: pdfs_core::control::Response::Transfers
pub fn snapshot(&self) -> Vec<TransferItem> {
let map = self.inner.lock().unwrap();
map.values()
let mut ids: Vec<_> = map.keys().copied().collect();
ids.sort_unstable();
ids.iter()
.map(|id| &map[id])
.map(|e| {
let done = e.done.load(Ordering::Relaxed);
let secs = e.started.elapsed().as_secs_f64();
@@ -120,6 +180,44 @@ impl Drop for TransferGuard {
}
}
/// Lifetime handle for a registered job: describe what it is doing with
/// [`detail`], size it with [`set_total`], advance it with [`step`], and the job
/// leaves the registry when this drops. Shared across the tasks of one batch as
/// an `Arc`, so a concurrent phase can report a single "N of M" line.
///
/// [`detail`]: JobGuard::detail
/// [`set_total`]: JobGuard::set_total
/// [`step`]: JobGuard::step
pub struct JobGuard {
reg: Arc<TransferRegistry>,
id: u64,
entry: Arc<JobEntry>,
}
impl JobGuard {
/// Say what the job is working on right now; empty clears it.
pub fn detail(&self, what: impl Into<String>) {
*self.entry.detail.lock().unwrap() = what.into();
}
/// Set how many steps the job now knows it has. Safe to raise mid-job as more
/// work is discovered.
pub fn set_total(&self, total: u64) {
self.entry.total.store(total, Ordering::Relaxed);
}
/// Mark one more step done.
pub fn step(&self) {
self.entry.done.fetch_add(1, Ordering::Relaxed);
}
}
impl Drop for JobGuard {
fn drop(&mut self) {
self.reg.jobs.lock().unwrap().remove(&self.id);
}
}
/// A [`Write`] that tallies bytes written to a [`TransferGuard`] (download path).
pub struct CountingWriter<'a, W> {
inner: W,
@@ -148,6 +246,31 @@ impl<W: Write> Write for CountingWriter<'_, W> {
}
}
/// A [`Read`] that tallies bytes read and *owns* its [`TransferGuard`], so the
/// transfer stays registered exactly as long as the reader lives. Used by the
/// concurrent bulk uploader, where each task hands its reader to the SDK and has
/// nowhere separate to park the guard; the transfer deregisters when the SDK
/// drops the reader after sealing the revision. Owning (rather than borrowing)
/// also keeps each upload future `Send + 'static` for [`tokio::task::spawn`].
pub struct OwnedCountingReader<R> {
inner: R,
guard: TransferGuard,
}
impl<R: Read> OwnedCountingReader<R> {
pub fn new(inner: R, guard: TransferGuard) -> Self {
Self { inner, guard }
}
}
impl<R: Read> Read for OwnedCountingReader<R> {
fn read(&mut self, buf: &mut [u8]) -> std::io::Result<usize> {
let n = self.inner.read(buf)?;
self.guard.add(n as u64);
Ok(n)
}
}
/// A [`Read`] that tallies bytes read through a [`TransferGuard`] (upload path).
pub struct CountingReader<'a, R> {
inner: R,
+28
View File
@@ -0,0 +1,28 @@
use std::env;
use std::process::Command;
fn main() {
let out_dir = env::var("OUT_DIR").unwrap();
let status = Command::new("glib-compile-resources")
.args([
"--target",
&format!("{}/pdfs.gresource", out_dir),
"--sourcedir",
"resources/icons",
"resources/pdfs.gresource.xml",
])
.status()
.expect("Failed to execute glib-compile-resources");
if !status.success() {
panic!(
"glib-compile-resources failed with exit code: {:?}",
status.code()
);
}
println!("cargo:rerun-if-changed=resources/pdfs.gresource.xml");
println!(
"cargo:rerun-if-changed=resources/icons/scalable/actions/pdfs-folder-upload-symbolic.svg"
);
}
@@ -0,0 +1,7 @@
<svg xmlns="http://www.w3.org/2000/svg" width="16" height="16" viewBox="0 0 16 16">
<!-- Upload-a-folder action: a folder silhouette with an up arrow knocked out of
it (one path, fill-rule evenodd, so symbolic recolouring keeps working).
Matches the Papirus folder shape (width 14, height 12) with a pixel-aligned,
proportionate up arrow cutout for size consistency. -->
<path fill="#222222" fill-rule="evenodd" d="M 1,3 V 14 C 1,14 1,15 2,15 H 14 C 14,15 15,15 15,14 V 5 C 15,4 14,4 14,4 H 9 L 7,2 H 2 C 2,2 1,2 1,3 Z M 7,12 V 8 H 5 L 8,5 L 11,8 H 9 V 12 Z"/>
</svg>

After

Width:  |  Height:  |  Size: 593 B

@@ -0,0 +1,9 @@
<?xml version="1.0" encoding="UTF-8"?>
<gresources>
<!-- Icons we ship ourselves because no icon theme has them. Registered as an
icon-theme resource path at startup, so they resolve by name (and recolour
with the theme) exactly like a stock symbolic icon. -->
<gresource prefix="/de/nils/protondrivelinux/icons">
<file preprocess="xml-stripblanks">scalable/actions/pdfs-folder-upload-symbolic.svg</file>
</gresource>
</gresources>
+2480 -75
View File
File diff suppressed because it is too large. Load diff
+12 -6
View File
@@ -18,7 +18,7 @@ use ksni::menu::StandardItem;
use ksni::{MenuItem, Tray, TrayService};
use pdfs_core::auth;
use pdfs_core::config::AppDirs;
use pdfs_core::control::{Request, Response, TransferDirection, TransferItem, send};
use pdfs_core::control::{JobItem, Request, Response, TransferDirection, TransferItem, send};
use pdfs_core::service;
/// How often the tray re-polls the daemon to refresh its menu.
@@ -38,12 +38,18 @@ struct DriveState {
sync: String,
}
/// Summarise a transfer snapshot into one menu line, or empty when idle. A single
/// Summarise a work snapshot into one menu line, or empty when idle. A single
/// transfer names the file and its percentage; several collapse to counts so the
/// menu stays one line regardless of queue depth.
fn sync_line(items: &[TransferItem]) -> String {
/// menu stays one line regardless of queue depth. With nothing moving bytes, a
/// running job speaks for itself — a scan or an index rebuild is still "busy",
/// and the tray saying nothing there reads as "finished".
fn sync_line(items: &[TransferItem], jobs: &[JobItem]) -> String {
match items {
[] => String::new(),
[] => match jobs.first() {
Some(j) if j.total > 0 => format!("{} ({} of {})", j.title, j.done, j.total),
Some(j) => format!("{}…", j.title),
None => String::new(),
},
[t] => {
let arrow = match t.direction {
TransferDirection::Download => "↓",
@@ -87,7 +93,7 @@ fn poll_state(socket: &Path, default_mountpoint: &Path) -> DriveState {
mountpoint: PathBuf::from(mountpoint),
// Same daemon is up, so a cheap follow-up poll gives the sync line.
sync: match send(socket, &Request::GetQueueStatus) {
Ok(Response::Transfers { items }) => sync_line(&items),
Ok(Response::Transfers { items, jobs }) => sync_line(&items, &jobs),
_ => String::new(),
},
},