diff --git a/src-tauri/src/sync/mod.rs b/src-tauri/src/sync/mod.rs index c6c3ba3..c048896 100644 --- a/src-tauri/src/sync/mod.rs +++ b/src-tauri/src/sync/mod.rs @@ -13,8 +13,24 @@ use crate::models::SyncTargetInfo; use crate::models::{MeetingId, SyncJobInfo, SyncKind}; use async_trait::async_trait; +#[cfg(feature = "sync")] +use std::collections::HashMap; use std::path::Path; +/// Below this, a plain streamed PUT (still disk-streamed, never buffered whole +/// into memory); above it, the backend's resumable/chunked protocol (T9.2, +/// M4.1). Small artifacts (transcript/notes/summary JSON) never cross this — +/// only `.wav` recordings realistically do. +#[cfg(feature = "sync")] +const LARGE_FILE_THRESHOLD: u64 = 8 * 1024 * 1024; // 8 MiB +/// Nextcloud/ownCloud chunking-v2 part size. +#[cfg(feature = "sync")] +const WEBDAV_CHUNK_SIZE: u64 = 10 * 1024 * 1024; // 10 MiB +/// Microsoft Graph upload-session part size — must be a multiple of 320 KiB; +/// 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 + #[cfg(feature = "sync")] pub mod oauth; @@ -137,6 +153,29 @@ impl WebDavTarget { } } + /// Nextcloud/ownCloud expose a chunking-v2 upload API (`…/dav/uploads//`, + /// sibling to `…/dav/files//`). Generic WebDAV (Seafile, Synology, + /// unbranded servers) has no equivalent, so large files there fall back to + /// one streamed-from-disk PUT — memory-safe, just not chunk-resumable. + fn supports_chunking(&self) -> bool { + matches!( + self.provider_hint.as_deref(), + Some("nextcloud") | Some("owncloud") + ) + } + + /// Root of the chunking-v2 upload collection for this user. + fn uploads_root(&self) -> String { + let origin = reqwest::Url::parse(&self.base_url) + .ok() + .map(|u| u.origin().ascii_serialization()) + .unwrap_or_else(|| self.base_url.trim_end_matches('/').to_string()); + format!( + "{origin}/remote.php/dav/uploads/{}", + self.username.trim_matches('/') + ) + } + /// Resolves the secret and returns an authenticated request builder, after /// enforcing TLS (or an explicit LAN plaintext opt-in). fn request( @@ -223,26 +262,29 @@ impl SyncTarget for WebDavTarget { Ok(resp.status().is_success()) } - /// Upload a file via PUT. - /// - /// ponytail: single PUT reading the file into memory — fine for - /// transcript/notes/summary. Chunked/resumable upload for large `.wav` - /// recordings is a later refinement (Nextcloud chunked upload). + /// Upload a file, streamed from disk (never buffered whole in memory). + /// Above `LARGE_FILE_THRESHOLD` on a Nextcloud/ownCloud target, uses the + /// chunking-v2 protocol (resumable: a retry skips chunks already sitting + /// server-side); otherwise a single streamed PUT. async fn put( &self, local: &Path, remote_path: &str, prog: ProgressSink, ) -> Result<(), SyncError> { - let bytes = tokio::fs::read(local) + let total = tokio::fs::metadata(local) .await - .map_err(|e| SyncError::Upload(e.to_string()))?; - let total = bytes.len() as u64; + .map_err(|e| SyncError::Upload(e.to_string()))? + .len(); + if total > LARGE_FILE_THRESHOLD && self.supports_chunking() { + return self.put_chunked(local, remote_path, total, prog).await; + } let url = self.url_for(remote_path); + let body = open_streaming_body(local, total, prog).await?; let resp = self .request(reqwest::Method::PUT, &url)? .header(reqwest::header::CONTENT_LENGTH, total) - .body(progress_body(bytes, prog)) + .body(body) .send() .await .map_err(|e| SyncError::Upload(e.to_string()))?; @@ -254,27 +296,221 @@ impl SyncTarget for WebDavTarget { } } -/// Wrap in-memory bytes as a reqwest streaming body that reports cumulative -/// bytes-sent to `prog` as the request drains it — the basis for live per-item -/// upload progress (FR-SYNC-11). Chunked so the UI sees steady movement on a -/// large recording instead of a single 0→100% jump at the end. #[cfg(feature = "sync")] -fn progress_body(bytes: Vec, prog: ProgressSink) -> reqwest::Body { - const CHUNK: usize = 64 * 1024; - let total = bytes.len() as u64; - let stream = futures_util::stream::unfold((bytes, 0usize), move |(bytes, pos)| { +impl WebDavTarget { + /// Nextcloud/ownCloud chunking-v2 upload: MKCOL an upload collection keyed + /// deterministically off `remote_path` (stable across retries — a job + /// retried after a crash lands on the *same* collection), PUT each part + /// under a lexically-sortable name, skipping any part a PROPFIND shows is + /// already there, then MOVE the virtual `.file` onto the real destination + /// to have the server assemble it (`OC-Total-Length` sanity-checks the + /// total). This is genuine mid-file resume, not just chunk-level retry. + async fn put_chunked( + &self, + local: &Path, + remote_path: &str, + total: u64, + prog: ProgressSink, + ) -> Result<(), SyncError> { + let upload_id = { + use sha2::{Digest, Sha256}; + let mut h = Sha256::new(); + h.update(remote_path.as_bytes()); + format!("{:x}", h.finalize()) + }; + let upload_dir = format!("{}/{}", self.uploads_root(), upload_id); + + let mkcol = reqwest::Method::from_bytes(b"MKCOL").expect("MKCOL is a valid method token"); + let resp = self + .request(mkcol, &upload_dir)? + .send() + .await + .map_err(|e| SyncError::Upload(e.to_string()))?; + match resp.status().as_u16() { + 201 | 405 => {} // created, or a prior attempt's collection is still there + 401 | 403 => return Err(SyncError::Auth), + other => return Err(SyncError::Upload(format!("MKCOL uploads {other}"))), + } + + let existing = self.list_upload_chunks(&upload_dir).await; + + let mut file = tokio::fs::File::open(local) + .await + .map_err(|e| SyncError::Upload(e.to_string()))?; + let mut offset = 0u64; + let mut sent: u64 = existing.values().sum::().min(total); + let _ = prog.send((sent, total)); + while offset < total { + let this_len = WEBDAV_CHUNK_SIZE.min(total - offset); + let name = chunk_name(offset); + if existing.get(&name).copied() == Some(this_len) { + // Already landed server-side from an earlier attempt — skip + // the network round trip entirely. + offset += this_len; + continue; + } + use tokio::io::{AsyncReadExt, AsyncSeekExt}; + file.seek(std::io::SeekFrom::Start(offset)) + .await + .map_err(|e| SyncError::Upload(e.to_string()))?; + let mut buf = vec![0u8; this_len as usize]; + file.read_exact(&mut buf) + .await + .map_err(|e| SyncError::Upload(e.to_string()))?; + let chunk_url = format!("{upload_dir}/{name}"); + let resp = self + .request(reqwest::Method::PUT, &chunk_url)? + .header(reqwest::header::CONTENT_LENGTH, this_len) + .body(buf) + .send() + .await + .map_err(|e| SyncError::Upload(e.to_string()))?; + match resp.status().as_u16() { + code if (200..300).contains(&code) => {} + 401 | 403 => return Err(SyncError::Auth), + other => return Err(SyncError::Upload(format!("PUT chunk {other}"))), + } + offset += this_len; + sent += this_len; + let _ = prog.send((sent.min(total), total)); + } + + // Assemble: MOVE-ing the collection's virtual `.file` member onto the + // real destination is what tells the server to concatenate the parts + // (in name-sorted order) and materialize the final file. + let dest_url = self.url_for(remote_path); + let move_method = + reqwest::Method::from_bytes(b"MOVE").expect("MOVE is a valid method token"); + let resp = self + .request(move_method, &format!("{upload_dir}/.file"))? + .header("Destination", dest_url) + .header("OC-Total-Length", total.to_string()) + .header("Overwrite", "T") + .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!("MOVE assemble {other}"))), + } + } + + /// Best-effort PROPFIND (depth 1) over the upload collection, returning + /// `{chunk name -> size}` for whatever's already there. Any transport or + /// parse hiccup just yields an empty map — the caller then re-uploads + /// every chunk, which is always correct, just not optimally resumed. + async fn list_upload_chunks(&self, upload_dir: &str) -> HashMap { + let method = match reqwest::Method::from_bytes(b"PROPFIND") { + Ok(m) => m, + Err(_) => return HashMap::new(), + }; + let Ok(builder) = self.request(method, upload_dir) else { + return HashMap::new(); + }; + let resp = builder + .header("Depth", "1") + .header(reqwest::header::CONTENT_TYPE, "application/xml") + .body( + r#""#, + ) + .send() + .await; + let Ok(resp) = resp else { + return HashMap::new(); + }; + if resp.status().as_u16() != 207 { + return HashMap::new(); + } + let Ok(body) = resp.text().await else { + return HashMap::new(); + }; + parse_chunk_sizes(&body) + } +} + +/// Open `local` and wrap it as a reqwest streaming request body that reads +/// fixed-size buffers off disk and reports cumulative bytes-sent to `prog` as +/// the request drains it (FR-SYNC-11) — memory use stays O(chunk), never +/// O(file size), and the UI sees steady movement on a large recording instead +/// of a single 0→100% jump at the end. +#[cfg(feature = "sync")] +async fn open_streaming_body( + local: &Path, + total: u64, + prog: ProgressSink, +) -> Result { + let file = tokio::fs::File::open(local) + .await + .map_err(|e| SyncError::Upload(e.to_string()))?; + let reader = tokio::io::BufReader::with_capacity(256 * 1024, file); + const READ_CHUNK: usize = 256 * 1024; + let stream = futures_util::stream::unfold((reader, 0u64), move |(mut reader, sent)| { let prog = prog.clone(); async move { - if pos >= bytes.len() { - return None; + use tokio::io::AsyncReadExt; + let mut buf = vec![0u8; READ_CHUNK]; + match reader.read(&mut buf).await { + Ok(0) => None, + Ok(n) => { + buf.truncate(n); + let sent = sent + n as u64; + let _ = prog.send((sent.min(total), total)); + Some((Ok::, std::io::Error>(buf), (reader, sent))) + } + Err(e) => Some((Err(e), (reader, sent))), } - let end = (pos + CHUNK).min(bytes.len()); - let chunk = bytes[pos..end].to_vec(); - let _ = prog.send((end as u64, total)); - Some((Ok::, std::io::Error>(chunk), (bytes, end))) } }); - reqwest::Body::wrap_stream(stream) + Ok(reqwest::Body::wrap_stream(stream)) +} + +/// Zero-padded (20-digit, comfortably covers any real file's byte offset) +/// decimal byte offset, used as a WebDAV chunk's name — chosen so lexical +/// sort order (what the server assembles by) matches upload order. +#[cfg(feature = "sync")] +fn chunk_name(offset: u64) -> String { + format!("{offset:020}") +} + +/// Best-effort scrape of a depth-1 PROPFIND multistatus response for +/// `{child name -> getcontentlength}`. Deliberately not a real XML parser — +/// namespace prefixes vary by server (`d:`, `D:`, `oc:`…) and we only need +/// two well-known leaf tags; any block that doesn't match is just skipped, so +/// a parse miss costs a redundant re-upload of that one chunk, never +/// correctness. +#[cfg(feature = "sync")] +fn parse_chunk_sizes(xml: &str) -> HashMap { + let mut out = HashMap::new(); + for block in xml.split("response>").skip(1) { + let Some(href) = extract_tag_text(block, "href") else { + continue; + }; + let Some(len) = + extract_tag_text(block, "getcontentlength").and_then(|s| s.trim().parse::().ok()) + else { + continue; + }; + let name = href.trim_end_matches('/').rsplit('/').next().unwrap_or(""); + if !name.is_empty() { + out.insert(name.to_string(), len); + } + } + out +} + +/// Extracts the text content of the first `<...:local_name>...` (or +/// unprefixed `...`) element in `block`. +#[cfg(feature = "sync")] +fn extract_tag_text(block: &str, local_name: &str) -> Option { + let needle = format!(":{local_name}>"); + let open_start = block.find(&needle).map(|i| i + needle.len()).or_else(|| { + let bare = format!("<{local_name}>"); + block.find(&bare).map(|i| i + bare.len()) + })?; + let rest = &block[open_start..]; + let end = rest.find(" + + + /remote.php/dav/uploads/u/abc/00000000000000000000 + + 10485760 + HTTP/1.1 200 OK + + + + /remote.php/dav/uploads/u/abc/00000000010485760 + + 5242880 + HTTP/1.1 200 OK + + + + + /remote.php/dav/uploads/u/abc/ + + + HTTP/1.1 200 OK + + +"#; + let sizes = parse_chunk_sizes(xml); + assert_eq!(sizes.get("00000000000000000000").copied(), Some(10_485_760)); + assert_eq!(sizes.get("00000000010485760").copied(), Some(5_242_880)); + assert_eq!( + sizes.len(), + 2, + "the length-less collection entry is skipped" + ); + } + + #[test] + fn chunk_name_is_zero_padded_and_lexically_sortable() { + assert_eq!(chunk_name(0), "00000000000000000000"); + assert!(chunk_name(0) < chunk_name(10_485_760)); + assert!(chunk_name(10_485_760) < chunk_name(20_971_520)); + } + + // ---- Minimal in-process mock servers (std::net only, no new crates) so + // the chunked/resumable protocols can be exercised end-to-end in `cargo + // test` without a live Nextcloud/Graph account — mirrors the loopback + // server pattern already used by `oauth::LoopbackRedirect`. + + struct MockRequest { + method: String, + path: String, + headers: HashMap, + body: Vec, + } + + fn read_mock_request(stream: &mut std::net::TcpStream) -> Option { + use std::io::Read; + let mut buf = Vec::new(); + let mut tmp = [0u8; 16384]; + let header_end = loop { + let n = stream.read(&mut tmp).ok()?; + if n == 0 { + return None; + } + buf.extend_from_slice(&tmp[..n]); + if let Some(pos) = buf.windows(4).position(|w| w == b"\r\n\r\n") { + break pos; + } + if buf.len() > 4 * 1024 * 1024 { + return None; // safety cap on header size + } + }; + let header_text = String::from_utf8_lossy(&buf[..header_end]).into_owned(); + let mut lines = header_text.split("\r\n"); + let request_line = lines.next()?; + let mut rl = request_line.split_whitespace(); + let method = rl.next()?.to_string(); + let path = rl.next()?.to_string(); + let mut headers = HashMap::new(); + for line in lines { + if let Some((k, v)) = line.split_once(':') { + headers.insert(k.trim().to_ascii_lowercase(), v.trim().to_string()); + } + } + let content_length: usize = headers + .get("content-length") + .and_then(|v| v.parse().ok()) + .unwrap_or(0); + let mut body = buf[header_end + 4..].to_vec(); + while body.len() < content_length { + let n = stream.read(&mut tmp).ok()?; + if n == 0 { + break; + } + body.extend_from_slice(&tmp[..n]); + } + body.truncate(content_length); + Some(MockRequest { + method, + path, + headers, + body, + }) + } + + fn write_mock_response( + stream: &mut std::net::TcpStream, + status: u16, + body: &[u8], + content_type: &str, + ) { + use std::io::Write; + let status_text = match status { + 200 => "OK", + 201 => "Created", + 202 => "Accepted", + 207 => "Multi-Status", + 400 => "Bad Request", + 404 => "Not Found", + 405 => "Method Not Allowed", + 416 => "Range Not Satisfiable", + _ => "OK", + }; + let _ = write!( + stream, + "HTTP/1.1 {status} {status_text}\r\nContent-Type: {content_type}\r\nContent-Length: {}\r\nConnection: close\r\n\r\n", + body.len() + ); + let _ = stream.write_all(body); + } + + /// A tiny Nextcloud-chunking-v2-aware WebDAV mock: MKCOL/PROPFIND/PUT/MOVE + /// against in-memory state, just enough to validate our client's protocol + /// sequence (including resume-skip) without a real Nextcloud server. + struct MockNextcloud { + port: u16, + files: std::sync::Arc>>>, + // `collections` (MKCOL bookkeeping) is only ever touched by the + // background request-handling threads, via their own clone — not + // read back through `Self` after construction. + chunk_put_count: std::sync::Arc, + } + + impl MockNextcloud { + fn start() -> Self { + let listener = std::net::TcpListener::bind("127.0.0.1:0").expect("bind mock nextcloud"); + let port = listener.local_addr().expect("local addr").port(); + let files = std::sync::Arc::new(std::sync::Mutex::new(HashMap::new())); + let collections = + std::sync::Arc::new(std::sync::Mutex::new(std::collections::HashSet::new())); + let chunk_put_count = std::sync::Arc::new(std::sync::atomic::AtomicU32::new(0)); + let (f, c, n) = (files.clone(), collections.clone(), chunk_put_count.clone()); + std::thread::spawn(move || { + for stream in listener.incoming() { + let Ok(mut stream) = stream else { continue }; + let (f, c, n) = (f.clone(), c.clone(), n.clone()); + std::thread::spawn(move || { + if let Some(req) = read_mock_request(&mut stream) { + handle_nextcloud_request(&mut stream, &req, &f, &c, &n); + } + }); + } + }); + Self { + port, + files, + chunk_put_count, + } + } + + fn base_url(&self) -> String { + format!("http://127.0.0.1:{}", self.port) + } + + fn chunk_put_count(&self) -> u32 { + self.chunk_put_count + .load(std::sync::atomic::Ordering::SeqCst) + } + + fn file(&self, path: &str) -> Option> { + self.files.lock().unwrap().get(path).cloned() + } + + fn seed_chunk(&self, path: &str, bytes: Vec) { + self.files.lock().unwrap().insert(path.to_string(), bytes); + } + } + + fn handle_nextcloud_request( + stream: &mut std::net::TcpStream, + req: &MockRequest, + files: &std::sync::Mutex>>, + collections: &std::sync::Mutex>, + chunk_put_count: &std::sync::atomic::AtomicU32, + ) { + match req.method.as_str() { + "MKCOL" => { + let mut cols = collections.lock().unwrap(); + if cols.insert(req.path.clone()) { + write_mock_response(stream, 201, b"", "text/plain"); + } else { + write_mock_response(stream, 405, b"", "text/plain"); + } + } + "PROPFIND" => { + if !collections.lock().unwrap().contains(&req.path) { + write_mock_response(stream, 404, b"", "text/plain"); + return; + } + let files = files.lock().unwrap(); + let prefix = format!("{}/", req.path.trim_end_matches('/')); + let mut body = + String::from(r#""#); + for (path, bytes) in files.iter() { + if let Some(rest) = path.strip_prefix(&prefix) { + if !rest.contains('/') { + body.push_str(&format!( + "{path}{}HTTP/1.1 200 OK", + bytes.len() + )); + } + } + } + body.push_str(""); + write_mock_response(stream, 207, body.as_bytes(), "application/xml"); + } + "PUT" => { + files + .lock() + .unwrap() + .insert(req.path.clone(), req.body.clone()); + if req.path.contains("/uploads/") { + chunk_put_count.fetch_add(1, std::sync::atomic::Ordering::SeqCst); + } + write_mock_response(stream, 201, b"", "text/plain"); + } + "MOVE" => { + let Some(dest_header) = req.headers.get("destination") else { + write_mock_response(stream, 400, b"", "text/plain"); + return; + }; + let dest_path = reqwest::Url::parse(dest_header) + .map(|u| u.path().to_string()) + .unwrap_or_else(|_| dest_header.clone()); + let collection = req.path.trim_end_matches("/.file").to_string(); + let mut files = files.lock().unwrap(); + let prefix = format!("{collection}/"); + let mut names: Vec = files + .keys() + .filter(|k| k.starts_with(&prefix) && !k[prefix.len()..].contains('/')) + .cloned() + .collect(); + names.sort(); + let mut assembled = Vec::new(); + for name in &names { + assembled.extend_from_slice(&files[name]); + } + files.insert(dest_path, assembled); + write_mock_response(stream, 201, b"", "text/plain"); + } + _ => write_mock_response(stream, 405, b"", "text/plain"), + } + } + + #[tokio::test] + async fn nextcloud_chunked_upload_assembles_and_resumes() { + let mock = MockNextcloud::start(); + let target = WebDavTarget { + base_url: mock.base_url(), + provider_hint: Some("nextcloud".to_string()), + remote_base_path: "/WhispAssist-test".to_string(), + username: "testuser".to_string(), + credential_ref: "wa-sync-test-nextcloud-mock".to_string(), + third_party: false, + allow_plaintext_lan: true, + }; + + // Three chunks (10 + 10 + 1 MiB) without needing a literal 250 MB file. + let total_len = (WEBDAV_CHUNK_SIZE * 2) + (1024 * 1024); + let mut content = vec![0u8; total_len as usize]; + for (i, b) in content.iter_mut().enumerate() { + *b = (i % 251) as u8; + } + let tmp = std::env::temp_dir().join(format!( + "wa_nextcloud_chunk_probe_{}.bin", + std::process::id() + )); + tokio::fs::write(&tmp, &content).await.unwrap(); + + let (tx, rx) = std::sync::mpsc::channel(); + target + .put(&tmp, "/WhispAssist-test/m1/audio.wav", tx) + .await + .expect("chunked put should succeed"); + + let events: Vec<_> = rx.try_iter().collect(); + assert!( + events.len() >= 3, + "expected per-chunk progress events, got {events:?}" + ); + + let assembled = mock + .file("/remote.php/dav/files/testuser/WhispAssist-test/m1/audio.wav") + .expect("assembled file should exist"); + assert_eq!( + assembled, content, + "assembled bytes must match the source file exactly" + ); + assert_eq!(mock.chunk_put_count(), 3, "expected exactly 3 chunk PUTs"); + + // Resume: pre-seed the first chunk as if a prior attempt already + // landed it, then confirm a fresh put() skips re-sending it. + let upload_id = { + use sha2::{Digest, Sha256}; + let mut h = Sha256::new(); + h.update(b"/WhispAssist-test/m1/audio2.wav"); + format!("{:x}", h.finalize()) + }; + mock.seed_chunk( + &format!( + "/remote.php/dav/uploads/testuser/{upload_id}/{}", + chunk_name(0) + ), + content[0..WEBDAV_CHUNK_SIZE as usize].to_vec(), + ); + let before = mock.chunk_put_count(); + let (tx2, _rx2) = std::sync::mpsc::channel(); + target + .put(&tmp, "/WhispAssist-test/m1/audio2.wav", tx2) + .await + .expect("resumed chunked put should succeed"); + assert_eq!( + mock.chunk_put_count() - before, + 2, + "resume should skip the chunk that already landed" + ); + let assembled2 = mock + .file("/remote.php/dav/files/testuser/WhispAssist-test/m1/audio2.wav") + .expect("resumed file should exist"); + assert_eq!(assembled2, content); + + let _ = tokio::fs::remove_file(&tmp).await; + } + + #[tokio::test] + async fn small_file_skips_chunking_even_on_a_chunking_capable_target() { + let mock = MockNextcloud::start(); + let target = WebDavTarget { + base_url: mock.base_url(), + provider_hint: Some("nextcloud".to_string()), + remote_base_path: "/WhispAssist-test".to_string(), + username: "testuser".to_string(), + credential_ref: "wa-sync-test-nextcloud-small".to_string(), + third_party: false, + allow_plaintext_lan: true, + }; + let tmp = std::env::temp_dir().join(format!("wa_small_probe_{}.bin", std::process::id())); + tokio::fs::write(&tmp, b"small transcript.json content") + .await + .unwrap(); + let (tx, _rx) = std::sync::mpsc::channel(); + target + .put(&tmp, "/WhispAssist-test/m1/transcript.json", tx) + .await + .expect("plain put should succeed"); + assert_eq!( + mock.chunk_put_count(), + 0, + "a small file must take the plain PUT path, never chunking" + ); + assert_eq!( + mock.file("/remote.php/dav/files/testuser/WhispAssist-test/m1/transcript.json") + .as_deref(), + Some(&b"small transcript.json content"[..]) + ); + let _ = tokio::fs::remove_file(&tmp).await; + } + + /// Minimal Microsoft Graph upload-session mock: `createUploadSession` + + /// `Content-Range` byte-range PUTs, enough to validate `OneDriveTarget`'s + /// chunked path without a live Graph/OAuth account. Wired in via + /// `WA_GRAPH_BASE_URL` (see `OneDriveTarget::graph_base`). + struct MockGraph { + port: u16, + uploads: std::sync::Arc>>>, + put_count: std::sync::Arc, + } + + impl MockGraph { + fn start() -> Self { + let listener = std::net::TcpListener::bind("127.0.0.1:0").expect("bind mock graph"); + let port = listener.local_addr().unwrap().port(); + let uploads = std::sync::Arc::new(std::sync::Mutex::new(HashMap::new())); + let put_count = std::sync::Arc::new(std::sync::atomic::AtomicU32::new(0)); + let base = format!("http://127.0.0.1:{port}"); + let (u, n) = (uploads.clone(), put_count.clone()); + std::thread::spawn(move || { + for stream in listener.incoming() { + let Ok(mut stream) = stream else { continue }; + let (u, n, base) = (u.clone(), n.clone(), base.clone()); + std::thread::spawn(move || { + if let Some(req) = read_mock_request(&mut stream) { + handle_graph_request(&mut stream, &req, &u, &n, &base); + } + }); + } + }); + Self { + port, + uploads, + put_count, + } + } + + fn base_url(&self) -> String { + format!("http://127.0.0.1:{}", self.port) + } + + fn put_count(&self) -> u32 { + self.put_count.load(std::sync::atomic::Ordering::SeqCst) + } + + fn assembled_any(&self) -> Option> { + self.uploads.lock().unwrap().values().next().cloned() + } + } + + fn handle_graph_request( + stream: &mut std::net::TcpStream, + req: &MockRequest, + uploads: &std::sync::Mutex>>, + put_count: &std::sync::atomic::AtomicU32, + base: &str, + ) { + if req.method == "POST" && req.path.ends_with(":/createUploadSession") { + let session_id = format!("session-{}", uuid::Uuid::new_v4().simple()); + uploads + .lock() + .unwrap() + .insert(session_id.clone(), Vec::new()); + let body = serde_json::json!({ + "uploadUrl": format!("{base}/upload-session/{session_id}"), + }); + write_mock_response(stream, 200, body.to_string().as_bytes(), "application/json"); + return; + } + if req.method == "PUT" && req.path.starts_with("/upload-session/") { + let session_id = req.path.trim_start_matches("/upload-session/").to_string(); + let Some(range) = req.headers.get("content-range") else { + write_mock_response(stream, 400, b"", "text/plain"); + return; + }; + let Some((ab, total_str)) = range.trim_start_matches("bytes ").split_once('/') else { + write_mock_response(stream, 400, b"", "text/plain"); + return; + }; + let Some((a, b)) = ab.split_once('-') else { + write_mock_response(stream, 400, b"", "text/plain"); + return; + }; + let (Ok(a), Ok(b), Ok(total)) = + (a.parse::(), b.parse::(), total_str.parse::()) + else { + write_mock_response(stream, 400, b"", "text/plain"); + return; + }; + let mut uploads = uploads.lock().unwrap(); + let Some(buf) = uploads.get_mut(&session_id) else { + write_mock_response(stream, 404, b"", "text/plain"); + return; + }; + if buf.len() as u64 != a { + write_mock_response(stream, 416, b"", "text/plain"); + return; + } + buf.extend_from_slice(&req.body); + let done = b + 1 == total; + drop(uploads); + put_count.fetch_add(1, std::sync::atomic::Ordering::SeqCst); + if done { + write_mock_response(stream, 201, b"{}", "application/json"); + } else { + write_mock_response(stream, 202, b"{}", "application/json"); + } + return; + } + write_mock_response(stream, 404, b"", "text/plain"); + } + + #[tokio::test] + async fn onedrive_chunked_upload_via_graph_session_mock() { + let mock = MockGraph::start(); + // SAFETY (test-only): no other test in this binary reads/writes these + // vars concurrently with OneDriveTarget network calls. A client id is + // required even though our token never needs refreshing (`bearer()` + // resolves it unconditionally before checking expiry). + unsafe { + std::env::set_var("WA_GRAPH_BASE_URL", mock.base_url()); + std::env::set_var("WA_OAUTH_ONEDRIVE_CLIENT_ID", "test-client-id"); + } + + let cred_ref = format!("wa-sync-test-onedrive-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 = OneDriveTarget { + credential_ref: cred_ref.clone(), + }; + + let total_len = (GRAPH_CHUNK_SIZE * 2) + (512 * 1024); + let mut content = vec![0u8; total_len as usize]; + for (i, b) in content.iter_mut().enumerate() { + *b = ((i * 7) % 253) as u8; + } + let tmp = + std::env::temp_dir().join(format!("wa_graph_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_GRAPH_BASE_URL"); + std::env::remove_var("WA_OAUTH_ONEDRIVE_CLIENT_ID"); + } + + result.expect("chunked graph put should succeed"); + let events: Vec<_> = rx.try_iter().collect(); + assert!( + events.len() >= 3, + "expected per-chunk progress events, got {events:?}" + ); + assert_eq!(mock.put_count(), 3, "expected exactly 3 range PUTs"); + let sent = mock.assembled_any().expect("session bytes recorded"); + assert_eq!( + sent, content, + "assembled bytes must match the source file exactly" + ); + } } #[cfg(feature = "sync")] @@ -593,11 +1377,21 @@ impl OneDriveTarget { resolve_access_token("onedrive", &self.credential_ref).await } + /// Graph API base — overridable via `WA_GRAPH_BASE_URL` so tests can point + /// this at a local mock instead of the real service. Unset (the normal + /// runtime case) always resolves to the real Graph endpoint. + fn graph_base() -> String { + std::env::var("WA_GRAPH_BASE_URL") + .ok() + .filter(|s| !s.is_empty()) + .unwrap_or_else(|| "https://graph.microsoft.com/v1.0".to_string()) + } + /// Graph path addressing: `…/root:/{path}{suffix}` (suffix `""` = the item, /// `:/content` = its bytes, `:/children` = its child collection). fn item_url(remote_path: &str, suffix: &str) -> String { let p = remote_path.trim_start_matches('/'); - format!("https://graph.microsoft.com/v1.0/me/drive/root:/{p}{suffix}") + format!("{}/me/drive/root:/{p}{suffix}", Self::graph_base()) } } @@ -614,7 +1408,7 @@ impl SyncTarget for OneDriveTarget { async fn test(&self) -> Result<(), SyncError> { let token = self.bearer().await?; let resp = http_client()? - .get("https://graph.microsoft.com/v1.0/me/drive") + .get(format!("{}/me/drive", Self::graph_base())) .bearer_auth(token) .send() .await @@ -631,9 +1425,9 @@ impl SyncTarget for OneDriveTarget { let mut parent = String::new(); for segment in remote_dir.split('/').filter(|s| !s.is_empty()) { let url = if parent.is_empty() { - "https://graph.microsoft.com/v1.0/me/drive/root/children".to_string() + format!("{}/me/drive/root/children", Self::graph_base()) } else { - format!("https://graph.microsoft.com/v1.0/me/drive/root:/{parent}:/children") + format!("{}/me/drive/root:/{parent}:/children", Self::graph_base()) }; let body = serde_json::json!({ "name": segment, @@ -672,24 +1466,30 @@ impl SyncTarget for OneDriveTarget { Ok(resp.status().is_success()) } - /// ponytail: simple content upload (Graph caps this at 250 MB). A resumable - /// upload session for larger recordings is a follow-up. + /// Upload, streamed from disk. Above `LARGE_FILE_THRESHOLD` (Graph's own + /// guidance: use an upload session above a few MB, and it hard-caps a + /// single PUT at 250 MB), uses a Graph upload session with ≤`GRAPH_CHUNK_SIZE` + /// byte-range PUTs; otherwise one streamed PUT. async fn put( &self, local: &Path, remote_path: &str, prog: ProgressSink, ) -> Result<(), SyncError> { - let token = self.bearer().await?; - let bytes = tokio::fs::read(local) + let total = tokio::fs::metadata(local) .await - .map_err(|e| SyncError::Upload(e.to_string()))?; - let total = bytes.len() as u64; + .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 resp = http_client()? .put(Self::item_url(remote_path, ":/content")) .bearer_auth(token) .header(reqwest::header::CONTENT_LENGTH, total) - .body(progress_body(bytes, prog)) + .body(body) .send() .await .map_err(|e| SyncError::Upload(e.to_string()))?; @@ -700,3 +1500,120 @@ impl SyncTarget for OneDriveTarget { } } } + +#[cfg(feature = "sync")] +impl OneDriveTarget { + /// Microsoft Graph large-file upload: create an upload session, then PUT + /// consecutive byte ranges with `Content-Range` (each ≤ `GRAPH_CHUNK_SIZE`, + /// itself a multiple of 320 KiB per Graph's requirement). Read errors and + /// network failures for one chunk are retried a few times in place before + /// giving up. + /// + /// ponytail (documented simplification, T9.2/M4.1 ceiling): a chunk is + /// retried within *this* `put()` call, but the session URL isn't persisted + /// on the `sync_jobs` row, so if the whole call fails and the durable queue + /// retries it later (a fresh `put()` invocation, e.g. after an app + /// restart), a new session starts and the upload resends from byte 0. + /// Graph sessions are queryable for `nextExpectedRanges` mid-flight, which + /// would let a *later* attempt resume too — that needs a place to persist + /// the session URL across attempts (a `sync_jobs` column), left as a + /// follow-up rather than a schema change for this milestone. + async fn put_chunked( + &self, + local: &Path, + remote_path: &str, + total: u64, + prog: ProgressSink, + ) -> Result<(), SyncError> { + let token = self.bearer().await?; + let session_url = self.create_upload_session(&token, remote_path).await?; + + let mut file = tokio::fs::File::open(local) + .await + .map_err(|e| SyncError::Upload(e.to_string()))?; + let mut offset = 0u64; + let mut sent = 0u64; + let _ = prog.send((0, total)); + while offset < total { + let this_len = GRAPH_CHUNK_SIZE.min(total - offset); + use tokio::io::{AsyncReadExt, AsyncSeekExt}; + file.seek(std::io::SeekFrom::Start(offset)) + .await + .map_err(|e| SyncError::Upload(e.to_string()))?; + 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 range = format!("bytes {offset}-{end}/{total}"); + + let mut last_err = None; + let mut ok = false; + for attempt in 0..3u32 { + if attempt > 0 { + tokio::time::sleep(std::time::Duration::from_millis(400 * attempt as u64)) + .await; + } + let resp = http_client()? + .put(&session_url) + .header(reqwest::header::CONTENT_LENGTH, this_len) + .header(reqwest::header::CONTENT_RANGE, range.clone()) + .body(buf.clone()) + .send() + .await; + match resp { + Ok(r) if (200..300).contains(&r.status().as_u16()) => { + ok = true; + break; + } + Ok(r) => last_err = Some(format!("chunk PUT {}", r.status().as_u16())), + Err(e) => last_err = Some(e.to_string()), + } + } + if !ok { + return Err(SyncError::Upload( + last_err.unwrap_or_else(|| "chunk upload failed".to_string()), + )); + } + offset += this_len; + sent += this_len; + let _ = prog.send((sent, total)); + } + Ok(()) + } + + /// POST `…/createUploadSession`, returning the (single-use, presigned) + /// `uploadUrl` chunk PUTs go to. + async fn create_upload_session( + &self, + token: &str, + remote_path: &str, + ) -> Result { + let url = Self::item_url(remote_path, ":/createUploadSession"); + let body = serde_json::json!({ + "item": { "@microsoft.graph.conflictBehavior": "replace" }, + }); + let resp = http_client()? + .post(&url) + .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) => {} + 401 | 403 => return Err(SyncError::Auth), + other => return Err(SyncError::Upload(format!("createUploadSession {other}"))), + } + #[derive(serde::Deserialize)] + struct SessionResp { + #[serde(rename = "uploadUrl")] + upload_url: String, + } + let parsed: SessionResp = resp + .json() + .await + .map_err(|e| SyncError::Upload(e.to_string()))?; + Ok(parsed.upload_url) + } +}