mirror of
https://github.com/narrrl/proton-drive-linux.git
synced 2026-11-03 16:41:26 +01:00
nvrl
3e3dd7db41
A user switching from a Windows PC to Linux had two half answers:
adopting the old device makes this machine back up as it, and restoring
its folders keeps them in the old device's tree. Neither puts the data
under this machine's own device.
Device roots are separate shares on the main volume, so each top-level
folder of the other device can move under this device's root in one
server-side move. Nothing is uploaded again, and the other device stays
registered, without those folders.
- core: `Request::MigrateDevice { uid }` and
`Db::sync_folder_set_share`
- fuse: `Core::migrate_device` moves every folder, suffixes a name
already taken here with the other device's name, keeps going past a
failed folder, and re-points sync folders that already tracked a
moved folder at the new share
- cli: `pdfs devices migrate UID [--yes]` confirms, moves, then runs
the ordinary restore picker
- gui: "Move to This Computer…" on another computer's row, followed by
the restore picker
- i18n: translate the new strings in every catalog
- docs: CLI.md, USER_GUIDE.md, RECOVERY.md, ARCHITECTURE.md, changelog
1953 lines
81 KiB
Rust
1953 lines
81 KiB
Rust
//! The control socket: the daemon`s IPC surface.
|
|
//!
|
|
//! The CLI and the GUI never touch the mount or the database directly — they
|
|
//! open this Unix socket, write one JSON [`CtlRequest`] line, and read one JSON
|
|
//! [`CtlResponse`] line back. That keeps the daemon the single writer of both
|
|
//! the FUSE state and `cache.db`.
|
|
//!
|
|
//! Connections are served concurrently, one thread each, because requests differ
|
|
//! wildly in cost: an `OpenPhoto` downloads a whole photo, and a serial loop
|
|
//! would stall the GUI`s 2s status poll behind it.
|
|
|
|
use std::io::{BufRead, BufReader, Write as _};
|
|
use std::os::unix::net::{UnixListener, UnixStream};
|
|
use std::path::{Component, Path, PathBuf};
|
|
use std::sync::atomic::{AtomicUsize, Ordering};
|
|
use std::time::{Duration, Instant};
|
|
|
|
use pdfs_core::config::AppDirs;
|
|
use pdfs_core::control::{
|
|
ActivityKind, Event, HEARTBEAT, PhotoMonth, RefreshScope, Request as CtlRequest,
|
|
Response as CtlResponse, Topic, TransferDirection, TrashFailure,
|
|
};
|
|
use pdfs_core::{CoreError, CoreResult};
|
|
use proton_drive_rs::proton_sdk::ids::NodeUid;
|
|
use tracing::{debug, info, warn};
|
|
|
|
use super::transfers::CountingReader;
|
|
use super::{Core, count_noun, human_bytes, human_duration, parse_uid};
|
|
|
|
const MAX_CONTROL_REQUEST_BYTES: u64 = 1024 * 1024;
|
|
const CONTROL_READ_TIMEOUT: Duration = Duration::from_secs(10);
|
|
/// Answer a [`CtlRequest::StopSharing`]: log it, and turn a partial removal
|
|
/// into an error that names what is still shared.
|
|
fn stop_sharing_reply(
|
|
core: &Core,
|
|
target: &str,
|
|
result: CoreResult<super::sharing::StopSharingOutcome>,
|
|
) -> CtlResponse {
|
|
match result.map(|outcome| outcome.into_result()) {
|
|
Ok(Ok(message)) => {
|
|
core.log_activity(ActivityKind::Unshare, target, "stopped sharing", true);
|
|
CtlResponse::Ok { message }
|
|
}
|
|
Ok(Err(left)) => {
|
|
core.log_activity(ActivityKind::Unshare, target, &left, false);
|
|
CtlResponse::error(CoreError::remote(left))
|
|
}
|
|
Err(e) => CtlResponse::error(e),
|
|
}
|
|
}
|
|
|
|
/// A client that connects, sends a request and then never reads its reply used
|
|
/// to park a handler thread forever in `write_all`; 64 of those exhaust
|
|
/// [`MAX_CONTROL_HANDLERS`] permanently. Generous enough for the largest
|
|
/// responses (photo bytes) over a local socket, finite so a wedged peer can't
|
|
/// hold a permit.
|
|
const CONTROL_WRITE_TIMEOUT: Duration = Duration::from_secs(30);
|
|
const MAX_CONTROL_HANDLERS: usize = 64;
|
|
/// Ceiling on any client-supplied row `limit`. Every in-tree caller asks for far
|
|
/// less (200 for activity, 60 for a photo page), but the value goes straight
|
|
/// into `SQL LIMIT ?` under the daemon's single connection — an unclamped one
|
|
/// materialises the whole account, and then serialises it into one JSON line.
|
|
const MAX_ROW_LIMIT: usize = 1000;
|
|
|
|
/// Clamp a client-supplied row limit to [`MAX_ROW_LIMIT`].
|
|
fn clamp_limit(limit: usize) -> usize {
|
|
limit.min(MAX_ROW_LIMIT)
|
|
}
|
|
/// Answer an album add or remove: parse the uids, run `change`, and report
|
|
/// per photo.
|
|
fn album_change(
|
|
uid: &str,
|
|
photos: &[String],
|
|
change: impl FnOnce(&NodeUid, &[NodeUid]) -> CoreResult<super::albums::AlbumOutcome>,
|
|
) -> CtlResponse {
|
|
let Some(album) = parse_uid(uid) else {
|
|
return CtlResponse::error(CoreError::invalid(format!("bad album uid: {uid}")));
|
|
};
|
|
let photos = match Core::parse_uids(photos) {
|
|
Ok(photos) => photos,
|
|
Err(e) => return CtlResponse::error(e),
|
|
};
|
|
match change(&album, &photos) {
|
|
Ok((changed, failed)) => CtlResponse::AlbumChanged {
|
|
changed,
|
|
failed: failed
|
|
.into_iter()
|
|
.map(|(uid, message)| TrashFailure { uid, message })
|
|
.collect(),
|
|
},
|
|
Err(e) => CtlResponse::error(e),
|
|
}
|
|
}
|
|
|
|
static ACTIVE_CONTROL_HANDLERS: AtomicUsize = AtomicUsize::new(0);
|
|
|
|
struct ControlHandlerPermit;
|
|
|
|
impl ControlHandlerPermit {
|
|
fn acquire() -> Option<Self> {
|
|
ACTIVE_CONTROL_HANDLERS
|
|
.fetch_update(Ordering::AcqRel, Ordering::Acquire, |active| {
|
|
(active < MAX_CONTROL_HANDLERS).then_some(active + 1)
|
|
})
|
|
.ok()
|
|
.map(|_| Self)
|
|
}
|
|
}
|
|
|
|
impl Drop for ControlHandlerPermit {
|
|
fn drop(&mut self) {
|
|
ACTIVE_CONTROL_HANDLERS.fetch_sub(1, Ordering::AcqRel);
|
|
}
|
|
}
|
|
|
|
/// Handler threads in use, for the diagnostics. At the limit the daemon refuses
|
|
/// connections, which a front-end cannot tell apart from a dead daemon.
|
|
pub(crate) fn active_handlers() -> usize {
|
|
ACTIVE_CONTROL_HANDLERS.load(Ordering::Acquire)
|
|
}
|
|
|
|
pub(crate) fn handler_limit() -> usize {
|
|
MAX_CONTROL_HANDLERS
|
|
}
|
|
|
|
fn read_request_line(reader: &mut impl BufRead) -> std::io::Result<Option<String>> {
|
|
let mut bytes = Vec::new();
|
|
loop {
|
|
let available = reader.fill_buf()?;
|
|
if available.is_empty() {
|
|
break;
|
|
}
|
|
let take = available
|
|
.iter()
|
|
.position(|byte| *byte == b'\n')
|
|
.map_or(available.len(), |index| index + 1);
|
|
if bytes.len().saturating_add(take) > MAX_CONTROL_REQUEST_BYTES as usize {
|
|
return Err(std::io::Error::new(
|
|
std::io::ErrorKind::InvalidData,
|
|
"control request exceeds size limit",
|
|
));
|
|
}
|
|
bytes.extend_from_slice(&available[..take]);
|
|
reader.consume(take);
|
|
if bytes.ends_with(b"\n") {
|
|
break;
|
|
}
|
|
}
|
|
if bytes.is_empty() {
|
|
return Ok(None);
|
|
}
|
|
if bytes.len() as u64 > MAX_CONTROL_REQUEST_BYTES || !bytes.ends_with(b"\n") {
|
|
return Err(std::io::Error::new(
|
|
std::io::ErrorKind::InvalidData,
|
|
"control request exceeds limit or is not newline terminated",
|
|
));
|
|
}
|
|
String::from_utf8(bytes)
|
|
.map(Some)
|
|
.map_err(|e| std::io::Error::new(std::io::ErrorKind::InvalidData, e))
|
|
}
|
|
|
|
/// Turn a CLI-supplied path into the mount that owns it and the path relative
|
|
/// to *that* mount's root.
|
|
///
|
|
/// [`rel_to_mount`] is the older, primary-only reading, and it is what most
|
|
/// requests still use. It rejects anything outside the primary
|
|
/// mountpoint, which leaves a secondary on-demand mount — a synced folder
|
|
/// switched out of mirror mode — with no way to be named at all: `pdfs ls` and
|
|
/// `pdfs refresh` both answered "is not under the mountpoint", so a folder
|
|
/// serving a stale listing had no manual escape hatch (`docs/BUGS.md` B86).
|
|
///
|
|
/// A relative path keeps the old meaning, relative to the primary mount, since
|
|
/// there is nothing in it to choose a mount by.
|
|
fn route_to_mount(core: &Core, mountpoint: &Path, path: &str) -> CoreResult<(Core, PathBuf)> {
|
|
let p = Path::new(path);
|
|
if !p.is_absolute() {
|
|
return Ok((core.clone(), p.to_path_buf()));
|
|
}
|
|
if let Some(routed) = core.rooted_at(p) {
|
|
return Ok(routed);
|
|
}
|
|
// Fall back to the primary reading so the error text stays the familiar one
|
|
// when the path is simply outside everything this daemon has mounted.
|
|
rel_to_mount(mountpoint, path).map(|rel| (core.clone(), rel))
|
|
}
|
|
|
|
/// [`route_to_mount`] for a request that changes what it names: trash, rename,
|
|
/// create a folder, resolve a conflict. Each has to run against the on-demand
|
|
/// mount serving the path, and `pdfs rm ~/Downloads/x` answered "is not under
|
|
/// the mountpoint" while `pdfs move` took the same path. The conflict listing
|
|
/// names a copy in a synced folder by its absolute local path (`docs/BUGS.md`
|
|
/// B99), so resolving one needs this too. A mirror folder has no mount to run
|
|
/// against: a file there is an ordinary local file.
|
|
fn route_change(core: &Core, mountpoint: &Path, path: &str) -> CoreResult<(Core, PathBuf)> {
|
|
let routed = route_to_mount(core, mountpoint, path).map_err(|e| {
|
|
let p = Path::new(path);
|
|
let mirrored = core
|
|
.db
|
|
.sync_folder_list()
|
|
.unwrap_or_default()
|
|
.iter()
|
|
.any(|folder| folder.mode == "mirror" && p.starts_with(&folder.local_path));
|
|
if mirrored {
|
|
CoreError::invalid(format!(
|
|
"{path} is in a mirrored folder; change the local file there instead"
|
|
))
|
|
} else {
|
|
e
|
|
}
|
|
})?;
|
|
if routed
|
|
.1
|
|
.components()
|
|
.all(|component| matches!(component, Component::Normal(_) | Component::CurDir))
|
|
{
|
|
Ok(routed)
|
|
} else {
|
|
Err(CoreError::invalid(format!("{path} escapes the mountpoint")))
|
|
}
|
|
}
|
|
|
|
/// [`route_change`] for a request that trashes or renames the node it names.
|
|
/// A mount's root is My files or a whole synced folder. An on-demand folder's
|
|
/// root has a parent on Drive, so nothing further down would stop `pdfs rm
|
|
/// ~/Downloads` from trashing the folder out from under its own mount.
|
|
fn route_node(core: &Core, mountpoint: &Path, path: &str) -> CoreResult<(Core, PathBuf)> {
|
|
let routed = route_change(core, mountpoint, path)?;
|
|
if below_root(&routed.1) {
|
|
Ok(routed)
|
|
} else {
|
|
Err(CoreError::invalid(format!(
|
|
"{path} is the root of a synced location and cannot be changed"
|
|
)))
|
|
}
|
|
}
|
|
|
|
/// Whether a mount-relative path names something below the mount's root.
|
|
fn below_root(rel: &Path) -> bool {
|
|
rel.components()
|
|
.any(|component| matches!(component, Component::Normal(_)))
|
|
}
|
|
|
|
/// 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.
|
|
fn rel_to_mount(mountpoint: &Path, path: &str) -> CoreResult<PathBuf> {
|
|
let p = Path::new(path);
|
|
let relative = if p.is_absolute() {
|
|
p.strip_prefix(mountpoint)
|
|
.map(Path::to_path_buf)
|
|
.map_err(|_| CoreError::invalid(format!("{path} is not under the mountpoint")))
|
|
} else {
|
|
Ok(p.to_path_buf())
|
|
}?;
|
|
|
|
if relative
|
|
.components()
|
|
.all(|component| matches!(component, Component::Normal(_) | Component::CurDir))
|
|
{
|
|
Ok(relative)
|
|
} else {
|
|
Err(CoreError::invalid(format!("{path} escapes the mountpoint")))
|
|
}
|
|
}
|
|
|
|
/// Handle one control-socket connection: read a single JSON request line,
|
|
/// dispatch it against `core`, and write back a JSON response line.
|
|
/// Refresh scopes being served right now, one entry per running refresh.
|
|
///
|
|
/// A `Vec` because there are three scopes and at most `MAX_CONTROL_HANDLERS`
|
|
/// handlers, so a set would buy nothing.
|
|
static REFRESHING: std::sync::Mutex<Vec<String>> = std::sync::Mutex::new(Vec::new());
|
|
|
|
/// What makes two refreshes the same work: the scope, and for a directory the
|
|
/// folder, since refreshing one folder says nothing about another. A full
|
|
/// photos refresh is not the same work as an ordinary one — it forgets what the
|
|
/// daemon knows first — so it keys apart.
|
|
fn scope_key(scope: &RefreshScope) -> String {
|
|
match scope {
|
|
RefreshScope::Dir { path } => format!("dir:{path}"),
|
|
RefreshScope::Trash => "trash".to_string(),
|
|
RefreshScope::Photos { full: true } => "photos:full".to_string(),
|
|
RefreshScope::Photos { full: false } => "photos".to_string(),
|
|
}
|
|
}
|
|
|
|
/// Holds one refresh scope for as long as it is being served. `acquire`
|
|
/// returns `None` when that scope is already in flight.
|
|
struct RefreshGuard(String);
|
|
|
|
impl RefreshGuard {
|
|
fn acquire(scope: &RefreshScope) -> Option<Self> {
|
|
let key = scope_key(scope);
|
|
let mut running = REFRESHING.lock().unwrap_or_else(|e| e.into_inner());
|
|
if running.contains(&key) {
|
|
return None;
|
|
}
|
|
running.push(key.clone());
|
|
Some(Self(key))
|
|
}
|
|
}
|
|
|
|
impl Drop for RefreshGuard {
|
|
fn drop(&mut self) {
|
|
let mut running = REFRESHING.lock().unwrap_or_else(|e| e.into_inner());
|
|
if let Some(at) = running.iter().position(|key| *key == self.0) {
|
|
running.swap_remove(at);
|
|
}
|
|
}
|
|
}
|
|
|
|
/// The reply to [`CtlRequest::Status`], which subscriptions are also sent
|
|
/// whenever it changes.
|
|
pub(crate) fn status_response(core: &Core, username: &str, mountpoint: &Path) -> CtlResponse {
|
|
let pins = core.cache.list_pins();
|
|
let queued = core.db.pending_op_counts().unwrap_or_default();
|
|
// Read after the queue: the drain counts an upload as landing before it
|
|
// retires the op, so one of the two always has it (B119).
|
|
let landing = core.landing_uploads.load(Ordering::SeqCst);
|
|
let staging = core.cache.staging_usage();
|
|
CtlResponse::Status {
|
|
parked_uploads: queued.parked.max(0) as u64,
|
|
failing_ops: queued.failing.max(0) as u64,
|
|
failing_error: queued.last_error.clone(),
|
|
staged_bytes: staging.bytes,
|
|
staged_oldest_secs: staging.oldest_secs,
|
|
paused: core.sync_pause().is_some(),
|
|
paused_until: core.sync_pause().flatten(),
|
|
username: username.to_string(),
|
|
mountpoint: mountpoint.display().to_string(),
|
|
pinned: pins.len(),
|
|
used: core.cache.usage(),
|
|
budget: core.cache.budget(),
|
|
pins,
|
|
online: core.online.load(Ordering::Relaxed),
|
|
pending_uploads: queued.uploads.max(0) as u64 + landing,
|
|
pending_changes: queued.changes.max(0) as u64,
|
|
}
|
|
}
|
|
|
|
/// What a front-end shows that `request` changes when it succeeds. Requests
|
|
/// from the CLI publish the same way as the app's, so an open window follows
|
|
/// a change made in a terminal.
|
|
fn touched_topics(request: &CtlRequest) -> &'static [Topic] {
|
|
use CtlRequest as R;
|
|
match request {
|
|
R::Pin { .. } | R::Unpin { .. } => &[Topic::Files],
|
|
R::CreateAlbum { .. }
|
|
| R::RenameAlbum { .. }
|
|
| R::DeleteAlbum { .. }
|
|
| R::AddToAlbum { .. }
|
|
| R::RemoveFromAlbum { .. }
|
|
| R::SetPhotoFavorite { .. }
|
|
| R::UploadPhoto { .. } => &[Topic::Photos],
|
|
R::TrashNodes { .. } => &[Topic::Files, Topic::Photos, Topic::Trash, Topic::Conflicts],
|
|
R::Rename { .. } | R::Move { .. } | R::CreateFolder { .. } | R::UploadPaths { .. } => {
|
|
&[Topic::Files, Topic::Conflicts]
|
|
}
|
|
R::Delete { .. } => &[Topic::Files, Topic::Trash, Topic::Conflicts],
|
|
R::Restore { .. } | R::DeleteForever { .. } | R::EmptyTrash => {
|
|
&[Topic::Files, Topic::Photos, Topic::Trash, Topic::Conflicts]
|
|
}
|
|
R::Refresh { scope } => match scope {
|
|
RefreshScope::Dir { .. } => &[Topic::Files],
|
|
RefreshScope::Trash => &[Topic::Trash],
|
|
// A photos refresh answers once it has started; the timeline
|
|
// publishes again when it lands.
|
|
RefreshScope::Photos { .. } => &[],
|
|
},
|
|
R::RenameDevice { .. } | R::DeleteDevice { .. } => &[Topic::Devices],
|
|
R::AdoptDevice { .. } | R::MigrateDevice { .. } => &[Topic::Devices, Topic::Locations],
|
|
R::AddSyncFolder { .. }
|
|
| R::RemoveSyncFolder { .. }
|
|
| R::SetSyncFolderMode { .. }
|
|
| R::SetSyncFolderPaused { .. }
|
|
| R::RestoreSyncFolders { .. }
|
|
| R::RestoreDeviceFolders { .. } => &[Topic::Locations, Topic::Devices],
|
|
R::RetryPendingOp { .. } => &[Topic::Queue],
|
|
R::DiscardPendingOp { .. } => &[Topic::Queue, Topic::Files],
|
|
R::ResolveConflict { .. } => &[Topic::Files, Topic::Trash, Topic::Conflicts],
|
|
R::ShareNode { .. }
|
|
| R::ShareNodeByUid { .. }
|
|
| R::UpdateShareRole { .. }
|
|
| R::UpdateShareRoleByUid { .. }
|
|
| R::RemoveShareEntry { .. }
|
|
| R::RemoveShareEntryByUid { .. }
|
|
| R::CreatePublicLink { .. }
|
|
| R::CreatePublicLinkByUid { .. }
|
|
| R::RemovePublicLink { .. }
|
|
| R::RemovePublicLinkByUid { .. }
|
|
| R::StopSharing { .. }
|
|
| R::StopSharingByUid { .. } => &[Topic::Shares, Topic::Files],
|
|
R::LeaveShared { .. }
|
|
| R::AcceptInvitation { .. }
|
|
| R::RejectInvitation { .. }
|
|
| R::CreateBookmark { .. }
|
|
| R::DeleteBookmark { .. } => &[Topic::Shares],
|
|
R::RestoreRevision { .. }
|
|
| R::RestoreRevisionByUid { .. }
|
|
| R::SaveRevisionAs { .. }
|
|
| R::SaveRevisionAsByUid { .. } => &[Topic::Files],
|
|
_ => &[],
|
|
}
|
|
}
|
|
|
|
/// How long a subscription's writes may block. A client that stops reading
|
|
/// is dropped after this rather than holding its handler forever.
|
|
const SUBSCRIPTION_WRITE_TIMEOUT: Duration = Duration::from_secs(10);
|
|
|
|
/// Serve a [`CtlRequest::Subscribe`]: send the current status and transfers
|
|
/// and say every topic changed, so the client catches up, then send whatever
|
|
/// the [`EventHub`](super::events::EventHub) collects, with a heartbeat when
|
|
/// nothing does. Ends when the client goes away or the daemon stops.
|
|
fn serve_subscription(core: &Core, username: &str, mountpoint: &Path, mut stream: UnixStream) {
|
|
let Some(mailbox) = core.events.subscribe() else {
|
|
warn!("control: refusing a subscription at the limit");
|
|
if let Ok(mut out) = serde_json::to_vec(&CtlResponse::error(CoreError::remote(
|
|
"too many subscriptions",
|
|
))) {
|
|
out.push(b'\n');
|
|
let _ = stream.write_all(&out);
|
|
}
|
|
return;
|
|
};
|
|
struct Unsubscribe<'a>(&'a Core, std::sync::Arc<super::events::Mailbox>);
|
|
impl Drop for Unsubscribe<'_> {
|
|
fn drop(&mut self) {
|
|
self.0.events.unsubscribe(&self.1);
|
|
}
|
|
}
|
|
let _unsubscribe = Unsubscribe(core, mailbox.clone());
|
|
if let Err(e) = stream.set_write_timeout(Some(SUBSCRIPTION_WRITE_TIMEOUT)) {
|
|
warn!(error = %e, "control: could not set subscription timeout");
|
|
return;
|
|
}
|
|
let heartbeat = super::events::event_line(&Event::Heartbeat);
|
|
let mut lines: Vec<_> = [
|
|
super::events::event_line(&Event::Status(Box::new(status_response(
|
|
core, username, mountpoint,
|
|
)))),
|
|
super::events::transfers_line(core.transfers.snapshot(), core.jobs_snapshot()),
|
|
super::events::event_line(&Event::Changed {
|
|
topics: Topic::ALL.to_vec(),
|
|
}),
|
|
]
|
|
.into_iter()
|
|
.flatten()
|
|
.collect();
|
|
loop {
|
|
for line in &lines {
|
|
if stream.write_all(line.as_bytes()).is_err() {
|
|
return;
|
|
}
|
|
}
|
|
if core.shutdown.is_stopping() {
|
|
return;
|
|
}
|
|
lines = mailbox.take(HEARTBEAT);
|
|
if lines.is_empty() {
|
|
lines.extend(heartbeat.clone());
|
|
}
|
|
}
|
|
}
|
|
|
|
fn handle_control_conn(core: &Core, username: &str, mountpoint: &Path, stream: UnixStream) {
|
|
if let Err(e) = stream.set_read_timeout(Some(CONTROL_READ_TIMEOUT)) {
|
|
warn!(error = %e, "control: could not set request timeout");
|
|
return;
|
|
}
|
|
if let Err(e) = stream.set_write_timeout(Some(CONTROL_WRITE_TIMEOUT)) {
|
|
warn!(error = %e, "control: could not set response timeout");
|
|
return;
|
|
}
|
|
let mut reader = BufReader::new(match stream.try_clone() {
|
|
Ok(s) => s,
|
|
Err(e) => {
|
|
warn!(error = %e, "control: clone stream failed");
|
|
return;
|
|
}
|
|
});
|
|
let line = match read_request_line(&mut reader) {
|
|
Ok(Some(line)) if !line.trim().is_empty() => line,
|
|
Ok(_) => return,
|
|
Err(e) => {
|
|
warn!(error = %e, "control: rejected request");
|
|
return;
|
|
}
|
|
};
|
|
let request = serde_json::from_str::<CtlRequest>(line.trim());
|
|
// A subscription lasts as long as its client, so it is not a request in
|
|
// flight and stays out of the diagnostics' list of them.
|
|
if let Ok(CtlRequest::Subscribe) = request {
|
|
serve_subscription(core, username, mountpoint, stream);
|
|
return;
|
|
}
|
|
let touched = request.as_ref().map_or(&[][..], touched_topics);
|
|
// Registered before the request is even parsed, so a handler that wedges
|
|
// inside the daemon shows up by name in `pdfs diagnostics`.
|
|
let _inflight = super::diagnostics::InflightGuard::new(super::diagnostics::request_kind(&line));
|
|
let response = match request {
|
|
Ok(CtlRequest::Status) => status_response(core, username, mountpoint),
|
|
Ok(CtlRequest::Subscribe) => {
|
|
CtlResponse::error(CoreError::invalid("a subscription is served on its own"))
|
|
}
|
|
Ok(CtlRequest::Pin { path }) => match rel_to_mount(mountpoint, &path) {
|
|
Ok(rel) => match core.pin(&rel) {
|
|
Ok(name) => {
|
|
core.touch_profile();
|
|
CtlResponse::Ok {
|
|
message: format!("pinned {name}"),
|
|
}
|
|
}
|
|
Err(e) => CtlResponse::error(e),
|
|
},
|
|
Err(e) => CtlResponse::error(e),
|
|
},
|
|
Ok(CtlRequest::Unpin { path }) => match rel_to_mount(mountpoint, &path) {
|
|
Ok(rel) => match core.unpin(&rel) {
|
|
Ok(name) => {
|
|
core.touch_profile();
|
|
CtlResponse::Ok {
|
|
message: format!("unpinned {name}"),
|
|
}
|
|
}
|
|
Err(e) => CtlResponse::error(e),
|
|
},
|
|
Err(e) => CtlResponse::error(e),
|
|
},
|
|
Ok(CtlRequest::ListPins) => CtlResponse::Pins {
|
|
pins: core.list_pins(),
|
|
},
|
|
Ok(CtlRequest::ListDir { path }) => match route_to_mount(core, mountpoint, &path) {
|
|
Ok((core, rel)) => match core.list_dir(&rel) {
|
|
Ok(entries) => CtlResponse::Entries { entries },
|
|
Err(e) => CtlResponse::error(e),
|
|
},
|
|
Err(e) => CtlResponse::error(e),
|
|
},
|
|
Ok(CtlRequest::Refresh { scope }) => match RefreshGuard::acquire(&scope) {
|
|
// The same scope is already being refreshed by another handler.
|
|
// Running it again cannot make the answer newer, and the duplicate
|
|
// would spend its whole run contending for the state and database
|
|
// locks the first one holds — which is how five stacked photo
|
|
// refreshes sat in flight for twenty-three minutes at mount time.
|
|
None => {
|
|
debug!(scope = %scope_key(&scope), "a refresh of this scope is already running");
|
|
CtlResponse::Ok {
|
|
message: "refreshing".to_string(),
|
|
}
|
|
}
|
|
Some(_refreshing) => {
|
|
let result = match &scope {
|
|
RefreshScope::Dir { path } => route_to_mount(core, mountpoint, path)
|
|
.and_then(|(core, rel)| core.refresh_dir(&rel)),
|
|
RefreshScope::Trash => {
|
|
core.invalidate_trash();
|
|
Ok(())
|
|
}
|
|
RefreshScope::Photos { full } if !full && core.timeline_refreshing() => {
|
|
// A refresh is already running and will publish exactly the
|
|
// timeline this request asks for. Doing the work again
|
|
// cannot make it newer, and the invalidation plus the
|
|
// album read contend with the running refresh for the
|
|
// database lock — which is how a handful of stacked photo
|
|
// refreshes at mount time left five control requests
|
|
// in flight for twenty-three minutes.
|
|
debug!("a photos refresh is already running; joining it");
|
|
Ok(())
|
|
}
|
|
RefreshScope::Photos { full } => {
|
|
core.invalidate_photos();
|
|
// A full refresh is "forget what you know", not a second code
|
|
// path: the ordinary refresh then finds every photo stale.
|
|
if let Some(Err(e)) = full.then(|| core.db.photos_mark_all_unresolved()) {
|
|
warn!(error = %e, "marking the whole timeline unresolved failed");
|
|
}
|
|
// Whatever changed the timeline can just as easily have
|
|
// changed an album; one scope covers the whole photos view.
|
|
core.invalidate_albums();
|
|
// The refresh is started here, not awaited. A library of
|
|
// twelve thousand photos is re-read in chunks of
|
|
// `TIMELINE_ENRICH_CHUNK` nodes, which takes minutes —
|
|
// longer than the client's read bound, so awaiting it
|
|
// returned `Resource temporarily unavailable` to a caller
|
|
// whose refresh was in fact running fine (B92). The front
|
|
// end follows it with `PhotosRefreshStatus`.
|
|
core.spawn_timeline_refresh();
|
|
Ok(())
|
|
}
|
|
};
|
|
match result {
|
|
Ok(()) => CtlResponse::Ok {
|
|
message: match scope {
|
|
RefreshScope::Photos { .. } => "refreshing".to_string(),
|
|
_ => "refreshed".to_string(),
|
|
},
|
|
},
|
|
Err(e) => CtlResponse::error(e),
|
|
}
|
|
}
|
|
},
|
|
Ok(CtlRequest::PhotosTimeline {
|
|
offset,
|
|
limit,
|
|
kind,
|
|
range,
|
|
favorites,
|
|
not_in_album,
|
|
}) => match core.photos_timeline(
|
|
offset,
|
|
clamp_limit(limit),
|
|
kind,
|
|
range,
|
|
favorites,
|
|
not_in_album,
|
|
) {
|
|
Ok(Some(items)) => CtlResponse::Photos {
|
|
available: true,
|
|
items,
|
|
counts: core.db.photos_counts().ok(),
|
|
},
|
|
Ok(None) => CtlResponse::Photos {
|
|
available: false,
|
|
items: Vec::new(),
|
|
counts: None,
|
|
},
|
|
Err(e) => CtlResponse::error(e),
|
|
},
|
|
Ok(CtlRequest::PhotoDuplicates) => match core.photo_duplicates() {
|
|
Ok(sets) => CtlResponse::PhotoDuplicates { sets },
|
|
Err(e) => CtlResponse::error(e),
|
|
},
|
|
Ok(CtlRequest::PhotoSimilar) => match core.photo_similar() {
|
|
Ok((sets, hashing, uncompared)) => CtlResponse::PhotoSimilar {
|
|
sets,
|
|
hashing,
|
|
uncompared,
|
|
},
|
|
Err(e) => CtlResponse::error(e),
|
|
},
|
|
Ok(CtlRequest::PhotoPlaces) => match core.photo_places() {
|
|
Ok(items) => CtlResponse::Places {
|
|
items,
|
|
mapping: core.timeline_progress(),
|
|
},
|
|
Err(e) => CtlResponse::error(e),
|
|
},
|
|
Ok(CtlRequest::PlacePhotos { id, offset, limit }) => {
|
|
match core.place_photos(id, offset, clamp_limit(limit)) {
|
|
Ok(items) => CtlResponse::Photos {
|
|
available: true,
|
|
items,
|
|
counts: None,
|
|
},
|
|
Err(e) => CtlResponse::error(e),
|
|
}
|
|
}
|
|
Ok(CtlRequest::PlaceLocations { id }) => match core.place_locations(id) {
|
|
Ok(items) => CtlResponse::PhotoLocations { items },
|
|
Err(e) => CtlResponse::error(e),
|
|
},
|
|
Ok(CtlRequest::PhotosOnThisDay { limit }) => {
|
|
match core.photos_on_this_day(clamp_limit(limit)) {
|
|
Ok(items) => CtlResponse::Photos {
|
|
available: true,
|
|
items,
|
|
counts: None,
|
|
},
|
|
Err(e) => CtlResponse::error(e),
|
|
}
|
|
}
|
|
Ok(CtlRequest::PhotoGroup { uid }) => match core.photo_group(&uid) {
|
|
Ok(items) => CtlResponse::Photos {
|
|
available: true,
|
|
items,
|
|
counts: None,
|
|
},
|
|
Err(e) => CtlResponse::error(e),
|
|
},
|
|
Ok(CtlRequest::SetPhotoFavorite { uid, favorite }) => match parse_uid(&uid) {
|
|
Some(node_uid) => match core.set_photo_favorite(&node_uid, favorite) {
|
|
Ok(()) => CtlResponse::Ok {
|
|
message: if favorite {
|
|
"added to favorites".to_string()
|
|
} else {
|
|
"removed from favorites".to_string()
|
|
},
|
|
},
|
|
Err(e) => CtlResponse::error(e),
|
|
},
|
|
None => CtlResponse::error(CoreError::invalid(format!("invalid uid: {uid}"))),
|
|
},
|
|
Ok(CtlRequest::PhotoAlbums) => match core.albums() {
|
|
Ok(Some(items)) => CtlResponse::Albums {
|
|
available: true,
|
|
items,
|
|
},
|
|
Ok(None) => CtlResponse::Albums {
|
|
available: false,
|
|
items: Vec::new(),
|
|
},
|
|
Err(e) => CtlResponse::error(e),
|
|
},
|
|
Ok(CtlRequest::AlbumPhotos { uid, offset, limit }) => match parse_uid(&uid) {
|
|
// An album page is a timeline page as far as the front-end is
|
|
// concerned, so it comes back in the same reply — with `counts`
|
|
// unset, which describe the whole timeline rather than this album.
|
|
Some(album) => match core.album_photos(&album, offset, clamp_limit(limit)) {
|
|
Ok(items) => CtlResponse::Photos {
|
|
available: true,
|
|
items,
|
|
counts: None,
|
|
},
|
|
Err(e) => CtlResponse::error(e),
|
|
},
|
|
None => CtlResponse::error(CoreError::invalid(format!("bad album uid: {uid}"))),
|
|
},
|
|
Ok(CtlRequest::CreateAlbum { name }) => match core.create_album(&name) {
|
|
Ok(uid) => CtlResponse::AlbumCreated { uid },
|
|
Err(e) => CtlResponse::error(e),
|
|
},
|
|
Ok(CtlRequest::RenameAlbum { uid, name }) => match parse_uid(&uid) {
|
|
Some(album) => match core.rename_album(&album, &name) {
|
|
Ok(()) => CtlResponse::Ok {
|
|
message: format!("renamed album to {name}"),
|
|
},
|
|
Err(e) => CtlResponse::error(e),
|
|
},
|
|
None => CtlResponse::error(CoreError::invalid(format!("bad album uid: {uid}"))),
|
|
},
|
|
Ok(CtlRequest::DeleteAlbum { uid, delete_photos }) => match parse_uid(&uid) {
|
|
Some(album) => match core.delete_album(&album, delete_photos) {
|
|
Ok(()) => CtlResponse::Ok {
|
|
message: "album deleted".to_string(),
|
|
},
|
|
Err(e) => CtlResponse::error(e),
|
|
},
|
|
None => CtlResponse::error(CoreError::invalid(format!("bad album uid: {uid}"))),
|
|
},
|
|
Ok(CtlRequest::AddToAlbum { uid, photos }) => {
|
|
album_change(&uid, &photos, |album, photos| {
|
|
core.add_to_album(album, photos)
|
|
})
|
|
}
|
|
Ok(CtlRequest::RemoveFromAlbum { uid, photos }) => {
|
|
album_change(&uid, &photos, |album, photos| {
|
|
core.remove_from_album(album, photos)
|
|
})
|
|
}
|
|
Ok(CtlRequest::PhotoMonths { kind }) => match core.db.photos_months(kind) {
|
|
Ok(months) => CtlResponse::PhotoMonths {
|
|
months: months
|
|
.into_iter()
|
|
.map(|(year, month, count)| PhotoMonth { year, month, count })
|
|
.collect(),
|
|
},
|
|
Err(e) => CtlResponse::error(e.into()),
|
|
},
|
|
Ok(CtlRequest::PhotoThumbs { uids }) => {
|
|
let parsed: Vec<NodeUid> = uids.iter().filter_map(|u| parse_uid(u)).collect();
|
|
CtlResponse::Thumbs {
|
|
items: core.photo_thumbs(&parsed),
|
|
}
|
|
}
|
|
Ok(CtlRequest::TrashNodes { uids, files_only }) => match Core::parse_uids(&uids) {
|
|
Ok(parsed) => match core.trash_photos(&parsed, files_only) {
|
|
Ok((trashed, failed)) => CtlResponse::Trashed {
|
|
trashed,
|
|
failed: failed
|
|
.into_iter()
|
|
.map(|(uid, message)| TrashFailure { uid, message })
|
|
.collect(),
|
|
},
|
|
Err(e) => CtlResponse::error(e),
|
|
},
|
|
Err(e) => CtlResponse::error(e),
|
|
},
|
|
Ok(CtlRequest::FileThumbs { items, generation }) => {
|
|
match core.file_thumbs(&items, generation) {
|
|
Some(items) => CtlResponse::Thumbs { items },
|
|
None => CtlResponse::FileThumbsStale,
|
|
}
|
|
}
|
|
Ok(CtlRequest::ReserveFileThumbGeneration) => CtlResponse::FileThumbGeneration {
|
|
generation: core.reserve_file_thumb_generation(),
|
|
},
|
|
Ok(CtlRequest::CancelFileThumbs { generation }) => {
|
|
core.cancel_file_thumbs(generation);
|
|
CtlResponse::Ok {
|
|
message: "thumbnail requests cancelled".to_string(),
|
|
}
|
|
}
|
|
Ok(CtlRequest::StartThumbnailBuild { path }) => match rel_to_mount(mountpoint, &path) {
|
|
Ok(rel) => match core.start_thumbnail_build(rel) {
|
|
Ok(status) => CtlResponse::ThumbnailBuild { status },
|
|
Err(error) => CtlResponse::error(error),
|
|
},
|
|
Err(error) => CtlResponse::error(error),
|
|
},
|
|
Ok(CtlRequest::CancelThumbnailBuild) => CtlResponse::ThumbnailBuild {
|
|
status: core.cancel_thumbnail_build(),
|
|
},
|
|
Ok(CtlRequest::ThumbnailBuildStatus) => CtlResponse::ThumbnailBuild {
|
|
status: core.thumbnail_build_status(),
|
|
},
|
|
Ok(CtlRequest::PhotosRefreshStatus) => CtlResponse::PhotosRefresh {
|
|
running: core.timeline_refreshing(),
|
|
},
|
|
Ok(CtlRequest::OpenPhoto { uid }) => match parse_uid(&uid) {
|
|
Some(u) => match core.open_photo(&u) {
|
|
Ok(p) => CtlResponse::FilePath {
|
|
path: p.display().to_string(),
|
|
},
|
|
Err(e) => CtlResponse::error(e),
|
|
},
|
|
None => CtlResponse::error(CoreError::invalid(format!("bad uid: {uid}"))),
|
|
},
|
|
// The scan alone walks every archive's central directory, so even the
|
|
// dry run goes to a background thread and is read back through
|
|
// `ImportStatus` rather than answered inline.
|
|
Ok(CtlRequest::ImportTakeout { archives, dry_run }) => {
|
|
if archives.is_empty() {
|
|
CtlResponse::error(CoreError::invalid("no Takeout archives given"))
|
|
} else {
|
|
let paths: Vec<PathBuf> = archives.into_iter().map(PathBuf::from).collect();
|
|
let missing: Vec<String> = paths
|
|
.iter()
|
|
.filter(|path| !path.is_file())
|
|
.map(|path| path.display().to_string())
|
|
.collect();
|
|
if !missing.is_empty() {
|
|
CtlResponse::error(CoreError::not_found(format!(
|
|
"cannot read: {}",
|
|
missing.join(", ")
|
|
)))
|
|
} else {
|
|
let count = paths.len();
|
|
match core.spawn_takeout_import(paths, dry_run) {
|
|
Ok(()) => CtlResponse::Ok {
|
|
message: format!(
|
|
"{} {count} archive(s)",
|
|
if dry_run { "scanning" } else { "importing" }
|
|
),
|
|
},
|
|
Err(e) => CtlResponse::error(e),
|
|
}
|
|
}
|
|
}
|
|
}
|
|
Ok(CtlRequest::ImportStatus) => {
|
|
let (running, summary) = crate::takeout::import_status();
|
|
CtlResponse::ImportStatus { running, summary }
|
|
}
|
|
Ok(CtlRequest::CancelImport) => {
|
|
if crate::takeout::cancel_import() {
|
|
CtlResponse::Ok {
|
|
message: "stopping the import".to_string(),
|
|
}
|
|
} else {
|
|
CtlResponse::error(CoreError::invalid("no import is running"))
|
|
}
|
|
}
|
|
// The scan reads the whole persisted timeline and every candidate is a
|
|
// download plus an upload, so this goes to a background thread and is
|
|
// read back through `RedateStatus`.
|
|
Ok(CtlRequest::RedatePhotos {
|
|
dry_run,
|
|
limit,
|
|
range,
|
|
}) => match core.spawn_redate(dry_run, limit, range) {
|
|
Ok(()) => CtlResponse::Ok {
|
|
message: if dry_run {
|
|
"checking photo dates".to_string()
|
|
} else {
|
|
"fixing photo dates".to_string()
|
|
},
|
|
},
|
|
Err(e) => CtlResponse::error(e),
|
|
},
|
|
Ok(CtlRequest::RedateStatus) => {
|
|
let (running, summary) = crate::redate::redate_status();
|
|
CtlResponse::RedateStatus { running, summary }
|
|
}
|
|
Ok(CtlRequest::CancelRedate) => {
|
|
if crate::redate::cancel_redate() {
|
|
CtlResponse::Ok {
|
|
message: "stopping the re-date".to_string(),
|
|
}
|
|
} else {
|
|
CtlResponse::error(CoreError::invalid("no photo re-date is running"))
|
|
}
|
|
}
|
|
Ok(CtlRequest::UploadPhoto {
|
|
name,
|
|
media_type,
|
|
source_path,
|
|
capture_time,
|
|
}) => {
|
|
// Streamed off disk rather than carried over the socket: see the note
|
|
// on `Request::UploadPhoto`. The length comes from the same handle the
|
|
// bytes do, so a file changing size under us cannot make the two
|
|
// disagree.
|
|
match std::fs::File::open(&source_path).and_then(|f| f.metadata().map(|m| (f, m.len())))
|
|
{
|
|
Err(e) => CtlResponse::error(CoreError::invalid(format!(
|
|
"cannot read {source_path}: {e}"
|
|
))),
|
|
Ok((file, len)) => {
|
|
let photos = core.photos();
|
|
let metadata = proton_drive_rs::PhotoUploadMetadata {
|
|
capture_time,
|
|
..Default::default()
|
|
};
|
|
let guard =
|
|
core.transfers
|
|
.begin(name.clone(), "", TransferDirection::Upload, len);
|
|
let reader = CountingReader::new(std::io::BufReader::new(file), &guard);
|
|
match core.rt.block_on(photos.upload_photo_from(
|
|
&name,
|
|
&media_type,
|
|
reader,
|
|
len as i64,
|
|
Vec::new(),
|
|
metadata,
|
|
false,
|
|
)) {
|
|
Ok(uid) => {
|
|
// The photo we just uploaded belongs at the head of the
|
|
// timeline, and the GUI reloads the gallery the moment this
|
|
// reply lands — so refresh now rather than leaving it to a
|
|
// background pass that would land just after that reload.
|
|
if let Err(e) = core.rt.block_on(core.refresh_timeline()) {
|
|
warn!(error = %e, "timeline refresh after upload failed");
|
|
}
|
|
CtlResponse::Ok {
|
|
message: format!("uploaded photo with uid {uid}"),
|
|
}
|
|
}
|
|
Err(e) => {
|
|
CtlResponse::error(CoreError::from_api(&e, "upload photo failed"))
|
|
}
|
|
}
|
|
}
|
|
}
|
|
}
|
|
// A uid addresses the node wherever it lives; the path only works for
|
|
// the primary mount's tree. Prefer the uid when the client sent one.
|
|
Ok(CtlRequest::OpenFile { path, uid }) => {
|
|
let opened = match uid.as_deref() {
|
|
Some(uid) => core.open_file_uid(uid),
|
|
None => rel_to_mount(mountpoint, &path).and_then(|rel| core.open_file(&rel)),
|
|
};
|
|
match opened {
|
|
Ok(p) => CtlResponse::FilePath {
|
|
path: p.display().to_string(),
|
|
},
|
|
Err(e) => CtlResponse::error(e),
|
|
}
|
|
}
|
|
Ok(CtlRequest::Search {
|
|
query,
|
|
limit,
|
|
scope,
|
|
}) => match core.search(&query, clamp_limit(limit), scope.as_deref()) {
|
|
Ok(hits) => CtlResponse::SearchResults { hits },
|
|
Err(e) => CtlResponse::error(e),
|
|
},
|
|
Ok(CtlRequest::SearchLocal { query, limit }) => {
|
|
match core.search_local(&query, clamp_limit(limit)) {
|
|
Ok(hits) => CtlResponse::LocalResults {
|
|
hits,
|
|
indexing: core.indexing.load(Ordering::Relaxed),
|
|
},
|
|
Err(e) => CtlResponse::error(e),
|
|
}
|
|
}
|
|
Ok(CtlRequest::SearchV2 {
|
|
query,
|
|
limit,
|
|
filters,
|
|
}) => match core.search_v2(&query, clamp_limit(limit), &filters) {
|
|
Ok((drive_hits, local_hits)) => CtlResponse::SearchResultsV2 {
|
|
drive_hits,
|
|
local_hits,
|
|
local_indexing: core.indexing.load(Ordering::Relaxed),
|
|
},
|
|
Err(e) => CtlResponse::error(e),
|
|
},
|
|
Ok(CtlRequest::Rename { path, new_name }) => match route_node(core, mountpoint, &path) {
|
|
Ok((core, rel)) => match core.rename(&rel, &new_name) {
|
|
Ok(name) => {
|
|
core.log_activity(ActivityKind::Rename, &name, format!("was {path}"), true);
|
|
CtlResponse::Ok {
|
|
message: format!("renamed to {name}"),
|
|
}
|
|
}
|
|
Err(e) => {
|
|
core.log_activity(ActivityKind::Rename, &path, &e, false);
|
|
CtlResponse::error(e)
|
|
}
|
|
},
|
|
Err(e) => CtlResponse::error(e),
|
|
},
|
|
// Two absolute paths may name two different locations, which only
|
|
// `move_between` can resolve; a relative pair keeps the old meaning.
|
|
Ok(CtlRequest::Move { path, new_parent })
|
|
if Path::new(&path).is_absolute() && Path::new(&new_parent).is_absolute() =>
|
|
{
|
|
match core.move_between(Path::new(&path), Path::new(&new_parent)) {
|
|
Ok(name) => {
|
|
core.log_activity(ActivityKind::Move, &name, format!("to {new_parent}"), true);
|
|
CtlResponse::Ok {
|
|
message: format!("moved {name}"),
|
|
}
|
|
}
|
|
Err(e) => {
|
|
core.log_activity(ActivityKind::Move, &path, &e, false);
|
|
CtlResponse::error(e)
|
|
}
|
|
}
|
|
}
|
|
Ok(CtlRequest::Move { path, new_parent }) => {
|
|
match (
|
|
rel_to_mount(mountpoint, &path),
|
|
rel_to_mount(mountpoint, &new_parent),
|
|
) {
|
|
(Ok(rel), Ok(parent_rel)) => match core.move_to(&rel, &parent_rel) {
|
|
Ok(name) => {
|
|
let dest = if new_parent.is_empty() {
|
|
"My files".to_string()
|
|
} else {
|
|
new_parent.clone()
|
|
};
|
|
core.log_activity(ActivityKind::Move, &name, format!("to {dest}"), true);
|
|
CtlResponse::Ok {
|
|
message: format!("moved {name}"),
|
|
}
|
|
}
|
|
Err(e) => {
|
|
core.log_activity(ActivityKind::Move, &path, &e, false);
|
|
CtlResponse::error(e)
|
|
}
|
|
},
|
|
(Err(e), _) | (_, Err(e)) => CtlResponse::error(e),
|
|
}
|
|
}
|
|
Ok(CtlRequest::Delete { path }) => match route_node(core, mountpoint, &path) {
|
|
Ok((core, rel)) => match core.delete(&rel) {
|
|
Ok(name) => {
|
|
core.log_activity(ActivityKind::Trash, &name, "", true);
|
|
CtlResponse::Ok {
|
|
message: format!("trashed {name}"),
|
|
}
|
|
}
|
|
Err(e) => {
|
|
core.log_activity(ActivityKind::Trash, &path, &e, false);
|
|
CtlResponse::error(e)
|
|
}
|
|
},
|
|
Err(e) => CtlResponse::error(e),
|
|
},
|
|
Ok(CtlRequest::CreateFolder { parent, name }) => {
|
|
match route_change(core, mountpoint, &parent) {
|
|
Ok((core, parent_rel)) => match core.create_folder(&parent_rel, &name) {
|
|
Ok(name) => {
|
|
core.log_activity(ActivityKind::CreateFolder, &name, "", true);
|
|
CtlResponse::Ok {
|
|
message: format!("created folder {name}"),
|
|
}
|
|
}
|
|
Err(e) => {
|
|
core.log_activity(ActivityKind::CreateFolder, &name, &e, false);
|
|
CtlResponse::error(e)
|
|
}
|
|
},
|
|
Err(e) => CtlResponse::error(e),
|
|
}
|
|
}
|
|
Ok(CtlRequest::UploadPaths { parent, sources }) => {
|
|
match rel_to_mount(mountpoint, &parent) {
|
|
// Ack immediately and upload on a background thread: a directory tree
|
|
// far outlasts the socket read timeout. Progress and completion are
|
|
// observed through GetQueueStatus; the activity log gets the summary
|
|
// when the whole batch finishes.
|
|
Ok(parent_rel) => {
|
|
let core = core.clone();
|
|
let paths: Vec<PathBuf> = sources.into_iter().map(PathBuf::from).collect();
|
|
let n = paths.len();
|
|
std::thread::spawn(move || {
|
|
let started = Instant::now();
|
|
match core.upload_paths(&parent_rel, &paths) {
|
|
Ok(stats) => {
|
|
// e.g. "Uploaded 700 files to Photos" — name the destination
|
|
// so the log reads like a sentence rather than a bare count.
|
|
let dest = parent_rel
|
|
.file_name()
|
|
.and_then(|n| n.to_str())
|
|
.unwrap_or("your Drive");
|
|
let target = format!(
|
|
"{} to {dest}",
|
|
count_noun(stats.uploaded, "file", "files")
|
|
);
|
|
// Size · folders · elapsed, with a trailing failure count
|
|
// when some files didn't make it.
|
|
let mut parts = vec![human_bytes(stats.bytes)];
|
|
if stats.folders > 0 {
|
|
parts.push(count_noun(stats.folders, "folder", "folders"));
|
|
}
|
|
parts.push(human_duration(started.elapsed()));
|
|
if stats.failed > 0 {
|
|
parts.push(format!("{} failed", stats.failed));
|
|
}
|
|
core.log_activity(
|
|
ActivityKind::Upload,
|
|
target,
|
|
parts.join(" · "),
|
|
stats.failed == 0,
|
|
);
|
|
}
|
|
Err(e) => {
|
|
warn!(error = %e, "bulk upload failed");
|
|
core.log_activity(ActivityKind::Upload, "bulk upload", &e, false);
|
|
}
|
|
}
|
|
});
|
|
CtlResponse::Ok {
|
|
message: format!("uploading {n} item(s)"),
|
|
}
|
|
}
|
|
Err(e) => CtlResponse::error(e),
|
|
}
|
|
}
|
|
Ok(CtlRequest::Diagnostics) => CtlResponse::Diagnostics {
|
|
diagnostics: core.diagnostics(),
|
|
},
|
|
Ok(CtlRequest::ListTrash) => match core.list_trash() {
|
|
Ok(entries) => CtlResponse::Entries { entries },
|
|
Err(e) => CtlResponse::error(e),
|
|
},
|
|
Ok(CtlRequest::Restore { uids }) => match core.restore(&uids) {
|
|
Ok(n) => {
|
|
core.log_activity(ActivityKind::Restore, format!("{n} item(s)"), "", true);
|
|
CtlResponse::Ok {
|
|
message: format!("restored {n} item(s)"),
|
|
}
|
|
}
|
|
Err(e) => CtlResponse::error(e),
|
|
},
|
|
Ok(CtlRequest::DeleteForever { uids }) => match core.delete_forever(&uids) {
|
|
Ok(n) => {
|
|
core.log_activity(
|
|
ActivityKind::DeleteForever,
|
|
format!("{n} item(s)"),
|
|
"",
|
|
true,
|
|
);
|
|
CtlResponse::Ok {
|
|
message: format!("permanently deleted {n} item(s)"),
|
|
}
|
|
}
|
|
Err(e) => CtlResponse::error(e),
|
|
},
|
|
Ok(CtlRequest::EmptyTrash) => match core.empty_trash() {
|
|
Ok(n) => {
|
|
let items = if n == 1 {
|
|
"1 item".to_string()
|
|
} else {
|
|
format!("{n} items")
|
|
};
|
|
core.log_activity(ActivityKind::EmptyTrash, items.clone(), "", true);
|
|
CtlResponse::Ok {
|
|
message: format!("emptied trash ({items})"),
|
|
}
|
|
}
|
|
Err(e) => CtlResponse::error(e),
|
|
},
|
|
Ok(CtlRequest::PurgeCache) => {
|
|
let freed = core.cache.clear_unpinned();
|
|
CtlResponse::Ok {
|
|
message: format!(
|
|
"purged {:.1} MiB of unpinned cache",
|
|
freed as f64 / 1_048_576.0
|
|
),
|
|
}
|
|
}
|
|
Ok(CtlRequest::CacheInspect { deep }) => match core.db.stats() {
|
|
Ok(stats) => {
|
|
// The deep check reads every page — tens of seconds on a real
|
|
// database, all of it holding the connection every FUSE handler
|
|
// needs — so it runs on a thread and this request returns
|
|
// whatever the last completed check found. `integrity_checked`
|
|
// tells the caller which answer it is holding, since "no
|
|
// problems found" and "never looked" are both an empty list;
|
|
// `integrity_running` says a fresh answer is on its way.
|
|
let (integrity_problems, integrity_checked, integrity_running) = if deep {
|
|
core.start_or_read_integrity_check()
|
|
} else {
|
|
(Vec::new(), false, false)
|
|
};
|
|
CtlResponse::CacheReport {
|
|
integrity_running,
|
|
schema_version: stats.schema_version,
|
|
db_bytes: stats.total_bytes(),
|
|
db_reclaimable_bytes: stats.reclaimable_bytes(),
|
|
tables: stats.tables,
|
|
cache_used: core.cache.usage(),
|
|
cache_budget: core.cache.budget(),
|
|
integrity_problems,
|
|
integrity_checked,
|
|
}
|
|
}
|
|
Err(e) => CtlResponse::error(CoreError::internal(format!("read db stats: {e:?}"))),
|
|
},
|
|
// Ack and vacuum on a thread. `VACUUM` rewrites the whole database
|
|
// under the connection mutex — minutes on a large one — and the
|
|
// request's own socket timeout is 10 s, so running it inline both froze
|
|
// the mount and lost its own answer. Progress shows up as a job in
|
|
// `GetQueueStatus`; the outcome goes to the log.
|
|
Ok(CtlRequest::CacheVacuum) => match core.start_vacuum() {
|
|
Ok(()) => CtlResponse::Ok {
|
|
message: "compacting the cache database in the background".into(),
|
|
},
|
|
Err(e) => CtlResponse::error(e),
|
|
},
|
|
Ok(CtlRequest::GetQueueStatus) => CtlResponse::Transfers {
|
|
items: core.transfers.snapshot(),
|
|
jobs: core.jobs_snapshot(),
|
|
},
|
|
Ok(CtlRequest::SetCacheBudget { bytes }) => {
|
|
core.cache.set_budget(bytes);
|
|
// Persist so the next mount keeps the new cap. Best-effort: a config
|
|
// write failure is reported but the live cap is already applied.
|
|
match AppDirs::new().map(|dirs| {
|
|
let mut cfg = dirs.load_config();
|
|
cfg.cache_budget = Some(bytes);
|
|
dirs.save_config(&cfg)
|
|
}) {
|
|
Ok(Ok(())) => {
|
|
core.touch_profile();
|
|
CtlResponse::Ok {
|
|
message: format!("cache budget set to {bytes} bytes"),
|
|
}
|
|
}
|
|
Ok(Err(e)) => CtlResponse::error(CoreError::internal(format!(
|
|
"budget applied but config write failed: {e}"
|
|
))),
|
|
Err(e) => CtlResponse::error(CoreError::internal(format!(
|
|
"budget applied but config unavailable: {e}"
|
|
))),
|
|
}
|
|
}
|
|
Ok(CtlRequest::SetBandwidthLimits { upload, download }) => {
|
|
core.transfers.set_limits(upload, download);
|
|
// Persisted like the cache budget: best-effort, the live caps are
|
|
// already applied.
|
|
match AppDirs::new().map(|dirs| {
|
|
let mut cfg = dirs.load_config();
|
|
cfg.upload_limit = Some(upload);
|
|
cfg.download_limit = Some(download);
|
|
dirs.save_config(&cfg)
|
|
}) {
|
|
Ok(Ok(())) => CtlResponse::Ok {
|
|
message: format!(
|
|
"bandwidth limits set to {upload} B/s up, {download} B/s down"
|
|
),
|
|
},
|
|
Ok(Err(e)) => CtlResponse::error(CoreError::internal(format!(
|
|
"limits applied but config write failed: {e}"
|
|
))),
|
|
Err(e) => CtlResponse::error(CoreError::internal(format!(
|
|
"limits applied but config unavailable: {e}"
|
|
))),
|
|
}
|
|
}
|
|
Ok(CtlRequest::ListDevices) => match core.list_devices() {
|
|
Ok(items) => CtlResponse::Devices { items },
|
|
Err(e) => CtlResponse::error(e),
|
|
},
|
|
Ok(CtlRequest::ListLocations) => match core.list_locations() {
|
|
Ok(items) => CtlResponse::Locations { items },
|
|
Err(e) => CtlResponse::error(e),
|
|
},
|
|
Ok(CtlRequest::RenameDevice { uid, name }) => match core.rename_device(&uid, &name) {
|
|
Ok(()) => CtlResponse::Ok {
|
|
message: format!("renamed device to {name}"),
|
|
},
|
|
Err(e) => CtlResponse::error(e),
|
|
},
|
|
Ok(CtlRequest::DeleteDevice { uid }) => match core.delete_device(&uid) {
|
|
Ok(()) => CtlResponse::Ok {
|
|
message: "device deleted".to_string(),
|
|
},
|
|
Err(e) => CtlResponse::error(e),
|
|
},
|
|
Ok(CtlRequest::AdoptDevice { uid }) => match core.adopt_device(uid.as_deref()) {
|
|
Ok(message) => CtlResponse::Ok { message },
|
|
Err(e) => CtlResponse::error(e),
|
|
},
|
|
Ok(CtlRequest::MigrateDevice { uid }) => match core.migrate_device(&uid) {
|
|
Ok(message) => CtlResponse::Ok { message },
|
|
Err(e) => CtlResponse::error(e),
|
|
},
|
|
Ok(CtlRequest::AddSyncFolder { local_path }) => {
|
|
// Registering the device and uploading a folder tree far outlasts the
|
|
// socket read timeout, so ack immediately and work on a background
|
|
// thread. The folder appears in ListSyncFolders once the row lands;
|
|
// completion (and any failures) go to the activity log.
|
|
let core = core.clone();
|
|
let path = PathBuf::from(&local_path);
|
|
std::thread::spawn(move || {
|
|
let started = Instant::now();
|
|
match core.add_sync_folder(&path) {
|
|
Ok(folder) => {
|
|
core.touch_profile();
|
|
let name = Path::new(&folder.local_path)
|
|
.file_name()
|
|
.and_then(|n| n.to_str())
|
|
.unwrap_or(&folder.local_path)
|
|
.to_string();
|
|
core.log_activity(
|
|
ActivityKind::Upload,
|
|
format!("synced {name}"),
|
|
human_duration(started.elapsed()),
|
|
folder.state != "error",
|
|
);
|
|
}
|
|
Err(e) => {
|
|
warn!(error = %e, "add sync folder failed");
|
|
core.log_activity(ActivityKind::Upload, "add sync folder", &e, false);
|
|
}
|
|
}
|
|
});
|
|
CtlResponse::Ok {
|
|
message: format!("adding {local_path}"),
|
|
}
|
|
}
|
|
Ok(CtlRequest::ListSyncFolders) => match core.list_sync_folders() {
|
|
Ok(items) => CtlResponse::SyncFolders { items },
|
|
Err(e) => CtlResponse::error(e),
|
|
},
|
|
Ok(CtlRequest::RemoveSyncFolder { id, delete_remote }) => {
|
|
match core.remove_sync_folder(id, delete_remote) {
|
|
Ok(()) => {
|
|
core.touch_profile();
|
|
CtlResponse::Ok {
|
|
message: "removed synced folder".to_string(),
|
|
}
|
|
}
|
|
Err(e) => CtlResponse::error(e),
|
|
}
|
|
}
|
|
Ok(CtlRequest::SetSyncFolderMode { id, mode }) => {
|
|
match core.request_sync_folder_mode(id, &mode) {
|
|
Ok(message) => {
|
|
core.touch_profile();
|
|
core.log_activity(ActivityKind::Upload, &message, "", true);
|
|
CtlResponse::Ok { message }
|
|
}
|
|
Err(e) => CtlResponse::error(e),
|
|
}
|
|
}
|
|
Ok(CtlRequest::SetSyncFolderPaused { id, paused }) => {
|
|
match core.set_sync_folder_paused(id, paused) {
|
|
Ok(()) => CtlResponse::Ok {
|
|
message: match paused {
|
|
true => format!("folder {id} paused"),
|
|
false => format!("folder {id} resumed"),
|
|
},
|
|
},
|
|
Err(e) => CtlResponse::error(e),
|
|
}
|
|
}
|
|
Ok(CtlRequest::SyncNow { id }) => {
|
|
core.sync_now(id);
|
|
CtlResponse::Ok {
|
|
message: match id {
|
|
Some(id) => format!("reconciling folder {id}"),
|
|
None => "reconciling all folders".to_string(),
|
|
},
|
|
}
|
|
}
|
|
Ok(CtlRequest::SetSyncPaused { paused, until }) => {
|
|
match core.set_sync_paused(paused, until) {
|
|
Ok(()) => CtlResponse::Ok {
|
|
message: match (paused, until) {
|
|
(false, _) => "sync resumed".to_string(),
|
|
(true, None) => "sync paused until resumed".to_string(),
|
|
(true, Some(_)) => "sync paused".to_string(),
|
|
},
|
|
},
|
|
Err(e) => CtlResponse::error(e),
|
|
}
|
|
}
|
|
Ok(CtlRequest::ListConflicts) => match core.list_conflicts() {
|
|
Ok(items) => CtlResponse::Conflicts { items },
|
|
Err(e) => CtlResponse::error(e),
|
|
},
|
|
Ok(CtlRequest::ResolveConflict { path, keep }) => {
|
|
match route_change(core, mountpoint, &path) {
|
|
Ok((core, rel)) => match core.resolve_conflict(&rel, &keep) {
|
|
Ok(message) => CtlResponse::Ok { message },
|
|
Err(e) => CtlResponse::error(e),
|
|
},
|
|
Err(e) => CtlResponse::error(e),
|
|
}
|
|
}
|
|
Ok(CtlRequest::ListPendingOps) => match core.pending_op_infos() {
|
|
Ok(items) => CtlResponse::PendingOps { items },
|
|
Err(e) => CtlResponse::error(e),
|
|
},
|
|
Ok(CtlRequest::ExportPendingOp { id, dest }) => {
|
|
match core.export_pending_op(id, Path::new(&dest)) {
|
|
Ok(written) => CtlResponse::Ok {
|
|
message: format!("wrote {} to {dest}", human_bytes(written)),
|
|
},
|
|
Err(e) => CtlResponse::error(e),
|
|
}
|
|
}
|
|
Ok(CtlRequest::DiscardPendingOp { id }) => match core.discard_pending_op(id) {
|
|
Ok(path) => CtlResponse::Ok {
|
|
message: format!("discarded the queued change to {path}"),
|
|
},
|
|
Err(e) => CtlResponse::error(e),
|
|
},
|
|
Ok(CtlRequest::RetryPendingOp { id }) => match core.retry_pending_ops(id) {
|
|
Ok(0) if id.is_some() => CtlResponse::Ok {
|
|
message: "nothing to retry: the operation is already due or waiting for a rename"
|
|
.to_string(),
|
|
},
|
|
Ok(moved) => CtlResponse::Ok {
|
|
message: format!("retrying {moved} queued operation(s) now"),
|
|
},
|
|
Err(e) => CtlResponse::error(e),
|
|
},
|
|
Ok(CtlRequest::ListRestorableFolders) => match core.list_restorable_folders(None) {
|
|
Ok(items) => CtlResponse::RestorableFolders { items },
|
|
Err(e) => CtlResponse::error(e),
|
|
},
|
|
Ok(CtlRequest::ListDeviceRestorableFolders { device }) => {
|
|
match core.list_restorable_folders(Some(&device)) {
|
|
Ok(items) => CtlResponse::RestorableFolders { items },
|
|
Err(e) => CtlResponse::error(e),
|
|
}
|
|
}
|
|
Ok(
|
|
request @ (CtlRequest::RestoreSyncFolders { .. }
|
|
| CtlRequest::RestoreDeviceFolders { .. }),
|
|
) => {
|
|
let (device, items) = match request {
|
|
CtlRequest::RestoreDeviceFolders { device, items } => (Some(device), items),
|
|
CtlRequest::RestoreSyncFolders { items } => (None, items),
|
|
_ => unreachable!("matched above"),
|
|
};
|
|
// Like AddSyncFolder: each restored folder downloads a whole tree,
|
|
// which outlasts the socket timeout. Ack, then work.
|
|
let core = core.clone();
|
|
std::thread::spawn(move || {
|
|
match core.restore_sync_folders(&items, device.as_deref()) {
|
|
Ok(message) => {
|
|
core.touch_profile();
|
|
core.log_activity(ActivityKind::Download, &message, "", true);
|
|
}
|
|
Err(e) => {
|
|
warn!(error = %e, "restore failed");
|
|
core.log_activity(ActivityKind::Download, "restore folders", &e, false);
|
|
}
|
|
}
|
|
});
|
|
CtlResponse::Ok {
|
|
message: "restoring folders".to_string(),
|
|
}
|
|
}
|
|
Ok(CtlRequest::ShareNode {
|
|
path,
|
|
emails,
|
|
role,
|
|
message,
|
|
}) => match rel_to_mount(mountpoint, &path) {
|
|
Ok(rel) => match core.share_node(&rel, &emails, &role, message.as_deref()) {
|
|
Ok((proton, external)) => {
|
|
core.log_activity(
|
|
ActivityKind::Share,
|
|
&path,
|
|
format!("{} recipient(s) as {role}", proton + external),
|
|
true,
|
|
);
|
|
CtlResponse::Ok {
|
|
message: format!("invited {proton} Proton and {external} external user(s)"),
|
|
}
|
|
}
|
|
Err(e) => {
|
|
core.log_activity(ActivityKind::Share, &path, &e, false);
|
|
CtlResponse::error(e)
|
|
}
|
|
},
|
|
Err(e) => CtlResponse::error(e),
|
|
},
|
|
Ok(CtlRequest::ShareNodeByUid {
|
|
uid,
|
|
emails,
|
|
role,
|
|
message,
|
|
}) => match core.share_node_by_uid(&uid, &emails, &role, message.as_deref()) {
|
|
Ok((proton, external)) => {
|
|
core.log_activity(
|
|
ActivityKind::Share,
|
|
&uid,
|
|
format!("{} recipient(s) as {role}", proton + external),
|
|
true,
|
|
);
|
|
CtlResponse::Ok {
|
|
message: format!("invited {proton} Proton and {external} external user(s)"),
|
|
}
|
|
}
|
|
Err(e) => {
|
|
core.log_activity(ActivityKind::Share, &uid, &e, false);
|
|
CtlResponse::error(e)
|
|
}
|
|
},
|
|
Ok(CtlRequest::ListShare { path }) => match rel_to_mount(mountpoint, &path) {
|
|
Ok(rel) => match core.list_share(&rel) {
|
|
Ok((entries, link)) => CtlResponse::Share { entries, link },
|
|
Err(e) => CtlResponse::error(e),
|
|
},
|
|
Err(e) => CtlResponse::error(e),
|
|
},
|
|
Ok(CtlRequest::ListShareByUid { uid }) => match core.list_share_by_uid(&uid) {
|
|
Ok((entries, link)) => CtlResponse::Share { entries, link },
|
|
Err(e) => CtlResponse::error(e),
|
|
},
|
|
Ok(CtlRequest::UpdateShareRole {
|
|
path,
|
|
id,
|
|
kind,
|
|
role,
|
|
}) => match rel_to_mount(mountpoint, &path) {
|
|
Ok(rel) => match core.update_share_role(&rel, &id, kind, &role) {
|
|
Ok(()) => CtlResponse::Ok {
|
|
message: format!("role updated to {role}"),
|
|
},
|
|
Err(e) => CtlResponse::error(e),
|
|
},
|
|
Err(e) => CtlResponse::error(e),
|
|
},
|
|
Ok(CtlRequest::UpdateShareRoleByUid {
|
|
uid,
|
|
id,
|
|
kind,
|
|
role,
|
|
}) => match core.update_share_role_by_uid(&uid, &id, kind, &role) {
|
|
Ok(()) => CtlResponse::Ok {
|
|
message: format!("role updated to {role}"),
|
|
},
|
|
Err(e) => CtlResponse::error(e),
|
|
},
|
|
Ok(CtlRequest::RemoveShareEntry { path, id, kind }) => {
|
|
match rel_to_mount(mountpoint, &path) {
|
|
Ok(rel) => match core.remove_share_entry(&rel, &id, kind) {
|
|
Ok(()) => {
|
|
core.log_activity(ActivityKind::Unshare, &path, "access removed", true);
|
|
CtlResponse::Ok {
|
|
message: "removed".to_string(),
|
|
}
|
|
}
|
|
Err(e) => CtlResponse::error(e),
|
|
},
|
|
Err(e) => CtlResponse::error(e),
|
|
}
|
|
}
|
|
Ok(CtlRequest::RemoveShareEntryByUid { uid, id, kind }) => {
|
|
match core.remove_share_entry_by_uid(&uid, &id, kind) {
|
|
Ok(()) => {
|
|
core.log_activity(ActivityKind::Unshare, &uid, "access removed", true);
|
|
CtlResponse::Ok {
|
|
message: "removed".to_string(),
|
|
}
|
|
}
|
|
Err(e) => CtlResponse::error(e),
|
|
}
|
|
}
|
|
Ok(CtlRequest::CreatePublicLink {
|
|
path,
|
|
role,
|
|
password,
|
|
expires,
|
|
}) => match rel_to_mount(mountpoint, &path) {
|
|
Ok(rel) => match core.create_public_link(&rel, &role, password.as_deref(), expires) {
|
|
Ok(link) => {
|
|
core.log_activity(ActivityKind::PublicLink, &path, "link created", true);
|
|
CtlResponse::PublicLink { link }
|
|
}
|
|
Err(e) => {
|
|
core.log_activity(ActivityKind::PublicLink, &path, &e, false);
|
|
CtlResponse::error(e)
|
|
}
|
|
},
|
|
Err(e) => CtlResponse::error(e),
|
|
},
|
|
Ok(CtlRequest::CreatePublicLinkByUid {
|
|
uid,
|
|
role,
|
|
password,
|
|
expires,
|
|
}) => match core.create_public_link_by_uid(&uid, &role, password.as_deref(), expires) {
|
|
Ok(link) => {
|
|
core.log_activity(ActivityKind::PublicLink, &uid, "link created", true);
|
|
CtlResponse::PublicLink { link }
|
|
}
|
|
Err(e) => {
|
|
core.log_activity(ActivityKind::PublicLink, &uid, &e, false);
|
|
CtlResponse::error(e)
|
|
}
|
|
},
|
|
Ok(CtlRequest::RemovePublicLink { path, id }) => match rel_to_mount(mountpoint, &path) {
|
|
Ok(rel) => match core.remove_public_link(&rel, &id) {
|
|
Ok(()) => {
|
|
core.log_activity(ActivityKind::Unshare, &path, "link removed", true);
|
|
CtlResponse::Ok {
|
|
message: "public link removed".to_string(),
|
|
}
|
|
}
|
|
Err(e) => CtlResponse::error(e),
|
|
},
|
|
Err(e) => CtlResponse::error(e),
|
|
},
|
|
Ok(CtlRequest::RemovePublicLinkByUid { uid, id }) => {
|
|
match core.remove_public_link_by_uid(&uid, &id) {
|
|
Ok(()) => {
|
|
core.log_activity(ActivityKind::Unshare, &uid, "link removed", true);
|
|
CtlResponse::Ok {
|
|
message: "public link removed".to_string(),
|
|
}
|
|
}
|
|
Err(e) => CtlResponse::error(e),
|
|
}
|
|
}
|
|
Ok(CtlRequest::StopSharing { path }) => match rel_to_mount(mountpoint, &path) {
|
|
Ok(rel) => stop_sharing_reply(core, &path, core.stop_sharing(&rel)),
|
|
Err(e) => CtlResponse::error(e),
|
|
},
|
|
Ok(CtlRequest::StopSharingByUid { uid }) => {
|
|
stop_sharing_reply(core, &uid, core.stop_sharing_by_uid(&uid))
|
|
}
|
|
Ok(CtlRequest::ListRevisions { path }) => match rel_to_mount(mountpoint, &path) {
|
|
Ok(rel) => match core.list_revisions(&rel) {
|
|
Ok(items) => CtlResponse::Revisions { items },
|
|
Err(e) => CtlResponse::error(e),
|
|
},
|
|
Err(e) => CtlResponse::error(e),
|
|
},
|
|
Ok(CtlRequest::ListRevisionsByUid { uid }) => match core.list_revisions_by_uid(&uid) {
|
|
Ok(items) => CtlResponse::Revisions { items },
|
|
Err(e) => CtlResponse::error(e),
|
|
},
|
|
Ok(CtlRequest::RestoreRevision { path, revision_id }) => {
|
|
match rel_to_mount(mountpoint, &path) {
|
|
Ok(rel) => match core.restore_revision(&rel, &revision_id) {
|
|
Ok(()) => CtlResponse::Ok {
|
|
message: format!("restoring version {revision_id} of {path}"),
|
|
},
|
|
Err(e) => CtlResponse::error(e),
|
|
},
|
|
Err(e) => CtlResponse::error(e),
|
|
}
|
|
}
|
|
Ok(CtlRequest::RestoreRevisionByUid { uid, revision_id }) => {
|
|
match core.restore_revision_by_uid(&uid, &revision_id) {
|
|
Ok(()) => CtlResponse::Ok {
|
|
message: format!("restoring version {revision_id}"),
|
|
},
|
|
Err(e) => CtlResponse::error(e),
|
|
}
|
|
}
|
|
Ok(CtlRequest::DeleteRevision { path, revision_id }) => {
|
|
match rel_to_mount(mountpoint, &path) {
|
|
Ok(rel) => match core.delete_revision(&rel, &revision_id) {
|
|
Ok(()) => CtlResponse::Ok {
|
|
message: format!("deleted version {revision_id} of {path}"),
|
|
},
|
|
Err(e) => CtlResponse::error(e),
|
|
},
|
|
Err(e) => CtlResponse::error(e),
|
|
}
|
|
}
|
|
Ok(CtlRequest::DeleteRevisionByUid { uid, revision_id }) => {
|
|
match core.delete_revision_by_uid(&uid, &revision_id) {
|
|
Ok(()) => CtlResponse::Ok {
|
|
message: format!("deleted version {revision_id}"),
|
|
},
|
|
Err(e) => CtlResponse::error(e),
|
|
}
|
|
}
|
|
Ok(CtlRequest::SaveRevisionAs {
|
|
path,
|
|
revision_id,
|
|
dest,
|
|
}) => match rel_to_mount(mountpoint, &path) {
|
|
Ok(rel) => match core.save_revision_as(&rel, &revision_id, Path::new(&dest)) {
|
|
Ok(written) => CtlResponse::Ok {
|
|
message: format!("wrote {} to {dest}", human_bytes(written)),
|
|
},
|
|
Err(e) => CtlResponse::error(e),
|
|
},
|
|
Err(e) => CtlResponse::error(e),
|
|
},
|
|
Ok(CtlRequest::SaveRevisionAsByUid {
|
|
uid,
|
|
revision_id,
|
|
dest,
|
|
}) => match core.save_revision_as_by_uid(&uid, &revision_id, Path::new(&dest)) {
|
|
Ok(written) => CtlResponse::Ok {
|
|
message: format!("wrote {} to {dest}", human_bytes(written)),
|
|
},
|
|
Err(e) => CtlResponse::error(e),
|
|
},
|
|
Ok(CtlRequest::ListSharedWithMe) => match core.list_shared_with_me() {
|
|
Ok(entries) => CtlResponse::Entries { entries },
|
|
Err(e) => CtlResponse::error(e),
|
|
},
|
|
Ok(CtlRequest::AccountQuota) => match core.account_quota() {
|
|
Ok((max_space, used_space)) => CtlResponse::AccountQuota {
|
|
max_space,
|
|
used_space,
|
|
},
|
|
Err(e) => CtlResponse::error(e),
|
|
},
|
|
Ok(CtlRequest::ListSharedFolder { uid }) => match core.list_shared_folder(&uid) {
|
|
Ok(entries) => CtlResponse::Entries { entries },
|
|
Err(e) => CtlResponse::error(e),
|
|
},
|
|
Ok(CtlRequest::OpenSharedFile { uid }) => match core.open_shared_file(&uid) {
|
|
Ok(p) => CtlResponse::FilePath {
|
|
path: p.display().to_string(),
|
|
},
|
|
Err(e) => CtlResponse::error(e),
|
|
},
|
|
Ok(CtlRequest::LeaveShared { uid }) => match core.leave_shared(&uid) {
|
|
Ok(()) => {
|
|
core.log_activity(ActivityKind::Unshare, "shared item", "left", true);
|
|
CtlResponse::Ok {
|
|
message: "left shared item".to_string(),
|
|
}
|
|
}
|
|
Err(e) => CtlResponse::error(e),
|
|
},
|
|
Ok(CtlRequest::ListInvitations) => match core.list_invitations() {
|
|
Ok(items) => CtlResponse::Invitations { items },
|
|
Err(e) => CtlResponse::error(e),
|
|
},
|
|
Ok(CtlRequest::AcceptInvitation { id }) => match core.accept_invitation(&id) {
|
|
Ok(()) => CtlResponse::Ok {
|
|
message: "invitation accepted".to_string(),
|
|
},
|
|
Err(e) => CtlResponse::error(e),
|
|
},
|
|
Ok(CtlRequest::RejectInvitation { id }) => match core.reject_invitation(&id) {
|
|
Ok(()) => CtlResponse::Ok {
|
|
message: "invitation rejected".to_string(),
|
|
},
|
|
Err(e) => CtlResponse::error(e),
|
|
},
|
|
Ok(CtlRequest::ListBookmarks) => match core.list_bookmarks() {
|
|
Ok(items) => CtlResponse::Bookmarks { items },
|
|
Err(e) => CtlResponse::error(e),
|
|
},
|
|
Ok(CtlRequest::CreateBookmark { url, password }) => {
|
|
match core.create_bookmark(&url, password.as_deref()) {
|
|
Ok(()) => CtlResponse::Ok {
|
|
message: "bookmark saved".to_string(),
|
|
},
|
|
Err(e) => CtlResponse::error(e),
|
|
}
|
|
}
|
|
Ok(CtlRequest::DeleteBookmark { token }) => match core.delete_bookmark(&token) {
|
|
Ok(()) => CtlResponse::Ok {
|
|
message: "bookmark removed".to_string(),
|
|
},
|
|
Err(e) => CtlResponse::error(e),
|
|
},
|
|
Ok(CtlRequest::ListSharedByMe) => match core.list_shared_by_me() {
|
|
Ok(items) => CtlResponse::SharedByMe { items },
|
|
Err(e) => CtlResponse::error(e),
|
|
},
|
|
Ok(CtlRequest::ListActivity { limit }) => CtlResponse::Activity {
|
|
items: core.list_activity(clamp_limit(limit)),
|
|
},
|
|
// The request did not parse: the caller sent something malformed.
|
|
Err(e) => CtlResponse::error(CoreError::invalid(format!("bad request: {e}"))),
|
|
};
|
|
if !touched.is_empty() && !matches!(response, CtlResponse::Error { .. }) {
|
|
core.events.publish(touched);
|
|
}
|
|
let mut out = match serde_json::to_vec(&response) {
|
|
Ok(v) => v,
|
|
Err(e) => {
|
|
warn!(error = %e, "control: serialize response failed");
|
|
return;
|
|
}
|
|
};
|
|
out.push(b'\n');
|
|
let mut stream = stream;
|
|
let _ = stream.write_all(&out);
|
|
}
|
|
|
|
/// Listen on the control socket, serving one command per connection, each on its
|
|
/// own thread. Runs on its own thread; returns only if the listener itself fails.
|
|
///
|
|
/// Concurrent rather than serial because requests differ wildly in cost: an
|
|
/// `OpenPhoto` downloads a whole photo, and while it ran the accept loop used to
|
|
/// stall every other caller behind it — the GUI's 2s status poll, and the
|
|
/// thumbnail batches the gallery needs to paint. [`Core`] is a bundle of handles
|
|
/// (`Arc`/`Clone`), so each connection gets its own copy of it.
|
|
pub(crate) fn run_control_socket(
|
|
core: Core,
|
|
username: String,
|
|
mountpoint: PathBuf,
|
|
listener: UnixListener,
|
|
) {
|
|
info!("control socket listening");
|
|
for conn in listener.incoming() {
|
|
// Checked on every accept, including the deliberate poke teardown sends
|
|
// to wake this thread out of a blocking accept (`docs/BUGS.md` B44).
|
|
if core.shutdown.is_stopping() {
|
|
info!("control socket closing");
|
|
return;
|
|
}
|
|
match conn {
|
|
Ok(stream) => {
|
|
let Some(permit) = ControlHandlerPermit::acquire() else {
|
|
warn!(
|
|
limit = MAX_CONTROL_HANDLERS,
|
|
"control: rejecting connection at handler limit"
|
|
);
|
|
continue;
|
|
};
|
|
let core = core.clone();
|
|
let username = username.clone();
|
|
let mountpoint = mountpoint.clone();
|
|
if let Err(e) = std::thread::Builder::new()
|
|
.name("pdfs-control".into())
|
|
.spawn(move || {
|
|
let _permit = permit;
|
|
handle_control_conn(&core, &username, &mountpoint, stream);
|
|
})
|
|
{
|
|
warn!(error = %e, "control: spawn handler failed");
|
|
}
|
|
}
|
|
Err(e) => {
|
|
warn!(error = %e, "control: accept failed");
|
|
}
|
|
}
|
|
}
|
|
}
|
|
|
|
#[cfg(test)]
|
|
mod request_limit_tests {
|
|
use super::*;
|
|
use std::io::Cursor;
|
|
|
|
#[test]
|
|
fn accepts_one_bounded_newline_terminated_request() {
|
|
let mut input = Cursor::new(b"{\"Status\":null}\n".to_vec());
|
|
assert_eq!(
|
|
read_request_line(&mut input).unwrap().as_deref(),
|
|
Some("{\"Status\":null}\n")
|
|
);
|
|
}
|
|
|
|
#[test]
|
|
fn rejects_a_request_without_a_terminating_newline() {
|
|
let mut input = Cursor::new(b"{\"Status\":null}".to_vec());
|
|
assert_eq!(
|
|
read_request_line(&mut input).unwrap_err().kind(),
|
|
std::io::ErrorKind::InvalidData
|
|
);
|
|
}
|
|
|
|
#[test]
|
|
fn rejects_a_request_larger_than_the_limit() {
|
|
let mut bytes = vec![b'x'; MAX_CONTROL_REQUEST_BYTES as usize + 1];
|
|
bytes.push(b'\n');
|
|
let mut input = Cursor::new(bytes);
|
|
assert_eq!(
|
|
read_request_line(&mut input).unwrap_err().kind(),
|
|
std::io::ErrorKind::InvalidData
|
|
);
|
|
}
|
|
|
|
#[test]
|
|
fn accepts_paths_inside_the_mountpoint() {
|
|
let mountpoint = Path::new("/mnt/proton");
|
|
assert_eq!(
|
|
rel_to_mount(mountpoint, "Photos/2026").unwrap(),
|
|
Path::new("Photos/2026")
|
|
);
|
|
assert_eq!(
|
|
rel_to_mount(mountpoint, "/mnt/proton/Photos/2026").unwrap(),
|
|
Path::new("Photos/2026")
|
|
);
|
|
assert_eq!(
|
|
rel_to_mount(mountpoint, "/mnt/proton").unwrap(),
|
|
Path::new("")
|
|
);
|
|
}
|
|
|
|
#[test]
|
|
fn rejects_paths_that_escape_the_mountpoint() {
|
|
let mountpoint = Path::new("/mnt/proton");
|
|
assert!(rel_to_mount(mountpoint, "../escape").is_err());
|
|
assert!(rel_to_mount(mountpoint, "Photos/../../escape").is_err());
|
|
assert!(rel_to_mount(mountpoint, "/mnt/elsewhere").is_err());
|
|
}
|
|
|
|
/// `pdfs rm` and `pdfs rename` reach on-demand mounts, whose root is a whole
|
|
/// synced folder with a parent on Drive. Only a path below a root may get
|
|
/// through to the trash or rename call.
|
|
#[test]
|
|
fn only_a_path_below_a_mount_root_names_a_node() {
|
|
assert!(below_root(Path::new("Photos")));
|
|
assert!(below_root(Path::new("./Photos/2026")));
|
|
assert!(!below_root(Path::new("")));
|
|
assert!(!below_root(Path::new(".")));
|
|
assert!(!below_root(Path::new("./.")));
|
|
}
|
|
}
|
|
|
|
#[cfg(test)]
|
|
mod tests {
|
|
use super::*;
|
|
|
|
/// Two refreshes of the same scope are one piece of work; two refreshes of
|
|
/// different scopes are not.
|
|
#[test]
|
|
fn a_scope_refreshes_once_at_a_time() {
|
|
let photos = RefreshScope::Photos { full: false };
|
|
let held = RefreshGuard::acquire(&photos).expect("first refresh runs");
|
|
assert!(RefreshGuard::acquire(&photos).is_none());
|
|
// A full refresh forgets what the daemon knows, so it is other work.
|
|
assert!(RefreshGuard::acquire(&RefreshScope::Photos { full: true }).is_some());
|
|
assert!(
|
|
RefreshGuard::acquire(&RefreshScope::Dir {
|
|
path: "a".to_string()
|
|
})
|
|
.is_some()
|
|
);
|
|
|
|
drop(held);
|
|
assert!(RefreshGuard::acquire(&photos).is_some());
|
|
}
|
|
}
|