mirror of
https://github.com/narrrl/proton-drive-linux.git
synced 2026-11-04 00:47:00 +01:00
nvrl
a06a8150fb
Refresh Cargo.lock to the newest compatible versions and move three crates to their next minor release: - parking_lot 0.12 -> 0.13 - rusqlite 0.40 -> 0.41 (newer bundled SQLite) - gettext-rs 0.7 -> 0.8 gettext-rs 0.8 marks setlocale as unsafe; the three calls in i18n.rs run before any thread starts, the same precondition the set_var calls next to them already state. rustc 1.99 deprecates AtomicX::fetch_update in favour of try_update (stable since 1.95, below the 1.96 minimum), which failed `clippy -D warnings` on main. Rename the two uses so clippy passes without `-A deprecated`. The open Dependabot alerts for zip (< 9.0.5) and rustls (< 0.23.46) are already resolved on main by #69 (zip 9.2.0, rustls 0.23.46).
2055 lines
85 KiB
Rust
2055 lines
85 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::FolderSize`].
|
|
fn folder_size_reply(size: CoreResult<proton_drive_rs::FolderSize>) -> CtlResponse {
|
|
match size {
|
|
Ok(size) => CtlResponse::FolderSize {
|
|
size: size.size,
|
|
items: size.descendants,
|
|
},
|
|
Err(e) => CtlResponse::error(e),
|
|
}
|
|
}
|
|
|
|
/// 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
|
|
.try_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),
|
|
session_expired: core.session_expired(),
|
|
tokens_unsaved: pdfs_core::auth::tokens_unsaved(),
|
|
version: env!("CARGO_PKG_VERSION").to_string(),
|
|
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::ResetSyncFolder { .. }
|
|
| 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::UpdatePublicLink { .. }
|
|
| R::UpdatePublicLinkByUid { .. }
|
|
| 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();
|
|
// Unless the caller knows better: the photo's own EXIF,
|
|
// then the file's modification time, which a copy resets.
|
|
let capture_time = capture_time.or_else(|| {
|
|
let path = Path::new(&source_path);
|
|
pdfs_core::exiftime::capture_time_of(path).or_else(|| {
|
|
let modified =
|
|
std::fs::metadata(path).and_then(|m| m.modified()).ok()?;
|
|
let since = modified.duration_since(std::time::UNIX_EPOCH).ok()?;
|
|
Some(since.as_secs() as i64)
|
|
})
|
|
});
|
|
let metadata = proton_drive_rs::PhotoUploadMetadata {
|
|
capture_time,
|
|
..Default::default()
|
|
};
|
|
let thumbnails = core.upload_thumbnails(Path::new(&source_path), &name);
|
|
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,
|
|
thumbnails,
|
|
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) => {
|
|
// The reply goes to the app's toast only; without
|
|
// this the journal shows a failure and no reason.
|
|
warn!(error = %e, "photo upload failed");
|
|
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::Copy { path, new_parent }) => {
|
|
match (
|
|
rel_to_mount(mountpoint, &path),
|
|
rel_to_mount(mountpoint, &new_parent),
|
|
) {
|
|
(Ok(rel), Ok(parent_rel)) => match core.copy_to(&rel, &parent_rel) {
|
|
Ok(name) => {
|
|
let dest = if new_parent.is_empty() {
|
|
"My files".to_string()
|
|
} else {
|
|
new_parent.clone()
|
|
};
|
|
core.log_activity(ActivityKind::Copy, &name, format!("to {dest}"), true);
|
|
CtlResponse::Ok {
|
|
message: format!("copied to {name}"),
|
|
}
|
|
}
|
|
Err(e) => {
|
|
core.log_activity(ActivityKind::Copy, &path, &e, false);
|
|
CtlResponse::error(e)
|
|
}
|
|
},
|
|
(Err(e), _) | (_, Err(e)) => CtlResponse::error(e),
|
|
}
|
|
}
|
|
Ok(CtlRequest::FolderSize { path }) => match rel_to_mount(mountpoint, &path) {
|
|
Ok(rel) => folder_size_reply(core.folder_size(&rel)),
|
|
Err(e) => CtlResponse::error(e),
|
|
},
|
|
Ok(CtlRequest::FolderSizeByUid { uid }) => folder_size_reply(core.folder_size_by_uid(&uid)),
|
|
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::ResetSyncFolder { id }) => match core.reset_sync_folder(id) {
|
|
Ok(message) => {
|
|
core.log_activity(ActivityKind::Sync, &message, "", true);
|
|
CtlResponse::Ok { message }
|
|
}
|
|
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(outcome) => {
|
|
core.touch_profile();
|
|
if !outcome.complete() {
|
|
warn!(skipped = ?outcome.skipped, "restore skipped folders");
|
|
}
|
|
core.log_activity(
|
|
ActivityKind::Download,
|
|
outcome.summary(),
|
|
outcome.skipped.join("; "),
|
|
outcome.complete(),
|
|
);
|
|
}
|
|
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::UpdatePublicLink {
|
|
path,
|
|
id,
|
|
role,
|
|
password,
|
|
expires,
|
|
}) => match rel_to_mount(mountpoint, &path) {
|
|
Ok(rel) => {
|
|
match core.update_public_link(&rel, &id, role.as_deref(), password, expires) {
|
|
Ok(link) => CtlResponse::PublicLink { link },
|
|
Err(e) => CtlResponse::error(e),
|
|
}
|
|
}
|
|
Err(e) => CtlResponse::error(e),
|
|
},
|
|
Ok(CtlRequest::UpdatePublicLinkByUid {
|
|
uid,
|
|
id,
|
|
role,
|
|
password,
|
|
expires,
|
|
}) => match core.update_public_link_by_uid(&uid, &id, role.as_deref(), password, expires) {
|
|
Ok(link) => CtlResponse::PublicLink { link },
|
|
Err(e) => 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());
|
|
}
|
|
}
|