diff --git a/src-tauri/src/sync/mod.rs b/src-tauri/src/sync/mod.rs index c048896..98ae45b 100644 --- a/src-tauri/src/sync/mod.rs +++ b/src-tauri/src/sync/mod.rs @@ -30,6 +30,9 @@ const WEBDAV_CHUNK_SIZE: u64 = 10 * 1024 * 1024; // 10 MiB /// Graph rejects a single chunk PUT above 60 MiB. #[cfg(feature = "sync")] const GRAPH_CHUNK_SIZE: u64 = 10 * 1024 * 1024; // 10 MiB == 32 x 320 KiB +/// Dropbox upload-session part size. +#[cfg(feature = "sync")] +const DROPBOX_CHUNK_SIZE: u64 = 8 * 1024 * 1024; // 8 MiB #[cfg(feature = "sync")] pub mod oauth; @@ -1307,6 +1310,452 @@ mod tests { "assembled bytes must match the source file exactly" ); } + + // ---- Dropbox mock (T9.10, M4.3) ---- + + struct MockDropbox { + port: u16, + files: std::sync::Arc>>>, + finish_count: std::sync::Arc, + } + + impl MockDropbox { + fn start() -> Self { + let listener = std::net::TcpListener::bind("127.0.0.1:0").expect("bind mock dropbox"); + let port = listener.local_addr().unwrap().port(); + let sessions = std::sync::Arc::new(std::sync::Mutex::new(HashMap::new())); + let files = std::sync::Arc::new(std::sync::Mutex::new(HashMap::new())); + let finish_count = std::sync::Arc::new(std::sync::atomic::AtomicU32::new(0)); + let (s, f, n) = (sessions.clone(), files.clone(), finish_count.clone()); + std::thread::spawn(move || { + for stream in listener.incoming() { + let Ok(mut stream) = stream else { continue }; + let (s, f, n) = (s.clone(), f.clone(), n.clone()); + std::thread::spawn(move || { + if let Some(req) = read_mock_request(&mut stream) { + handle_dropbox_request(&mut stream, &req, &s, &f, &n); + } + }); + } + }); + Self { + port, + files, + finish_count, + } + } + + fn base_url(&self) -> String { + format!("http://127.0.0.1:{}", self.port) + } + + fn finish_count(&self) -> u32 { + self.finish_count.load(std::sync::atomic::Ordering::SeqCst) + } + + fn file(&self, path: &str) -> Option> { + self.files.lock().unwrap().get(path).cloned() + } + } + + fn dropbox_api_arg(req: &MockRequest) -> serde_json::Value { + req.headers + .get("dropbox-api-arg") + .and_then(|s| serde_json::from_str(s).ok()) + .unwrap_or(serde_json::Value::Null) + } + + fn handle_dropbox_request( + stream: &mut std::net::TcpStream, + req: &MockRequest, + sessions: &std::sync::Mutex>>, + files: &std::sync::Mutex>>, + finish_count: &std::sync::atomic::AtomicU32, + ) { + match (req.method.as_str(), req.path.as_str()) { + ("POST", "/files/upload") => { + let arg = dropbox_api_arg(req); + let path = arg["path"].as_str().unwrap_or("").to_string(); + files.lock().unwrap().insert(path, req.body.clone()); + write_mock_response(stream, 200, b"{}", "application/json"); + } + ("POST", "/files/upload_session/start") => { + let session_id = format!("sess-{}", uuid::Uuid::new_v4().simple()); + sessions + .lock() + .unwrap() + .insert(session_id.clone(), req.body.clone()); + let body = serde_json::json!({ "session_id": session_id }); + write_mock_response(stream, 200, body.to_string().as_bytes(), "application/json"); + } + ("POST", "/files/upload_session/append_v2") => { + let arg = dropbox_api_arg(req); + let session_id = arg["cursor"]["session_id"].as_str().unwrap_or(""); + let mut sessions = sessions.lock().unwrap(); + let Some(buf) = sessions.get_mut(session_id) else { + write_mock_response(stream, 404, b"", "text/plain"); + return; + }; + buf.extend_from_slice(&req.body); + write_mock_response(stream, 200, b"{}", "application/json"); + } + ("POST", "/files/upload_session/finish") => { + let arg = dropbox_api_arg(req); + let session_id = arg["cursor"]["session_id"].as_str().unwrap_or(""); + let path = arg["commit"]["path"].as_str().unwrap_or("").to_string(); + let mut sessions = sessions.lock().unwrap(); + let Some(mut buf) = sessions.remove(session_id) else { + write_mock_response(stream, 404, b"", "text/plain"); + return; + }; + buf.extend_from_slice(&req.body); + files.lock().unwrap().insert(path, buf); + finish_count.fetch_add(1, std::sync::atomic::Ordering::SeqCst); + write_mock_response(stream, 200, b"{}", "application/json"); + } + ("POST", "/files/create_folder_v2") => { + write_mock_response(stream, 200, b"{}", "application/json"); + } + ("POST", "/users/get_current_account") => { + write_mock_response(stream, 200, b"{}", "application/json"); + } + _ => write_mock_response(stream, 404, b"", "text/plain"), + } + } + + #[tokio::test] + async fn dropbox_chunked_upload_via_session_mock() { + let mock = MockDropbox::start(); + // SAFETY (test-only): see the equivalent OneDrive test above. + unsafe { + std::env::set_var("WA_DROPBOX_CONTENT_BASE_URL", mock.base_url()); + std::env::set_var("WA_DROPBOX_API_BASE_URL", mock.base_url()); + std::env::set_var("WA_OAUTH_DROPBOX_CLIENT_ID", "test-client-id"); + } + + let cred_ref = format!("wa-sync-test-dropbox-mock-{}", std::process::id()); + let token_json = serde_json::json!({ + "access_token": "fake-token", + "refresh_token": null, + "expires_at": 0, + }) + .to_string(); + credentials::set(&cred_ref, &token_json).expect("store fake token"); + + let target = DropboxTarget { + credential_ref: cred_ref.clone(), + }; + + let total_len = (DROPBOX_CHUNK_SIZE * 2) + (512 * 1024); + let mut content = vec![0u8; total_len as usize]; + for (i, b) in content.iter_mut().enumerate() { + *b = ((i * 11) % 251) as u8; + } + let tmp = + std::env::temp_dir().join(format!("wa_dropbox_chunk_probe_{}.bin", std::process::id())); + tokio::fs::write(&tmp, &content).await.unwrap(); + + let (tx, rx) = std::sync::mpsc::channel(); + let result = target.put(&tmp, "/m1/audio.wav", tx).await; + + let _ = tokio::fs::remove_file(&tmp).await; + let _ = credentials::delete(&cred_ref); + unsafe { + std::env::remove_var("WA_DROPBOX_CONTENT_BASE_URL"); + std::env::remove_var("WA_DROPBOX_API_BASE_URL"); + std::env::remove_var("WA_OAUTH_DROPBOX_CLIENT_ID"); + } + + result.expect("chunked dropbox put should succeed"); + let events: Vec<_> = rx.try_iter().collect(); + assert!( + events.len() >= 3, + "expected per-chunk progress events, got {events:?}" + ); + assert_eq!(mock.finish_count(), 1, "expected exactly one finish call"); + let stored = mock.file("/m1/audio.wav").expect("file committed"); + assert_eq!( + stored, content, + "assembled bytes must match the source file exactly" + ); + } + + // ---- Box mock (T9.10, M4.3) ---- + + /// `folder_id -> {name -> (id, is_folder)}`. + type BoxFolderMap = HashMap>; + + struct MockBox { + port: u16, + folders: std::sync::Arc>, + // committed files: file id -> bytes + files: std::sync::Arc>>>, + } + + impl MockBox { + fn start(part_size: u64) -> Self { + let listener = std::net::TcpListener::bind("127.0.0.1:0").expect("bind mock box"); + let port = listener.local_addr().unwrap().port(); + let mut root = HashMap::new(); + root.insert("0".to_string(), HashMap::new()); + let folders = std::sync::Arc::new(std::sync::Mutex::new(root)); + // upload session id -> assembled bytes so far + let sessions: std::sync::Arc>>> = + std::sync::Arc::new(std::sync::Mutex::new(HashMap::new())); + // upload session id -> (folder_id, file_name), so the commit + // handler knows where to file the assembled bytes without + // needing to decode anything from the session id itself (which + // goes through a URL and could be percent-encoded by the + // client's HTTP stack). + let session_meta: std::sync::Arc>> = + std::sync::Arc::new(std::sync::Mutex::new(HashMap::new())); + let files = std::sync::Arc::new(std::sync::Mutex::new(HashMap::new())); + let next_id = std::sync::Arc::new(std::sync::atomic::AtomicU32::new(1)); + let (fo, se, sm, fi, ni) = ( + folders.clone(), + sessions.clone(), + session_meta.clone(), + files.clone(), + next_id.clone(), + ); + std::thread::spawn(move || { + for stream in listener.incoming() { + let Ok(mut stream) = stream else { continue }; + let (fo, se, sm, fi, ni) = + (fo.clone(), se.clone(), sm.clone(), fi.clone(), ni.clone()); + std::thread::spawn(move || { + if let Some(req) = read_mock_request(&mut stream) { + handle_box_request( + &mut stream, + &req, + &fo, + &se, + &sm, + &fi, + &ni, + part_size, + ); + } + }); + } + }); + Self { + port, + folders, + files, + } + } + + fn base_url(&self) -> String { + format!("http://127.0.0.1:{}", self.port) + } + + fn find_file_bytes(&self, dir_segments: &[&str], name: &str) -> Option> { + let folders = self.folders.lock().unwrap(); + let mut parent = "0".to_string(); + for seg in dir_segments { + let (id, is_folder) = folders.get(&parent)?.get(*seg)?; + if !is_folder { + return None; + } + parent = id.clone(); + } + let (file_id, _) = folders.get(&parent)?.get(name)?; + self.files.lock().unwrap().get(file_id).cloned() + } + } + + #[allow(clippy::too_many_arguments)] + fn handle_box_request( + stream: &mut std::net::TcpStream, + req: &MockRequest, + folders: &std::sync::Mutex, + sessions: &std::sync::Mutex>>, + session_meta: &std::sync::Mutex>, + files: &std::sync::Mutex>>, + next_id: &std::sync::atomic::AtomicU32, + part_size: u64, + ) { + let path = req.path.split('?').next().unwrap_or(&req.path).to_string(); + + if req.method == "GET" && path.starts_with("/folders/") && path.ends_with("/items") { + let folder_id = path + .trim_start_matches("/folders/") + .trim_end_matches("/items") + .to_string(); + let folders = folders.lock().unwrap(); + let entries: Vec<_> = folders + .get(&folder_id) + .map(|children| { + children + .iter() + .map(|(name, (id, is_folder))| { + serde_json::json!({ + "id": id, + "name": name, + "type": if *is_folder { "folder" } else { "file" }, + }) + }) + .collect() + }) + .unwrap_or_default(); + let body = serde_json::json!({ "entries": entries }); + write_mock_response(stream, 200, body.to_string().as_bytes(), "application/json"); + return; + } + + if req.method == "GET" && path == "/users/me" { + write_mock_response(stream, 200, b"{}", "application/json"); + return; + } + + if req.method == "POST" && path == "/folders" { + let parsed: serde_json::Value = serde_json::from_slice(&req.body).unwrap_or_default(); + let name = parsed["name"].as_str().unwrap_or("").to_string(); + let parent_id = parsed["parent"]["id"].as_str().unwrap_or("0").to_string(); + let new_id = next_id + .fetch_add(1, std::sync::atomic::Ordering::SeqCst) + .to_string(); + let mut folders = folders.lock().unwrap(); + folders.entry(new_id.clone()).or_default(); + folders + .entry(parent_id) + .or_default() + .insert(name, (new_id.clone(), true)); + let body = serde_json::json!({ "id": new_id }); + write_mock_response(stream, 201, body.to_string().as_bytes(), "application/json"); + return; + } + + if req.method == "POST" + && (path == "/files/upload_sessions" || path.ends_with("/upload_sessions")) + { + let parsed: serde_json::Value = serde_json::from_slice(&req.body).unwrap_or_default(); + let folder_id = parsed["folder_id"].as_str().unwrap_or("0").to_string(); + let name = parsed["file_name"].as_str().unwrap_or("").to_string(); + let session_id = format!("box-sess-{}", uuid::Uuid::new_v4().simple()); + sessions + .lock() + .unwrap() + .insert(session_id.clone(), Vec::new()); + session_meta + .lock() + .unwrap() + .insert(session_id.clone(), (folder_id, name)); + let body = serde_json::json!({ "id": session_id, "part_size": part_size }); + write_mock_response(stream, 200, body.to_string().as_bytes(), "application/json"); + return; + } + + if req.method == "PUT" && path.starts_with("/files/upload_sessions/") { + let session_id = path + .trim_start_matches("/files/upload_sessions/") + .to_string(); + let mut sessions = sessions.lock().unwrap(); + let Some(buf) = sessions.get_mut(&session_id) else { + write_mock_response(stream, 404, b"", "text/plain"); + return; + }; + buf.extend_from_slice(&req.body); + let body = serde_json::json!({ + "part": { "part_id": uuid::Uuid::new_v4().simple().to_string(), "offset": 0, "size": req.body.len() }, + }); + write_mock_response(stream, 200, body.to_string().as_bytes(), "application/json"); + return; + } + + if req.method == "POST" + && path.starts_with("/files/upload_sessions/") + && path.ends_with("/commit") + { + let session_id = path + .trim_start_matches("/files/upload_sessions/") + .trim_end_matches("/commit") + .to_string(); + let mut sessions = sessions.lock().unwrap(); + let Some(bytes) = sessions.remove(&session_id) else { + write_mock_response(stream, 404, b"", "text/plain"); + return; + }; + drop(sessions); + let Some((folder_id, name)) = session_meta.lock().unwrap().remove(&session_id) else { + write_mock_response(stream, 404, b"", "text/plain"); + return; + }; + let file_id = format!("file-{}", uuid::Uuid::new_v4().simple()); + files.lock().unwrap().insert(file_id.clone(), bytes); + folders + .lock() + .unwrap() + .entry(folder_id) + .or_default() + .insert(name, (file_id, false)); + write_mock_response(stream, 201, b"{}", "application/json"); + return; + } + + write_mock_response(stream, 404, b"", "text/plain"); + } + + #[tokio::test] + async fn box_chunked_upload_via_session_mock() { + let part_size: u64 = 4 * 1024 * 1024; // 4 MiB, small enough to force multiple parts + let mock = MockBox::start(part_size); + // SAFETY (test-only): see the equivalent OneDrive test above. + unsafe { + std::env::set_var("WA_BOX_API_BASE_URL", mock.base_url()); + std::env::set_var("WA_BOX_UPLOAD_BASE_URL", mock.base_url()); + std::env::set_var("WA_OAUTH_BOX_CLIENT_ID", "test-client-id"); + } + + let cred_ref = format!("wa-sync-test-box-mock-{}", std::process::id()); + let token_json = serde_json::json!({ + "access_token": "fake-token", + "refresh_token": null, + "expires_at": 0, + }) + .to_string(); + credentials::set(&cred_ref, &token_json).expect("store fake token"); + + let target = BoxTarget { + credential_ref: cred_ref.clone(), + }; + + let total_len = (part_size * 2) + (256 * 1024); + let mut content = vec![0u8; total_len as usize]; + for (i, b) in content.iter_mut().enumerate() { + *b = ((i * 13) % 249) as u8; + } + let tmp = + std::env::temp_dir().join(format!("wa_box_chunk_probe_{}.bin", std::process::id())); + tokio::fs::write(&tmp, &content).await.unwrap(); + + target.ensure_dir("/m1").await.expect("ensure_dir"); + let (tx, rx) = std::sync::mpsc::channel(); + let result = target.put(&tmp, "/m1/audio.wav", tx).await; + + let _ = tokio::fs::remove_file(&tmp).await; + let _ = credentials::delete(&cred_ref); + unsafe { + std::env::remove_var("WA_BOX_API_BASE_URL"); + std::env::remove_var("WA_BOX_UPLOAD_BASE_URL"); + std::env::remove_var("WA_OAUTH_BOX_CLIENT_ID"); + } + + result.expect("chunked box put should succeed"); + let events: Vec<_> = rx.try_iter().collect(); + assert!( + events.len() >= 3, + "expected per-chunk progress events, got {events:?}" + ); + let stored = mock + .find_file_bytes(&["m1"], "audio.wav") + .expect("file committed under /m1"); + assert_eq!( + stored, content, + "assembled bytes must match the source file exactly" + ); + } } #[cfg(feature = "sync")] @@ -1346,11 +1795,8 @@ pub fn build_sync_target( match row.kind.as_str() { "webdav" => Ok(Box::new(WebDavTarget::from_row(row))), "onedrive" => Ok(Box::new(OneDriveTarget::from_row(row))), - // Linking works (tokens stored); upload lands next, same trait. - "dropbox" | "box" => Err(SyncError::Upload(format!( - "{} upload is not implemented yet (OAuth linking works)", - row.kind - ))), + "dropbox" => Ok(Box::new(DropboxTarget::from_row(row))), + "box" => Ok(Box::new(BoxTarget::from_row(row))), other => Err(SyncError::Upload(format!( "unknown sync target kind {other}" ))), @@ -1617,3 +2063,659 @@ impl OneDriveTarget { Ok(parsed.upload_url) } } + +/// Dropbox via its v2 HTTP API (T9.10, M4.3). Path-addressed like +/// OneDrive/WebDAV (no folder-ID indirection, unlike Box below) — +/// `files/create_folder_v2` creates every missing intermediate directory in +/// one call, unlike WebDAV's per-segment MKCOL or Graph's per-segment POST. +#[cfg(feature = "sync")] +pub struct DropboxTarget { + pub credential_ref: String, +} + +#[cfg(feature = "sync")] +impl DropboxTarget { + pub fn from_row(row: &crate::storage::SyncTargetRow) -> Self { + Self { + credential_ref: row.credential_ref.clone(), + } + } + + async fn bearer(&self) -> Result { + resolve_access_token("dropbox", &self.credential_ref).await + } + + /// RPC/metadata host — overridable for tests, same pattern as + /// `OneDriveTarget::graph_base`. + fn api_base() -> String { + std::env::var("WA_DROPBOX_API_BASE_URL") + .ok() + .filter(|s| !s.is_empty()) + .unwrap_or_else(|| "https://api.dropboxapi.com/2".to_string()) + } + + /// Content (upload/download) host — a separate host on the real service. + fn content_base() -> String { + std::env::var("WA_DROPBOX_CONTENT_BASE_URL") + .ok() + .filter(|s| !s.is_empty()) + .unwrap_or_else(|| "https://content.dropboxapi.com/2".to_string()) + } + + fn dropbox_path(remote_path: &str) -> String { + let p = remote_path.trim_end_matches('/'); + if p.starts_with('/') { + p.to_string() + } else { + format!("/{p}") + } + } +} + +#[cfg(feature = "sync")] +#[async_trait] +impl SyncTarget for DropboxTarget { + fn kind(&self) -> SyncKind { + SyncKind::Dropbox + } + fn is_third_party(&self) -> bool { + true + } + + async fn test(&self) -> Result<(), SyncError> { + let token = self.bearer().await?; + let resp = http_client()? + .post(format!("{}/users/get_current_account", Self::api_base())) + .bearer_auth(token) + .header(reqwest::header::CONTENT_TYPE, "application/json") + .body("null") + .send() + .await + .map_err(|e| SyncError::Unreachable(e.to_string()))?; + match resp.status().as_u16() { + code if (200..300).contains(&code) => Ok(()), + 401 => Err(SyncError::Auth), + other => Err(SyncError::Unreachable(format!("dropbox {other}"))), + } + } + + /// `create_folder_v2` creates every missing intermediate directory in one + /// call — an existing folder comes back as a `409` conflict, which we + /// treat as success (idempotent, matching WebDAV MKCOL's 405-means-exists + /// and Graph's 409-means-exists handling). + async fn ensure_dir(&self, remote_dir: &str) -> Result<(), SyncError> { + if remote_dir.trim_matches('/').is_empty() { + return Ok(()); // root always exists + } + let token = self.bearer().await?; + let body = + serde_json::json!({ "path": Self::dropbox_path(remote_dir), "autorename": false }); + let resp = http_client()? + .post(format!("{}/files/create_folder_v2", Self::api_base())) + .bearer_auth(token) + .json(&body) + .send() + .await + .map_err(|e| SyncError::Upload(e.to_string()))?; + match resp.status().as_u16() { + code if (200..300).contains(&code) => Ok(()), + 401 => Err(SyncError::Auth), + 409 => Ok(()), // already exists + other => Err(SyncError::Upload(format!("create_folder_v2 {other}"))), + } + } + + async fn exists(&self, remote_path: &str, _sha256: &str) -> Result { + let token = self.bearer().await?; + let body = serde_json::json!({ "path": Self::dropbox_path(remote_path) }); + let resp = http_client()? + .post(format!("{}/files/get_metadata", Self::api_base())) + .bearer_auth(token) + .json(&body) + .send() + .await + .map_err(|e| SyncError::Unreachable(e.to_string()))?; + match resp.status().as_u16() { + code if (200..300).contains(&code) => Ok(true), + 409 => Ok(false), // path/not_found + 401 => Err(SyncError::Auth), + other => Err(SyncError::Upload(format!("get_metadata {other}"))), + } + } + + /// Upload, streamed from disk. Above `LARGE_FILE_THRESHOLD`, uses + /// Dropbox's upload-session protocol (`put_chunked`); otherwise one + /// streamed `files/upload` call. + async fn put( + &self, + local: &Path, + remote_path: &str, + prog: ProgressSink, + ) -> Result<(), SyncError> { + let total = tokio::fs::metadata(local) + .await + .map_err(|e| SyncError::Upload(e.to_string()))? + .len(); + if total > LARGE_FILE_THRESHOLD { + return self.put_chunked(local, remote_path, total, prog).await; + } + let token = self.bearer().await?; + let body = open_streaming_body(local, total, prog).await?; + let arg = serde_json::json!({ + "path": Self::dropbox_path(remote_path), + "mode": "overwrite", + "autorename": false, + "mute": true, + }); + let resp = http_client()? + .post(format!("{}/files/upload", Self::content_base())) + .bearer_auth(token) + .header("Dropbox-API-Arg", arg.to_string()) + .header(reqwest::header::CONTENT_TYPE, "application/octet-stream") + .body(body) + .send() + .await + .map_err(|e| SyncError::Upload(e.to_string()))?; + match resp.status().as_u16() { + code if (200..300).contains(&code) => Ok(()), + 401 => Err(SyncError::Auth), + other => Err(SyncError::Upload(format!("upload {other}"))), + } + } +} + +#[cfg(feature = "sync")] +impl DropboxTarget { + /// Dropbox upload-session chunking: `upload_session/start` (first + /// chunk), `append_v2` per subsequent chunk, `finish` with the last chunk + /// plus commit info. + /// + /// ponytail (documented simplification, T9.2/M4.1/M4.3 ceiling, same + /// class as `OneDriveTarget::put_chunked`): the session id lives only for + /// this call, not persisted on the `sync_jobs` row — a durable-queue + /// retry after an app restart starts a fresh session rather than + /// resuming Dropbox's session mid-file. + async fn put_chunked( + &self, + local: &Path, + remote_path: &str, + total: u64, + prog: ProgressSink, + ) -> Result<(), SyncError> { + let token = self.bearer().await?; + let mut file = tokio::fs::File::open(local) + .await + .map_err(|e| SyncError::Upload(e.to_string()))?; + use tokio::io::AsyncReadExt; + + let mut first = vec![0u8; DROPBOX_CHUNK_SIZE.min(total) as usize]; + file.read_exact(&mut first) + .await + .map_err(|e| SyncError::Upload(e.to_string()))?; + let start_arg = serde_json::json!({ "close": false }); + let resp = http_client()? + .post(format!( + "{}/files/upload_session/start", + Self::content_base() + )) + .bearer_auth(&token) + .header("Dropbox-API-Arg", start_arg.to_string()) + .header(reqwest::header::CONTENT_TYPE, "application/octet-stream") + .body(first.clone()) + .send() + .await + .map_err(|e| SyncError::Upload(e.to_string()))?; + if !(200..300).contains(&resp.status().as_u16()) { + return Err(SyncError::Upload(format!( + "upload_session/start {}", + resp.status() + ))); + } + #[derive(serde::Deserialize)] + struct StartResp { + session_id: String, + } + let session: StartResp = resp + .json() + .await + .map_err(|e| SyncError::Upload(e.to_string()))?; + + let mut offset = first.len() as u64; + let _ = prog.send((offset, total)); + while offset < total { + let this_len = DROPBOX_CHUNK_SIZE.min(total - offset) as usize; + let mut buf = vec![0u8; this_len]; + file.read_exact(&mut buf) + .await + .map_err(|e| SyncError::Upload(e.to_string()))?; + let is_last = offset + this_len as u64 == total; + if !is_last { + let arg = serde_json::json!({ + "cursor": { "session_id": session.session_id, "offset": offset }, + "close": false, + }); + let resp = http_client()? + .post(format!( + "{}/files/upload_session/append_v2", + Self::content_base() + )) + .bearer_auth(&token) + .header("Dropbox-API-Arg", arg.to_string()) + .header(reqwest::header::CONTENT_TYPE, "application/octet-stream") + .body(buf) + .send() + .await + .map_err(|e| SyncError::Upload(e.to_string()))?; + if !(200..300).contains(&resp.status().as_u16()) { + return Err(SyncError::Upload(format!( + "upload_session/append_v2 {}", + resp.status() + ))); + } + } else { + let arg = serde_json::json!({ + "cursor": { "session_id": session.session_id, "offset": offset }, + "commit": { + "path": Self::dropbox_path(remote_path), + "mode": "overwrite", + "autorename": false, + "mute": true, + }, + }); + let resp = http_client()? + .post(format!( + "{}/files/upload_session/finish", + Self::content_base() + )) + .bearer_auth(&token) + .header("Dropbox-API-Arg", arg.to_string()) + .header(reqwest::header::CONTENT_TYPE, "application/octet-stream") + .body(buf) + .send() + .await + .map_err(|e| SyncError::Upload(e.to_string()))?; + if !(200..300).contains(&resp.status().as_u16()) { + return Err(SyncError::Upload(format!( + "upload_session/finish {}", + resp.status() + ))); + } + } + offset += this_len as u64; + let _ = prog.send((offset, total)); + } + Ok(()) + } +} + +/// Box via its v2 API (T9.10, M4.3). Unlike Dropbox/OneDrive/WebDAV, Box +/// addresses items by numeric ID, not path — every path-based operation here +/// first walks/creates the folder chain from the root (id `"0"`) by listing +/// each level's children. Box's simple (non-session) upload endpoint requires +/// `multipart/form-data`, a `reqwest` feature nothing else here needs; its +/// session-based upload takes plain `PUT` bodies like every other target in +/// this file, and works fine for one chunk, so `put` always goes through the +/// session path regardless of file size — one code path, no new dependency +/// feature. +#[cfg(feature = "sync")] +pub struct BoxTarget { + pub credential_ref: String, +} + +#[cfg(feature = "sync")] +impl BoxTarget { + pub fn from_row(row: &crate::storage::SyncTargetRow) -> Self { + Self { + credential_ref: row.credential_ref.clone(), + } + } + + async fn bearer(&self) -> Result { + resolve_access_token("box", &self.credential_ref).await + } + + fn api_base() -> String { + std::env::var("WA_BOX_API_BASE_URL") + .ok() + .filter(|s| !s.is_empty()) + .unwrap_or_else(|| "https://api.box.com/2.0".to_string()) + } + + fn upload_base() -> String { + std::env::var("WA_BOX_UPLOAD_BASE_URL") + .ok() + .filter(|s| !s.is_empty()) + .unwrap_or_else(|| "https://upload.box.com/api/2.0".to_string()) + } + + /// One folder's children, `{name -> (id, is_folder)}`. + /// + /// ponytail (documented ceiling, M4.3): a single page (up to the 1000-item + /// max we request) — a target folder with more entries than that would + /// miss later ones. WhispAssist's own upload layout (one subfolder per + /// meeting, a handful of files each) never approaches this in practice; + /// paginate via `offset` if it ever does. + async fn list_children( + &self, + token: &str, + folder_id: &str, + ) -> Result, SyncError> { + let resp = http_client()? + .get(format!( + "{}/folders/{folder_id}/items?fields=id,name,type&limit=1000", + Self::api_base() + )) + .bearer_auth(token) + .send() + .await + .map_err(|e| SyncError::Unreachable(e.to_string()))?; + match resp.status().as_u16() { + 401 | 403 => return Err(SyncError::Auth), + code if !(200..300).contains(&code) => { + return Err(SyncError::Upload(format!("folder items {code}"))) + } + _ => {} + } + #[derive(serde::Deserialize)] + struct Entry { + id: String, + name: String, + #[serde(rename = "type")] + kind: String, + } + #[derive(serde::Deserialize)] + struct Items { + entries: Vec, + } + let items: Items = resp + .json() + .await + .map_err(|e| SyncError::Upload(e.to_string()))?; + Ok(items + .entries + .into_iter() + .map(|e| (e.name, (e.id, e.kind == "folder"))) + .collect()) + } + + /// Walks `path` from the root (id `"0"`), creating any missing folder + /// segment. Returns the deepest folder's id. + async fn resolve_or_create_folder(&self, path: &str) -> Result { + let token = self.bearer().await?; + let mut parent_id = "0".to_string(); + for segment in path.split('/').filter(|s| !s.is_empty()) { + let children = self.list_children(&token, &parent_id).await?; + if let Some((id, true)) = children.get(segment) { + parent_id = id.clone(); + continue; + } + let body = serde_json::json!({ "name": segment, "parent": { "id": parent_id } }); + let resp = http_client()? + .post(format!("{}/folders", Self::api_base())) + .bearer_auth(&token) + .json(&body) + .send() + .await + .map_err(|e| SyncError::Upload(e.to_string()))?; + match resp.status().as_u16() { + 201 => { + #[derive(serde::Deserialize)] + struct Created { + id: String, + } + let created: Created = resp + .json() + .await + .map_err(|e| SyncError::Upload(e.to_string()))?; + parent_id = created.id; + } + 409 => { + // Created concurrently (e.g. by another upload_job in the + // same pump run) between our list and create — re-list + // and use the now-existing id. + let children = self.list_children(&token, &parent_id).await?; + parent_id = + children + .get(segment) + .map(|(id, _)| id.clone()) + .ok_or_else(|| { + SyncError::Upload(format!( + "folder {segment} conflict but not found" + )) + })?; + } + 401 | 403 => return Err(SyncError::Auth), + other => return Err(SyncError::Upload(format!("create folder {other}"))), + } + } + Ok(parent_id) + } + + /// Looks up an existing file's id in `parent_id` by name, without + /// creating anything. + async fn find_file( + &self, + token: &str, + parent_id: &str, + name: &str, + ) -> Result, SyncError> { + let children = self.list_children(token, parent_id).await?; + Ok(children.get(name).and_then( + |(id, is_folder)| { + if *is_folder { + None + } else { + Some(id.clone()) + } + }, + )) + } + + fn split_path(remote_path: &str) -> (String, String) { + match remote_path.trim_start_matches('/').rsplit_once('/') { + Some((dir, name)) => (dir.to_string(), name.to_string()), + None => ( + String::new(), + remote_path.trim_start_matches('/').to_string(), + ), + } + } +} + +#[cfg(feature = "sync")] +#[async_trait] +impl SyncTarget for BoxTarget { + fn kind(&self) -> SyncKind { + SyncKind::Box + } + fn is_third_party(&self) -> bool { + true + } + + async fn test(&self) -> Result<(), SyncError> { + let token = self.bearer().await?; + let resp = http_client()? + .get(format!("{}/users/me", Self::api_base())) + .bearer_auth(token) + .send() + .await + .map_err(|e| SyncError::Unreachable(e.to_string()))?; + match resp.status().as_u16() { + code if (200..300).contains(&code) => Ok(()), + 401 | 403 => Err(SyncError::Auth), + other => Err(SyncError::Unreachable(format!("box {other}"))), + } + } + + async fn ensure_dir(&self, remote_dir: &str) -> Result<(), SyncError> { + self.resolve_or_create_folder(remote_dir).await?; + Ok(()) + } + + async fn exists(&self, remote_path: &str, _sha256: &str) -> Result { + let (dir, name) = Self::split_path(remote_path); + let token = self.bearer().await?; + // Read-only walk: a missing intermediate segment means the file + // can't exist either, without creating anything. + let mut parent_id = "0".to_string(); + for segment in dir.split('/').filter(|s| !s.is_empty()) { + let children = self.list_children(&token, &parent_id).await?; + match children.get(segment) { + Some((id, true)) => parent_id = id.clone(), + _ => return Ok(false), + } + } + Ok(self.find_file(&token, &parent_id, &name).await?.is_some()) + } + + async fn put( + &self, + local: &Path, + remote_path: &str, + prog: ProgressSink, + ) -> Result<(), SyncError> { + let total = tokio::fs::metadata(local) + .await + .map_err(|e| SyncError::Upload(e.to_string()))? + .len(); + let (dir, name) = Self::split_path(remote_path); + let parent_id = self.resolve_or_create_folder(&dir).await?; + self.put_via_session(local, &parent_id, &name, total, prog) + .await + } +} + +#[cfg(feature = "sync")] +impl BoxTarget { + /// Box chunked upload, used for every file regardless of size (see the + /// struct doc comment for why): create a session (new file, or a new + /// version of the existing file if one is already there), PUT each part + /// with `Content-Range`, collect the `part` descriptor Box returns per + /// chunk (it computes the digest for us — no local hashing needed), then + /// commit with the collected parts. + /// + /// ponytail (documented ceiling, M4.3, same class as + /// `OneDriveTarget::put_chunked`): the session id lives only for this + /// call — a durable-queue retry after a restart starts a fresh session + /// rather than resuming Box's session mid-file. + async fn put_via_session( + &self, + local: &Path, + parent_id: &str, + name: &str, + total: u64, + prog: ProgressSink, + ) -> Result<(), SyncError> { + let token = self.bearer().await?; + let existing = self.find_file(&token, parent_id, name).await?; + let (session_url, part_size) = self + .create_upload_session(&token, parent_id, name, total, existing.as_deref()) + .await?; + + let mut file = tokio::fs::File::open(local) + .await + .map_err(|e| SyncError::Upload(e.to_string()))?; + use tokio::io::AsyncReadExt; + let mut offset = 0u64; + let mut parts = Vec::new(); + let _ = prog.send((0, total)); + while offset < total { + let this_len = part_size.min(total - offset); + let mut buf = vec![0u8; this_len as usize]; + file.read_exact(&mut buf) + .await + .map_err(|e| SyncError::Upload(e.to_string()))?; + let end = offset + this_len - 1; + let resp = http_client()? + .put(&session_url) + .bearer_auth(&token) + .header( + reqwest::header::CONTENT_RANGE, + format!("bytes {offset}-{end}/{total}"), + ) + .header(reqwest::header::CONTENT_LENGTH, this_len) + .body(buf) + .send() + .await + .map_err(|e| SyncError::Upload(e.to_string()))?; + if !(200..300).contains(&resp.status().as_u16()) { + return Err(SyncError::Upload(format!("upload part {}", resp.status()))); + } + let part_body: serde_json::Value = resp + .json() + .await + .map_err(|e| SyncError::Upload(e.to_string()))?; + parts.push(part_body.get("part").cloned().ok_or_else(|| { + SyncError::Upload("box part response missing 'part'".to_string()) + })?); + offset += this_len; + let _ = prog.send((offset, total)); + } + + let commit = serde_json::json!({ "parts": parts }); + let resp = http_client()? + .post(format!("{session_url}/commit")) + .bearer_auth(&token) + .json(&commit) + .send() + .await + .map_err(|e| SyncError::Upload(e.to_string()))?; + match resp.status().as_u16() { + code if (200..300).contains(&code) => Ok(()), + 401 | 403 => Err(SyncError::Auth), + other => Err(SyncError::Upload(format!("commit {other}"))), + } + } + + /// Creates an upload session (new file, or a new version of + /// `existing_file_id` if `Some`). Returns the session's part-upload URL + /// and the `part_size` Box wants us to use. + async fn create_upload_session( + &self, + token: &str, + parent_id: &str, + name: &str, + total: u64, + existing_file_id: Option<&str>, + ) -> Result<(String, u64), SyncError> { + let url = match existing_file_id { + Some(file_id) => format!("{}/files/{file_id}/upload_sessions", Self::upload_base()), + None => format!("{}/files/upload_sessions", Self::upload_base()), + }; + let body = serde_json::json!({ + "folder_id": parent_id, + "file_size": total, + "file_name": name, + }); + let resp = http_client()? + .post(&url) + .bearer_auth(token) + .json(&body) + .send() + .await + .map_err(|e| SyncError::Upload(e.to_string()))?; + if !(200..300).contains(&resp.status().as_u16()) { + return Err(SyncError::Upload(format!( + "upload_sessions {}", + resp.status() + ))); + } + #[derive(serde::Deserialize)] + struct SessionResp { + id: String, + part_size: u64, + } + let session: SessionResp = resp + .json() + .await + .map_err(|e| SyncError::Upload(e.to_string()))?; + Ok(( + format!( + "{}/files/upload_sessions/{}", + Self::upload_base(), + session.id + ), + session.part_size.max(1), + )) + } +}