mirror of
https://github.com/narrrl/proton-drive-linux.git
synced 2026-11-04 16:56:59 +01:00
Compare commits
| Author | SHA1 | Date | |
|---|---|---|---|
 nvrl
|
6c67226ab0 | ||
|
nvrl
|
6f7269b9b6 | ||
|
nvrl
|
dacacf51bb | ||
|
nvrl
|
8f28f5fd22 | ||
|
nvrl
|
924e07e85f | ||
|
nvrl
|
e77ab0ffa9 | ||
|
nvrl
|
d131d3dca8 | ||
|
nvrl
|
5766bc4ed1 | ||
|
nvrl
|
35bc957cd5 | ||
|
nvrl
|
f9000f8540 |
No files matched your search
@@ -62,6 +62,7 @@ jobs:
|
||||
cp target/release/pdfs dist/binaries/
|
||||
cp target/release/pdfs-tray dist/binaries/
|
||||
cp target/release/pdfs-app dist/binaries/
|
||||
cp target/release/pdfs-prompt dist/binaries/
|
||||
|
||||
# Create tarball
|
||||
tar -czf proton-drive-linux-${{ env.VERSION }}-x86_64.tar.gz -C dist/binaries .
|
||||
@@ -78,6 +79,7 @@ jobs:
|
||||
cp target/release/pdfs deb/usr/bin/
|
||||
cp target/release/pdfs-tray deb/usr/bin/
|
||||
cp target/release/pdfs-app deb/usr/bin/
|
||||
cp target/release/pdfs-prompt deb/usr/bin/
|
||||
|
||||
# Copy desktop integration files
|
||||
cp packaging/io.narl.proton-drive-linux.desktop deb/usr/share/applications/
|
||||
@@ -101,7 +103,7 @@ jobs:
|
||||
Proton Drive client for Linux featuring:
|
||||
- FUSE files-on-demand mount
|
||||
- Command-line interface (CLI)
|
||||
- GTK4 system tray and desktop application
|
||||
- GTK4 system tray, desktop application, and search launcher
|
||||
EOF
|
||||
|
||||
# Build Debian package
|
||||
|
||||
Generated
+83
-10
@@ -468,6 +468,16 @@ dependencies = [
|
||||
"cipher 0.5.2",
|
||||
]
|
||||
|
||||
[[package]]
|
||||
name = "bstr"
|
||||
version = "1.12.3"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "5cee35f73844aa3014bb606320a6c1f010249dbdf43342fe54b5a4f6a8ed4b79"
|
||||
dependencies = [
|
||||
"memchr",
|
||||
"serde_core",
|
||||
]
|
||||
|
||||
[[package]]
|
||||
name = "buffer-redux"
|
||||
version = "1.1.0"
|
||||
@@ -804,6 +814,25 @@ version = "0.1.6"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "fd121741cf3eb82c08dd3023eb55bf2665e5f60ec20f89760cf836ae4562e6a0"
|
||||
|
||||
[[package]]
|
||||
name = "crossbeam-deque"
|
||||
version = "0.8.7"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "5181e0de7b61eb03a81e347d6dd8797bae9da5146707b51077e2d71a54ec0ceb"
|
||||
dependencies = [
|
||||
"crossbeam-epoch",
|
||||
"crossbeam-utils",
|
||||
]
|
||||
|
||||
[[package]]
|
||||
name = "crossbeam-epoch"
|
||||
version = "0.9.20"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "2d6914041f254d6e9176c01941b21115dcfb7089e55135a35411081bd106ef3f"
|
||||
dependencies = [
|
||||
"crossbeam-utils",
|
||||
]
|
||||
|
||||
[[package]]
|
||||
name = "crossbeam-utils"
|
||||
version = "0.8.21"
|
||||
@@ -1689,6 +1718,19 @@ dependencies = [
|
||||
"system-deps",
|
||||
]
|
||||
|
||||
[[package]]
|
||||
name = "globset"
|
||||
version = "0.4.18"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "52dfc19153a48bde0cbd630453615c8151bce3a5adfac7a0aebfbf0a1e1f57e3"
|
||||
dependencies = [
|
||||
"aho-corasick",
|
||||
"bstr",
|
||||
"log",
|
||||
"regex-automata",
|
||||
"regex-syntax",
|
||||
]
|
||||
|
||||
[[package]]
|
||||
name = "gobject-sys"
|
||||
version = "0.20.10"
|
||||
@@ -2152,6 +2194,22 @@ dependencies = [
|
||||
"icu_properties",
|
||||
]
|
||||
|
||||
[[package]]
|
||||
name = "ignore"
|
||||
version = "0.4.28"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "2adf14691c72bcfc1058740436a35bdd3ae9c07d1a941ef00b749e9ea16aefa7"
|
||||
dependencies = [
|
||||
"crossbeam-deque",
|
||||
"globset",
|
||||
"log",
|
||||
"memchr",
|
||||
"regex-automata",
|
||||
"same-file",
|
||||
"walkdir",
|
||||
"winapi-util",
|
||||
]
|
||||
|
||||
[[package]]
|
||||
name = "indexmap"
|
||||
version = "2.14.0"
|
||||
@@ -2278,6 +2336,15 @@ dependencies = [
|
||||
"signature",
|
||||
]
|
||||
|
||||
[[package]]
|
||||
name = "kamadak-exif"
|
||||
version = "0.6.1"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "1130d80c7374efad55a117d715a3af9368f0fa7a2c54573afc15a188cd984837"
|
||||
dependencies = [
|
||||
"mutate_once",
|
||||
]
|
||||
|
||||
[[package]]
|
||||
name = "keccak"
|
||||
version = "0.1.6"
|
||||
@@ -2502,6 +2569,12 @@ dependencies = [
|
||||
"windows-sys 0.61.2",
|
||||
]
|
||||
|
||||
[[package]]
|
||||
name = "mutate_once"
|
||||
version = "0.1.2"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "13d2233c9842d08cfe13f9eac96e207ca6a2ea10b80259ebe8ad0268be27d2af"
|
||||
|
||||
[[package]]
|
||||
name = "nix"
|
||||
version = "0.29.0"
|
||||
@@ -2825,7 +2898,7 @@ dependencies = [
|
||||
|
||||
[[package]]
|
||||
name = "pdfs-cli"
|
||||
version = "0.2.2"
|
||||
version = "0.3.0"
|
||||
dependencies = [
|
||||
"anyhow",
|
||||
"clap 4.6.1",
|
||||
@@ -2840,9 +2913,10 @@ dependencies = [
|
||||
|
||||
[[package]]
|
||||
name = "pdfs-core"
|
||||
version = "0.2.2"
|
||||
version = "0.3.0"
|
||||
dependencies = [
|
||||
"directories",
|
||||
"ignore",
|
||||
"keyring",
|
||||
"proton-drive-rs",
|
||||
"proton-sdk",
|
||||
@@ -2857,7 +2931,7 @@ dependencies = [
|
||||
|
||||
[[package]]
|
||||
name = "pdfs-fuse"
|
||||
version = "0.2.2"
|
||||
version = "0.3.0"
|
||||
dependencies = [
|
||||
"fuser",
|
||||
"libc",
|
||||
@@ -2870,11 +2944,12 @@ dependencies = [
|
||||
|
||||
[[package]]
|
||||
name = "pdfs-gui"
|
||||
version = "0.2.2"
|
||||
version = "0.3.0"
|
||||
dependencies = [
|
||||
"anyhow",
|
||||
"async-channel",
|
||||
"gtk4",
|
||||
"kamadak-exif",
|
||||
"ksni",
|
||||
"libadwaita",
|
||||
"pdfs-core",
|
||||
@@ -3085,9 +3160,8 @@ dependencies = [
|
||||
|
||||
[[package]]
|
||||
name = "proton-drive-rs"
|
||||
version = "0.1.2"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "c9380c88a032e2ff079b2d5a6221ccc40ebeee3ebbb4ea60d4d65ef770e4ca8d"
|
||||
version = "0.1.4"
|
||||
source = "git+https://github.com/narrrl/proton-sdk-rs?branch=main#2e3c23803eecbdcb9f442c1e49229347f272596b"
|
||||
dependencies = [
|
||||
"async-trait",
|
||||
"base64",
|
||||
@@ -3110,9 +3184,8 @@ dependencies = [
|
||||
|
||||
[[package]]
|
||||
name = "proton-sdk"
|
||||
version = "0.1.2"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "1da22604d2568af2d15be6a45f23f8c5b95fa775bdac1540e833c2eafb587888"
|
||||
version = "0.1.4"
|
||||
source = "git+https://github.com/narrrl/proton-sdk-rs?branch=main#2e3c23803eecbdcb9f442c1e49229347f272596b"
|
||||
dependencies = [
|
||||
"aes-gcm",
|
||||
"async-trait",
|
||||
|
||||
+11
-4
@@ -3,16 +3,15 @@ resolver = "2"
|
||||
members = ["crates/pdfs-core", "crates/pdfs-fuse", "crates/pdfs-cli", "crates/pdfs-gui"]
|
||||
|
||||
[workspace.package]
|
||||
version = "0.2.2"
|
||||
version = "0.3.0"
|
||||
edition = "2024"
|
||||
license = "MIT"
|
||||
authors = ["Nils Pukropp <contact@narl.io>"]
|
||||
rust-version = "1.96"
|
||||
|
||||
[workspace.dependencies]
|
||||
# Local Proton SDK crates (path deps until published; switch to versioned later).
|
||||
proton-sdk = "0.1.2"
|
||||
proton-drive-rs = "0.1.2"
|
||||
proton-sdk = "0.1.4"
|
||||
proton-drive-rs = "0.1.4"
|
||||
|
||||
# Internal crates.
|
||||
pdfs-core = { path = "crates/pdfs-core" }
|
||||
@@ -26,6 +25,7 @@ anyhow = "1"
|
||||
tracing = "0.1"
|
||||
tracing-subscriber = { version = "0.3", features = ["env-filter"] }
|
||||
directories = "5"
|
||||
ignore = "0.4"
|
||||
sha2 = "0.10"
|
||||
rusqlite = { version = "0.32", features = ["bundled"] }
|
||||
keyring = { version = "3", features = ["sync-secret-service", "crypto-rust"] }
|
||||
@@ -33,3 +33,10 @@ clap = { version = "4", features = ["derive"] }
|
||||
rpassword = "7"
|
||||
fuser = "0.17"
|
||||
libc = "0.2"
|
||||
|
||||
[patch.crates-io]
|
||||
# Track the SDK's `main` on GitHub rather than a sibling checkout, so a clean
|
||||
# clone (and CI, and the PKGBUILD) builds without the working tree next door.
|
||||
# `cargo update -p proton-sdk -p proton-drive-rs` pulls newer pushes.
|
||||
proton-sdk = { git = "https://github.com/narrrl/proton-sdk-rs", branch = "main" }
|
||||
proton-drive-rs = { git = "https://github.com/narrrl/proton-sdk-rs", branch = "main" }
|
||||
@@ -8,6 +8,7 @@ A fast, unofficial Proton Drive client for Linux. This client features an advanc
|
||||
- **Command-Line Interface (CLI)**: Manage your drive, authenticate, and monitor sync status directly from the terminal.
|
||||
- **Non-Blocking GTK4 Desktop App**: Browse files, manage pins, and configure options through a modern, native GUI with a fully non-blocking asynchronous main loop.
|
||||
- **System Tray Integration**: Background indicator for status monitoring, quick actions, and fast sync controls.
|
||||
- **Fuzzy Search Launcher Prompt (HUD)**: A Google Drive-style search launcher (`pdfs-prompt`) for finding and opening files or folders instantly, ideal for binding to a system-wide hotkey.
|
||||
- **Secure Credential Storage**: Integrates with the system Secret Service (GNOME Keyring, KWallet, etc.) with smart in-memory credential caching to avoid UI thread blockages.
|
||||
- **Proton Photos Support**: Access your Proton Photos timeline, view thumbnails, and download backed-up media natively (available in the GUI as a navigation tab and via the CLI).
|
||||
|
||||
@@ -23,15 +24,15 @@ The client includes several optimizations designed for high efficiency, a low me
|
||||
|
||||
## Screenshots
|
||||
|
||||
### GUI Application
|
||||
### GUI Application & Launcher
|
||||
|
||||
| Login Screen | Account Overview |
|
||||
| Login Screen | Files Browser |
|
||||
| :---: | :---: |
|
||||
|  |  |
|
||||
|  |  |
|
||||
|
||||
| Files (Grid View) | Files (List View) |
|
||||
| Photos Timeline | Search Launcher Prompt |
|
||||
| :---: | :---: |
|
||||
|  |  |
|
||||
|  |  |
|
||||
|
||||
---
|
||||
|
||||
@@ -76,6 +77,7 @@ The compiled binaries will be available under `target/release/`:
|
||||
- `pdfs`: The CLI utility.
|
||||
- `pdfs-app`: The GTK4 application.
|
||||
- `pdfs-tray`: The tray status notifier.
|
||||
- `pdfs-prompt`: The launcher prompt for quick HUD search.
|
||||
|
||||
---
|
||||
|
||||
@@ -108,7 +110,7 @@ This project has a GitHub Actions CI workflow configured under `.github/workflow
|
||||
- Sets up the Rust compiler and caches build targets to speed up runs.
|
||||
- Compiles the workspace members in release mode.
|
||||
3. **Artifact Packaging**:
|
||||
- Generates a `.tar.gz` containing the raw binaries (`pdfs`, `pdfs-app`, `pdfs-tray`).
|
||||
- Generates a `.tar.gz` containing the raw binaries (`pdfs`, `pdfs-app`, `pdfs-tray`, `pdfs-prompt`).
|
||||
- Packs them into a Debian package (`.deb`).
|
||||
4. **Publishing**:
|
||||
- Creates a GitHub Release matching the pushed tag and uploads the `.deb` and `.tar.gz` packages as release assets.
|
||||
|
||||
+44
-38
@@ -119,6 +119,8 @@ enum Command {
|
||||
/// Destination folder path, inside the mountpoint or relative to it.
|
||||
parent: PathBuf,
|
||||
},
|
||||
/// Show the daemon's in-flight transfers (active uploads/downloads).
|
||||
Transfers,
|
||||
}
|
||||
|
||||
fn main() -> Result<()> {
|
||||
@@ -151,6 +153,7 @@ fn main() -> Result<()> {
|
||||
Command::Rm { path } => cmd_rm(path),
|
||||
Command::Mkdir { parent, name } => cmd_mkdir(parent, name),
|
||||
Command::Upload { file, parent } => cmd_upload(file, parent),
|
||||
Command::Transfers => cmd_transfers(),
|
||||
}
|
||||
}
|
||||
|
||||
@@ -248,7 +251,9 @@ fn cmd_daemon(mountpoint: Option<PathBuf>) -> Result<()> {
|
||||
std::thread::sleep(std::time::Duration::from_secs(2));
|
||||
}
|
||||
Err(e) => {
|
||||
tracing::error!(error = %e, "mount failed; retrying in 5s");
|
||||
// `{:#}` so the whole anyhow chain lands in the log — the top
|
||||
// context alone ("mount failed") never says *why*.
|
||||
tracing::error!(error = format!("{e:#}"), "mount failed; retrying in 5s");
|
||||
std::thread::sleep(std::time::Duration::from_secs(5));
|
||||
}
|
||||
}
|
||||
@@ -289,39 +294,11 @@ fn mount_once(mountpoint: Option<PathBuf>) -> Result<pdfs_fuse::MountOutcome> {
|
||||
.block_on(auth::resume_client())
|
||||
.context("resume session (run `pdfs login` first)")?;
|
||||
|
||||
// Persist rotated tokens for the daemon's lifetime. Proton refresh tokens are
|
||||
// single-use: the 401-refresh path swaps in a new access+refresh pair, and if
|
||||
// that pair is never written back the keyring keeps a now-dead refresh token,
|
||||
// so the next `pdfs mount` fails with `InvalidRefreshToken`. The session
|
||||
// shares its token store with the Drive client (one `Arc`-backed token store),
|
||||
// so both the poll below and the shutdown flush see every rotation.
|
||||
//
|
||||
// `ProtonApiSession` is `Clone` over that shared store, so the polling task
|
||||
// gets its own handle and we keep `session` here for a final flush. The poll
|
||||
// is a backstop for long-lived mounts; the flush after `mount` returns is what
|
||||
// makes the common case correct — a refresh shortly before a clean unmount
|
||||
// would otherwise be lost in the poll gap, leaving a dead token behind.
|
||||
let poll_session = session.clone();
|
||||
rt.spawn(async move {
|
||||
let mut last = poll_session.current_tokens().await;
|
||||
loop {
|
||||
tokio::time::sleep(std::time::Duration::from_secs(10)).await;
|
||||
let current = poll_session.current_tokens().await;
|
||||
if current.access_token == last.access_token
|
||||
&& current.refresh_token == last.refresh_token
|
||||
{
|
||||
continue;
|
||||
}
|
||||
match auth::persist(&poll_session).await {
|
||||
Ok(()) => {
|
||||
tracing::info!("persisted refreshed session tokens");
|
||||
last = current;
|
||||
}
|
||||
Err(e) => tracing::warn!(error = %e, "failed to persist refreshed tokens"),
|
||||
}
|
||||
}
|
||||
});
|
||||
|
||||
// Token updates are persisted automatically in real-time. The SDK session has
|
||||
// a token refresh callback registered during `resume_client` that writes updated
|
||||
// credentials back to the OS keyring immediately upon rotation. This prevents
|
||||
// the system keyring from holding stale refresh tokens even if the daemon crashes,
|
||||
// the system is suspended, or the network changes.
|
||||
let handle = rt.handle().clone();
|
||||
let result = pdfs_fuse::mount(
|
||||
client,
|
||||
@@ -333,11 +310,9 @@ fn mount_once(mountpoint: Option<PathBuf>) -> Result<pdfs_fuse::MountOutcome> {
|
||||
username,
|
||||
);
|
||||
|
||||
// Clean unmount: flush whatever tokens the store holds now so the keyring
|
||||
// always ends a session with the live refresh token, not a stale one. Runs
|
||||
// even if `mount` errored — a rotation may still have happened first.
|
||||
// Clean unmount check (as a best-effort final backup).
|
||||
if let Err(e) = rt.block_on(auth::persist(&session)) {
|
||||
tracing::warn!(error = %e, "failed to persist tokens on shutdown");
|
||||
tracing::debug!(error = %e, "no new tokens to persist on shutdown");
|
||||
}
|
||||
|
||||
let outcome = result.context("mount failed")?;
|
||||
@@ -366,6 +341,37 @@ fn cmd_unpin(path: PathBuf) -> Result<()> {
|
||||
Ok(())
|
||||
}
|
||||
|
||||
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 } => {
|
||||
for t in items {
|
||||
let arrow = match t.direction {
|
||||
TransferDirection::Download => "↓",
|
||||
TransferDirection::Upload => "↑",
|
||||
};
|
||||
let pct = if t.bytes_total > 0 {
|
||||
format!(
|
||||
"{:.0}%",
|
||||
100.0 * t.bytes_completed as f64 / t.bytes_total as f64
|
||||
)
|
||||
} else {
|
||||
"—".to_string()
|
||||
};
|
||||
let speed = t.speed_bytes_sec as f64 / 1_048_576.0;
|
||||
println!(
|
||||
"{arrow} {:>4} {:>8} / {:>8} {speed:>6.2} MiB/s {}",
|
||||
pct, t.bytes_completed, t.bytes_total, t.name
|
||||
);
|
||||
}
|
||||
}
|
||||
CtlResponse::Error { message } => bail!("{message}"),
|
||||
other => bail!("unexpected response: {other:?}"),
|
||||
}
|
||||
Ok(())
|
||||
}
|
||||
|
||||
fn cmd_pins() -> Result<()> {
|
||||
match control_request(CtlRequest::ListPins)? {
|
||||
CtlResponse::Pins { pins } if pins.is_empty() => println!("No pinned files."),
|
||||
|
||||
@@ -16,6 +16,7 @@ serde_json.workspace = true
|
||||
thiserror.workspace = true
|
||||
tracing.workspace = true
|
||||
directories.workspace = true
|
||||
ignore.workspace = true
|
||||
sha2.workspace = true
|
||||
rusqlite.workspace = true
|
||||
keyring.workspace = true
|
||||
@@ -84,6 +84,8 @@ pub async fn login(
|
||||
session.apply_second_factor_code(code.trim()).await?;
|
||||
}
|
||||
|
||||
register_refresh_handler(&session, password.to_owned());
|
||||
|
||||
save(&session, password).await?;
|
||||
Ok(())
|
||||
}
|
||||
@@ -137,6 +139,9 @@ pub fn logout() -> Result<()> {
|
||||
pub async fn resume_client() -> Result<(ProtonDriveClient, ProtonApiSession)> {
|
||||
let stored = load()?;
|
||||
let session = ProtonApiSession::resume(client_config(), stored.to_params())?;
|
||||
|
||||
register_refresh_handler(&session, stored.mailbox_password.clone());
|
||||
|
||||
let client = ProtonDriveClient::new(&session, stored.mailbox_password.clone().into_bytes());
|
||||
Ok((client, session))
|
||||
}
|
||||
@@ -151,3 +156,35 @@ pub async fn persist(session: &ProtonApiSession) -> Result<()> {
|
||||
let mailbox_password = load()?.mailbox_password;
|
||||
save(session, &mailbox_password).await
|
||||
}
|
||||
|
||||
fn register_refresh_handler(session: &ProtonApiSession, mailbox_password: String) {
|
||||
let session_id = session.session_id().as_str().to_owned();
|
||||
let username = session.username().to_owned();
|
||||
let user_id = session.user_id().as_str().to_owned();
|
||||
let scopes = session.scopes().to_vec();
|
||||
let password_mode = match session.password_mode() {
|
||||
PasswordMode::Single => 1,
|
||||
PasswordMode::Dual => 2,
|
||||
};
|
||||
|
||||
session.http().set_on_tokens_refreshed(move |tokens| {
|
||||
let stored = StoredSession {
|
||||
session_id: session_id.clone(),
|
||||
username: username.clone(),
|
||||
user_id: user_id.clone(),
|
||||
access_token: tokens.access_token,
|
||||
refresh_token: tokens.refresh_token,
|
||||
scopes: scopes.clone(),
|
||||
password_mode,
|
||||
mailbox_password: mailbox_password.clone(),
|
||||
};
|
||||
if let Ok(json) = serde_json::to_string(&stored)
|
||||
&& let Ok(entry) = keyring_entry() {
|
||||
if let Err(e) = entry.set_password(&json) {
|
||||
tracing::warn!(error = %e, "failed to auto-persist refreshed tokens in keyring");
|
||||
} else {
|
||||
tracing::info!("successfully auto-persisted refreshed tokens in keyring");
|
||||
}
|
||||
}
|
||||
});
|
||||
}
|
||||
@@ -142,6 +142,12 @@ impl AppDirs {
|
||||
self.state_dir().join("tray.sock")
|
||||
}
|
||||
|
||||
/// The user's home directory — the root the daemon indexes for local
|
||||
/// (non-Drive) file search. `None` if it cannot be resolved.
|
||||
pub fn home_dir(&self) -> Option<PathBuf> {
|
||||
directories::UserDirs::new().map(|u| u.home_dir().to_path_buf())
|
||||
}
|
||||
|
||||
/// Default mountpoint when the user does not pass one explicitly.
|
||||
pub fn default_mountpoint(&self) -> PathBuf {
|
||||
directories::UserDirs::new()
|
||||
|
||||
@@ -63,6 +63,10 @@ pub enum Request {
|
||||
/// Full-text search node names against the daemon's local metadata index.
|
||||
/// `limit` caps the number of hits returned. Replies with [`Response::SearchResults`].
|
||||
Search { query: String, limit: usize },
|
||||
/// Search the daemon's index of *local* (non-Drive) files on this machine.
|
||||
/// Independent of [`Request::Search`] so a front-end can fire both at once and
|
||||
/// render whichever lands first. Replies with [`Response::LocalResults`].
|
||||
SearchLocal { query: String, limit: usize },
|
||||
/// Rename a file or folder. `path` is mountpoint-relative; `new_name` is a
|
||||
/// single path component (no separators). Replies with [`Response::Ok`].
|
||||
Rename { path: String, new_name: String },
|
||||
@@ -89,6 +93,34 @@ 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.
|
||||
GetQueueStatus,
|
||||
}
|
||||
|
||||
/// Which way an active transfer is moving, in a [`TransferItem`].
|
||||
#[derive(Serialize, Deserialize, Debug, Clone, Copy, PartialEq, Eq)]
|
||||
pub enum TransferDirection {
|
||||
Download,
|
||||
Upload,
|
||||
}
|
||||
|
||||
/// One in-flight transfer in a [`Response::Transfers`] snapshot.
|
||||
#[derive(Serialize, Deserialize, Debug, Clone)]
|
||||
pub struct TransferItem {
|
||||
/// Node uid in `volume~link` form (empty for an upload whose uid isn't known
|
||||
/// until the draft is sealed).
|
||||
pub uid: String,
|
||||
/// File name being transferred.
|
||||
pub name: String,
|
||||
pub direction: TransferDirection,
|
||||
/// Bytes moved so far.
|
||||
pub bytes_completed: u64,
|
||||
/// Total bytes expected, or `0` when unknown (indeterminate progress).
|
||||
pub bytes_total: u64,
|
||||
/// Average throughput since the transfer began, bytes per second.
|
||||
pub speed_bytes_sec: u64,
|
||||
}
|
||||
|
||||
/// One entry in a [`Request::ListDir`] listing.
|
||||
@@ -134,6 +166,20 @@ pub struct SearchHit {
|
||||
pub uid: String,
|
||||
}
|
||||
|
||||
/// One hit in a [`Request::SearchLocal`] result: a file on this machine, outside
|
||||
/// Proton Drive. Unlike [`SearchHit`] there is no uid or pin state — the file is
|
||||
/// already local, so the front-end opens `path` directly.
|
||||
#[derive(Serialize, Deserialize, Debug, Clone)]
|
||||
pub struct LocalHit {
|
||||
pub name: String,
|
||||
/// Absolute path on this machine.
|
||||
pub path: String,
|
||||
pub is_dir: bool,
|
||||
pub size: u64,
|
||||
/// Modification time, epoch seconds.
|
||||
pub modified: i64,
|
||||
}
|
||||
|
||||
/// One photo in a [`Request::PhotosTimeline`] page.
|
||||
#[derive(Serialize, Deserialize, Debug, Clone)]
|
||||
pub struct PhotoItem {
|
||||
@@ -178,6 +224,12 @@ pub enum Response {
|
||||
FilePath { path: String },
|
||||
/// Full-text search results (reply to [`Request::Search`]).
|
||||
SearchResults { hits: Vec<SearchHit> },
|
||||
/// Local-file search results (reply to [`Request::SearchLocal`]). `indexing`
|
||||
/// 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> },
|
||||
/// The request failed.
|
||||
Error { message: String },
|
||||
}
|
||||
|
||||
+264
-4
@@ -18,9 +18,10 @@ use proton_drive_rs::{Node, NodeKind};
|
||||
use rusqlite::{Connection, OptionalExtension, params};
|
||||
|
||||
use crate::Result;
|
||||
use crate::localindex::LocalEntry;
|
||||
|
||||
/// Current schema version. Bump on every forward migration added below.
|
||||
const SCHEMA_VERSION: i64 = 5;
|
||||
const SCHEMA_VERSION: i64 = 6;
|
||||
|
||||
/// Below this length the trigram tokenizer indexes nothing (it needs 3-char
|
||||
/// grams), so short queries fall back to a `LIKE` scan over `nodes.name`.
|
||||
@@ -119,6 +120,9 @@ impl Db {
|
||||
if current < 5 {
|
||||
tx.execute_batch(MIGRATION_V5)?;
|
||||
}
|
||||
if current < 6 {
|
||||
tx.execute_batch(MIGRATION_V6)?;
|
||||
}
|
||||
tx.execute(
|
||||
"INSERT INTO sync_state (key, value) VALUES ('schema_version', ?1)
|
||||
ON CONFLICT(key) DO UPDATE SET value = excluded.value",
|
||||
@@ -213,9 +217,13 @@ impl Db {
|
||||
)?;
|
||||
collect_pairs(stmt.query_map(params![pat, limit as i64], pair)?)?
|
||||
} else {
|
||||
// Quote the user input as a single FTS phrase so its characters are
|
||||
// matched literally, never parsed as FTS query syntax.
|
||||
let phrase = format!("\"{}\"", query.replace('"', "\"\""));
|
||||
// Escape double quotes and quote each term, then combine with AND so
|
||||
// all terms must match but can appear in any order or position.
|
||||
let phrase = query
|
||||
.split_whitespace()
|
||||
.map(|word| format!("\"{}\"", word.replace('"', "\"\"")))
|
||||
.collect::<Vec<_>>()
|
||||
.join(" AND ");
|
||||
let mut stmt = conn.prepare(
|
||||
"SELECT n.node_json, n.uid
|
||||
FROM nodes_fts f JOIN nodes n ON n.uid = f.uid
|
||||
@@ -522,6 +530,161 @@ impl Db {
|
||||
}
|
||||
Ok(out)
|
||||
}
|
||||
|
||||
// ---- local (non-Drive) file index (schema v6) ----
|
||||
|
||||
/// Open a new local-index scan generation. Rows written under it survive
|
||||
/// [`local_finish_scan`](Self::local_finish_scan); rows still carrying an
|
||||
/// older generation are pruned there as "no longer on disk".
|
||||
pub fn local_begin_scan(&self) -> Result<i64> {
|
||||
let conn = self.conn.lock().unwrap();
|
||||
let current: i64 = conn
|
||||
.query_row(
|
||||
"SELECT value FROM sync_state WHERE key = 'local_scan_gen'",
|
||||
[],
|
||||
|row| row.get::<_, String>(0),
|
||||
)
|
||||
.optional()?
|
||||
.and_then(|v| v.parse().ok())
|
||||
.unwrap_or(0);
|
||||
let next = current + 1;
|
||||
conn.execute(
|
||||
"INSERT INTO sync_state (key, value) VALUES ('local_scan_gen', ?1)
|
||||
ON CONFLICT(key) DO UPDATE SET value = excluded.value",
|
||||
[next.to_string()],
|
||||
)?;
|
||||
Ok(next)
|
||||
}
|
||||
|
||||
/// Write one batch of walked entries under scan generation `generation`. The
|
||||
/// FTS index is *not* touched here — it is rebuilt once in
|
||||
/// [`local_finish_scan`](Self::local_finish_scan).
|
||||
pub fn local_upsert_batch(&self, generation: i64, entries: &[LocalEntry]) -> Result<()> {
|
||||
let mut conn = self.conn.lock().unwrap();
|
||||
let tx = conn.transaction()?;
|
||||
{
|
||||
let mut stmt = tx.prepare_cached(
|
||||
"INSERT INTO local_files (path, name, is_dir, size, mtime, scan_gen)
|
||||
VALUES (?1, ?2, ?3, ?4, ?5, ?6)
|
||||
ON CONFLICT(path) DO UPDATE SET
|
||||
name = excluded.name,
|
||||
is_dir = excluded.is_dir,
|
||||
size = excluded.size,
|
||||
mtime = excluded.mtime,
|
||||
scan_gen = excluded.scan_gen",
|
||||
)?;
|
||||
for e in entries {
|
||||
stmt.execute(params![
|
||||
e.path,
|
||||
e.name,
|
||||
e.is_dir as i64,
|
||||
e.size,
|
||||
e.mtime,
|
||||
generation
|
||||
])?;
|
||||
}
|
||||
}
|
||||
tx.commit()?;
|
||||
Ok(())
|
||||
}
|
||||
|
||||
/// Close scan generation `generation`: drop every row an older scan wrote
|
||||
/// (those paths are gone from disk), rebuild the FTS index over what remains,
|
||||
/// and stamp the completion time. Returns the number of indexed entries.
|
||||
pub fn local_finish_scan(&self, generation: i64, finished_at: i64) -> Result<i64> {
|
||||
let mut conn = self.conn.lock().unwrap();
|
||||
let tx = conn.transaction()?;
|
||||
tx.execute(
|
||||
"DELETE FROM local_files WHERE scan_gen != ?1",
|
||||
params![generation],
|
||||
)?;
|
||||
tx.execute_batch("INSERT INTO local_fts(local_fts) VALUES('rebuild');")?;
|
||||
tx.execute(
|
||||
"INSERT INTO sync_state (key, value) VALUES ('local_indexed_at', ?1)
|
||||
ON CONFLICT(key) DO UPDATE SET value = excluded.value",
|
||||
[finished_at.to_string()],
|
||||
)?;
|
||||
let count: i64 = tx.query_row("SELECT COUNT(*) FROM local_files", [], |r| r.get(0))?;
|
||||
tx.commit()?;
|
||||
Ok(count)
|
||||
}
|
||||
|
||||
/// When the last local scan completed (epoch seconds), or `None` if the index
|
||||
/// has never been built. The daemon uses this to decide whether a fresh mount
|
||||
/// needs an immediate rescan or can serve the existing index.
|
||||
pub fn local_indexed_at(&self) -> Result<Option<i64>> {
|
||||
let conn = self.conn.lock().unwrap();
|
||||
Ok(conn
|
||||
.query_row(
|
||||
"SELECT value FROM sync_state WHERE key = 'local_indexed_at'",
|
||||
[],
|
||||
|row| row.get::<_, String>(0),
|
||||
)
|
||||
.optional()?
|
||||
.and_then(|v| v.parse().ok()))
|
||||
}
|
||||
|
||||
/// Substring search over indexed local file names, newest-modified first
|
||||
/// within a relevance tier. Mirrors [`search`](Self::search): the trigram
|
||||
/// index handles queries of at least [`TRIGRAM_MIN`] chars, shorter ones fall
|
||||
/// back to a `LIKE` scan.
|
||||
pub fn search_local(&self, query: &str, limit: usize) -> Result<Vec<LocalFileHit>> {
|
||||
let query = query.trim();
|
||||
if query.is_empty() {
|
||||
return Ok(Vec::new());
|
||||
}
|
||||
let conn = self.conn.lock().unwrap();
|
||||
let mut stmt;
|
||||
let rows = if query.chars().count() < TRIGRAM_MIN {
|
||||
let pat = format!("%{}%", like_escape(query));
|
||||
stmt = conn.prepare(
|
||||
"SELECT path, name, is_dir, size, mtime FROM local_files
|
||||
WHERE name LIKE ?1 ESCAPE '\\'
|
||||
ORDER BY mtime DESC LIMIT ?2",
|
||||
)?;
|
||||
stmt.query_map(params![pat, limit as i64], local_hit)?
|
||||
} else {
|
||||
let phrase = query
|
||||
.split_whitespace()
|
||||
.map(|word| format!("\"{}\"", word.replace('"', "\"\"")))
|
||||
.collect::<Vec<_>>()
|
||||
.join(" AND ");
|
||||
stmt = conn.prepare(
|
||||
"SELECT f.path, f.name, f.is_dir, f.size, f.mtime
|
||||
FROM local_fts x JOIN local_files f ON f.id = x.rowid
|
||||
WHERE x.name MATCH ?1
|
||||
ORDER BY x.rank LIMIT ?2",
|
||||
)?;
|
||||
stmt.query_map(params![phrase, limit as i64], local_hit)?
|
||||
};
|
||||
let mut out = Vec::new();
|
||||
for r in rows {
|
||||
out.push(r?);
|
||||
}
|
||||
Ok(out)
|
||||
}
|
||||
}
|
||||
|
||||
/// One hit from [`Db::search_local`]: an indexed file on the machine itself, not
|
||||
/// in Drive. `path` is absolute.
|
||||
pub struct LocalFileHit {
|
||||
pub path: String,
|
||||
pub name: String,
|
||||
pub is_dir: bool,
|
||||
pub size: i64,
|
||||
pub mtime: i64,
|
||||
}
|
||||
|
||||
/// Row mapper for [`Db::search_local`]'s two query paths, which select the same
|
||||
/// columns in the same order.
|
||||
fn local_hit(row: &rusqlite::Row) -> rusqlite::Result<LocalFileHit> {
|
||||
Ok(LocalFileHit {
|
||||
path: row.get(0)?,
|
||||
name: row.get(1)?,
|
||||
is_dir: row.get::<_, i64>(2)? != 0,
|
||||
size: row.get(3)?,
|
||||
mtime: row.get(4)?,
|
||||
})
|
||||
}
|
||||
|
||||
/// Row mapper for the `(node_json, uid)` pairs both search paths return.
|
||||
@@ -660,6 +823,32 @@ CREATE TABLE pins (
|
||||
);
|
||||
";
|
||||
|
||||
/// Schema v6: the index of *local* (non-Drive) files, so the launcher prompt can
|
||||
/// search the machine alongside Drive. `local_files` is keyed by absolute path;
|
||||
/// `scan_gen` stamps the scan that last saw a row, so a rescan prunes vanished
|
||||
/// files with one `DELETE` instead of diffing.
|
||||
///
|
||||
/// `local_fts` is an *external-content* FTS5 index over `local_files.name`: the
|
||||
/// text lives once in the base table and the index is rebuilt in bulk at the end
|
||||
/// of a scan (`INSERT INTO local_fts(local_fts) VALUES('rebuild')`). That is far
|
||||
/// cheaper than the delete-then-insert-per-row dance `nodes_fts` needs, because
|
||||
/// a scan rewrites most rows at once rather than trickling single updates.
|
||||
const MIGRATION_V6: &str = "
|
||||
CREATE TABLE local_files (
|
||||
id INTEGER PRIMARY KEY,
|
||||
path TEXT NOT NULL UNIQUE,
|
||||
name TEXT NOT NULL,
|
||||
is_dir INTEGER NOT NULL,
|
||||
size INTEGER NOT NULL DEFAULT 0,
|
||||
mtime INTEGER NOT NULL DEFAULT 0,
|
||||
scan_gen INTEGER NOT NULL DEFAULT 0
|
||||
);
|
||||
|
||||
CREATE VIRTUAL TABLE local_fts USING fts5(
|
||||
name, content='local_files', content_rowid='id', tokenize='trigram'
|
||||
);
|
||||
";
|
||||
|
||||
#[cfg(test)]
|
||||
mod tests {
|
||||
use super::*;
|
||||
@@ -784,6 +973,62 @@ mod tests {
|
||||
);
|
||||
}
|
||||
|
||||
fn local(path: &str, name: &str, is_dir: bool) -> LocalEntry {
|
||||
LocalEntry {
|
||||
path: path.into(),
|
||||
name: name.into(),
|
||||
is_dir,
|
||||
size: 10,
|
||||
mtime: 5,
|
||||
}
|
||||
}
|
||||
|
||||
/// A local scan is searchable by substring, and a *later* scan prunes the
|
||||
/// paths it no longer sees — including out of the FTS index, so a deleted
|
||||
/// file cannot keep surfacing in the prompt.
|
||||
#[test]
|
||||
fn local_scan_indexes_then_prunes_stale_paths() {
|
||||
let db = Db::open_in_memory().unwrap();
|
||||
|
||||
let gen1 = db.local_begin_scan().unwrap();
|
||||
db.local_upsert_batch(
|
||||
gen1,
|
||||
&[
|
||||
local("/home/u/docs/report.pdf", "report.pdf", false),
|
||||
local("/home/u/docs/notes.md", "notes.md", false),
|
||||
],
|
||||
)
|
||||
.unwrap();
|
||||
assert_eq!(db.local_finish_scan(gen1, 1_000).unwrap(), 2);
|
||||
|
||||
// Trigram index gives substring (not just prefix) matches.
|
||||
let hits = db.search_local("port", 10).unwrap();
|
||||
assert_eq!(hits.len(), 1);
|
||||
assert_eq!(hits[0].path, "/home/u/docs/report.pdf");
|
||||
assert!(!hits[0].is_dir);
|
||||
assert_eq!(db.local_indexed_at().unwrap(), Some(1_000));
|
||||
|
||||
// Second scan sees only notes.md → report.pdf is gone from disk.
|
||||
let gen2 = db.local_begin_scan().unwrap();
|
||||
db.local_upsert_batch(gen2, &[local("/home/u/docs/notes.md", "notes.md", false)])
|
||||
.unwrap();
|
||||
assert_eq!(db.local_finish_scan(gen2, 2_000).unwrap(), 1);
|
||||
assert!(db.search_local("report", 10).unwrap().is_empty());
|
||||
assert_eq!(db.search_local("notes", 10).unwrap().len(), 1);
|
||||
}
|
||||
|
||||
/// Queries below the trigram minimum still match, via the `LIKE` fallback.
|
||||
#[test]
|
||||
fn local_search_short_query_like_fallback() {
|
||||
let db = Db::open_in_memory().unwrap();
|
||||
let g = db.local_begin_scan().unwrap();
|
||||
db.local_upsert_batch(g, &[local("/home/u/a.txt", "a.txt", false)])
|
||||
.unwrap();
|
||||
db.local_finish_scan(g, 1).unwrap();
|
||||
assert_eq!(db.search_local("a", 10).unwrap().len(), 1);
|
||||
assert!(db.search_local("", 10).unwrap().is_empty());
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn event_cursor_roundtrip() {
|
||||
let db = Db::open_in_memory().unwrap();
|
||||
@@ -805,6 +1050,13 @@ mod tests {
|
||||
db.upsert_node(&file("f1", "docs", "report.pdf", 1))
|
||||
.unwrap();
|
||||
db.upsert_node(&file("f2", "root", "notes.txt", 2)).unwrap();
|
||||
db.upsert_node(&file(
|
||||
"f3",
|
||||
"root",
|
||||
"Rampage Open Air 2026 - order 166765244.pdf",
|
||||
3,
|
||||
))
|
||||
.unwrap();
|
||||
|
||||
// Substring match (trigram), not just prefix.
|
||||
let hits = db.search("port", 10).unwrap();
|
||||
@@ -816,6 +1068,14 @@ mod tests {
|
||||
// Top-level file → bare name.
|
||||
let hits = db.search("notes", 10).unwrap();
|
||||
assert_eq!(hits[0].path, "notes.txt");
|
||||
|
||||
// Multi-term FTS5 query matching (out of order, separated terms)
|
||||
let hits = db.search("rampage 2026", 10).unwrap();
|
||||
assert_eq!(hits.len(), 1);
|
||||
assert_eq!(
|
||||
hits[0].node.name,
|
||||
"Rampage Open Air 2026 - order 166765244.pdf"
|
||||
);
|
||||
}
|
||||
|
||||
#[test]
|
||||
|
||||
@@ -8,6 +8,7 @@ pub mod config;
|
||||
pub mod control;
|
||||
pub mod db;
|
||||
pub mod error;
|
||||
pub mod localindex;
|
||||
pub mod service;
|
||||
|
||||
pub use error::{Error, Result};
|
||||
|
||||
@@ -0,0 +1,224 @@
|
||||
//! Filesystem walker behind the daemon's index of *local* (non-Drive) files.
|
||||
//!
|
||||
//! The launcher prompt searches Proton Drive and the machine's own files side by
|
||||
//! side. Drive names come from the `nodes` FTS index; local names come from the
|
||||
//! `local_files` index this module feeds. The daemon runs [`scan`] on a
|
||||
//! background thread — never on a FUSE or control-socket thread — and streams
|
||||
//! batches into [`crate::db::Db::local_upsert_batch`].
|
||||
//!
|
||||
//! Walking is deliberately shallow on cost: we stat each entry once (the walker
|
||||
//! already has the `DirEntry` metadata) and skip the directories that dominate a
|
||||
//! home directory's inode count without ever being interesting to search
|
||||
//! (`node_modules`, `target`, caches, VCS internals). The Drive mountpoint is
|
||||
//! always excluded — walking it would fault in every remote node through FUSE.
|
||||
|
||||
use std::path::{Path, PathBuf};
|
||||
use std::time::UNIX_EPOCH;
|
||||
|
||||
use ignore::{WalkBuilder, WalkState};
|
||||
|
||||
/// Directory names skipped anywhere in the tree. These hold build artefacts,
|
||||
/// dependency trees, and caches: high inode count, no search value.
|
||||
const SKIP_DIRS: &[&str] = &[
|
||||
"node_modules",
|
||||
"target",
|
||||
"__pycache__",
|
||||
"venv",
|
||||
".venv",
|
||||
"vendor",
|
||||
"dist-newstyle",
|
||||
"Trash",
|
||||
];
|
||||
|
||||
/// Hard cap on indexed entries, so a pathological home directory cannot grow the
|
||||
/// database without bound. Reaching it stops the walk early.
|
||||
const MAX_ENTRIES: usize = 400_000;
|
||||
|
||||
/// Rows handed to the sink at a time. Large enough that the per-transaction cost
|
||||
/// amortises, small enough that the writer lock is released often (FUSE
|
||||
/// write-through shares the same connection).
|
||||
const BATCH: usize = 2_000;
|
||||
|
||||
/// One indexed local file or directory.
|
||||
#[derive(Debug, Clone)]
|
||||
pub struct LocalEntry {
|
||||
/// Absolute path.
|
||||
pub path: String,
|
||||
/// Final path component.
|
||||
pub name: String,
|
||||
pub is_dir: bool,
|
||||
/// Size in bytes (0 for directories).
|
||||
pub size: i64,
|
||||
/// Modification time, epoch seconds.
|
||||
pub mtime: i64,
|
||||
}
|
||||
|
||||
/// Walk `roots` in parallel, calling `sink` with batches of entries.
|
||||
///
|
||||
/// Hidden files and anything under `excludes` (the Drive mountpoint, our own
|
||||
/// state/cache dirs) are skipped, as are the [`SKIP_DIRS`] names at any depth.
|
||||
/// Symlinks are not followed, so a link loop cannot hang the scan. Returns the
|
||||
/// number of entries handed to `sink`.
|
||||
///
|
||||
/// `sink` is called from the walker's worker threads, serialized by an internal
|
||||
/// mutex; it must not block for long.
|
||||
pub fn scan(
|
||||
roots: &[PathBuf],
|
||||
excludes: &[PathBuf],
|
||||
sink: impl FnMut(Vec<LocalEntry>) + Send,
|
||||
) -> usize {
|
||||
let Some((first, rest)) = roots.split_first() else {
|
||||
return 0;
|
||||
};
|
||||
|
||||
let mut builder = WalkBuilder::new(first);
|
||||
for root in rest {
|
||||
builder.add(root);
|
||||
}
|
||||
let excludes: Vec<PathBuf> = excludes.to_vec();
|
||||
builder
|
||||
// `standard_filters` would honour .gitignore/.ignore files: a source
|
||||
// tree's ignored-but-present files are still files the user may want to
|
||||
// find, so only the hidden filter stays on.
|
||||
.standard_filters(false)
|
||||
.hidden(true)
|
||||
.follow_links(false)
|
||||
.same_file_system(true)
|
||||
.threads(std::thread::available_parallelism().map_or(4, |n| n.get().min(8)))
|
||||
.filter_entry(move |entry| {
|
||||
let path = entry.path();
|
||||
if excludes.iter().any(|e| path.starts_with(e)) {
|
||||
return false;
|
||||
}
|
||||
!matches!(entry.file_name().to_str(), Some(name) if SKIP_DIRS.contains(&name))
|
||||
});
|
||||
|
||||
let state = std::sync::Mutex::new(SinkState {
|
||||
buf: Vec::with_capacity(BATCH),
|
||||
total: 0,
|
||||
sink,
|
||||
});
|
||||
|
||||
builder.build_parallel().run(|| {
|
||||
Box::new(|result| {
|
||||
let Ok(entry) = result else {
|
||||
return WalkState::Continue;
|
||||
};
|
||||
// Depth 0 is a root itself; indexing it adds nothing to search.
|
||||
if entry.depth() == 0 {
|
||||
return WalkState::Continue;
|
||||
}
|
||||
let Some(local) = to_entry(&entry) else {
|
||||
return WalkState::Continue;
|
||||
};
|
||||
|
||||
let mut state = state.lock().unwrap();
|
||||
state.push(local);
|
||||
if state.total >= MAX_ENTRIES {
|
||||
return WalkState::Quit;
|
||||
}
|
||||
WalkState::Continue
|
||||
})
|
||||
});
|
||||
|
||||
let mut state = state.into_inner().unwrap();
|
||||
state.flush();
|
||||
state.total
|
||||
}
|
||||
|
||||
/// Batching buffer shared by the walker's worker threads.
|
||||
struct SinkState<F: FnMut(Vec<LocalEntry>)> {
|
||||
buf: Vec<LocalEntry>,
|
||||
total: usize,
|
||||
sink: F,
|
||||
}
|
||||
|
||||
impl<F: FnMut(Vec<LocalEntry>)> SinkState<F> {
|
||||
fn push(&mut self, entry: LocalEntry) {
|
||||
self.buf.push(entry);
|
||||
self.total += 1;
|
||||
if self.buf.len() >= BATCH {
|
||||
self.flush();
|
||||
}
|
||||
}
|
||||
|
||||
fn flush(&mut self) {
|
||||
if !self.buf.is_empty() {
|
||||
let batch = std::mem::take(&mut self.buf);
|
||||
(self.sink)(batch);
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
/// Convert a walker entry into a [`LocalEntry`], reusing the metadata the walker
|
||||
/// already fetched. Entries with non-UTF-8 paths are dropped: the index (and the
|
||||
/// JSON wire format) is UTF-8 only.
|
||||
fn to_entry(entry: &ignore::DirEntry) -> Option<LocalEntry> {
|
||||
let path = entry.path().to_str()?.to_string();
|
||||
let name = entry.file_name().to_str()?.to_string();
|
||||
let meta = entry.metadata().ok()?;
|
||||
let is_dir = meta.is_dir();
|
||||
let mtime = meta
|
||||
.modified()
|
||||
.ok()
|
||||
.and_then(|t| t.duration_since(UNIX_EPOCH).ok())
|
||||
.map_or(0, |d| d.as_secs() as i64);
|
||||
Some(LocalEntry {
|
||||
path,
|
||||
name,
|
||||
is_dir,
|
||||
size: if is_dir { 0 } else { meta.len() as i64 },
|
||||
mtime,
|
||||
})
|
||||
}
|
||||
|
||||
/// The paths a scan should never descend into: the Drive mountpoint (walking it
|
||||
/// would fault every remote node in through FUSE) plus our own state and cache
|
||||
/// dirs, which hold blobs no user searches for by name.
|
||||
pub fn default_excludes(mountpoint: &Path, state_dir: &Path, cache_dir: &Path) -> Vec<PathBuf> {
|
||||
vec![
|
||||
mountpoint.to_path_buf(),
|
||||
state_dir.to_path_buf(),
|
||||
cache_dir.to_path_buf(),
|
||||
]
|
||||
}
|
||||
|
||||
#[cfg(test)]
|
||||
mod tests {
|
||||
use super::*;
|
||||
|
||||
/// A scan indexes plain files, skips the excluded subtree and the junk dirs,
|
||||
/// and reports names/paths the search index can match on.
|
||||
#[test]
|
||||
fn scan_indexes_files_and_honours_excludes() {
|
||||
let tmp = std::env::temp_dir().join(format!("pdfs-scan-{}", std::process::id()));
|
||||
let _ = std::fs::remove_dir_all(&tmp);
|
||||
std::fs::create_dir_all(tmp.join("docs")).unwrap();
|
||||
std::fs::create_dir_all(tmp.join("node_modules/pkg")).unwrap();
|
||||
std::fs::create_dir_all(tmp.join("mnt")).unwrap();
|
||||
std::fs::write(tmp.join("docs/report.pdf"), b"x").unwrap();
|
||||
std::fs::write(tmp.join("node_modules/pkg/index.js"), b"x").unwrap();
|
||||
std::fs::write(tmp.join("mnt/remote.txt"), b"x").unwrap();
|
||||
|
||||
let mut got = Vec::new();
|
||||
let n = scan(std::slice::from_ref(&tmp), &[tmp.join("mnt")], |batch| {
|
||||
got.extend(batch)
|
||||
});
|
||||
|
||||
let names: Vec<&str> = got.iter().map(|e| e.name.as_str()).collect();
|
||||
assert_eq!(n, got.len());
|
||||
assert!(names.contains(&"report.pdf"));
|
||||
assert!(names.contains(&"docs"));
|
||||
// Junk dir pruned, mountpoint-style exclude pruned.
|
||||
assert!(!names.contains(&"index.js"));
|
||||
assert!(!names.contains(&"node_modules"));
|
||||
assert!(!names.contains(&"remote.txt"));
|
||||
|
||||
let report = got.iter().find(|e| e.name == "report.pdf").unwrap();
|
||||
assert!(!report.is_dir);
|
||||
assert_eq!(report.size, 1);
|
||||
assert!(report.path.ends_with("docs/report.pdf"));
|
||||
|
||||
std::fs::remove_dir_all(&tmp).unwrap();
|
||||
}
|
||||
}
|
||||
+230
-20
@@ -38,6 +38,7 @@ use std::io::{BufRead, BufReader, Write as _};
|
||||
use std::os::unix::fs::FileExt;
|
||||
use std::os::unix::net::{UnixListener, UnixStream};
|
||||
use std::path::{Component, Path, PathBuf};
|
||||
use std::sync::atomic::{AtomicBool, Ordering};
|
||||
use std::sync::{Arc, Mutex};
|
||||
use std::time::{Duration, Instant, SystemTime, UNIX_EPOCH};
|
||||
|
||||
@@ -51,15 +52,21 @@ use fuser::{
|
||||
use pdfs_core::cache::{BLOCK_SIZE, ContentCache};
|
||||
use pdfs_core::config::AppDirs;
|
||||
use pdfs_core::control::{
|
||||
DirEntry, PhotoItem, Request as CtlRequest, Response as CtlResponse, SearchHit,
|
||||
DirEntry, LocalHit, PhotoItem, Request as CtlRequest, Response as CtlResponse, SearchHit,
|
||||
TransferDirection,
|
||||
};
|
||||
use pdfs_core::db::{Db, StoredNode};
|
||||
use pdfs_core::localindex;
|
||||
use proton_drive_rs::proton_sdk::error::ProtonError;
|
||||
use proton_drive_rs::proton_sdk::ids::{DriveEventId, LinkId, NodeUid, VolumeId};
|
||||
use proton_drive_rs::{
|
||||
DriveEvent, DriveEventScopeId, Node, NodeKind, PhotosTimelineItem, ProtonDriveClient,
|
||||
ProtonPhotosClient, ThumbnailType,
|
||||
};
|
||||
|
||||
mod transfers;
|
||||
use tracing::{debug, error, info, warn};
|
||||
use transfers::{CountingReader, CountingWriter, TransferRegistry};
|
||||
|
||||
/// Attribute/entry cache lifetime handed back to the kernel. Long because the
|
||||
/// Phase 2 event poller actively invalidates changed inodes; without a remote
|
||||
@@ -73,6 +80,15 @@ const POLL_INTERVAL: Duration = Duration::from_secs(10);
|
||||
/// in [`Core`] and serve every page from memory until it goes stale.
|
||||
const TIMELINE_TTL: Duration = Duration::from_secs(60);
|
||||
|
||||
/// How stale the local-file index may get before the background scanner rebuilds
|
||||
/// it. A rescan is a full walk of `$HOME`, so this trades index freshness against
|
||||
/// disk churn; the scanner also always rebuilds once per daemon start when the
|
||||
/// index is older than this.
|
||||
const LOCAL_INDEX_TTL: Duration = Duration::from_secs(15 * 60);
|
||||
|
||||
/// How often the scanner thread wakes to check whether the index went stale.
|
||||
const LOCAL_INDEX_CHECK: Duration = Duration::from_secs(60);
|
||||
|
||||
/// The FUSE root inode is always 1.
|
||||
const ROOT_INO: u64 = 1;
|
||||
|
||||
@@ -293,6 +309,12 @@ struct Core {
|
||||
/// requests slice memory instead of re-fetching the whole timeline per page.
|
||||
/// `None` until first fetched; refreshed once older than [`TIMELINE_TTL`].
|
||||
timeline: Arc<Mutex<Option<TimelineCache>>>,
|
||||
/// In-flight upload/download progress, served to `GetQueueStatus`. Shared
|
||||
/// across the FUSE session and the control-socket task.
|
||||
transfers: Arc<TransferRegistry>,
|
||||
/// True while the background scanner is rebuilding the local-file index, so
|
||||
/// `SearchLocal` can tell a front-end "still indexing" apart from "no match".
|
||||
indexing: Arc<AtomicBool>,
|
||||
}
|
||||
|
||||
/// The whole photos timeline plus when it was fetched, for TTL freshness.
|
||||
@@ -681,6 +703,17 @@ impl Core {
|
||||
error!(%uid, error = %e, "reopen scratch file failed");
|
||||
Errno::EIO
|
||||
})?;
|
||||
let name = {
|
||||
let st = self.state.lock().unwrap();
|
||||
st.entries
|
||||
.get(&ino)
|
||||
.map(|e| e.node.name.clone())
|
||||
.unwrap_or_default()
|
||||
};
|
||||
let guard = self
|
||||
.transfers
|
||||
.begin(name, uid.to_string(), TransferDirection::Upload, len);
|
||||
let reader = CountingReader::new(reader, &guard);
|
||||
self.rt
|
||||
.block_on(self.client.upload_new_revision_from(
|
||||
&uid,
|
||||
@@ -693,6 +726,7 @@ impl Core {
|
||||
error!(%uid, error = %e, "upload new revision failed");
|
||||
Errno::EIO
|
||||
})?;
|
||||
drop(guard);
|
||||
let now = now_secs();
|
||||
{
|
||||
let mut st = self.state.lock().unwrap();
|
||||
@@ -719,6 +753,47 @@ impl Core {
|
||||
Ok(())
|
||||
}
|
||||
|
||||
/// Download a whole file's plaintext, registering the transfer so
|
||||
/// `GetQueueStatus` can report its progress. `total` is the expected size
|
||||
/// (`0` if unknown). Streams through [`download_file_to`] so each block ticks
|
||||
/// the progress counter.
|
||||
///
|
||||
/// [`download_file_to`]: ProtonDriveClient::download_file_to
|
||||
fn download_file_tracked(
|
||||
&self,
|
||||
uid: &NodeUid,
|
||||
name: &str,
|
||||
total: u64,
|
||||
) -> std::result::Result<Vec<u8>, ProtonError> {
|
||||
let guard = self
|
||||
.transfers
|
||||
.begin(name, uid.to_string(), TransferDirection::Download, total);
|
||||
let mut out = CountingWriter::new(Vec::with_capacity(total as usize), &guard);
|
||||
self.rt
|
||||
.block_on(self.client.download_file_to(uid, &mut out))?;
|
||||
Ok(out.into_inner())
|
||||
}
|
||||
|
||||
/// Like [`download_file_tracked`] for a photo, streaming through the photos
|
||||
/// client's [`download_photo_to`].
|
||||
///
|
||||
/// [`download_file_tracked`]: Core::download_file_tracked
|
||||
/// [`download_photo_to`]: ProtonPhotosClient::download_photo_to
|
||||
fn download_photo_tracked(
|
||||
&self,
|
||||
photos: &ProtonPhotosClient,
|
||||
uid: &NodeUid,
|
||||
name: &str,
|
||||
total: u64,
|
||||
) -> std::result::Result<Vec<u8>, ProtonError> {
|
||||
let guard = self
|
||||
.transfers
|
||||
.begin(name, uid.to_string(), TransferDirection::Download, total);
|
||||
let mut out = CountingWriter::new(Vec::with_capacity(total as usize), &guard);
|
||||
self.rt.block_on(photos.download_photo_to(uid, &mut out))?;
|
||||
Ok(out.into_inner())
|
||||
}
|
||||
|
||||
/// Pin the node at mountpoint-relative `rel`. A file downloads its full
|
||||
/// plaintext into the content cache; a folder records a recursive pin and
|
||||
/// downloads every descendant file (selective sync). Returns a human message.
|
||||
@@ -746,8 +821,7 @@ impl Core {
|
||||
return Ok(format!("{name} ({n} files)"));
|
||||
}
|
||||
let bytes = self
|
||||
.rt
|
||||
.block_on(self.client.download_file(&uid))
|
||||
.download_file_tracked(&uid, &name, size)
|
||||
.map_err(|e| format!("download: {e}"))?;
|
||||
self.cache
|
||||
.store(&uid, mtime, size, &bytes)
|
||||
@@ -763,7 +837,7 @@ impl Core {
|
||||
/// Walks the tree depth-first, enumerating each folder so a cold subtree is
|
||||
/// fully discovered; the lock is dropped before each network download.
|
||||
fn pin_subtree(&self, ino: u64) -> Result<usize, String> {
|
||||
let mut files: Vec<(NodeUid, i64, u64)> = Vec::new();
|
||||
let mut files: Vec<(NodeUid, String, i64, u64)> = Vec::new();
|
||||
let mut stack = vec![ino];
|
||||
while let Some(dir) = stack.pop() {
|
||||
self.ensure_children(dir)
|
||||
@@ -777,6 +851,7 @@ impl Core {
|
||||
} else {
|
||||
files.push((
|
||||
e.uid.clone(),
|
||||
e.node.name.clone(),
|
||||
e.node.modification_time,
|
||||
node_size(&e.node),
|
||||
));
|
||||
@@ -786,12 +861,12 @@ impl Core {
|
||||
}
|
||||
}
|
||||
let mut count = 0;
|
||||
for (uid, mtime, size) in files {
|
||||
for (uid, name, mtime, size) in files {
|
||||
if self.cache.is_cached(&uid, mtime, size) {
|
||||
count += 1;
|
||||
continue;
|
||||
}
|
||||
match self.rt.block_on(self.client.download_file(&uid)) {
|
||||
match self.download_file_tracked(&uid, &name, size) {
|
||||
Ok(bytes) => {
|
||||
if self.cache.store(&uid, mtime, size, &bytes).is_ok() {
|
||||
count += 1;
|
||||
@@ -940,6 +1015,26 @@ impl Core {
|
||||
.collect())
|
||||
}
|
||||
|
||||
/// Search the index of files on this machine (outside Drive), built by the
|
||||
/// background scanner in [`run_local_index`]. Pure local lookup, never hits
|
||||
/// the network — and never touches the FUSE mount, which the scanner excludes.
|
||||
fn search_local(&self, query: &str, limit: usize) -> Result<Vec<LocalHit>, String> {
|
||||
let hits = self
|
||||
.db
|
||||
.search_local(query, limit)
|
||||
.map_err(|e| format!("local search: {e}"))?;
|
||||
Ok(hits
|
||||
.into_iter()
|
||||
.map(|h| LocalHit {
|
||||
name: h.name,
|
||||
path: h.path,
|
||||
is_dir: h.is_dir,
|
||||
size: h.size.max(0) as u64,
|
||||
modified: h.mtime,
|
||||
})
|
||||
.collect())
|
||||
}
|
||||
|
||||
/// A page of the photos timeline (newest first), with the page's thumbnails
|
||||
/// fetched into the cache and their paths attached. `Ok(None)` when the
|
||||
/// account has no photos volume.
|
||||
@@ -1050,8 +1145,7 @@ impl Core {
|
||||
return Ok(p);
|
||||
}
|
||||
let bytes = self
|
||||
.rt
|
||||
.block_on(photos.download_photo(uid))
|
||||
.download_photo_tracked(&photos, uid, &node.name, size)
|
||||
.map_err(|e| format!("download photo: {e}"))?;
|
||||
self.cache
|
||||
.store(uid, mtime, size, &bytes)
|
||||
@@ -1067,20 +1161,23 @@ impl Core {
|
||||
let (ino, uid) = self
|
||||
.resolve_path(rel)
|
||||
.map_err(|e| format!("resolve path: {e:?}"))?;
|
||||
let (mtime, size) = {
|
||||
let (name, mtime, size) = {
|
||||
let st = self.state.lock().unwrap();
|
||||
let e = st.entries.get(&ino).ok_or("node vanished")?;
|
||||
if !e.node.is_file() {
|
||||
return Err("not a regular file".into());
|
||||
}
|
||||
(e.node.modification_time, node_size(&e.node))
|
||||
(
|
||||
e.node.name.clone(),
|
||||
e.node.modification_time,
|
||||
node_size(&e.node),
|
||||
)
|
||||
};
|
||||
if let Some(p) = self.cache.cached_content_path(&uid, mtime, size) {
|
||||
return Ok(p);
|
||||
}
|
||||
let bytes = self
|
||||
.rt
|
||||
.block_on(self.client.download_file(&uid))
|
||||
.download_file_tracked(&uid, &name, size)
|
||||
.map_err(|e| format!("download: {e}"))?;
|
||||
self.cache
|
||||
.store(&uid, mtime, size, &bytes)
|
||||
@@ -1206,13 +1303,24 @@ impl Core {
|
||||
.map_err(|e| format!("resolve path: {e:?}"))?;
|
||||
self.ensure_children(pino)
|
||||
.map_err(|e| format!("enumerate: {e:?}"))?;
|
||||
let guard = self
|
||||
.transfers
|
||||
.begin(name, "", TransferDirection::Upload, bytes.len() as u64);
|
||||
let reader = CountingReader::new(std::io::Cursor::new(bytes), &guard);
|
||||
let new_uid = self
|
||||
.rt
|
||||
.block_on(
|
||||
self.client
|
||||
.upload_file(&parent_uid, name, media_type_for(name), bytes),
|
||||
)
|
||||
.block_on(self.client.upload_file_from(
|
||||
&parent_uid,
|
||||
name,
|
||||
media_type_for(name),
|
||||
reader,
|
||||
bytes.len() as i64,
|
||||
Vec::new(),
|
||||
None,
|
||||
false,
|
||||
))
|
||||
.map_err(|e| format!("upload: {e}"))?;
|
||||
drop(guard);
|
||||
let node = self
|
||||
.fetch_node(&new_uid)
|
||||
.map_err(|e| format!("fetch node: {e:?}"))?;
|
||||
@@ -2219,6 +2327,73 @@ async fn run_event_sync(
|
||||
}
|
||||
}
|
||||
|
||||
/// Keep the local-file index fresh for the launcher prompt's "This computer"
|
||||
/// results. Rebuilds the index whenever it is older than [`LOCAL_INDEX_TTL`],
|
||||
/// then sleeps; runs on its own thread for the life of the daemon.
|
||||
///
|
||||
/// The walk is the one part of the daemon that touches the wider filesystem, so
|
||||
/// it is deliberately kept off every hot path: it never runs on a FUSE or
|
||||
/// control-socket thread, and it excludes the mountpoint (walking it would fault
|
||||
/// every remote node in through FUSE, defeating on-demand hydration).
|
||||
fn run_local_index(db: Arc<Db>, indexing: Arc<AtomicBool>, mountpoint: PathBuf) {
|
||||
loop {
|
||||
let age = db.local_indexed_at().ok().flatten();
|
||||
let stale =
|
||||
age.is_none_or(|at| now_secs().saturating_sub(at) >= LOCAL_INDEX_TTL.as_secs() as i64);
|
||||
if stale {
|
||||
scan_local_once(&db, &indexing, &mountpoint);
|
||||
}
|
||||
std::thread::sleep(LOCAL_INDEX_CHECK);
|
||||
}
|
||||
}
|
||||
|
||||
/// Walk `$HOME` once and replace the local-file index with what it finds.
|
||||
/// Batches stream straight into SQLite, so peak memory is one batch — not the
|
||||
/// whole home directory.
|
||||
fn scan_local_once(db: &Db, indexing: &AtomicBool, mountpoint: &Path) {
|
||||
let dirs = match AppDirs::new() {
|
||||
Ok(d) => d,
|
||||
Err(e) => {
|
||||
warn!(error = %e, "local index: cannot resolve app dirs");
|
||||
return;
|
||||
}
|
||||
};
|
||||
let Some(home) = dirs.home_dir() else {
|
||||
warn!("local index: cannot resolve home directory");
|
||||
return;
|
||||
};
|
||||
let generation = match db.local_begin_scan() {
|
||||
Ok(g) => g,
|
||||
Err(e) => {
|
||||
warn!(error = %e, "local index: cannot open scan generation");
|
||||
return;
|
||||
}
|
||||
};
|
||||
|
||||
let excludes = localindex::default_excludes(mountpoint, &dirs.state_dir(), &dirs.cache_dir());
|
||||
indexing.store(true, Ordering::Relaxed);
|
||||
let started = Instant::now();
|
||||
|
||||
let walked = localindex::scan(&[home], &excludes, |batch| {
|
||||
if let Err(e) = db.local_upsert_batch(generation, &batch) {
|
||||
warn!(error = %e, "local index: batch write failed");
|
||||
}
|
||||
});
|
||||
|
||||
// Prune what this scan did not see and rebuild the FTS index over the rest,
|
||||
// even if some batches failed — a partial index still beats none.
|
||||
match db.local_finish_scan(generation, now_secs()) {
|
||||
Ok(indexed) => info!(
|
||||
walked,
|
||||
indexed,
|
||||
elapsed_ms = started.elapsed().as_millis() as u64,
|
||||
"local index rebuilt"
|
||||
),
|
||||
Err(e) => warn!(error = %e, "local index: finish failed"),
|
||||
}
|
||||
indexing.store(false, Ordering::Relaxed);
|
||||
}
|
||||
|
||||
/// Turn a CLI-supplied path into a mountpoint-relative path. An absolute path
|
||||
/// must live under `mountpoint`; a relative path is taken as already relative to
|
||||
/// the mount root.
|
||||
@@ -2322,10 +2497,22 @@ fn handle_control_conn(core: &Core, username: &str, mountpoint: &Path, stream: U
|
||||
capture_time,
|
||||
..Default::default()
|
||||
};
|
||||
match core
|
||||
.rt
|
||||
.block_on(photos.upload_photo(&name, &media_type, &bytes, metadata))
|
||||
{
|
||||
let guard = core.transfers.begin(
|
||||
name.clone(),
|
||||
"",
|
||||
TransferDirection::Upload,
|
||||
bytes.len() as u64,
|
||||
);
|
||||
let reader = CountingReader::new(std::io::Cursor::new(&bytes), &guard);
|
||||
match core.rt.block_on(photos.upload_photo_from(
|
||||
&name,
|
||||
&media_type,
|
||||
reader,
|
||||
bytes.len() as i64,
|
||||
Vec::new(),
|
||||
metadata,
|
||||
false,
|
||||
)) {
|
||||
Ok(uid) => {
|
||||
*core.timeline.lock().unwrap() = None;
|
||||
CtlResponse::Ok {
|
||||
@@ -2350,6 +2537,13 @@ fn handle_control_conn(core: &Core, username: &str, mountpoint: &Path, stream: U
|
||||
Ok(hits) => CtlResponse::SearchResults { hits },
|
||||
Err(e) => CtlResponse::Error { message: e },
|
||||
},
|
||||
Ok(CtlRequest::SearchLocal { query, limit }) => match core.search_local(&query, limit) {
|
||||
Ok(hits) => CtlResponse::LocalResults {
|
||||
hits,
|
||||
indexing: core.indexing.load(Ordering::Relaxed),
|
||||
},
|
||||
Err(e) => CtlResponse::Error { message: e },
|
||||
},
|
||||
Ok(CtlRequest::Rename { path, new_name }) => match rel_to_mount(mountpoint, &path) {
|
||||
Ok(rel) => match core.rename(&rel, &new_name) {
|
||||
Ok(name) => CtlResponse::Ok {
|
||||
@@ -2413,6 +2607,9 @@ fn handle_control_conn(core: &Core, username: &str, mountpoint: &Path, stream: U
|
||||
),
|
||||
}
|
||||
}
|
||||
Ok(CtlRequest::GetQueueStatus) => CtlResponse::Transfers {
|
||||
items: core.transfers.snapshot(),
|
||||
},
|
||||
Ok(CtlRequest::SetCacheBudget { bytes }) => {
|
||||
core.cache.set_budget(bytes);
|
||||
// Persist so the next mount keeps the new cap. Best-effort: a config
|
||||
@@ -2539,8 +2736,21 @@ pub fn mount(
|
||||
cache: Arc::new(cache),
|
||||
db,
|
||||
timeline: Arc::new(Mutex::new(None)),
|
||||
transfers: TransferRegistry::new(),
|
||||
indexing: Arc::new(AtomicBool::new(false)),
|
||||
};
|
||||
|
||||
// Keep the launcher prompt's "This computer" index fresh. Its own thread:
|
||||
// the walk is I/O-heavy and must never sit in front of a FUSE callback.
|
||||
{
|
||||
let db = core.db.clone();
|
||||
let indexing = core.indexing.clone();
|
||||
let mountpoint = mountpoint.to_path_buf();
|
||||
std::thread::Builder::new()
|
||||
.name("pdfs-localindex".into())
|
||||
.spawn(move || run_local_index(db, indexing, mountpoint))?;
|
||||
}
|
||||
|
||||
let mut config = Config::default();
|
||||
config.mount_options = vec![
|
||||
MountOption::FSName("protondrive".to_string()),
|
||||
|
||||
@@ -0,0 +1,169 @@
|
||||
//! In-memory registry of in-flight transfers, behind [`Request::GetQueueStatus`].
|
||||
//!
|
||||
//! The daemon downloads and uploads whole files through the SDK's streaming
|
||||
//! [`Read`]/[`Write`] variants. 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.
|
||||
//!
|
||||
//! [`Request::GetQueueStatus`]: pdfs_core::control::Request::GetQueueStatus
|
||||
|
||||
use std::collections::HashMap;
|
||||
use std::io::{Read, Write};
|
||||
use std::sync::atomic::{AtomicU64, Ordering};
|
||||
use std::sync::{Arc, Mutex};
|
||||
use std::time::Instant;
|
||||
|
||||
use pdfs_core::control::{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.
|
||||
struct Entry {
|
||||
name: String,
|
||||
uid: String,
|
||||
direction: TransferDirection,
|
||||
total: u64,
|
||||
done: AtomicU64,
|
||||
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`]).
|
||||
#[derive(Default)]
|
||||
pub struct TransferRegistry {
|
||||
inner: Mutex<HashMap<u64, Arc<Entry>>>,
|
||||
next: AtomicU64,
|
||||
}
|
||||
|
||||
impl TransferRegistry {
|
||||
pub fn new() -> Arc<Self> {
|
||||
Arc::new(Self::default())
|
||||
}
|
||||
|
||||
/// Register a new transfer of `total` bytes (`0` = unknown), returning a
|
||||
/// guard that tracks progress and deregisters the transfer when dropped.
|
||||
pub fn begin(
|
||||
self: &Arc<Self>,
|
||||
name: impl Into<String>,
|
||||
uid: impl Into<String>,
|
||||
direction: TransferDirection,
|
||||
total: u64,
|
||||
) -> TransferGuard {
|
||||
let id = self.next.fetch_add(1, Ordering::Relaxed);
|
||||
let entry = Arc::new(Entry {
|
||||
name: name.into(),
|
||||
uid: uid.into(),
|
||||
direction,
|
||||
total,
|
||||
done: AtomicU64::new(0),
|
||||
started: Instant::now(),
|
||||
});
|
||||
self.inner.lock().unwrap().insert(id, entry.clone());
|
||||
TransferGuard {
|
||||
reg: self.clone(),
|
||||
id,
|
||||
entry,
|
||||
}
|
||||
}
|
||||
|
||||
/// Snapshot every in-flight transfer for [`Response::Transfers`]. 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()
|
||||
.map(|e| {
|
||||
let done = e.done.load(Ordering::Relaxed);
|
||||
let secs = e.started.elapsed().as_secs_f64();
|
||||
let speed = if secs > 0.0 {
|
||||
(done as f64 / secs) as u64
|
||||
} else {
|
||||
0
|
||||
};
|
||||
TransferItem {
|
||||
uid: e.uid.clone(),
|
||||
name: e.name.clone(),
|
||||
direction: e.direction,
|
||||
bytes_completed: done,
|
||||
bytes_total: e.total,
|
||||
speed_bytes_sec: speed,
|
||||
}
|
||||
})
|
||||
.collect()
|
||||
}
|
||||
}
|
||||
|
||||
/// Lifetime handle for a registered transfer: tick progress with [`add`], and
|
||||
/// the transfer leaves the registry when this drops.
|
||||
///
|
||||
/// [`add`]: TransferGuard::add
|
||||
pub struct TransferGuard {
|
||||
reg: Arc<TransferRegistry>,
|
||||
id: u64,
|
||||
entry: Arc<Entry>,
|
||||
}
|
||||
|
||||
impl TransferGuard {
|
||||
/// Record `n` more bytes moved.
|
||||
pub fn add(&self, n: u64) {
|
||||
self.entry.done.fetch_add(n, Ordering::Relaxed);
|
||||
}
|
||||
}
|
||||
|
||||
impl Drop for TransferGuard {
|
||||
fn drop(&mut self) {
|
||||
self.reg.inner.lock().unwrap().remove(&self.id);
|
||||
}
|
||||
}
|
||||
|
||||
/// A [`Write`] that tallies bytes written to a [`TransferGuard`] (download path).
|
||||
pub struct CountingWriter<'a, W> {
|
||||
inner: W,
|
||||
guard: &'a TransferGuard,
|
||||
}
|
||||
|
||||
impl<'a, W: Write> CountingWriter<'a, W> {
|
||||
pub fn new(inner: W, guard: &'a TransferGuard) -> Self {
|
||||
Self { inner, guard }
|
||||
}
|
||||
|
||||
pub fn into_inner(self) -> W {
|
||||
self.inner
|
||||
}
|
||||
}
|
||||
|
||||
impl<W: Write> Write for CountingWriter<'_, W> {
|
||||
fn write(&mut self, buf: &[u8]) -> std::io::Result<usize> {
|
||||
let n = self.inner.write(buf)?;
|
||||
self.guard.add(n as u64);
|
||||
Ok(n)
|
||||
}
|
||||
|
||||
fn flush(&mut self) -> std::io::Result<()> {
|
||||
self.inner.flush()
|
||||
}
|
||||
}
|
||||
|
||||
/// A [`Read`] that tallies bytes read through a [`TransferGuard`] (upload path).
|
||||
pub struct CountingReader<'a, R> {
|
||||
inner: R,
|
||||
guard: &'a TransferGuard,
|
||||
}
|
||||
|
||||
impl<'a, R: Read> CountingReader<'a, R> {
|
||||
pub fn new(inner: R, guard: &'a TransferGuard) -> Self {
|
||||
Self { inner, guard }
|
||||
}
|
||||
}
|
||||
|
||||
impl<R: Read> Read for CountingReader<'_, 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)
|
||||
}
|
||||
}
|
||||
@@ -14,6 +14,10 @@ path = "src/main.rs"
|
||||
name = "pdfs-app"
|
||||
path = "src/app.rs"
|
||||
|
||||
[[bin]]
|
||||
name = "pdfs-prompt"
|
||||
path = "src/prompt.rs"
|
||||
|
||||
[dependencies]
|
||||
pdfs-core.workspace = true
|
||||
tracing.workspace = true
|
||||
@@ -24,3 +28,4 @@ ksni = "0.2"
|
||||
gtk4 = { version = "0.9", features = ["v4_12"] }
|
||||
adw = { package = "libadwaita", version = "0.7", features = ["v1_5"] }
|
||||
async-channel = "2"
|
||||
exif = { version = "0.6", package = "kamadak-exif" }
|
||||
+680
-213
File diff suppressed because it is too large.
Load diff
@@ -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, send};
|
||||
use pdfs_core::control::{Request, Response, TransferDirection, TransferItem, send};
|
||||
use pdfs_core::service;
|
||||
|
||||
/// How often the tray re-polls the daemon to refresh its menu.
|
||||
@@ -33,6 +33,42 @@ struct DriveState {
|
||||
mounted: bool,
|
||||
/// Mountpoint to act on (from the daemon when mounted, else the default).
|
||||
mountpoint: PathBuf,
|
||||
/// One-line sync summary of in-flight transfers, empty when idle. Shown as a
|
||||
/// disabled menu line under the status line.
|
||||
sync: String,
|
||||
}
|
||||
|
||||
/// Summarise a transfer 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 {
|
||||
match items {
|
||||
[] => String::new(),
|
||||
[t] => {
|
||||
let arrow = match t.direction {
|
||||
TransferDirection::Download => "↓",
|
||||
TransferDirection::Upload => "↑",
|
||||
};
|
||||
if t.bytes_total == 0 {
|
||||
format!("{arrow} {}…", t.name)
|
||||
} else {
|
||||
let pct = (t.bytes_completed * 100 / t.bytes_total.max(1)).min(100);
|
||||
format!("{arrow} {} ({pct}%)", t.name)
|
||||
}
|
||||
}
|
||||
_ => {
|
||||
let down = items
|
||||
.iter()
|
||||
.filter(|t| t.direction == TransferDirection::Download)
|
||||
.count();
|
||||
let up = items.len() - down;
|
||||
match (down, up) {
|
||||
(d, 0) => format!("↓ {d} downloading"),
|
||||
(0, u) => format!("↑ {u} uploading"),
|
||||
(d, u) => format!("↓ {d} · ↑ {u}"),
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
struct DriveTray {
|
||||
@@ -49,12 +85,18 @@ fn poll_state(socket: &Path, default_mountpoint: &Path) -> DriveState {
|
||||
line: format!("Mounted at {mountpoint} ({pinned} pinned)"),
|
||||
mounted: true,
|
||||
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),
|
||||
_ => String::new(),
|
||||
},
|
||||
},
|
||||
// Socket answered but with something unexpected — treat as up but odd.
|
||||
Ok(_) => DriveState {
|
||||
line: "Mount: unexpected daemon response".into(),
|
||||
mounted: true,
|
||||
mountpoint: default_mountpoint.to_path_buf(),
|
||||
sync: String::new(),
|
||||
},
|
||||
// No daemon: describe login state instead so the menu is still useful.
|
||||
Err(_) => {
|
||||
@@ -67,6 +109,7 @@ fn poll_state(socket: &Path, default_mountpoint: &Path) -> DriveState {
|
||||
line,
|
||||
mounted: false,
|
||||
mountpoint: default_mountpoint.to_path_buf(),
|
||||
sync: String::new(),
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -116,6 +159,21 @@ impl Tray for DriveTray {
|
||||
..Default::default()
|
||||
}
|
||||
.into(),
|
||||
];
|
||||
|
||||
// Live sync status, shown only while something is transferring.
|
||||
if !self.state.sync.is_empty() {
|
||||
items.push(
|
||||
StandardItem {
|
||||
label: self.state.sync.clone(),
|
||||
enabled: false,
|
||||
..Default::default()
|
||||
}
|
||||
.into(),
|
||||
);
|
||||
}
|
||||
|
||||
items.extend([
|
||||
MenuItem::Separator,
|
||||
StandardItem {
|
||||
label: "Open Manager".into(),
|
||||
@@ -123,7 +181,7 @@ impl Tray for DriveTray {
|
||||
..Default::default()
|
||||
}
|
||||
.into(),
|
||||
];
|
||||
]);
|
||||
|
||||
if self.state.mounted {
|
||||
items.push(
|
||||
|
||||
File diff suppressed because it is too large.
Load diff
Binary file not shown.
|
Before Width: | Height: | Size: 215 KiB |
Binary file not shown.
|
After Width: | Height: | Size: 187 KiB |
Binary file not shown.
|
Before Width: | Height: | Size: 200 KiB |
Binary file not shown.
|
Before Width: | Height: | Size: 211 KiB |
Binary file not shown.
|
Before Width: | Height: | Size: 191 KiB After Width: | Height: | Size: 193 KiB |
Binary file not shown.
|
After Width: | Height: | Size: 1.1 MiB |
Binary file not shown.
|
After Width: | Height: | Size: 2.4 MiB |
+12
-7
@@ -1,21 +1,22 @@
|
||||
# Maintainer: Nils Pukropp <contact@narl.io>
|
||||
#
|
||||
# Local-build PKGBUILD. Builds the proton-drive-linux workspace from the
|
||||
# checkout this file lives in. Path deps point at a sibling ../proton-sdk-rs
|
||||
# checkout (the Proton SDK is unpublished), so this is NOT a source-fetch
|
||||
# package: it compiles the working tree in place.
|
||||
# checkout this file lives in: this is NOT a source-fetch package, it compiles
|
||||
# the working tree in place. The Proton SDK is unpublished and comes from its
|
||||
# GitHub repo (pinned by Cargo.lock), so the build needs network access — no
|
||||
# sibling checkout required any more.
|
||||
#
|
||||
# Build + install: cd packaging && makepkg -fi
|
||||
|
||||
pkgname=proton-drive-linux
|
||||
pkgver=0.2.1
|
||||
pkgver=0.2.5
|
||||
pkgrel=1
|
||||
pkgdesc="Proton Drive client for Linux: FUSE files-on-demand mount, CLI, GTK4 app + tray"
|
||||
arch=('x86_64')
|
||||
url="https://github.com/narl/proton-drive-linux"
|
||||
license=('MIT')
|
||||
depends=('fuse3' 'gtk4' 'libadwaita')
|
||||
makedepends=('cargo')
|
||||
makedepends=('cargo' 'git')
|
||||
provides=('pdfs')
|
||||
# Files-on-demand mount + keyring need these at runtime.
|
||||
optdepends=('gnome-keyring: credential storage via Secret Service'
|
||||
@@ -40,10 +41,13 @@ build() {
|
||||
cd "$_workspace"
|
||||
export RUSTUP_TOOLCHAIN=stable
|
||||
export CARGO_TARGET_DIR="$_workspace/target"
|
||||
cargo build --release --frozen \
|
||||
# `--locked` not `--frozen`: the lock is authoritative, but cargo must still be
|
||||
# allowed to fetch the git-sourced SDK.
|
||||
cargo build --release --locked \
|
||||
--bin pdfs \
|
||||
--bin pdfs-tray \
|
||||
--bin pdfs-app
|
||||
--bin pdfs-app \
|
||||
--bin pdfs-prompt
|
||||
}
|
||||
|
||||
package() {
|
||||
@@ -53,6 +57,7 @@ package() {
|
||||
install -Dm755 "$rel/pdfs" "$pkgdir/usr/bin/pdfs"
|
||||
install -Dm755 "$rel/pdfs-tray" "$pkgdir/usr/bin/pdfs-tray"
|
||||
install -Dm755 "$rel/pdfs-app" "$pkgdir/usr/bin/pdfs-app"
|
||||
install -Dm755 "$rel/pdfs-prompt" "$pkgdir/usr/bin/pdfs-prompt"
|
||||
|
||||
install -Dm644 "packaging/io.narl.proton-drive-linux.desktop" \
|
||||
"$pkgdir/usr/share/applications/io.narl.proton-drive-linux.desktop"
|
||||
|
||||
Reference in new issue
Block a user