diff --git a/apps/notescraft-sync-server/tests/chunk_client_e2e.rs b/apps/notescraft-sync-server/tests/chunk_client_e2e.rs new file mode 100644 index 0000000..9844068 --- /dev/null +++ b/apps/notescraft-sync-server/tests/chunk_client_e2e.rs @@ -0,0 +1,239 @@ +//! The real engine sync client against the real server: a big file travels in +//! chunks, a cut connection resumes with the chunks that never arrived. + +use std::path::{Path, PathBuf}; +use std::sync::Mutex; + +use notescraft_engine::{Changes, HttpRemote, Limits, Progress, PutReply, Remote, SyncError, SyncOptions, SyncReport, SyncResult, SyncState, sync_once_with}; +use notescraft_sync_server::{AppState, Store, router}; +use notescraft_vault::Vault; + +const ADMIN: &str = "test-admin-token-0123456789abcdef"; +/// The size of one chunk the client sends. +const CHUNK: usize = 32 * 1024; +/// The size above which the client sends a file in chunks. +const THRESHOLD: u64 = 16 * 1024; +/// The biggest chunk the server takes. +const SERVER_CHUNK_MAX: usize = 64 * 1024; +/// The biggest file the server assembles. +const SERVER_FILE_CAP: u64 = 1024 * 1024; +/// The size of the file that travels. +const FILE_BYTES: usize = 300 * 1024; + +/// A temp folder that is empty when the test starts. +fn temp(name: &str) -> PathBuf { + let dir = std::env::temp_dir().join(format!("nc-chunk-e2e-{name}-{}", std::process::id())); + std::fs::remove_dir_all(&dir).ok(); + std::fs::create_dir_all(&dir).expect("dir"); + dir +} + +/// Starts the server on an ephemeral port and returns its base url. +fn start(data: &Path) -> String { + std::fs::create_dir_all(data.join("blobs")).expect("blobs"); + let state = AppState::new(Store::open(&data.join("sync.db")).expect("db"), data.join("blobs"), ADMIN); + // The chunk limits are set before the router is built: the route takes its + // body cap from the state. + state.set_chunk_limits(SERVER_CHUNK_MAX, SERVER_FILE_CAP); + let listener = std::net::TcpListener::bind("127.0.0.1:0").expect("bind"); + listener.set_nonblocking(true).expect("nonblocking"); + let port = listener.local_addr().expect("addr").port(); + std::thread::spawn(move || { + let runtime = tokio::runtime::Builder::new_multi_thread().enable_all().build().expect("runtime"); + runtime.block_on(async { + let listener = tokio::net::TcpListener::from_std(listener).expect("listener"); + axum::serve(listener, router(state)).await.ok(); + }); + }); + format!("http://127.0.0.1:{port}") +} + +fn admin(base: &str, path: &str) -> serde_json::Value { + let agent: ureq::Agent = ureq::Agent::config_builder().http_status_as_error(false).build().into(); + let mut response = agent.post(format!("{base}{path}")).header("Authorization", format!("Bearer {ADMIN}")).send_empty().expect("admin call"); + response.body_mut().read_json().expect("json") +} + +/// Issues a pairing code for the vault, bound to `user` when given. +fn issue(base: &str, vault_id: &str, user: Option<&str>) -> String { + let path = user.map_or_else(|| format!("/v1/vaults/{vault_id}/pairing"), |user| format!("/v1/vaults/{vault_id}/pairing?user={user}")); + admin(base, &path)["code"].as_str().expect("code").to_owned() +} + +/// Redeems a code with raw HTTP to also learn the account the device joined. +fn pair_raw(base: &str, code: &str, name: &str) -> (String, String) { + let agent: ureq::Agent = ureq::Agent::config_builder().http_status_as_error(false).build().into(); + let mut response = agent.post(format!("{base}/v1/devices")).send_json(serde_json::json!({"code": code, "name": name})).expect("pair"); + let device: serde_json::Value = response.body_mut().read_json().expect("json"); + (device["token"].as_str().expect("token").to_owned(), device["user_id"].as_str().expect("user id").to_owned()) +} + +/// A remote that can cut the connection in the middle of a chunked upload. +struct CutRemote { + inner: HttpRemote, + /// How many chunks the client sent. + sent: Mutex, + /// From this send on, the connection is cut. + cut_from: Mutex>, +} + +impl CutRemote { + fn new(inner: HttpRemote) -> Self { + Self { inner, sent: Mutex::new(0), cut_from: Mutex::new(None) } + } + + /// Cuts the connection from the given chunk send on. + fn cut_from(&self, call: usize) { + *self.cut_from.lock().unwrap_or_else(std::sync::PoisonError::into_inner) = Some(call); + } + + /// Lets the connection through again. + fn mend(&self) { + *self.cut_from.lock().unwrap_or_else(std::sync::PoisonError::into_inner) = None; + } + + /// How many chunks were sent since the last count. + fn take_sent(&self) -> usize { + let mut sent = self.sent.lock().unwrap_or_else(std::sync::PoisonError::into_inner); + std::mem::take(&mut *sent) + } +} + +impl Remote for CutRemote { + fn changes(&self, since: u64, limit: usize, wait: u64) -> SyncResult { + self.inner.changes(since, limit, wait) + } + + fn get(&self, path: &str) -> SyncResult> { + self.inner.get(path) + } + + fn put(&self, path: &str, bytes: &[u8], base_seq: u64, mtime_ms: u64) -> SyncResult { + self.inner.put(path, bytes, base_seq, mtime_ms) + } + + fn delete(&self, path: &str, base_seq: u64) -> SyncResult { + self.inner.delete(path, base_seq) + } + + fn missing_chunks(&self, hashes: &[String]) -> SyncResult> { + self.inner.missing_chunks(hashes) + } + + fn put_chunk(&self, hash: &str, bytes: &[u8]) -> SyncResult<()> { + let mut sent = self.sent.lock().unwrap_or_else(std::sync::PoisonError::into_inner); + *sent = sent.saturating_add(1); + let cut = *self.cut_from.lock().unwrap_or_else(std::sync::PoisonError::into_inner); + if cut.is_some_and(|from| *sent >= from) { + return Err(SyncError::Http("the connection was cut".to_owned())); + } + drop(sent); + self.inner.put_chunk(hash, bytes) + } + + fn put_chunked(&self, path: &str, chunk_hashes: &[String], content_hash: &str, base_seq: u64, mtime_ms: u64) -> SyncResult { + self.inner.put_chunked(path, chunk_hashes, content_hash, base_seq, mtime_ms) + } +} + +/// A device with its own vault, its own state and a remote that can cut. +struct Device { + vault: Vault, + remote: CutRemote, + state: SyncState, + name: &'static str, +} + +impl Device { + fn new(base: &str, vault_id: &str, token: &str, name: &'static str) -> Self { + let root = temp(name); + let vault = Vault::open(&root).expect("vault"); + Self { vault, remote: CutRemote::new(HttpRemote::new(base, vault_id, token)), state: SyncState::default(), name } + } + + fn write(&self, path: &str, bytes: &[u8]) { + self.vault.write(path, bytes).expect("write"); + } + + fn read(&self, path: &str) -> Vec { + self.vault.read(path).expect("read") + } + + /// Syncs once and returns the report. + fn sync(&mut self) -> SyncReport { + let options = SyncOptions { limits: Limits { threshold: THRESHOLD, chunk: CHUNK }, progress: Box::new(|_: &Progress| {}) }; + sync_once_with(&self.vault, &self.remote, &mut self.state, self.name, "2026-10-20", options).expect("sync") + } +} + +/// Bytes that look random but come from a small generator, so the test is stable. +fn random_bytes(len: usize, seed: u64) -> Vec { + let mut state = seed; + (0..len) + .map(|_| { + state = state.wrapping_mul(6_364_136_223_846_793_005).wrapping_add(1_442_695_040_888_963_407); + u8::try_from((state >> 33) & 0xff).expect("byte") + }) + .collect() +} + +/// How many chunks the server holds for the vault. +fn stored_chunks(blobs: &Path, vault_id: &str) -> usize { + let dir = blobs.join(vault_id).join("chunks"); + let Ok(entries) = std::fs::read_dir(dir) else { return 0 }; + entries.filter_map(std::result::Result::ok).filter(|entry| entry.path().is_file()).count() +} + +#[test] +fn a_big_file_travels_in_chunks_and_a_cut_resumes_with_the_rest() { + let data = temp("server-data"); + let base = start(&data); + let blobs = data.join("blobs"); + let vault_id = format!("v-chunk-{}", std::time::SystemTime::now().duration_since(std::time::UNIX_EPOCH).map_or(0, |d| d.subsec_nanos())); + let (phone_token, user_id) = pair_raw(&base, &issue(&base, &vault_id, None), "phone"); + let laptop = pair_raw(&base, &issue(&base, &vault_id, Some(&user_id)), "laptop"); + let mut phone = Device::new(&base, &vault_id, &phone_token, "phone"); + let mut laptop = Device::new(&base, &vault_id, &laptop.0, "laptop"); + + // A file that is too big for one request travels in chunks. + let bytes = random_bytes(FILE_BYTES, 42); + phone.write("big.bin", &bytes); + let report = phone.sync(); + assert_eq!(report.uploaded, vec!["big.bin".to_owned()]); + let chunks = FILE_BYTES.div_ceil(CHUNK); + assert_eq!(stored_chunks(&blobs, &vault_id), chunks, "the server holds every chunk"); + + // The other device gets the very same bytes. + let report = laptop.sync(); + assert_eq!(report.downloaded, vec!["big.bin".to_owned()]); + assert_eq!(laptop.read("big.bin"), bytes); + + // The connection dies on the fourth chunk of the next file. + let other = random_bytes(FILE_BYTES, 7); + phone.write("other.bin", &other); + phone.remote.take_sent(); + phone.remote.cut_from(4); + let error = sync_once_with( + &phone.vault, + &phone.remote, + &mut phone.state, + "phone", + "2026-10-20", + SyncOptions { limits: Limits { threshold: THRESHOLD, chunk: CHUNK }, progress: Box::new(|_: &Progress| {}) }, + ) + .expect_err("cut"); + assert!(matches!(error, SyncError::Http(_)), "{error:?}"); + phone.remote.take_sent(); + assert_eq!(stored_chunks(&blobs, &vault_id), chunks + 3, "the three chunks that arrived are kept"); + + // The next pass sends the chunks that never arrived and nothing else. + phone.remote.mend(); + phone.sync(); + assert_eq!(phone.remote.take_sent(), chunks - 3, "only the missing chunks travel again"); + assert_eq!(stored_chunks(&blobs, &vault_id), chunks * 2, "the server holds every chunk of both files"); + + // The other device converges on the second file too. + laptop.sync(); + assert_eq!(laptop.read("other.bin"), other); + assert_eq!(laptop.read("big.bin"), bytes); +} diff --git a/crates/engine/src/ai/assistant_gate.rs b/crates/engine/src/ai/assistant_gate.rs index 811b7b1..ca26589 100644 --- a/crates/engine/src/ai/assistant_gate.rs +++ b/crates/engine/src/ai/assistant_gate.rs @@ -199,6 +199,7 @@ pub const READ_COMMANDS: &[&str] = &[ "sync.status", "sync.conflicts", "server.vaults", + "server.devices", "share.members", "share.invites", "settings.get", diff --git a/crates/engine/src/lib.rs b/crates/engine/src/lib.rs index fdd19a9..9ec0eaf 100644 --- a/crates/engine/src/lib.rs +++ b/crates/engine/src/lib.rs @@ -68,7 +68,7 @@ pub(crate) use vault::{attachments, bookmarks, buffers, content_search, daily, e pub use crate::app_settings::{ APP_ID, AppSettings, AssistantSettings, EditorSettings, GitSettings, GrammarInfo, MAX_FONT_SIZE, MAX_INDENT_WIDTH, MAX_RECENT_VAULTS, MAX_SHORTCUT_LEN, - MAX_UI_BYTES, MIN_FONT_SIZE, MIN_INDENT_WIDTH, PluginSettings, PreviewSettings, SCHEMA_VERSION, ViewDefault, + MAX_UI_BYTES, MIN_FONT_SIZE, MIN_INDENT_WIDTH, PluginSettings, PreviewSettings, SCHEMA_VERSION, SyncSettings, ViewDefault, }; pub use crate::assistant::{is_gated_source, run_as_assistant}; pub use crate::assistant_chat::{ @@ -113,13 +113,15 @@ pub use crate::registry::{MAX_COMMANDS, Registry, current_command_id}; pub use crate::reminders_config::RemindersConfig; pub use crate::session::Session; pub use crate::settings_store::{load, load_at, load_checked, load_checked_at, save, save_at}; -pub use crate::share::{Accepted, Invite, InviteInfo, Member, ServerClient, ServerVault, ShareError, ShareResult, bind_folder, connect_with_code}; +pub use crate::share::{ + Accepted, Invite, InviteInfo, Member, ServerClient, ServerDevice, ServerVault, ShareError, ShareResult, bind_folder, connect_with_code, +}; pub use crate::slash::{SlashEntry, filter_entries, trigger_ok}; pub use crate::spec::{CommandResult, CommandSpec}; pub use crate::sync::{ - Binding, Changes, ConflictPair, FileRecord, HttpRemote, Keep, Pairing, PutReply, Remote, ReportCounts, SyncError, SyncPhase, SyncReport, SyncResult, - SyncState, SyncStatus, SyncWorker, Timings, clear_binding, list_conflicts, load_binding, load_token, pair, resolve_conflict, save_binding, save_token, - sync_once, + Binding, CHUNK_SIZE, CHUNK_THRESHOLD, Changes, ConflictPair, FileRecord, HttpRemote, Keep, Limits, Pairing, Progress, PutReply, Remote, ReportCounts, + SyncError, SyncOptions, SyncPhase, SyncReport, SyncResult, SyncState, SyncStatus, SyncWorker, Timings, Transfer, clear_binding, list_conflicts, + load_binding, load_token, pair, resolve_conflict, save_binding, save_token, sync_once, sync_once_with, upload_chunked, }; pub use crate::vault::snapshots::Snapshot; pub use notescraft_vault::{FileCount, LARGE_FOLDER_FILES, count_files}; diff --git a/crates/engine/src/prefs/app_settings.rs b/crates/engine/src/prefs/app_settings.rs index f438226..0deb744 100644 --- a/crates/engine/src/prefs/app_settings.rs +++ b/crates/engine/src/prefs/app_settings.rs @@ -73,6 +73,8 @@ pub struct AppSettings { /// launcher; a list, for the same reason `grammars` is: a forgotten vault /// must not come back. pub vault_catalog: Vec, + /// The three sync options the sync sheet and the Sync settings page share. + pub sync: SyncSettings, } /// One installed syntax grammar. @@ -108,6 +110,7 @@ impl Default for AppSettings { hotkeys: BTreeMap::new(), grammars: Vec::new(), vault_catalog: Vec::new(), + sync: SyncSettings::default(), } } } @@ -164,6 +167,28 @@ pub fn is_grammar_id(id: &str) -> bool { (1..=MAX_GRAMMAR_ID).contains(&id.chars().count()) && id.chars().all(|c| c.is_ascii_lowercase() || c.is_ascii_digit() || c == '-') } +/// The three options a sync vault may offer (`design/mobile/03-sync-sheet.png`). +/// +/// Booleans cannot hold a bad value, so there is nothing to repair: the +/// `#[serde(default)]` fields keep a missing key at its default when an older +/// `config.toml` is read. +#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)] +#[serde(default)] +pub struct SyncSettings { + /// Sync only when the machine is on Wi-Fi. + pub wifi_only: bool, + /// Sync files that are not Markdown. + pub attachments: bool, + /// Sync the settings and the themes of the app between the devices. + pub settings_and_themes: bool, +} + +impl Default for SyncSettings { + fn default() -> Self { + Self { wifi_only: true, attachments: true, settings_and_themes: false } + } +} + /// Drops grammar entries whose id is not a grammar id, is repeated, or whose file is not /// `.sublime-syntax`, and keeps the list sorted by id. fn repair_grammars(grammars: &mut Vec) -> Vec { @@ -438,6 +463,14 @@ mod tests { assert_eq!(settings.editor.font_size, MIN_FONT_SIZE); } + #[test] + fn sync_options_carry_their_defaults_and_an_old_file_reads() { + assert_eq!(SyncSettings::default(), SyncSettings { wifi_only: true, attachments: true, settings_and_themes: false }); + let empty: AppSettings = serde_json::from_str("{}").unwrap_or_default(); + assert_eq!(empty.sync, SyncSettings::default(), "a file without the section keeps the defaults"); + assert!(empty.sync.wifi_only); + } + #[test] fn session_repairs_are_prefixed_and_applied() { let mut settings = AppSettings::default(); diff --git a/crates/engine/src/prefs/settings.rs b/crates/engine/src/prefs/settings.rs index d9a71f2..e779f5b 100644 --- a/crates/engine/src/prefs/settings.rs +++ b/crates/engine/src/prefs/settings.rs @@ -187,7 +187,7 @@ fn write_key(doc: &mut Value, segments: &[String], value: Value) -> std::result: fn write_section(section: &mut Value, name: &str, rest: &[String], value: Value) -> std::result::Result<(), CommandError> { match name { "session" => session_path(section, rest, value), - "git" | "editor" => match rest { + "git" | "editor" | "sync" => match rest { [] => { whole_section(section, value); Ok(()) diff --git a/crates/engine/src/share/client.rs b/crates/engine/src/share/client.rs index 00e1031..22f7e7b 100644 --- a/crates/engine/src/share/client.rs +++ b/crates/engine/src/share/client.rs @@ -310,6 +310,21 @@ impl ServerClient { self.post_json("/v1/invites/accept", &json!({ "code": code })) } + /// The devices paired with a vault (`GET /v1/vaults/{vault}/devices`). + /// + /// The rows carry no token and no token hash: the server never keeps one + /// for a listing, and this request sends none either. + /// + /// # Errors + /// [`ShareError::Protocol`] when the vault id cannot travel in a URL, + /// [`ShareError::Http`] when the server cannot be reached and + /// [`ShareError::Status`] when it refuses the request. + pub fn devices(&self, vault: &str) -> ShareResult> { + refuse_id(vault)?; + let path = format!("/v1/vaults/{}/devices", encode_segment(vault)); + self.get_json(&path) + } + /// The URL of a path on the server. fn url(&self, path: &str) -> String { format!("{}{path}", self.base) diff --git a/crates/engine/src/share/commands.rs b/crates/engine/src/share/commands.rs index 91d9bc1..6db55f5 100644 --- a/crates/engine/src/share/commands.rs +++ b/crates/engine/src/share/commands.rs @@ -37,6 +37,7 @@ const DEVICE_NAME: &str = "NotesCraft"; pub(crate) fn register_share(reg: &mut Registry) -> crate::error::Result<()> { let specs = vec![ CommandSpec::new("server.vaults", "Server vaults", "Lists the vaults of the server account this vault is bound to", |_| true, server_vaults), + CommandSpec::new("server.devices", "Vault devices", "Lists the devices paired with the bound vault; never a token", |_| true, server_devices), CommandSpec::new( "server.create_vault", "Create server vault", @@ -119,6 +120,13 @@ fn server_vaults(session: &mut Session, _params: &Value) -> CommandResult { crate::commands::json(json!({ "vaults": vaults })) } +/// `server.devices`: the devices of the bound vault; never a token. +fn server_devices(session: &mut Session, _params: &Value) -> CommandResult { + let (client, binding) = client_of(session)?; + let devices = client.devices(&binding.vault_id).map_err(share_error)?; + crate::commands::json(json!({ "vault_id": binding.vault_id, "devices": devices })) +} + /// `server.create_vault`: a new vault, owned by this account. fn server_create_vault(session: &mut Session, params: &Value) -> CommandResult { let name = name_param(params, "name")?; diff --git a/crates/engine/src/share/devices.rs b/crates/engine/src/share/devices.rs new file mode 100644 index 0000000..6e1aebe --- /dev/null +++ b/crates/engine/src/share/devices.rs @@ -0,0 +1,42 @@ +//! The devices of a bound vault, as the server lists them. +//! +//! The server names `id`, `name`, `created_ms`, `last_seen_ms` and `current` +//! (design `03-sync-sheet`); a token or its hash is never part of the +//! request, the answer or a row. + +use serde::{Deserialize, Serialize}; + +/// One device paired with the bound vault, as `GET /v1/vaults/{vault}/devices` +/// answers it. +#[derive(Debug, Clone, Default, PartialEq, Eq, Serialize, Deserialize)] +pub struct ServerDevice { + /// The device's id. + #[serde(default)] + pub id: String, + /// The name the device paired under. + #[serde(default)] + pub name: String, + /// When the device was paired, in Unix milliseconds. + #[serde(default)] + pub created_ms: u64, + /// When the server saw the device last, if ever. + #[serde(default)] + pub last_seen_ms: Option, + /// True for the device that asked; the server decides. + #[serde(default)] + pub current: bool, +} + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn a_row_of_the_server_reads_whole_or_defaults() { + let row = serde_json::from_str::(r#"{"id":"d-1","name":"Pixel 9","current":true}"#).unwrap_or_default(); + assert_eq!(row, ServerDevice { id: "d-1".to_owned(), name: "Pixel 9".to_owned(), created_ms: 0, last_seen_ms: None, current: true }); + let full = + serde_json::from_str::(r#"{"id":"d-2","name":"Laptop","created_ms":50,"last_seen_ms":60,"current":false}"#).unwrap_or_default(); + assert_eq!(full.last_seen_ms, Some(60)); + } +} diff --git a/crates/engine/src/share/mod.rs b/crates/engine/src/share/mod.rs index 7f5194c..5c8abcf 100644 --- a/crates/engine/src/share/mod.rs +++ b/crates/engine/src/share/mod.rs @@ -15,9 +15,11 @@ pub(crate) mod bind; pub(crate) mod client; pub(crate) mod commands; +mod devices; pub use bind::{bind_folder, connect_with_code}; pub use client::{Accepted, Invite, InviteInfo, Member, ServerClient, ServerVault}; +pub use devices::ServerDevice; use serde::Deserialize; use thiserror::Error as ThisError; diff --git a/crates/engine/src/sync/chunks.rs b/crates/engine/src/sync/chunks.rs new file mode 100644 index 0000000..598b9b7 --- /dev/null +++ b/crates/engine/src/sync/chunks.rs @@ -0,0 +1,223 @@ +//! Chunked uploads: a big file travels to the server in content-addressed chunks. +//! +//! The server keeps every chunk by its own hash, so a file that was cut off half +//! way resumes with the chunks that never arrived: the client asks which ones +//! the server is missing and sends only those. One chunk at a time is read from +//! disk, so the memory a big file needs does not grow with the file. +//! +//! The client hashes the file once to learn the hash of every chunk and of the +//! whole content, asks the server which chunks it is missing, and reads the file +//! a second time to send the missing ones. The second pass checks every chunk +//! against the hash of the first pass, so a file that changed in between is +//! refused instead of being uploaded in pieces. + +use std::collections::HashSet; +use std::fs::File; +use std::io::Read; +use std::path::Path; +use std::time::Duration; + +use notescraft_sync::{hash_hex, hasher}; +use notescraft_vault::Vault; + +use crate::sync::{PutReply, Remote, SyncError, SyncResult}; + +/// Files above this size upload in chunks. +pub const CHUNK_THRESHOLD: u64 = 8 * 1024 * 1024; +/// How many bytes one chunk holds. +pub const CHUNK_SIZE: usize = 4 * 1024 * 1024; +/// The most hashes one "which chunks are missing" question holds. +const MAX_CHECK_BATCH: usize = 5000; +/// How many times a chunk is sent again before the upload is given up. +const CHUNK_RETRIES: usize = 3; +/// The waits between the tries of a chunk, in milliseconds. +const BACKOFF_MS: [u64; CHUNK_RETRIES] = [200, 400, 800]; + +/// How far a big file has moved. +#[derive(Debug, Clone, PartialEq, Eq)] +pub struct Progress { + /// The vault path of the file. + pub path: String, + /// The bytes of the file the server holds now. + pub done_bytes: u64, + /// The bytes the file has. + pub total_bytes: u64, +} + +/// When files upload in chunks and how big a chunk is. +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub struct Limits { + /// Files above this size upload in chunks. + pub threshold: u64, + /// How many bytes one chunk holds. + pub chunk: usize, +} + +impl Default for Limits { + fn default() -> Self { + Self { threshold: CHUNK_THRESHOLD, chunk: CHUNK_SIZE } + } +} + +/// Uploads `path` to the server in chunks. +/// +/// The file is read from `vault` twice: once to learn the hash of every chunk +/// and of the whole content, once to send the chunks the server is missing. +/// `progress` is called after every chunk, with the bytes the server holds. +/// +/// The answer is the one of a plain write: [`PutReply::Applied`] when the +/// server took the file, [`PutReply::Unchanged`] when it already held it and +/// [`PutReply::Conflict`] when somebody else wrote first. +/// +/// # Errors +/// [`SyncError::Vault`] when the file cannot be read, [`SyncError::BadRemote`] +/// when the file changed while it was uploaded, and whatever the transport +/// reports for the chunks and the write itself. +pub fn upload_chunked( + remote: &dyn Remote, + vault: &Vault, + path: &str, + base_seq: u64, + mtime_ms: u64, + limits: Limits, + progress: &mut dyn FnMut(&Progress), +) -> SyncResult { + if limits.chunk == 0 || limits.chunk > CHUNK_SIZE { + return Err(SyncError::Io(format!("a chunk must hold between one byte and {CHUNK_SIZE} bytes"))); + } + let file = vault.resolve(path).map_err(|error| SyncError::Vault(error.to_string()))?; + let plan = plan_chunks(&file, limits.chunk)?; + if plan.chunks.is_empty() { + // An empty file has no chunk: it goes as a plain write. + return remote.put(path, &[], base_seq, mtime_ms); + } + let missing = missing_set(remote, &plan.chunks)?; + send_chunks(remote, &file, &plan, &missing, path, limits, progress)?; + match remote.put_chunked(path, &plan.chunks, &plan.content_hash, base_seq, mtime_ms) { + Ok(reply) => Ok(reply), + Err(SyncError::MissingChunks(list)) => { + // The server lost a chunk on its way: send those again, try once more. + let gone: HashSet = list.into_iter().collect(); + send_chunks(remote, &file, &plan, &gone, path, limits, progress)?; + remote.put_chunked(path, &plan.chunks, &plan.content_hash, base_seq, mtime_ms) + } + Err(error) => Err(error), + } +} + +/// What the hash pass learned about a file. +struct Plan { + /// The hash of every chunk, in the order they make up the file. + chunks: Vec, + /// The hash of the whole content. + content_hash: String, + /// How many bytes the file has. + total: u64, +} + +/// Reads the file once and answers the hash of every chunk and of the whole +/// content. +fn plan_chunks(file: &Path, chunk: usize) -> SyncResult { + let mut handle = open(file)?; + let mut content = hasher(); + let mut chunks = Vec::new(); + let mut total = 0u64; + let mut buffer = vec![0u8; chunk]; + loop { + let read = read_upto(&mut handle, &mut buffer)?; + if read == 0 { + break; + } + let bytes = buffer.get(..read).ok_or_else(|| SyncError::Io("the chunk buffer is shorter than the bytes read".to_owned()))?; + content.update(bytes); + chunks.push(hash_hex(bytes)); + total = total + .checked_add(u64::try_from(read).map_err(|_| SyncError::Io("a chunk holds too many bytes".to_owned()))?) + .ok_or_else(|| SyncError::Io("the file is too long to count".to_owned()))?; + } + Ok(Plan { chunks, content_hash: content.finish_hex(), total }) +} + +/// The chunks of the list the server does not hold yet. +fn missing_set(remote: &dyn Remote, chunks: &[String]) -> SyncResult> { + let mut missing = HashSet::new(); + for batch in chunks.chunks(MAX_CHECK_BATCH) { + for hash in remote.missing_chunks(batch)? { + missing.insert(hash); + } + } + Ok(missing) +} + +/// Sends the chunks of `wanted` the server is missing and reports the progress. +/// +/// The file is read again: every chunk is checked against the hash the first +/// pass gave it, so a file that changed in between is refused before any of it +/// is sent. +fn send_chunks( + remote: &dyn Remote, + file: &Path, + plan: &Plan, + wanted: &HashSet, + path: &str, + limits: Limits, + progress: &mut dyn FnMut(&Progress), +) -> SyncResult<()> { + let mut handle = open(file)?; + let mut buffer = vec![0u8; limits.chunk]; + let mut sent: HashSet = HashSet::new(); + let mut done = 0u64; + for expected in &plan.chunks { + let read = read_upto(&mut handle, &mut buffer)?; + if read == 0 { + return Err(SyncError::BadRemote(format!("the file `{path}` shrank while it was uploaded"))); + } + let bytes = buffer.get(..read).ok_or_else(|| SyncError::Io("the chunk buffer is shorter than the bytes read".to_owned()))?; + let hash = hash_hex(bytes); + if hash != *expected { + return Err(SyncError::BadRemote(format!("the file `{path}` changed while it was uploaded"))); + } + if wanted.contains(&hash) && !sent.contains(&hash) { + put_with_retry(remote, &hash, bytes)?; + sent.insert(hash.clone()); + } + done = done.saturating_add(u64::try_from(read).map_err(|_| SyncError::Io("a chunk holds too many bytes".to_owned()))?); + progress(&Progress { path: path.to_owned(), done_bytes: done, total_bytes: plan.total }); + } + Ok(()) +} + +/// Sends one chunk, trying again with a growing wait before giving up. +fn put_with_retry(remote: &dyn Remote, hash: &str, bytes: &[u8]) -> SyncResult<()> { + let mut attempt = 0usize; + loop { + match remote.put_chunk(hash, bytes) { + Ok(()) => return Ok(()), + Err(error) => { + let Some(wait) = BACKOFF_MS.get(attempt).copied() else { return Err(error) }; + std::thread::sleep(Duration::from_millis(wait)); + attempt = attempt.saturating_add(1); + } + } + } +} + +/// Opens a file of the vault. +fn open(file: &Path) -> SyncResult { + File::open(file).map_err(|error| SyncError::Io(format!("{}: {error}", file.display()))) +} + +/// Reads into `buffer` until it is full or the file ends; at most a full buffer. +fn read_upto(file: &mut File, buffer: &mut [u8]) -> SyncResult { + let mut read = 0usize; + while read < buffer.len() { + let Some(rest) = buffer.get_mut(read..) else { break }; + match file.read(rest) { + Ok(0) => break, + Ok(count) => read = read.saturating_add(count), + Err(error) if error.kind() == std::io::ErrorKind::Interrupted => {} + Err(error) => return Err(SyncError::Io(error.to_string())), + } + } + Ok(read) +} diff --git a/crates/engine/src/sync/driver.rs b/crates/engine/src/sync/driver.rs index fe93189..a887583 100644 --- a/crates/engine/src/sync/driver.rs +++ b/crates/engine/src/sync/driver.rs @@ -5,6 +5,7 @@ use std::collections::BTreeMap; use notescraft_sync::{Action, Base, FileRecord, check_hash, check_syncable, hash_hex, plan}; use notescraft_vault::Vault; +use crate::sync::chunks::{self, Limits, Progress}; use crate::sync::{PutReply, Remote, SyncError, SyncResult, SyncState, reconcile, scan, state}; /// How many records one `/changes` page asks for. @@ -13,6 +14,9 @@ const PAGE_LIMIT: usize = 1000; const MAX_PAGES: usize = 10_000; /// Most names a conflict copy gets before the first one is used again. const MAX_COPY_TRIES: usize = 10; +/// The largest file that is fetched into memory; bigger ones are left for a +/// later day instead of being pulled into the memory of the app. +pub(crate) const MAX_DOWNLOAD_IN_MEMORY: u64 = 64 * 1024 * 1024; /// What one sync pass did. #[derive(Debug, Default, PartialEq, Eq)] @@ -29,10 +33,26 @@ pub struct SyncReport { pub merged: Vec, /// Paths of the conflict copies written. pub conflicts: Vec, + /// Paths left alone because the client cannot fetch them yet. + pub skipped: Vec, /// How many paths were already in step. pub adopted: usize, } +/// How one sync pass behaves. +pub struct SyncOptions { + /// When files upload in chunks and how big a chunk is. + pub limits: Limits, + /// Called after every chunk of a big file, with the bytes that moved. + pub progress: Box, +} + +impl Default for SyncOptions { + fn default() -> Self { + Self { limits: Limits::default(), progress: Box::new(|_: &Progress| {}) } + } +} + /// Syncs `vault` with `remote`, once. /// /// The pass reads the vault, reads the change log of the server, asks @@ -51,12 +71,40 @@ pub struct SyncReport { /// [`SyncError`] when the server cannot be reached, when it sends something /// unusable, or when the vault refuses a read or a write. pub fn sync_once(vault: &Vault, remote: &dyn Remote, state: &mut SyncState, device: &str, stamp: &str) -> SyncResult { + sync_once_with(vault, remote, state, device, stamp, SyncOptions::default()) +} + +/// [`sync_once`] with the options of the pass: the chunk limits and the +/// progress callback that watches a big file travel. +/// +/// # Errors +/// [`SyncError`] when the server cannot be reached, when it sends something +/// unusable, or when the vault refuses a read or a write. +pub fn sync_once_with(vault: &Vault, remote: &dyn Remote, state: &mut SyncState, device: &str, stamp: &str, options: SyncOptions) -> SyncResult { + sync_once_scanned(vault, remote, state, device, stamp, options, scan::ScanOptions::default()) +} + +/// [`sync_once_with`] also naming what the scan of the vault leaves out: with +/// `attachments: false` the pass sees only the Markdown notes. +/// +/// # Errors +/// [`SyncError`] when the server cannot be reached, when it sends something +/// unusable, or when the vault refuses a read or a write. +pub(crate) fn sync_once_scanned( + vault: &Vault, + remote: &dyn Remote, + state: &mut SyncState, + device: &str, + stamp: &str, + options: SyncOptions, + scan_options: scan::ScanOptions, +) -> SyncResult { let mut report = SyncReport::default(); - let local = scan::scan(vault)?; + let local = scan::scan(vault, scan_options)?; let hashes: BTreeMap = local.iter().map(|(path, file)| (path.clone(), file.hash.clone())).collect(); let (records, cursor) = fetch(remote, state.last_seq)?; let actions = plan(&hashes, &state.bases, &records); - let mut ctx = Ctx { vault, remote, state, report: &mut report, local: &local, device, stamp }; + let mut ctx = Ctx { vault, remote, state, report: &mut report, local: &local, device, stamp, options }; for action in actions { perform(&mut ctx, action)?; ctx.state.save(ctx.vault)?; @@ -76,12 +124,19 @@ struct Ctx<'a> { local: &'a BTreeMap, device: &'a str, stamp: &'a str, + options: SyncOptions, } /// Performs one planned action. fn perform(ctx: &mut Ctx<'_>, action: Action) -> SyncResult<()> { match action { Action::Download(record) => { + if record.size > MAX_DOWNLOAD_IN_MEMORY { + // Too big to fetch here yet: the file stays as it is, without a + // conflict and without a half-written copy. + ctx.report.skipped.push(record.path.clone()); + return Ok(()); + } let bytes = fetch_blob(ctx.remote, &record)?; ctx.vault.write(&record.path, &bytes).map_err(|e| vault_error(&e))?; ctx.state.bases.insert(record.path.clone(), Base { hash: record.hash.clone(), seq: record.seq }); @@ -94,19 +149,11 @@ fn perform(ctx: &mut Ctx<'_>, action: Action) -> SyncResult<()> { ctx.report.deleted_local.push(record.path.clone()); } Action::Upload { path, base_seq } => { - let bytes = ctx.vault.read(&path).map_err(|e| vault_error(&e))?; - let hash = hash_hex(&bytes); - let mtime_ms = ctx.local.get(&path).map_or(0, |file| file.mtime_ms); - match ctx.remote.put(&path, &bytes, base_seq, mtime_ms)? { - PutReply::Applied(reply) | PutReply::Unchanged(reply) => { - ctx.state.bases.insert(path.clone(), Base { hash, seq: reply.seq }); - ctx.report.uploaded.push(path.clone()); - state::store_base_copy(ctx.vault, &path, &bytes)?; - } - PutReply::Conflict(current) => { - let base_hash = ctx.state.bases.get(&path).map(|base| base.hash.clone()); - reconcile_path(ctx, ¤t, &bytes, base_hash.as_deref())?; - } + let (size, mtime_ms) = ctx.local.get(&path).map_or((0, 0), |file| (file.size, file.mtime_ms)); + if size > ctx.options.limits.threshold { + upload_big(ctx, &path, base_seq, mtime_ms)?; + } else { + upload_small(ctx, &path, base_seq, mtime_ms)?; } } Action::DeleteRemote { path, base_seq } => match ctx.remote.delete(&path, base_seq)? { @@ -143,6 +190,44 @@ fn perform(ctx: &mut Ctx<'_>, action: Action) -> SyncResult<()> { Ok(()) } +/// Uploads a small file in one piece. +fn upload_small(ctx: &mut Ctx<'_>, path: &str, base_seq: u64, mtime_ms: u64) -> SyncResult<()> { + let bytes = ctx.vault.read(path).map_err(|e| vault_error(&e))?; + let hash = hash_hex(&bytes); + match ctx.remote.put(path, &bytes, base_seq, mtime_ms)? { + PutReply::Applied(reply) | PutReply::Unchanged(reply) => { + ctx.state.bases.insert(path.to_owned(), Base { hash, seq: reply.seq }); + ctx.report.uploaded.push(path.to_owned()); + state::store_base_copy(ctx.vault, path, &bytes)?; + Ok(()) + } + PutReply::Conflict(current) => { + let base_hash = ctx.state.bases.get(path).map(|base| base.hash.clone()); + reconcile_path(ctx, ¤t, &bytes, base_hash.as_deref()) + } + } +} + +/// Uploads a big file in chunks, so a cut connection only loses one chunk. +fn upload_big(ctx: &mut Ctx<'_>, path: &str, base_seq: u64, mtime_ms: u64) -> SyncResult<()> { + let limits = ctx.options.limits; + let progress: &mut dyn FnMut(&Progress) = &mut ctx.options.progress; + let reply = chunks::upload_chunked(ctx.remote, ctx.vault, path, base_seq, mtime_ms, limits, progress)?; + match reply { + PutReply::Applied(record) | PutReply::Unchanged(record) => { + ctx.state.bases.insert(path.to_owned(), Base { hash: record.hash.clone(), seq: record.seq }); + ctx.report.uploaded.push(path.to_owned()); + Ok(()) + } + PutReply::Conflict(current) => { + // The other side wrote first: the content comes back for the merge. + let bytes = ctx.vault.read(path).map_err(|e| vault_error(&e))?; + let base_hash = ctx.state.bases.get(path).map(|base| base.hash.clone()); + reconcile_path(ctx, ¤t, &bytes, base_hash.as_deref()) + } + } +} + /// Settles a file both sides changed. fn reconcile_path(ctx: &mut Ctx<'_>, record: &FileRecord, ours: &[u8], base_hash: Option<&str>) -> SyncResult<()> { let theirs = fetch_blob(ctx.remote, record)?; diff --git a/crates/engine/src/sync/http.rs b/crates/engine/src/sync/http.rs index 61960bc..102f3f9 100644 --- a/crates/engine/src/sync/http.rs +++ b/crates/engine/src/sync/http.rs @@ -8,14 +8,15 @@ use std::time::Duration; use notescraft_sync::{Changes, FileRecord}; use serde::{Deserialize, Serialize}; +use crate::sync::driver::MAX_DOWNLOAD_IN_MEMORY; use crate::sync::{PutReply, Remote, SyncError, SyncResult}; /// Seconds a request may take when the caller does not ask for a long poll. const DEFAULT_TIMEOUT_SECS: u64 = 60; /// Seconds added to the long poll, to cover the round trip. const POLL_MARGIN_SECS: u64 = 15; -/// Largest body the client reads. -const MAX_BODY_BYTES: u64 = 25 * 1024 * 1024; +/// Largest body the client reads: the biggest file it fetches. +const MAX_BODY_BYTES: u64 = MAX_DOWNLOAD_IN_MEMORY; /// Longest body kept in an error message. const MAX_ERROR_BODY: usize = 200; /// The hexadecimal digits of a percent escape. @@ -121,6 +122,53 @@ impl Remote for HttpRemote { other => Err(status_error(other, &body)), } } + + fn missing_chunks(&self, hashes: &[String]) -> SyncResult> { + let url = format!("{}/chunks/check", self.vault_url()); + let body = serde_json::json!({ "hashes": hashes }); + let response = self.agent.post(url.as_str()).header("Authorization", self.auth()).send_json(body).map_err(|e| SyncError::Http(e.to_string()))?; + let (status, body) = read(response)?; + if status != 200 { + return Err(status_error(status, &body)); + } + let missing: MissingChunks = serde_json::from_slice(&body).map_err(|e| SyncError::BadRemote(format!("the chunk list is not readable: {e}")))?; + Ok(missing.missing) + } + + fn put_chunk(&self, hash: &str, bytes: &[u8]) -> SyncResult<()> { + let url = format!("{}/chunks/{}", self.vault_url(), encode_segment(hash)); + let response = self.agent.put(url.as_str()).header("Authorization", self.auth()).send(bytes).map_err(|e| SyncError::Http(e.to_string()))?; + let (status, body) = read(response)?; + // 201 when the chunk was stored, 200 when the server already had it. + if status == 200 || status == 201 { + return Ok(()); + } + Err(status_error(status, &body)) + } + + fn put_chunked(&self, path: &str, chunk_hashes: &[String], content_hash: &str, base_seq: u64, mtime_ms: u64) -> SyncResult { + let url = self.file_url(path); + let response = self + .agent + .put(url.as_str()) + .header("Authorization", self.auth()) + .header("X-Chunks", chunk_hashes.join(",")) + .header("X-Content-Hash", content_hash) + .header("X-Base-Seq", base_seq.to_string()) + .header("X-Mtime-Ms", mtime_ms.to_string()) + .send_empty() + .map_err(|e| SyncError::Http(e.to_string()))?; + let (status, body) = read(response)?; + match status { + 201 => Ok(PutReply::Applied(parse_record(&body)?)), + 200 => Ok(PutReply::Unchanged(parse_record(&body)?)), + 409 => match parse_missing(&body) { + Some(missing) => Err(SyncError::MissingChunks(missing)), + None => Ok(PutReply::Conflict(parse_current(&body)?)), + }, + other => Err(status_error(other, &body)), + } + } } /// The device a pairing code buys. @@ -181,6 +229,19 @@ fn parse_current(body: &[u8]) -> SyncResult { Ok(current.current) } +/// Reads the `{"missing": [...]}` body of a chunked write whose chunks are not +/// all there; `None` when the body is a conflict about the revision instead. +fn parse_missing(body: &[u8]) -> Option> { + serde_json::from_slice::(body).ok().map(|missing| missing.missing) +} + +/// The `{"missing": [...]}` answer of the chunk check. +#[derive(Deserialize)] +struct MissingChunks { + /// The hashes the server does not hold. + missing: Vec, +} + /// Turns an unexpected status into an error. fn status_error(status: u16, body: &[u8]) -> SyncError { let text: String = String::from_utf8_lossy(body).chars().take(MAX_ERROR_BODY).collect(); diff --git a/crates/engine/src/sync/mod.rs b/crates/engine/src/sync/mod.rs index 4d63c03..abff503 100644 --- a/crates/engine/src/sync/mod.rs +++ b/crates/engine/src/sync/mod.rs @@ -12,6 +12,7 @@ //! file cannot be merged, both are kept. mod binding; +mod chunks; pub(crate) mod commands; mod conflicts; mod driver; @@ -23,18 +24,22 @@ mod state; mod worker; pub use binding::{Binding, clear_binding, load_binding, load_token, save_binding, save_token}; +pub use chunks::{CHUNK_SIZE, CHUNK_THRESHOLD, Limits, Progress, upload_chunked}; pub use conflicts::{ConflictPair, Keep, list_conflicts, resolve as resolve_conflict}; -pub use driver::{SyncReport, sync_once}; +pub use driver::{SyncOptions, SyncReport, sync_once, sync_once_with}; pub use http::{HttpRemote, Pairing, pair}; pub use remote::{PutReply, Remote}; pub use state::SyncState; -pub use worker::{ReportCounts, SyncPhase, SyncStatus, SyncWorker, Timings}; +pub use worker::{ReportCounts, SyncPhase, SyncStatus, SyncWorker, Timings, Transfer}; /// The HTTP helpers the share client shares with the sync transport. pub(crate) use http::{agent, encode_segment}; pub use notescraft_sync::{Changes, FileRecord}; +/// The scan options of the worker's pass; the session sets them from the user's options. +pub(crate) use scan::ScanOptions; + use thiserror::Error as ThisError; /// Everything that can go wrong while syncing. @@ -59,6 +64,10 @@ pub enum SyncError { /// vault, a broken hash or bytes that do not match the hash they came with. #[error("the sync server sent something unusable: {0}")] BadRemote(String), + /// The server does not hold every chunk of a file: the hashes it is + /// missing come with the answer, so the client can send them again. + #[error("the sync server is missing {} chunk(s) of the file", .0.len())] + MissingChunks(Vec), /// The sync state could not be read or written. #[error("sync state: {0}")] State(String), diff --git a/crates/engine/src/sync/remote.rs b/crates/engine/src/sync/remote.rs index db9cca9..d2ac2f5 100644 --- a/crates/engine/src/sync/remote.rs +++ b/crates/engine/src/sync/remote.rs @@ -2,7 +2,7 @@ use notescraft_sync::{Changes, FileRecord}; -use crate::sync::SyncResult; +use crate::sync::{SyncError, SyncResult}; /// What the server did with a write. #[derive(Debug, Clone, PartialEq, Eq)] @@ -51,4 +51,41 @@ pub trait Remote { /// [`SyncError::Http`] when the server cannot be reached and /// [`SyncError::Status`] when it refuses the delete. fn delete(&self, path: &str, base_seq: u64) -> SyncResult; + + /// The chunks of `hashes` the server does not hold yet. + /// + /// The answer holds only the hashes that are missing, so a client that was + /// cut off sends the rest and nothing else. + /// + /// # Errors + /// [`SyncError::Http`] when the server cannot be reached and + /// [`SyncError::BadRemote`] when the server does not know chunks. + fn missing_chunks(&self, _hashes: &[String]) -> SyncResult> { + Err(SyncError::BadRemote("chunks unsupported".to_owned())) + } + + /// Stores one chunk under its own hash. + /// + /// The server keeps the chunk by content, so sending it twice costs + /// nothing. + /// + /// # Errors + /// [`SyncError::Http`] when the server cannot be reached and + /// [`SyncError::BadRemote`] when the server does not know chunks. + fn put_chunk(&self, _hash: &str, _bytes: &[u8]) -> SyncResult<()> { + Err(SyncError::BadRemote("chunks unsupported".to_owned())) + } + + /// Writes a file whose content the server already holds as chunks. + /// + /// `chunk_hashes` are the hashes of the chunks in the order they make up + /// the file, `content_hash` the hash of the whole content. + /// + /// # Errors + /// [`SyncError::Http`] when the server cannot be reached, + /// [`SyncError::MissingChunks`] when the server is missing a chunk of the + /// list, and [`SyncError::BadRemote`] when the server does not know chunks. + fn put_chunked(&self, _path: &str, _chunk_hashes: &[String], _content_hash: &str, _base_seq: u64, _mtime_ms: u64) -> SyncResult { + Err(SyncError::BadRemote("chunks unsupported".to_owned())) + } } diff --git a/crates/engine/src/sync/scan.rs b/crates/engine/src/sync/scan.rs index 09d0847..b1eb81e 100644 --- a/crates/engine/src/sync/scan.rs +++ b/crates/engine/src/sync/scan.rs @@ -1,14 +1,18 @@ //! The disk side of a sync: the files the vault holds right now. use std::collections::BTreeMap; +use std::fs::File; +use std::io::Read; -use notescraft_sync::{check_syncable, hash_hex}; +use notescraft_sync::{check_syncable, hasher}; use notescraft_vault::Vault; use crate::sync::{SyncError, SyncResult}; /// Files above this size are left alone; the server refuses them too. -const MAX_FILE_BYTES: u64 = 25 * 1024 * 1024; +const MAX_FILE_BYTES: u64 = 1024 * 1024 * 1024; +/// How many bytes are read at a time while a file is hashed. +const READ_BUFFER: usize = 64 * 1024; /// One file of the vault. pub(crate) struct LocalFile { @@ -16,6 +20,22 @@ pub(crate) struct LocalFile { pub hash: String, /// Last modification in Unix milliseconds. pub mtime_ms: u64, + /// How many bytes the file has. + pub size: u64, +} + +/// What a scan leaves out of the vault picture. +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub(crate) struct ScanOptions { + /// Whether files that are not Markdown travel; the sync sheet calls them + /// the attachments (`SyncSettings.attachments`). + pub attachments: bool, +} + +impl Default for ScanOptions { + fn default() -> Self { + Self { attachments: true } + } } /// Reads every syncable file of `vault` and hashes it. @@ -24,21 +44,79 @@ pub(crate) struct LocalFile { /// `.git`) are left out, and so are the ones above [`MAX_FILE_BYTES`]. A file /// that vanishes while the vault is being read is left out too. /// +/// The content of a file is never held in memory: it is hashed as it is read, +/// so a file of any size costs the same memory. +/// +/// An `attachments` of `false` leaves out every file that is not Markdown. +/// /// # Errors /// [`SyncError::Vault`] when a file cannot be read. -pub(crate) fn scan(vault: &Vault) -> SyncResult> { +pub(crate) fn scan(vault: &Vault, options: ScanOptions) -> SyncResult> { let entries = vault.list().map_err(|e| SyncError::Vault(e.to_string()))?; let mut files = BTreeMap::new(); for entry in entries { - if entry.is_dir || entry.size > MAX_FILE_BYTES || check_syncable(&entry.path).is_err() { + if entry.is_dir || entry.size > MAX_FILE_BYTES || check_syncable(&entry.path).is_err() || (!is_markdown(&entry.path) && !options.attachments) { continue; } - let bytes = match vault.read(&entry.path) { - Ok(bytes) => bytes, - Err(notescraft_vault::Error::Io { source, .. }) if source.kind() == std::io::ErrorKind::NotFound => continue, - Err(error) => return Err(SyncError::Vault(error.to_string())), + let full = vault.resolve(&entry.path).map_err(|e| SyncError::Vault(e.to_string()))?; + let hash = match hash_file(&full) { + Ok(hash) => hash, + Err(error) if error.kind() == std::io::ErrorKind::NotFound => continue, + Err(error) => return Err(SyncError::Io(format!("{}: {error}", entry.path))), }; - files.insert(entry.path.clone(), LocalFile { hash: hash_hex(&bytes), mtime_ms: entry.modified.unwrap_or(0) }); + files.insert(entry.path.clone(), LocalFile { hash, mtime_ms: entry.modified.unwrap_or(0), size: entry.size }); } Ok(files) } + +/// Whether the path ends in `.md`, read case-insensitively; a file without an +/// extension is not Markdown. +fn is_markdown(path: &str) -> bool { + std::path::Path::new(path).extension().is_some_and(|extension| extension.eq_ignore_ascii_case("md")) +} + +/// The SHA-256 hex of the content of a file, read in pieces. +fn hash_file(file: &std::path::Path) -> std::result::Result { + let mut handle = File::open(file)?; + let mut hash = hasher(); + let mut buffer = vec![0u8; READ_BUFFER]; + loop { + let read = handle.read(&mut buffer)?; + if read == 0 { + break; + } + let bytes = buffer.get(..read).ok_or_else(|| std::io::Error::other("the read buffer is shorter than the bytes read"))?; + hash.update(bytes); + } + Ok(hash.finish_hex()) +} + +#[cfg(test)] +mod tests { + use super::*; + + use notescraft_vault::Vault; + + /// A vault with one note and one attachment, under a fresh temp folder. + fn seeded(name: &str) -> Vault { + let root = std::env::temp_dir().join(format!("notescraft-scan-{}-{name}", std::process::id())); + let _ = std::fs::remove_dir_all(&root); + std::fs::create_dir_all(&root).expect("create the temp vault"); + let vault = Vault::open(&root).expect("open the temp vault"); + vault.write("notes/idea.md", b"text").expect("write the note"); + vault.write("notes/diagram.png", b"bytes").expect("write the attachment"); + vault + } + + #[test] + fn attachments_on_or_off_changes_what_the_scan_sees() { + let vault = seeded("on-off"); + let note_only = scan(&vault, ScanOptions { attachments: false }).expect("scan without attachments"); + assert!(!note_only.contains_key("notes/diagram.png"), "the png is left out"); + assert_eq!(note_only.len(), 1, "the note stays"); + let with_all = scan(&vault, ScanOptions { attachments: true }).expect("scan with attachments"); + assert_eq!(with_all.len(), 2, "the note and the png both travel"); + let defaulted = scan(&vault, ScanOptions::default()).expect("scan with the defaults"); + assert_eq!(defaulted.len(), 2, "the default scan keeps the attachments"); + } +} diff --git a/crates/engine/src/sync/worker.rs b/crates/engine/src/sync/worker.rs index 5f48b7c..da09eab 100644 --- a/crates/engine/src/sync/worker.rs +++ b/crates/engine/src/sync/worker.rs @@ -21,7 +21,8 @@ use notescraft_templates::Date; use notescraft_vault::Vault; use crate::sync::binding::{load_binding, load_token}; -use crate::sync::driver::{SyncReport, sync_once}; +use crate::sync::chunks::{Limits, Progress}; +use crate::sync::driver::{SyncOptions, SyncReport, sync_once_scanned}; use crate::sync::http::HttpRemote; use crate::sync::remote::Remote; use crate::sync::scan; @@ -106,6 +107,17 @@ pub struct ReportCounts { pub conflicts: usize, } +/// The file that is moving right now. +#[derive(Debug, Clone, PartialEq, Eq, serde::Serialize)] +pub struct Transfer { + /// The vault path of the file. + pub path: String, + /// The bytes of the file the server holds now. + pub done_bytes: u64, + /// The bytes the file has. + pub total_bytes: u64, +} + /// What a worker knows: a cheap clone of its shared status. #[derive(Debug, Clone, Default, PartialEq, Eq)] pub struct SyncStatus { @@ -130,13 +142,26 @@ pub struct SyncStatus { pub last_report: Option, /// The conflict copies the last cycle wrote. pub conflicts: Vec, + /// The big file that is moving right now, while one does. + pub transfer: Option, +} + +/// What the worker thread and the readers of its status share. +#[derive(Debug)] +struct Shared { + /// The status every reader sees. + status: Mutex, + /// The waits the loop uses. + timings: RwLock, + /// What the scan of every cycle leaves out (the sync options); the app pushes + /// the user's toggles here. + scan_options: RwLock, } /// A background thread syncing one vault. #[derive(Debug)] pub struct SyncWorker { - status: Arc>, - timings: Arc>, + shared: Arc, thread: Option>, tx: Option>, } @@ -153,7 +178,7 @@ impl SyncWorker { let (server, vault_id) = (binding.server, binding.vault_id); let worker = Self::start_with(vault, device, Box::new(HttpRemote::new(&server, &vault_id, &token)), clock_millis); { - let mut status = lock_status(&worker.status); + let mut status = lock_status(&worker.shared.status); status.server = Some(server); status.vault_id = Some(vault_id); } @@ -164,18 +189,21 @@ impl SyncWorker { #[must_use] pub fn start_with(vault: &Vault, device: &str, remote: Box, clock: fn() -> u64) -> Self { let (tx, rx) = channel(); - let status = Arc::new(Mutex::new(SyncStatus { bound: true, ..SyncStatus::default() })); - let timings = Arc::new(RwLock::new(Timings::default())); + let shared = Arc::new(Shared { + status: Mutex::new(SyncStatus { bound: true, ..SyncStatus::default() }), + timings: RwLock::new(Timings::default()), + scan_options: RwLock::new(scan::ScanOptions::default()), + }); let vault = vault.sibling(); let device = device.to_owned(); - let (thread_status, thread_timings) = (Arc::clone(&status), Arc::clone(&timings)); + let thread_shared = Arc::clone(&shared); let thread = std::thread::Builder::new() .name("notescraft-sync".to_owned()) - .spawn(move || run(&vault, &device, remote.as_ref(), clock, &thread_status, &thread_timings, &rx)) + .spawn(move || run(&vault, &device, remote.as_ref(), clock, &thread_shared, &rx)) .ok(); - let worker = Self { status, timings, thread, tx: Some(tx) }; + let worker = Self { shared, thread, tx: Some(tx) }; if worker.thread.is_none() { - set_status(&worker.status, |status| { + set_status(&worker.shared.status, |status| { status.phase = SyncPhase::Error; status.last_error = Some("the sync thread could not start".to_owned()); }); @@ -185,11 +213,19 @@ impl SyncWorker { /// Replaces the waits; they take effect on the next loop wake-up. Tests shorten them. pub fn set_timings(&self, timings: Timings) { - if let Ok(mut guard) = self.timings.write() { + if let Ok(mut guard) = self.shared.timings.write() { *guard = timings; } } + /// Replaces what the scan of the cycles leaves out, the way the engine + /// applies the user's sync options; it takes effect on the next cycle. + pub(crate) fn set_scan_options(&self, options: scan::ScanOptions) { + if let Ok(mut guard) = self.shared.scan_options.write() { + *guard = options; + } + } + /// Wakes the worker up. Requests of the same run coalesce: the thread waits for the /// debounce after the last one. pub fn request_sync(&self) { @@ -201,7 +237,7 @@ impl SyncWorker { /// The status as it stands. #[must_use] pub fn status(&self) -> SyncStatus { - lock_status(&self.status).clone() + lock_status(&self.shared.status).clone() } /// Signals and joins the thread, giving it [`STOP_CAP`]; the same work [`Drop`] does. @@ -241,7 +277,7 @@ enum Fate { /// The loop of the worker thread. It owns a vault sibling and a receiver; everything else /// it borrows. -fn run(vault: &Vault, device: &str, remote: &dyn Remote, clock: fn() -> u64, status: &Mutex, timings: &RwLock, rx: &Receiver<()>) { +fn run(vault: &Vault, device: &str, remote: &dyn Remote, clock: fn() -> u64, shared: &Arc, rx: &Receiver<()>) { // The start counts as one request, so the first cycle runs right after the debounce. let mut last_request = Some(Instant::now()); let mut last_cycle: Option = None; @@ -250,7 +286,7 @@ fn run(vault: &Vault, device: &str, remote: &dyn Remote, clock: fn() -> u64, sta let mut wait_for_request = false; let mut cycles: u64 = 0; loop { - let Some(due) = next_due(timings, last_request, last_cycle, failed_at, retries, wait_for_request) else { + let Some(due) = next_due(&shared.timings, last_request, last_cycle, failed_at, retries, wait_for_request) else { // Sign-in-again: nothing times out, only a request or the end of the worker. match rx.recv() { Ok(()) => { @@ -278,7 +314,7 @@ fn run(vault: &Vault, device: &str, remote: &dyn Remote, clock: fn() -> u64, sta last_request = None; last_cycle = Some(now); cycles = cycles.wrapping_add(1); - let fate = run_cycle(vault, device, remote, clock, status, cycles); + let fate = run_cycle(vault, device, remote, clock, shared, cycles); match fate { Fate::Success | Fate::Fail => { failed_at = None; @@ -325,18 +361,18 @@ fn next_due( /// Runs one cycle and reports what the loop should do next. `cycle` is the number of /// cycles the thread has run, carried in the status file. -fn run_cycle(vault: &Vault, device: &str, remote: &dyn Remote, clock: fn() -> u64, status: &Mutex, cycle: u64) -> Fate { - set_status(status, |status| status.phase = SyncPhase::Syncing); +fn run_cycle(vault: &Vault, device: &str, remote: &dyn Remote, clock: fn() -> u64, shared: &Arc, cycle: u64) -> Fate { + set_status(&shared.status, |status| status.phase = SyncPhase::Syncing); match SyncState::load(vault) { - Ok(mut state) => one_pass(vault, device, remote, clock, status, &mut state, cycle), + Ok(mut state) => one_pass(vault, device, remote, clock, shared, &mut state, cycle), Err(error) => { - set_status(status, |status| { + set_status(&shared.status, |status| { status.phase = SyncPhase::Error; status.last_error = Some(short(&error.to_string())); status.last_report = None; status.pending = 0; }); - write_status_file(vault, &fresh_status(status), cycle); + write_status_file(vault, &fresh_status(&shared.status), cycle); Fate::Fail } } @@ -344,9 +380,10 @@ fn run_cycle(vault: &Vault, device: &str, remote: &dyn Remote, clock: fn() -> u6 /// The sync run of one cycle. The one status update of the cycle carries everything the /// run decided, so a reader never sees a half-updated status. -fn one_pass(vault: &Vault, device: &str, remote: &dyn Remote, clock: fn() -> u64, status: &Mutex, state: &mut SyncState, cycle: u64) -> Fate { - let run = sync_once(vault, remote, state, device, &stamp_from_millis(clock())); - let pending = pending_count(vault, state); +fn one_pass(vault: &Vault, device: &str, remote: &dyn Remote, clock: fn() -> u64, shared: &Arc, state: &mut SyncState, cycle: u64) -> Fate { + let scan = *shared.scan_options.read().unwrap_or_else(PoisonError::into_inner); + let run = sync_once_scanned(vault, remote, state, device, &stamp_from_millis(clock()), options_for(shared, vault, cycle), scan); + let pending = pending_count(vault, state, scan); let seq = state.last_seq; let (fate, phase, error, counts, conflicts, last_sync): (Fate, SyncPhase, Option, Option, Vec, Option) = match run { Ok(report) => (Fate::Success, SyncPhase::Idle, None, Some(counts_of(&report)), report.conflicts.clone(), Some(clock())), @@ -356,7 +393,7 @@ fn one_pass(vault: &Vault, device: &str, remote: &dyn Remote, clock: fn() -> u64 } Err(other) => (Fate::Fail, SyncPhase::Error, Some(short(&other.to_string())), None, Vec::new(), None), }; - set_status(status, |status| { + set_status(&shared.status, |status| { status.last_seq = seq; status.pending = pending; status.phase = phase; @@ -364,15 +401,33 @@ fn one_pass(vault: &Vault, device: &str, remote: &dyn Remote, clock: fn() -> u64 status.last_report = counts; status.conflicts = conflicts; status.last_sync_ms = last_sync; + status.transfer = None; }); - write_status_file(vault, &fresh_status(status), cycle); + write_status_file(vault, &fresh_status(&shared.status), cycle); fate } +/// The options of one pass: the default chunk limits and a callback that shows +/// the big file that is moving. +fn options_for(shared: &Arc, vault: &Vault, cycle: u64) -> SyncOptions { + let watching = Arc::clone(shared); + let file_vault = vault.sibling(); + SyncOptions { + limits: Limits::default(), + progress: Box::new(move |progress: &Progress| { + set_status(&watching.status, |status| { + status.transfer = Some(Transfer { path: progress.path.clone(), done_bytes: progress.done_bytes, total_bytes: progress.total_bytes }); + }); + write_status_file(&file_vault, &fresh_status(&watching.status), cycle); + }), + } +} + /// The local files whose content differs from the synced base of their path; a scan that /// fails leaves the count unknown, which reads as nothing waiting. -fn pending_count(vault: &Vault, state: &SyncState) -> usize { - scan::scan(vault).map_or(0, |local| local.iter().filter(|(path, file)| state.bases.get(*path).is_none_or(|base| base.hash != file.hash)).count()) +fn pending_count(vault: &Vault, state: &SyncState, scan_options: scan::ScanOptions) -> usize { + scan::scan(vault, scan_options) + .map_or(0, |local| local.iter().filter(|(path, file)| state.bases.get(*path).is_none_or(|base| base.hash != file.hash)).count()) } /// The counts of a report. @@ -446,6 +501,8 @@ struct StatusFile { last_error: Option, pending: usize, conflicts: Vec, + #[serde(skip_serializing_if = "Option::is_none")] + transfer: Option, } /// Writes the status file of the last cycle so other processes can show it. A vault that @@ -458,6 +515,7 @@ fn write_status_file(vault: &Vault, status: &SyncStatus, cycle: u64) { last_error: status.last_error.clone(), pending: status.pending, conflicts: status.conflicts.clone(), + transfer: status.transfer.clone(), }; if let Ok(bytes) = serde_json::to_vec(&file) { let _ = vault.write_private(STATUS_FILE, &bytes); diff --git a/crates/engine/src/vault/session.rs b/crates/engine/src/vault/session.rs index 9eb80b5..b837bea 100644 --- a/crates/engine/src/vault/session.rs +++ b/crates/engine/src/vault/session.rs @@ -63,6 +63,9 @@ pub struct Session { pub(crate) pending: Vec, /// The highest pending-change id handed out so far. pub(crate) next_pending_id: u64, + /// Whether the sync scan keeps the non-Markdown files (the applied user + /// option; the next started worker reads it too). + pub(crate) sync_attachments: bool, /// The notes still to read after a lazy open. pub(crate) scan: crate::indexing::ScanState, /// The format/lint batch: its queue and the proposals read so far. @@ -140,10 +143,18 @@ impl Session { pub fn start_sync(&mut self, device_name: &str) -> crate::sync::SyncResult<()> { self.stop_sync(); let worker = crate::sync::SyncWorker::start(&self.vault, device_name)?; + worker.set_scan_options(self.sync_scan_options()); self.sync = Some(worker); Ok(()) } + /// The options the scan of the sync worker uses (the user's sync options; + /// the session remembers the last applied value for the next start). + #[must_use] + fn sync_scan_options(&self) -> crate::sync::ScanOptions { + crate::sync::ScanOptions { attachments: self.sync_attachments } + } + /// Stops the sync worker, if one runs; the notes are the worker's last known state. pub fn stop_sync(&mut self) { self.sync = None; @@ -209,6 +220,12 @@ impl Session { self.set_config(app.session.clone())?; self.plugin_settings = app.plugins.clone(); self.assistant = app.assistant.clone(); + // The sync worker scans with the user's options; a worker started later + // (the next open, the next pairing) reads the remembered value. + self.sync_attachments = app.sync.attachments; + if let Some(worker) = self.sync.as_ref() { + worker.set_scan_options(self.sync_scan_options()); + } Ok(()) } @@ -373,6 +390,7 @@ impl Session { batch: crate::batch::BatchState::default(), grammar_transport: Box::new(notescraft_syntax::fetch::Curl), sync: None, + sync_attachments: true, }; session.begin_scan(); session diff --git a/crates/engine/tests/core_commands.rs b/crates/engine/tests/core_commands.rs index a746ee5..5fe04cf 100644 --- a/crates/engine/tests/core_commands.rs +++ b/crates/engine/tests/core_commands.rs @@ -207,6 +207,7 @@ const CORE_IDS: &[&str] = &[ "sync.resolve", // The server account, its vaults and the sharing surface (SPEC 19, spec 5-7). "server.vaults", + "server.devices", "server.create_vault", "server.rename_vault", "server.delete_vault", diff --git a/crates/engine/tests/share_devices.rs b/crates/engine/tests/share_devices.rs new file mode 100644 index 0000000..45f38f1 --- /dev/null +++ b/crates/engine/tests/share_devices.rs @@ -0,0 +1,217 @@ +//! The `server.devices` command against a stub server: the request it makes, +//! the rows it lists and the secrets it never carries. +//! +//! The stub is a plain TCP listener that answers one canned JSON and remembers +//! the request line, like the stub in `share_commands.rs`, kept smaller here +//! because the command needs one route. + +mod common; + +use std::io::{BufRead, BufReader, Write}; +use std::net::{TcpListener, TcpStream}; +use std::sync::atomic::{AtomicBool, Ordering}; +use std::sync::{Arc, Mutex}; +use std::time::Duration; + +use common::temp_vault; +use notescraft_engine::{Error, Registry, Session, bind_folder, register_core}; +use serde_json::{Value, json}; + +/// The token of the bound vault; it must never reach an answer or an error. +const TOKEN: &str = "secret-token"; + +/// One request the stub received. +#[derive(Debug, Clone)] +struct Request { + /// The HTTP method. + method: String, + /// The path of the request line. + path: String, + /// The `Authorization` header, when there was one. + auth: Option, +} + +/// A stub server: one canned answer, the requests it saw. +struct Stub { + /// The base URL of the server. + base: String, + /// Every request the stub received. + seen: Arc>>, + /// Whether the listener thread should stop. + stop: Arc, +} + +impl Stub { + /// Starts a stub on an ephemeral port that answers one route. + fn spawn(status: u16, body: &str) -> Self { + let listener = TcpListener::bind("127.0.0.1:0").expect("bind the stub"); + let port = listener.local_addr().expect("addr").port(); + let seen: Arc>> = Arc::new(Mutex::new(Vec::new())); + let stop = Arc::new(AtomicBool::new(false)); + let head = format!("HTTP/1.1 {status} OK\r\nContent-Type: application/json\r\nContent-Length: {}\r\n\r\n{body}", body.len()); + let thread_seen = Arc::clone(&seen); + let thread_stop = Arc::clone(&stop); + std::thread::spawn(move || { + while !thread_stop.load(Ordering::Relaxed) { + match listener.accept() { + Ok((stream, _)) => { + let Ok(mut writer) = stream.try_clone() else { continue }; + let mut reader = BufReader::new(stream); + if let Some(request) = read_request(&mut reader) { + thread_seen.lock().expect("seen").push(request); + let _ = writer.write_all(head.as_bytes()); + let _ = writer.flush(); + } + } + Err(_) => std::thread::sleep(Duration::from_millis(2)), + } + } + }); + Self { base: format!("http://127.0.0.1:{port}"), seen, stop } + } + + /// Every request the stub received. + fn requests(&self) -> Vec { + self.seen.lock().expect("seen").clone() + } +} + +impl Drop for Stub { + fn drop(&mut self) { + self.stop.store(true, Ordering::Relaxed); + } +} + +/// Reads one request: method, path, the authorization header; no body is expected. +fn read_request(reader: &mut BufReader) -> Option { + let mut line = String::new(); + reader.read_line(&mut line).ok()?; + let mut parts = line.split_whitespace(); + let method = parts.next()?.to_owned(); + let path = parts.next()?.to_owned(); + let mut auth = None; + loop { + let mut header = String::new(); + if reader.read_line(&mut header).ok()? == 0 { + break; + } + let trimmed = header.trim_end(); + if trimmed.is_empty() { + break; + } + if let Some((name, value)) = trimmed.split_once(':') + && name.trim().eq_ignore_ascii_case("authorization") + { + auth = Some(value.trim().to_owned()); + } + } + Some(Request { method, path, auth }) +} + +/// A vault bound to `stub`, with the registry, the session and the vault guard. +fn bound(stub: &Stub, name: &str) -> (Registry, Session, common::TempVault) { + let vault = temp_vault(name); + bind_folder(vault.path(), &stub.base, "v-1", TOKEN, "d-1", Some("u-1")).expect("bind the vault"); + let mut reg = Registry::new(); + register_core(&mut reg).expect("register the core commands"); + let session = Session::open(vault.path()).expect("open the vault"); + (reg, session, vault) +} + +/// The code of a command error. +fn code(result: &Result) -> String { + match result { + Err(Error::Command { source, .. }) => source.code.clone(), + other => panic!("expected a command error, got {other:?}"), + } +} + +/// The message of a command error. +fn message(result: &Result) -> String { + match result { + Err(Error::Command { source, .. }) => source.message.clone(), + other => panic!("expected a command error, got {other:?}"), + } +} + +/// `server.devices` asks the server with the token and lists its rows. +#[test] +fn the_device_list_goes_to_the_server_with_the_token() { + let stub = Stub::spawn( + 200, + r#"[{"id":"d-1","name":"Pixel 9","created_ms":100,"last_seen_ms":null,"current":true},{"id":"d-2","name":"Laptop","created_ms":50,"last_seen_ms":60,"current":false}]"#, + ); + let (reg, mut session, _vault) = bound(&stub, "devices"); + + let answer = reg.run("server.devices", &mut session, &Value::Null).expect("server.devices"); + let devices = answer.get("devices").and_then(Value::as_array).expect("the devices"); + assert_eq!(devices.len(), 2); + assert_eq!(devices.first().and_then(|device| device.get("id")), Some(&json!("d-1"))); + assert_eq!(devices.first().and_then(|device| device.get("name")), Some(&json!("Pixel 9"))); + assert_eq!(devices.first().and_then(|device| device.get("current")), Some(&json!(true))); + assert_eq!(devices.first().and_then(|device| device.get("last_seen_ms")), Some(&Value::Null)); + assert_eq!(devices.get(1).and_then(|device| device.get("last_seen_ms")), Some(&json!(60))); + + let requests = stub.requests(); + assert_eq!(requests.len(), 1); + assert_eq!(requests.first().map(|request| request.method.as_str()), Some("GET")); + assert_eq!(requests.first().map(|request| request.path.as_str()), Some("/v1/vaults/v-1/devices")); + assert_eq!(requests.first().and_then(|request| request.auth.clone()), Some(format!("Bearer {TOKEN}"))); + assert!(!answer.to_string().contains(TOKEN), "the token never reaches the answer: {answer}"); +} + +#[test] +fn no_token_reaches_the_answer_or_the_rows() { + let stub = Stub::spawn(200, r#"[{"id":"d-2","name":"Laptop","created_ms":50,"last_seen_ms":60,"current":false}]"#); + let (reg, mut session, _vault) = bound(&stub, "no-token"); + + let answer = reg.run("server.devices", &mut session, &Value::Null).expect("server.devices"); + assert!(!answer.to_string().contains(TOKEN), "{}", answer); + assert!(!answer.to_string().contains("token"), "{}", answer); + assert!(!answer.to_string().contains("secret"), "{}", answer); +} + +/// A vault that is not bound says so instead of asking the server. +#[test] +fn an_unbound_vault_reports_not_bound() { + let stub = Stub::spawn(200, "[]"); + let mut reg = Registry::new(); + register_core(&mut reg).expect("register the core commands"); + let vault = temp_vault("devices-unbound"); + let mut session = Session::open(vault.path()).expect("open the vault"); + + let result = reg.run("server.devices", &mut session, &Value::Null); + assert_eq!(code(&result), "share"); + assert!(message(&result).contains("not bound"), "{}", message(&result)); + assert!(stub.requests().is_empty(), "no request was made"); +} + +/// The message of the server reaches the user, the token never does. +#[test] +fn the_server_answer_refuses_like_the_others() { + let stub = Stub::spawn(401, r#"{"error":"token is not valid for this vault"}"#); + let (reg, mut session, _vault) = bound(&stub, "devices-401"); + + let result = reg.run("server.devices", &mut session, &Value::Null); + assert_eq!(code(&result), "share"); + let text = message(&result); + assert!(text.contains("token is not valid for this vault"), "{text}"); + assert!(!text.contains(TOKEN), "{text}"); +} + +/// `server.devices` never panics on hostile params, and asks nothing. +#[test] +fn hostile_params_never_panic_the_devices_command() { + let stub = Stub::spawn(200, "[]"); + let (reg, mut session, _vault) = bound(&stub, "devices-hostile"); + let previous = std::panic::take_hook(); + std::panic::set_hook(Box::new(|_| {})); + let mut panicked = Vec::new(); + for params in notescraft_engine::adversarial_params() { + if let Err(Error::Panicked(other)) = reg.run("server.devices", &mut session, ¶ms) { + panicked.push(other); + } + } + std::panic::set_hook(previous); + assert!(panicked.is_empty(), "commands panicked: {panicked:?}"); +} diff --git a/crates/engine/tests/sync_chunks.rs b/crates/engine/tests/sync_chunks.rs new file mode 100644 index 0000000..7664ae1 --- /dev/null +++ b/crates/engine/tests/sync_chunks.rs @@ -0,0 +1,360 @@ +//! Chunked uploads: the client against a server that keeps chunks in memory. +//! +//! The fake server holds the chunks it was sent in a map and can be told to +//! cut the connection, to lose a chunk or to change the file under the +//! client's feet. + +use std::collections::BTreeMap; +use std::path::PathBuf; +use std::sync::{Arc, Mutex, MutexGuard}; + +use notescraft_engine::{Changes, Limits, Progress, PutReply, Remote, SyncError, SyncOptions, SyncResult, SyncState, sync_once_with, upload_chunked}; +use notescraft_sync::{Ledger, Outcome, Write, hash_hex}; +use notescraft_vault::Vault; + +/// The size of one chunk in the tests. +const CHUNK: usize = 1024; +/// The size above which a file uploads in chunks. +const THRESHOLD: u64 = 2 * 1024; + +/// The server the client talks to. +struct Server { + inner: Mutex, +} + +/// What the server keeps. +struct Inner { + ledger: Ledger, + /// The chunks the server holds, by their hash. + chunks: BTreeMap>, + /// The files the server holds, by their path. + files: BTreeMap>, + /// The hashes of every chunk that was sent. + chunk_calls: Vec, + /// The chunk list of every chunked write. + chunked_calls: Vec>, + /// How many chunk sends there were. + calls: usize, + /// From this send on, the connection is cut. + cut_from: Option, + /// The file that changes while the client is uploading it. + corrupt: Option<(PathBuf, Vec)>, +} + +impl Inner { + fn new() -> Self { + Self { + ledger: Ledger::new(), + chunks: BTreeMap::new(), + files: BTreeMap::new(), + chunk_calls: Vec::new(), + chunked_calls: Vec::new(), + calls: 0, + cut_from: None, + corrupt: None, + } + } +} + +impl Server { + fn new() -> Arc { + Arc::new(Server { inner: Mutex::new(Inner::new()) }) + } + + fn lock(&self) -> MutexGuard<'_, Inner> { + self.inner.lock().unwrap_or_else(std::sync::PoisonError::into_inner) + } + + /// The hashes of the chunks that were sent. + fn chunk_calls(&self) -> Vec { + self.lock().chunk_calls.clone() + } + + /// How many chunked writes the server saw. + fn chunked_calls(&self) -> usize { + self.lock().chunked_calls.len() + } + + /// The file the server holds. + fn file(&self, path: &str) -> Option> { + self.lock().files.get(path).cloned() + } + + /// Cuts the connection from the given chunk send on. + fn cut_from(&self, call: usize) { + self.lock().cut_from = Some(call); + } + + /// Lets the connection through again. + fn mend(&self) { + self.lock().cut_from = None; + } + + /// Forgets the chunks that were sent, so a test can look at one pass only. + fn forget_calls(&self) { + self.lock().chunk_calls.clear(); + } + + /// Changes the file while the client uploads it. + fn corrupt(&self, path: PathBuf, bytes: Vec) { + self.lock().corrupt = Some((path, bytes)); + } +} + +/// The client the driver talks to. +struct FakeRemote { + server: Arc, +} + +impl Remote for FakeRemote { + fn changes(&self, since: u64, limit: usize, _wait: u64) -> SyncResult { + let inner = self.server.lock(); + Ok(inner.ledger.changes_since(since, limit)) + } + + fn get(&self, path: &str) -> SyncResult> { + let inner = self.server.lock(); + inner.files.get(path).cloned().ok_or_else(|| SyncError::Status { status: 404, body: path.to_owned() }) + } + + fn put(&self, path: &str, bytes: &[u8], base_seq: u64, mtime_ms: u64) -> SyncResult { + let mut inner = self.server.lock(); + let outcome = inner + .ledger + .apply(Write { path: path.to_owned(), hash: Some(hash_hex(bytes)), size: u64::try_from(bytes.len()).expect("size"), mtime_ms, base_seq }) + .map_err(|error| SyncError::Status { status: 400, body: error.to_string() })?; + match outcome { + Outcome::Applied(record) => { + inner.files.insert(path.to_owned(), bytes.to_vec()); + Ok(PutReply::Applied(record)) + } + Outcome::Unchanged(record) => Ok(PutReply::Unchanged(record)), + Outcome::Conflict(record) => Ok(PutReply::Conflict(record)), + } + } + + fn delete(&self, path: &str, base_seq: u64) -> SyncResult { + let mut inner = self.server.lock(); + let outcome = inner + .ledger + .apply(Write { path: path.to_owned(), hash: None, size: 0, mtime_ms: 0, base_seq }) + .map_err(|error| SyncError::Status { status: 400, body: error.to_string() })?; + match outcome { + Outcome::Applied(record) => { + inner.files.remove(path); + Ok(PutReply::Applied(record)) + } + Outcome::Unchanged(record) => Ok(PutReply::Unchanged(record)), + Outcome::Conflict(record) => Ok(PutReply::Conflict(record)), + } + } + + fn missing_chunks(&self, hashes: &[String]) -> SyncResult> { + let mut inner = self.server.lock(); + if let Some((path, bytes)) = inner.corrupt.take() { + std::fs::write(&path, &bytes).map_err(|error| SyncError::Io(error.to_string()))?; + } + Ok(hashes.iter().filter(|hash| !inner.chunks.contains_key(*hash)).cloned().collect()) + } + + fn put_chunk(&self, hash: &str, bytes: &[u8]) -> SyncResult<()> { + let mut inner = self.server.lock(); + inner.calls = inner.calls.saturating_add(1); + inner.chunk_calls.push(hash.to_owned()); + if inner.cut_from.is_some_and(|from| inner.calls >= from) { + return Err(SyncError::Http("the connection was cut".to_owned())); + } + inner.chunks.insert(hash.to_owned(), bytes.to_vec()); + Ok(()) + } + + fn put_chunked(&self, path: &str, chunk_hashes: &[String], content_hash: &str, base_seq: u64, mtime_ms: u64) -> SyncResult { + let mut inner = self.server.lock(); + inner.chunked_calls.push(chunk_hashes.to_vec()); + let mut bytes = Vec::new(); + for hash in chunk_hashes { + let Some(chunk) = inner.chunks.get(hash) else { return Err(SyncError::MissingChunks(vec![hash.clone()])) }; + bytes.extend_from_slice(chunk); + } + let outcome = inner + .ledger + .apply(Write { path: path.to_owned(), hash: Some(content_hash.to_owned()), size: u64::try_from(bytes.len()).expect("size"), mtime_ms, base_seq }) + .map_err(|error| SyncError::Status { status: 400, body: error.to_string() })?; + match outcome { + Outcome::Applied(record) => { + inner.files.insert(path.to_owned(), bytes); + Ok(PutReply::Applied(record)) + } + Outcome::Unchanged(record) => Ok(PutReply::Unchanged(record)), + Outcome::Conflict(record) => Ok(PutReply::Conflict(record)), + } + } +} + +/// A vault in a fresh folder. +fn temp_vault(name: &str) -> Vault { + let root = std::env::temp_dir().join(format!("nc-engine-chunks-{name}-{}", std::process::id())); + let _ = std::fs::remove_dir_all(&root); + std::fs::create_dir_all(&root).expect("temp folder"); + Vault::open(&root).expect("vault") +} + +/// The limits the tests use: small chunks, so a small file has many of them. +fn limits() -> Limits { + Limits { threshold: THRESHOLD, chunk: CHUNK } +} + +/// Ten kilobytes whose chunks differ from one another. +fn ten_chunks() -> Vec { + (0..10u8).flat_map(|index| vec![index; CHUNK]).collect() +} + +/// The hash of the chunk number `index` of `bytes`. +fn chunk_hash(bytes: &[u8], index: usize) -> String { + let start = index * CHUNK; + let end = start.saturating_add(CHUNK); + hash_hex(bytes.get(start..end).expect("chunk")) +} + +#[test] +fn a_big_file_travels_in_chunks_and_is_written_once() { + let server = Server::new(); + let vault = temp_vault("big"); + let bytes = ten_chunks(); + vault.write("big.bin", &bytes).expect("write"); + let remote = FakeRemote { server: server.clone() }; + + let reply = upload_chunked(&remote, &vault, "big.bin", 0, 0, limits(), &mut |_: &Progress| {}).expect("upload"); + + assert!(matches!(reply, PutReply::Applied(_)), "{reply:?}"); + assert_eq!(server.chunk_calls().len(), 10, "one send per chunk"); + assert_eq!(server.chunked_calls(), 1, "the file is written once"); + assert_eq!(server.file("big.bin"), Some(bytes), "the server holds the whole file"); +} + +#[test] +fn a_cut_connection_resumes_with_the_chunks_that_never_arrived() { + let server = Server::new(); + let vault = temp_vault("resume"); + let bytes = ten_chunks(); + vault.write("big.bin", &bytes).expect("write"); + let remote = FakeRemote { server: server.clone() }; + + // The connection dies on the sixth chunk; the chunk is tried again and fails. + server.cut_from(6); + let error = upload_chunked(&remote, &vault, "big.bin", 0, 0, limits(), &mut |_: &Progress| {}).expect_err("cut"); + assert!(matches!(error, SyncError::Http(_)), "{error:?}"); + assert_eq!(server.chunked_calls(), 0, "the file was never written"); + + // The next pass sends only the chunks that never arrived. + server.mend(); + server.forget_calls(); + upload_chunked(&remote, &vault, "big.bin", 0, 0, limits(), &mut |_: &Progress| {}).expect("resumed"); + + let sent: Vec = (5..10).map(|index| chunk_hash(&bytes, index)).collect(); + assert_eq!(server.chunk_calls(), sent, "only the missing chunks are sent again"); + assert_eq!(server.file("big.bin"), Some(bytes), "the server holds the whole file"); +} + +#[test] +fn the_same_chunk_of_a_file_is_sent_once() { + let server = Server::new(); + let vault = temp_vault("same"); + let bytes = vec![7u8; CHUNK * 4]; + vault.write("same.bin", &bytes).expect("write"); + let remote = FakeRemote { server: server.clone() }; + + upload_chunked(&remote, &vault, "same.bin", 0, 0, limits(), &mut |_: &Progress| {}).expect("upload"); + + assert_eq!(server.chunk_calls().len(), 1, "the four equal chunks travel once"); + assert_eq!(server.file("same.bin"), Some(bytes), "the file still holds the four blocks"); +} + +#[test] +fn the_progress_grows_and_ends_at_the_size_of_the_file() { + let server = Server::new(); + let vault = temp_vault("progress"); + let bytes = ten_chunks(); + vault.write("big.bin", &bytes).expect("write"); + let remote = FakeRemote { server: server.clone() }; + + let mut seen: Vec = Vec::new(); + upload_chunked(&remote, &vault, "big.bin", 0, 0, limits(), &mut |progress: &Progress| seen.push(progress.clone())).expect("upload"); + + assert_eq!(seen.len(), 10, "one report per chunk"); + assert!(seen.windows(2).all(|pair| pair[0].done_bytes <= pair[1].done_bytes), "the progress never goes back"); + let last = seen.last().expect("a report"); + assert_eq!(last.done_bytes, last.total_bytes); + assert_eq!(last.total_bytes, u64::try_from(bytes.len()).expect("size")); + assert_eq!(last.path, "big.bin"); +} + +#[test] +fn a_small_file_never_touches_the_chunk_methods() { + let server = Server::new(); + let vault = temp_vault("small"); + vault.write("small.md", b"# one\n").expect("write"); + let remote = FakeRemote { server: server.clone() }; + let mut state = SyncState::default(); + + let report = sync_once_with(&vault, &remote, &mut state, "phone", "2026-10-20", SyncOptions { limits: limits(), progress: Box::new(|_: &Progress| {}) }) + .expect("sync"); + + assert_eq!(report.uploaded, vec!["small.md".to_owned()]); + assert_eq!(server.chunk_calls(), Vec::::new(), "a small file is not chunked"); + assert_eq!(server.chunked_calls(), 0); + assert_eq!(server.file("small.md"), Some(b"# one\n".to_vec())); +} + +#[test] +fn a_file_that_changes_while_it_is_uploaded_is_refused() { + let server = Server::new(); + let vault = temp_vault("changed"); + let bytes = ten_chunks(); + vault.write("big.bin", &bytes).expect("write"); + let remote = FakeRemote { server: server.clone() }; + // The file changes between the hash pass and the upload pass. + server.corrupt(vault.root().join("big.bin"), vec![1u8; CHUNK * 10]); + + let error = upload_chunked(&remote, &vault, "big.bin", 0, 0, limits(), &mut |_: &Progress| {}).expect_err("changed"); + + assert!(matches!(error, SyncError::BadRemote(_)), "{error:?}"); + assert_eq!(server.chunked_calls(), 0, "no mixed file reaches the server"); + assert_eq!(server.file("big.bin"), None); +} + +#[test] +fn an_empty_file_goes_as_a_plain_write() { + let server = Server::new(); + let vault = temp_vault("empty"); + vault.write("empty.bin", b"").expect("write"); + let remote = FakeRemote { server: server.clone() }; + + let reply = upload_chunked(&remote, &vault, "empty.bin", 0, 0, limits(), &mut |_: &Progress| {}).expect("upload"); + + assert!(matches!(reply, PutReply::Applied(_)), "{reply:?}"); + assert_eq!(server.chunk_calls(), Vec::::new(), "an empty file has no chunk"); + assert_eq!(server.chunked_calls(), 0); + assert_eq!(server.file("empty.bin"), Some(Vec::new())); +} + +#[test] +fn the_threshold_decides_what_is_chunked() { + let server = Server::new(); + let vault = temp_vault("threshold"); + let at_threshold: Vec = (0..THRESHOLD).map(|index| u8::try_from(index % 251).expect("byte")).collect(); + let over_threshold: Vec = (0..=THRESHOLD).map(|index| u8::try_from(index % 251).expect("byte")).collect(); + vault.write("at.bin", &at_threshold).expect("write"); + vault.write("over.bin", &over_threshold).expect("write"); + let remote = FakeRemote { server: server.clone() }; + let mut state = SyncState::default(); + + let report = sync_once_with(&vault, &remote, &mut state, "phone", "2026-10-20", SyncOptions { limits: limits(), progress: Box::new(|_: &Progress| {}) }) + .expect("sync"); + + assert_eq!(report.uploaded.len(), 2); + assert_eq!(server.chunked_calls(), 1, "only the file over the threshold is chunked"); + assert_eq!(server.chunk_calls().len(), 3, "the file over the threshold has three chunks"); + assert_eq!(server.file("at.bin"), Some(at_threshold)); + assert_eq!(server.file("over.bin"), Some(over_threshold)); +} diff --git a/crates/i18n/locales/en.toml b/crates/i18n/locales/en.toml index 5f9b73d..e4e43eb 100644 --- a/crates/i18n/locales/en.toml +++ b/crates/i18n/locales/en.toml @@ -1328,6 +1328,31 @@ ago-minutes = "{n} min ago" ago-hours = "{n} h ago" ago-days = "{n} d ago" +[ui.sync-sheet] +title = "Sync" +status-up-to-date = "Everything is up to date" +status-syncing = "Syncing..." +status-offline = "Offline, will retry" +status-sign-in = "Sign in again" +status-error = "Sync stopped" +last-sync = "Last sync {when} · {host}" +never-synced = "Not yet synced" +devices = "Devices" +this-phone = "This phone" +this-computer = "This computer" +last-seen = "last seen {when}" +never-seen = "last seen never" +chip-online = "online" +chip-synced = "synced" +chip-stale = "stale" +conflicts = "{n} conflicts to review" +wifi-only = "Sync on Wi-Fi only" +attachments = "Sync attachments" +settings-and-themes = "Sync settings and themes" +sync-now = "Sync now" +loading = "Loading devices..." +pairing-failed = "The pairing failed. Check the server address and the code." + [ui.live] pill-live = "Live" go-live = "Go live" diff --git a/crates/sync/src/hash.rs b/crates/sync/src/hash.rs index 38f390f..4488b87 100644 --- a/crates/sync/src/hash.rs +++ b/crates/sync/src/hash.rs @@ -9,12 +9,33 @@ const HEX: &[u8; 16] = b"0123456789abcdef"; /// SHA-256 of `bytes` as 64 lowercase hex digits. #[must_use] pub fn hash_hex(bytes: &[u8]) -> String { - let mut out = String::with_capacity(64); - for byte in &Sha256::digest(bytes) { - out.push(char::from(HEX.get(usize::from(byte >> 4)).copied().unwrap_or(b'0'))); - out.push(char::from(HEX.get(usize::from(byte & 0x0f)).copied().unwrap_or(b'0'))); + hex_digest(&Sha256::digest(bytes)) +} + +/// An incremental SHA-256 hash: the pieces are added one at a time, so content +/// too large to hold in memory can still be hashed. +#[derive(Debug, Clone, Default)] +pub struct Hasher { + inner: Sha256, +} + +impl Hasher { + /// Adds `bytes` to the hash. + pub fn update(&mut self, bytes: &[u8]) { + self.inner.update(bytes); } - out + + /// The hash of everything added so far, as 64 lowercase hex digits. + #[must_use] + pub fn finish_hex(self) -> String { + hex_digest(&self.inner.finalize()) + } +} + +/// A fresh incremental hasher, for content that never sits in memory whole. +#[must_use] +pub fn hasher() -> Hasher { + Hasher::default() } /// Checks the shape of a hash received from outside. @@ -28,3 +49,38 @@ pub fn check_hash(hash: &str) -> Result<()> { Err(Error::InvalidHash(hash.chars().take(80).collect())) } } + +/// A digest as 64 lowercase hex digits. +fn hex_digest(digest: &[u8]) -> String { + let mut out = String::with_capacity(digest.len() * 2); + for byte in digest { + out.push(char::from(HEX.get(usize::from(byte >> 4)).copied().unwrap_or(b'0'))); + out.push(char::from(HEX.get(usize::from(byte & 0x0f)).copied().unwrap_or(b'0'))); + } + out +} + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn the_incremental_hash_is_the_hash_of_the_joined_bytes() { + let mut hasher = hasher(); + hasher.update(b"hello "); + hasher.update(b"world"); + assert_eq!(hasher.finish_hex(), hash_hex(b"hello world")); + } + + #[test] + fn an_empty_hash_is_the_hash_of_no_bytes() { + assert_eq!(hasher().finish_hex(), hash_hex(b"")); + } + + #[test] + fn a_broken_hash_is_refused() { + assert!(check_hash(&hash_hex(b"x")).is_ok()); + assert!(check_hash("not a hash").is_err()); + assert!(check_hash(&hash_hex(b"x").to_uppercase()).is_err()); + } +} diff --git a/crates/sync/src/lib.rs b/crates/sync/src/lib.rs index 26ee097..8a3e37b 100644 --- a/crates/sync/src/lib.rs +++ b/crates/sync/src/lib.rs @@ -9,7 +9,7 @@ //! of each path and a `/changes` page to a list of [`Action`]s. //! * [`merge3`] — three-way line merge; anything doubtful is a conflict, never a guess. //! * [`conflict_path`] — the name of the copy a conflict leaves behind. -//! * [`check_path`], [`check_syncable`], [`hash_hex`] — the gate every path and hash passes. +//! * [`check_path`], [`check_syncable`], [`hash_hex`], [`hasher`] — the gate every path and hash passes. //! //! ``` //! use notescraft_sync::{Ledger, Outcome, Write, hash_hex}; @@ -33,7 +33,7 @@ mod record; pub use conflict::conflict_path; pub use error::{Error, Result}; -pub use hash::{check_hash, hash_hex}; +pub use hash::{Hasher, check_hash, hash_hex, hasher}; pub use ledger::{Ledger, Outcome, Write}; pub use merge::{Merge, merge3}; pub use path::{MAX_PATH_BYTES, check_path, check_syncable, is_excluded}; diff --git a/crates/ui-egui/src/panels/mod.rs b/crates/ui-egui/src/panels/mod.rs index 421f7c2..031ed46 100644 --- a/crates/ui-egui/src/panels/mod.rs +++ b/crates/ui-egui/src/panels/mod.rs @@ -13,6 +13,7 @@ pub mod sidebar; pub mod sidebar_tabs; pub mod source_panel; pub mod sync_pill; +pub mod sync_sheet; pub mod tags_panel; pub mod tree; #[cfg(test)] diff --git a/crates/ui-egui/src/panels/sync_pill/mod.rs b/crates/ui-egui/src/panels/sync_pill/mod.rs index b6dbcfe..1ec20d5 100644 --- a/crates/ui-egui/src/panels/sync_pill/mod.rs +++ b/crates/ui-egui/src/panels/sync_pill/mod.rs @@ -106,7 +106,13 @@ pub fn show(ui: &mut Ui, tokens: &Tokens, status: &SyncStatus, right: f32, middl ui.painter().galley(text, galley, tint); let response = ui.interact(rect, ui.id().with("sync-pill"), Sense::click()); let _ = crate::motion::hover(ui.ctx(), "sync-pill", &response); - response.clicked() + let clicked = response.clicked(); + // Phone width: the click asks the shell for the sheet, not for Settings > + // Sync; the flag is a same-frame egui temp the shell reads and clears. + if clicked && crate::style::compact::is_compact(ui.ctx()) { + ui.ctx().data_mut(|data| data.insert_temp(egui::Id::new("sync-sheet-request"), true)); + } + clicked } #[cfg(test)] diff --git a/crates/ui-egui/src/panels/sync_sheet/devices.rs b/crates/ui-egui/src/panels/sync_sheet/devices.rs new file mode 100644 index 0000000..03fe993 --- /dev/null +++ b/crates/ui-egui/src/panels/sync_sheet/devices.rs @@ -0,0 +1,284 @@ +//! The state half of the sync sheet: the status card, the device rows and the +//! conflict row — everything the vault's standing on the server decides. + +use egui::{Align2, CornerRadius, FontId, Rect, Sense, Ui, Vec2, vec2}; +use notescraft_engine::{ServerDevice, SyncPhase, SyncStatus}; + +use super::{Facts, SheetAction, SyncSheet}; +use crate::text::sync_sheet as strings; +use crate::theme::{Tokens, color}; + +/// Radius of one card, points. +const CARD_RADIUS: u8 = 12; +/// Height of one device row, points. +const DEVICE_ROW: f32 = 72.0; +/// Height of the conflict row, points. +const CONFLICT_ROW: f32 = 56.0; +/// Height of the status card, points. +const STATUS_ROW: f32 = 96.0; +/// Side of the status icon, points. +const STATUS_ICON: f32 = 24.0; +/// Left pad inside a card, points. +const CARD_PAD: f32 = 16.0; +/// Gap between the icon and the text block, points. +const TEXT_GAP: f32 = 16.0; +/// Side of the chip's state dot, points. +const CHIP_DOT: f32 = 8.0; +/// Space between the chip's dot and its word, points. +const CHIP_GAP: f32 = 6.0; +/// Text size of a name, points. +const NAME: f32 = 15.0; +/// Text size of muted detail lines, points. +const DETAIL: f32 = 13.0; +/// Gap between the headline and the detail of the status card. +const LINE_GAP: f32 = 16.0; +/// Icons of the status card, by phase; a phase without one shows its error. +const UP_TO_DATE_ICON: &str = "circle-check"; +const SYNCING_ICON: &str = "refresh-cw"; +const ATTENTION_ICON: &str = "circle-alert"; +const ERROR_ICON: &str = "triangle-alert"; +/// Milliseconds a device may be out of sight to still count as online. +const ONLINE_WINDOW_MS: u64 = 2 * 60 * 1000; +/// Milliseconds in one day. +const MS_PER_DAY: u64 = 24 * 60 * 60 * 1000; + +/// The icon of the status card, from the phase. +fn status_icon(status: Option<&SyncStatus>) -> &'static str { + match status.map(|status| status.phase) { + Some(SyncPhase::Idle) => UP_TO_DATE_ICON, + Some(SyncPhase::Syncing) => SYNCING_ICON, + Some(SyncPhase::Offline | SyncPhase::SignInAgain) => ATTENTION_ICON, + Some(SyncPhase::Error) | None => ERROR_ICON, + } +} + +/// The tint of the status card's icon. +fn status_tint(tokens: &Tokens, status: Option<&SyncStatus>) -> egui::Color32 { + let colors = &tokens.theme.colors; + match status.map(|status| status.phase) { + Some(SyncPhase::Idle) => color(colors.success), + Some(SyncPhase::Syncing | SyncPhase::SignInAgain) => color(colors.warning), + Some(SyncPhase::Offline) => color(colors.text_muted), + Some(SyncPhase::Error) | None => color(colors.danger), + } +} + +/// The headline of the status card, from the phase; the error speaks for itself. +#[must_use] +pub fn headline(status: Option<&SyncStatus>) -> String { + match status.map(|status| status.phase) { + Some(SyncPhase::Idle) => strings::status_up_to_date().to_owned(), + Some(SyncPhase::Syncing) => strings::status_syncing().to_owned(), + Some(SyncPhase::Offline) => strings::status_offline().to_owned(), + Some(SyncPhase::SignInAgain) => strings::status_sign_in().to_owned(), + Some(SyncPhase::Error) | None => status.and_then(|status| status.last_error.clone()).unwrap_or_else(|| strings::status_error().to_owned()), + } +} + +/// The second line of the status card: "Last sync 2 min ago · host". +fn status_detail(status: Option<&SyncStatus>, now_ms: u64) -> String { + let host = status.and_then(|status| status.server.as_deref()).map(crate::views::settings_view::sync::server_host).unwrap_or_default(); + let Some(last) = status.and_then(|status| status.last_sync_ms) else { + // Never synced: no "Last sync" prefix, and the host only when there is one. + let never = strings::never_synced(); + return if host.is_empty() { never.to_owned() } else { format!("{never} · {host}") }; + }; + let text = strings::last_sync().replace("{when}", &crate::views::settings_view::sync::ago(last, now_ms)).replace("{host}", host); + text.trim_end_matches([' ', '·']).to_owned() +} + +/// The status card (design: the check and the two lines). +pub fn status_card(ui: &mut Ui, tokens: &Tokens, facts: &Facts<'_>) { + let colors = &tokens.theme.colors; + let (rect, _) = ui.allocate_exact_size(Vec2::new(ui.available_width(), STATUS_ROW), Sense::hover()); + ui.painter().rect_filled(rect, CornerRadius::same(CARD_RADIUS), color(colors.surface_2)); + let icon = Rect::from_center_size(rect.left_top() + vec2(CARD_PAD + STATUS_ICON / 2.0, rect.height() / 2.0), Vec2::splat(STATUS_ICON)); + crate::icons::paint(ui, status_icon(facts.status.as_ref()), icon, status_tint(tokens, facts.status.as_ref())); + let text_left = rect.left() + CARD_PAD + STATUS_ICON + TEXT_GAP; + let width = rect.right() - text_left - CARD_PAD; + ui.painter().text( + egui::pos2(text_left, rect.center().y - LINE_GAP), + Align2::LEFT_CENTER, + headline(facts.status.as_ref()), + FontId::new(NAME, crate::fonts::semibold()), + color(colors.text), + ); + let font = FontId::proportional(DETAIL); + let galley = ui.painter().layout(status_detail(facts.status.as_ref(), facts.now_ms), font, color(colors.text_muted), width); + ui.painter().galley(egui::pos2(text_left, rect.center().y + LINE_GAP - galley.size().y / 2.0), galley, color(colors.text_muted)); +} + +/// The devices: the heading, the rows (current first) and the line of a load. +pub fn section(sheet: &mut SyncSheet, ui: &mut Ui, tokens: &Tokens, facts: &Facts<'_>, action: &mut Option) { + let colors = &tokens.theme.colors; + heading(ui, tokens, strings::devices()); + if sheet.loading() { + ui.horizontal(|ui| { + ui.add(egui::Spinner::new().size(14.0).color(color(colors.accent))); + ui.add_space(8.0); + ui.label(egui::RichText::new(strings::loading()).size(DETAIL).color(color(colors.text_muted))); + }); + } else if let Some(error) = sheet.error.as_deref() { + ui.label(egui::RichText::new(error).size(DETAIL).color(color(colors.danger))); + } else { + for device in ordered(&sheet.devices) { + device_row(ui, tokens, &device, facts); + } + } + conflict_row(ui, tokens, facts, action); +} + +/// The devices with the current one first, the others in the order of the server. +fn ordered(devices: &[ServerDevice]) -> Vec { + let mut sorted = devices.to_vec(); + sorted.sort_by_key(|device| !device.current); + sorted +} + +/// The card of one device (design: name, "last seen ago", state chip). +fn device_row(ui: &mut Ui, tokens: &Tokens, device: &ServerDevice, facts: &Facts<'_>) { + let colors = &tokens.theme.colors; + let (rect, _) = ui.allocate_exact_size(Vec2::new(ui.available_width(), DEVICE_ROW), Sense::hover()); + ui.painter().rect_filled(rect, CornerRadius::same(CARD_RADIUS), color(colors.surface_2)); + let name = match device.current { + true if facts.compact => strings::this_phone(), + true => strings::this_computer(), + false => device.name.as_str(), + }; + ui.painter().text( + egui::pos2(rect.left() + CARD_PAD, rect.center().y - 12.0), + Align2::LEFT_CENTER, + name, + FontId::new(NAME, crate::fonts::semibold()), + color(colors.text), + ); + let seen = match device.last_seen_ms { + Some(seen) => strings::last_seen().replace("{when}", &crate::views::settings_view::sync::ago(seen, facts.now_ms)), + None => strings::never_seen().to_owned(), + }; + ui.painter().text( + egui::pos2(rect.left() + CARD_PAD, rect.center().y + 12.0), + Align2::LEFT_CENTER, + seen, + FontId::proportional(DETAIL), + color(colors.text_muted), + ); + paint_chip(ui, tokens, rect, device, facts.now_ms); +} + +/// The state chip at the right end of a device row. +fn paint_chip(ui: &Ui, tokens: &Tokens, row: Rect, device: &ServerDevice, now_ms: u64) { + let (label, tint) = chip_state(device.last_seen_ms, now_ms, tokens); + let galley = ui.painter().layout_no_wrap(label.to_owned(), FontId::new(DETAIL, crate::fonts::semibold()), tint); + let chip_width = CHIP_DOT + CHIP_GAP + galley.size().x + CARD_PAD; + let left = row.right() - chip_width; + let centre_y = row.center().y; + ui.painter().circle_filled(egui::pos2(left + CHIP_DOT / 2.0, centre_y), CHIP_DOT / 2.0, tint); + ui.painter().galley(egui::pos2(left + CHIP_DOT + CHIP_GAP, centre_y - galley.size().y / 2.0), galley, tint); +} + +/// The word and colour of the state chip: online, synced, stale. +fn chip_state(last_seen_ms: Option, now_ms: u64, tokens: &Tokens) -> (&'static str, egui::Color32) { + let colors = &tokens.theme.colors; + let seen_ago = last_seen_ms.map(|seen| now_ms.saturating_sub(seen)); + let label = match seen_ago { + Some(ago) if ago <= ONLINE_WINDOW_MS => strings::chip_online(), + Some(ago) if ago <= MS_PER_DAY => strings::chip_synced(), + _ => strings::chip_stale(), + }; + ( + label, + match seen_ago { + Some(ago) if ago <= MS_PER_DAY => color(colors.success), + Some(_) => color(colors.warning), + None => color(colors.text_muted), + }, + ) +} + +/// The row "N conflicts to review"; hidden unless there is something to review. +fn conflict_row(ui: &mut Ui, tokens: &Tokens, facts: &Facts<'_>, action: &mut Option) { + if facts.conflicts == 0 { + return; + } + let colors = &tokens.theme.colors; + let label = strings::conflicts().replace("{n}", &facts.conflicts.to_string()); + let (rect, hit) = ui.allocate_exact_size(Vec2::new(ui.available_width(), CONFLICT_ROW), Sense::click()); + hit.widget_info(|| egui::WidgetInfo::selected(egui::WidgetType::Button, true, false, "conflicts")); + ui.painter().rect_filled(rect, CornerRadius::same(CARD_RADIUS), color(colors.surface_2)); + ui.painter().text( + egui::pos2(rect.left() + CARD_PAD, rect.center().y), + Align2::LEFT_CENTER, + label, + FontId::new(NAME, crate::fonts::semibold()), + color(colors.text), + ); + crate::icons::paint( + ui, + "chevron-right", + Rect::from_center_size(egui::pos2(rect.right() - CARD_PAD - 8.0, rect.center().y), vec2(16.0, 16.0)), + color(colors.text_muted), + ); + if hit.clicked() { + *action = Some(SheetAction::ReviewConflicts); + } +} + +/// The small caps heading over a group of cards. +fn heading(ui: &mut Ui, tokens: &Tokens, text: &str) { + let colors = &tokens.theme.colors; + ui.add_space(4.0); + ui.label(egui::RichText::new(text).font(FontId::new(12.0, crate::fonts::semibold())).color(color(colors.text_muted))); +} + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn the_headlines_follow_the_phases() { + let status = SyncStatus { phase: SyncPhase::Idle, ..SyncStatus::default() }; + assert_eq!(headline(Some(&status)), "Everything is up to date"); + let syncing = SyncStatus { phase: SyncPhase::Syncing, ..SyncStatus::default() }; + assert_eq!(headline(Some(&syncing)), "Syncing..."); + assert_eq!(headline(None), "Sync stopped", "a sheet without a status speaks its own state"); + } + + #[test] + fn the_error_speaks_as_the_headline() { + let status = SyncStatus { phase: SyncPhase::Error, last_error: Some("quota is full".to_owned()), ..SyncStatus::default() }; + assert_eq!(headline(Some(&status)), "quota is full"); + } + + #[test] + fn the_detail_carries_the_last_sync_and_the_host() { + let status = SyncStatus { server: Some("https://sync.example.com".to_owned()), last_sync_ms: Some(998_000), ..SyncStatus::default() }; + assert_eq!(status_detail(Some(&status), 1_000_000), "Last sync 2 s ago · sync.example.com"); + assert_eq!(status_detail(None, 1_000_000), strings::never_synced()); + } + + #[test] + fn the_chips_age() { + let tokens = Tokens::new(notescraft_themes::builtin("catppuccin-mocha").expect("theme")); + let now = 1_000_000_000_u64; + let (online, _) = chip_state(Some(now - 1000), now, &tokens); + assert_eq!(online, strings::chip_online()); + let (synced, _) = chip_state(Some(now - 3_600_000), now, &tokens); + assert_eq!(synced, strings::chip_synced()); + let (stale, _) = chip_state(Some(now - MS_PER_DAY - 1), now, &tokens); + assert_eq!(stale, strings::chip_stale()); + let (never, _) = chip_state(None, now, &tokens); + assert_eq!(never, strings::chip_stale()); + } + + #[test] + fn the_current_device_comes_first() { + let devices = [ + ServerDevice { id: "d-2".to_owned(), name: "Laptop".to_owned(), ..ServerDevice::default() }, + ServerDevice { id: "d-1".to_owned(), name: "Phone".to_owned(), current: true, ..ServerDevice::default() }, + ]; + let listed = ordered(&devices); + assert_eq!(listed.first().map(|device| device.id.as_str()), Some("d-1"), "the current device first"); + assert_eq!(listed.get(1).map(|device| device.id.as_str()), Some("d-2")); + } +} diff --git a/crates/ui-egui/src/panels/sync_sheet/mod.rs b/crates/ui-egui/src/panels/sync_sheet/mod.rs new file mode 100644 index 0000000..f9cca3d --- /dev/null +++ b/crates/ui-egui/src/panels/sync_sheet/mod.rs @@ -0,0 +1,224 @@ +//! The sync sheet (design `03-sync-sheet`): everything about sync of the +//! bound vault in one bottom sheet of the compact layout. +//! +//! The compact-layout sync pill opens it; on a wider window the pill keeps +//! opening Settings > Sync (see [`crate::panels::sync_pill`]). The sheet holds +//! the state (devices, the running load job) and gets the live facts — the +//! status, the conflicts and the user's options — handed in every frame. +//! +//! No ask waits on the UI thread: the devices load on a worker thread that +//! holds the one [`ServerClient`] (the token never crosses back) and answers +//! through a channel the sheet polls, exactly like the Share dialog. + +mod devices; +mod options; + +use std::sync::mpsc::{Receiver, TryRecvError}; +use std::sync::{Arc, Mutex}; + +use egui::{Context, ScrollArea}; +use notescraft_engine::{ServerClient, ServerDevice, SyncSettings, SyncStatus, load_binding, load_token}; + +use crate::text::sync_sheet as strings; +use crate::theme::Tokens; + +/// The id of the sheet, also the salt of its widgets. +const SHEET_ID: &str = "notescraft-sync-sheet"; +/// Most of the screen the open sheet may take, as a fraction, so the content +/// scrolls instead of reaching the top edge of the window; the sheet of the +/// design (status, three devices, the conflict row, three options and the +/// button) is about 700 pt tall on a 390 x 844 phone and shows whole. +const MAX_SHEET_HEIGHT: f32 = 0.9; +/// How long a sheet without a job waits before it repaints for the spinner. +const LOAD_MILLIS: u64 = 40; + +/// What the shell hands to the sheet every frame. +pub struct Facts<'a> { + /// The status of the sync worker, `None` when the vault has no worker. + pub status: Option, + /// The conflict copies waiting for a decision. + pub conflicts: usize, + /// The user's three sync options. + pub options: &'a SyncSettings, + /// Now, in Unix milliseconds, for the ages. + pub now_ms: u64, + /// Whether the window is in the compact (phone) layout. + pub compact: bool, +} + +/// What the sheet asks the app to do. +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub enum SheetAction { + /// Close the sheet (the ×, a tap on the dimmed area or Escape). + Close, + /// Run `sync.now`. + SyncNow, + /// Open Settings > Sync's conflict list. + ReviewConflicts, + /// One of the three options flipped; the shell stores the new value. + Toggle(SheetOption, bool), +} + +/// Which of the three options a [`SheetAction::Toggle`] names. +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub enum SheetOption { + /// `sync.wifi_only`, stored but not yet honoured by the worker. + WifiOnly, + /// `sync.attachments`: `false` leaves the non-Markdown files out of the scan. + Attachments, + /// `sync.settings_and_themes`, stored but not yet honoured by the worker. + SettingsAndThemes, +} + +impl SheetOption { + /// The settings key of the option. + #[must_use] + pub fn key(self) -> &'static str { + match self { + SheetOption::WifiOnly => "sync.wifi_only", + SheetOption::Attachments => "sync.attachments", + SheetOption::SettingsAndThemes => "sync.settings_and_themes", + } + } +} + +/// The state of the open sheet between frames. +pub struct SyncSheet { + /// The client of the account, `None` when the vault is not bound. + client: Option>>, + /// The vault id on the server. + vault_id: String, + /// The devices of the last load, current first. + devices: Vec, + /// What the last failed load said. + error: Option, + /// The running load job, when there is one. + job: Option>>>, +} + +/// The answer of a load job. +enum DevicesOutcome { + /// The devices the server named. + Listed(Vec), + /// What the server or the connection said. + Failed(String), +} + +impl SyncSheet { + /// Builds the sheet over the vault of `session` and asks for its devices. + #[must_use] + pub fn open(session: ¬escraft_engine::Session) -> Self { + let mut sheet = Self { client: None, vault_id: String::new(), devices: Vec::new(), error: None, job: None }; + let binding = load_binding(session.vault()).ok().flatten(); + let token = load_token(session.vault()).ok().flatten(); + if let Some(binding) = binding + && let Some(token) = token + { + sheet.client = Some(Arc::new(Mutex::new(ServerClient::new(&binding.server, &token)))); + sheet.vault_id.clone_from(&binding.vault_id); + sheet.start_load(); + } + sheet + } + + /// Draws the sheet; returns what it asked the app to do. + pub fn show(&mut self, ctx: &Context, tokens: &Tokens, facts: &Facts<'_>) -> Option { + self.poll(); + let max_height = ctx.content_rect().height() * MAX_SHEET_HEIGHT; + let mut action = None; + let closed = crate::widgets::bottom_sheet::show( + ctx, + tokens, + &crate::widgets::bottom_sheet::Sheet { + id: egui::Id::new(SHEET_ID), + icon: crate::icons::image("refresh-cw", 16.0, egui::Color32::TRANSPARENT), + title: strings::title(), + chip: None, + }, + |ui| { + ScrollArea::vertical() + .id_salt("notescraft-sync-sheet-scroll") + .auto_shrink([false, true]) + .min_scrolled_height(max_height) + .max_height(max_height) + .show(ui, |ui| { + devices::status_card(ui, tokens, facts); + let mut sheet_action = None; + devices::section(self, ui, tokens, facts, &mut sheet_action); + options::section(ui, tokens, facts, &mut sheet_action); + if sheet_action.is_some() { + action = sheet_action; + } + }); + }, + ); + if self.job.is_some() { + ctx.request_repaint_after(std::time::Duration::from_millis(LOAD_MILLIS)); + } + if closed { + return Some(action.unwrap_or(SheetAction::Close)); + } + action + } + + /// Whether a load job is running. + #[must_use] + pub fn loading(&self) -> bool { + self.job.is_some() + } + + /// Starts a load of the devices on a fresh worker thread; the client moves + /// in, so the token never reaches the screen. + fn start_load(&mut self) { + if self.job.is_some() { + return; + } + let Some(client) = self.client.clone() else { return }; + let vault_id = self.vault_id.clone(); + let (sender, receiver) = std::sync::mpsc::channel(); + match std::thread::Builder::new().name("notescraft-sync-sheet".to_owned()).spawn(move || { + let outcome = match client.lock().unwrap_or_else(std::sync::PoisonError::into_inner).devices(&vault_id) { + Ok(devices) => DevicesOutcome::Listed(devices), + Err(error) => DevicesOutcome::Failed(error.to_string()), + }; + let _ = sender.send(outcome); + }) { + Ok(_) => { + self.job = Some(Arc::new(Mutex::new(receiver))); + self.error = None; + } + Err(error) => self.error = Some(error.to_string()), + } + } + + /// Takes the answer of the load job, when it is there. + fn poll(&mut self) { + let Some(job) = self.job.clone() else { return }; + let outcome = { + let guard = job.lock().unwrap_or_else(std::sync::PoisonError::into_inner); + match guard.try_recv() { + Ok(outcome) => Some(outcome), + Err(TryRecvError::Empty) => None, + Err(TryRecvError::Disconnected) => Some(DevicesOutcome::Failed(strings::pairing_failed().to_owned())), + } + }; + match outcome { + None => {} + Some(DevicesOutcome::Listed(devices)) => { + self.job = None; + self.devices = devices; + self.error = None; + } + Some(DevicesOutcome::Failed(error)) => { + self.job = None; + self.error = Some(error); + } + } + } +} + +/// The current Unix time in milliseconds; the one clock of the sheet. +#[must_use] +pub fn now_millis() -> u64 { + std::time::SystemTime::now().duration_since(std::time::UNIX_EPOCH).map_or(0, |since| u64::try_from(since.as_millis()).unwrap_or(u64::MAX)) +} diff --git a/crates/ui-egui/src/panels/sync_sheet/options.rs b/crates/ui-egui/src/panels/sync_sheet/options.rs new file mode 100644 index 0000000..3594fb4 --- /dev/null +++ b/crates/ui-egui/src/panels/sync_sheet/options.rs @@ -0,0 +1,96 @@ +//! The options half of the sync sheet: the three toggles and Sync now. +//! +//! The toggles report a flip; the shell stores it through `settings.set` and +//! the next frame's facts show it. The Sync now button is the primary action +//! of the sheet; it asks for `sync.now` and stays open. + +use egui::{Align2, CornerRadius, FontId, Rect, Sense, Ui, Vec2, vec2}; + +use super::{Facts, SHEET_ID, SheetAction, SheetOption}; +use crate::text::sync_sheet as strings; +use crate::theme::{Tokens, color}; + +/// Height of one toggle row, points. +const TOGGLE_ROW: f32 = 48.0; +/// Height of the Sync now button, points. +const BUTTON_ROW: f32 = 52.0; +/// Left pad inside a row, points. +const ROW_PAD: f32 = 16.0; +/// Radius of the Sync now card, points. +const BUTTON_RADIUS: u8 = 12; +/// Track of the switch: how wide and how high, points. +const TRACK: Vec2 = vec2(40.0, 22.0); +/// Radius of the switch's knob, points. +const KNOB: f32 = 9.0; +/// Space between a row and the button, points. +const BUTTON_GAP: f32 = 8.0; + +/// The three rows and the primary button. +pub fn section(ui: &mut Ui, tokens: &Tokens, facts: &Facts<'_>, action: &mut Option) { + toggle(ui, tokens, SheetOption::WifiOnly, facts.options.wifi_only, action); + toggle(ui, tokens, SheetOption::Attachments, facts.options.attachments, action); + toggle(ui, tokens, SheetOption::SettingsAndThemes, facts.options.settings_and_themes, action); + ui.add_space(BUTTON_GAP); + button(ui, tokens, action); +} + +/// The card of one option; `true` when this frame pressed it. +fn toggle(ui: &mut Ui, tokens: &Tokens, option: SheetOption, on: bool, action: &mut Option) { + let colors = &tokens.theme.colors; + let label = match option { + SheetOption::WifiOnly => strings::wifi_only(), + SheetOption::Attachments => strings::attachments(), + SheetOption::SettingsAndThemes => strings::settings_and_themes(), + }; + let (rect, hit) = ui.allocate_exact_size(Vec2::new(ui.available_width(), TOGGLE_ROW), Sense::click()); + hit.widget_info(|| egui::WidgetInfo::selected(egui::WidgetType::Checkbox, true, on, label)); + let glow = crate::motion::hover(ui.ctx(), ui.id().with((SHEET_ID, "toggle", label)), &hit); + ui.painter().rect_filled(rect, CornerRadius::same(u8::MAX / 2), color(colors.selection).gamma_multiply(glow * 0.6)); + ui.painter().text(egui::pos2(rect.left() + ROW_PAD, rect.center().y), Align2::LEFT_CENTER, label, FontId::proportional(15.0), color(colors.text)); + paint_switch(ui, tokens, rect, on); + if hit.clicked() { + *action = Some(SheetAction::Toggle(option, !on)); + } +} + +/// The switch at the right end of a row (design: mauve track, dark knob). +fn paint_switch(ui: &Ui, tokens: &Tokens, row: Rect, on: bool) { + let colors = &tokens.theme.colors; + let track = Rect::from_center_size(egui::pos2(row.right() - ROW_PAD - TRACK.x / 2.0, row.center().y), TRACK); + ui.painter().rect_filled(track, CornerRadius::same(u8::MAX / 2), if on { color(colors.accent) } else { color(colors.surface_2) }); + let knob_x = if on { track.right() - KNOB } else { track.left() + KNOB }; + ui.painter().circle_filled(egui::pos2(knob_x, track.center().y), KNOB, color(colors.text_muted)); +} + +/// The Sync now button, full width, the accent behind the primary word. +fn button(ui: &mut Ui, tokens: &Tokens, action: &mut Option) { + let colors = &tokens.theme.colors; + let (rect, hit) = ui.allocate_exact_size(Vec2::new(ui.available_width(), BUTTON_ROW), Sense::click()); + hit.widget_info(|| egui::WidgetInfo::labeled(egui::WidgetType::Button, true, strings::sync_now())); + let glow = crate::motion::hover(ui.ctx(), ui.id().with((SHEET_ID, "sync-now")), &hit); + let base = crate::motion::mix(color(colors.accent), color(colors.text), glow * 0.12); + ui.painter().rect_filled(rect, CornerRadius::same(BUTTON_RADIUS), base); + ui.painter().text(rect.center(), Align2::CENTER_CENTER, strings::sync_now(), FontId::new(16.0, crate::fonts::semibold()), color(colors.text)); + if hit.clicked() { + *action = Some(SheetAction::SyncNow); + } +} + +#[cfg(test)] +mod tests { + use super::*; + + use notescraft_engine::SyncSettings; + + #[test] + fn the_toggles_target_the_settings_keys() { + assert_eq!(SheetOption::WifiOnly.key(), "sync.wifi_only"); + assert_eq!(SheetOption::Attachments.key(), "sync.attachments"); + assert_eq!(SheetOption::SettingsAndThemes.key(), "sync.settings_and_themes"); + } + + #[test] + fn the_sync_settings_carry_their_defaults() { + assert_eq!(SyncSettings::default(), SyncSettings { wifi_only: true, attachments: true, settings_and_themes: false }); + } +} diff --git a/crates/ui-egui/src/shell/app.rs b/crates/ui-egui/src/shell/app.rs index 52e1cbe..06f7882 100644 --- a/crates/ui-egui/src/shell/app.rs +++ b/crates/ui-egui/src/shell/app.rs @@ -106,6 +106,8 @@ pub struct NotesApp { pub quit_prompt: Option, /// The Share dialog and its worker result, while it is open. pub share_dialog: Option, + /// The sync bottom sheet of the compact layout, while it is open. + pub sync_sheet: Option, /// PDF export dialog and its worker result receiver. pub export_dialog: Option, pub(crate) export_result: Option>, @@ -263,6 +265,7 @@ impl NotesApp { quit_prompt: None, export_dialog: None, share_dialog: None, + sync_sheet: None, export_result: None, export_open_path: None, right_tab: RightTab::default(), diff --git a/crates/ui-egui/src/shell/app/sync.rs b/crates/ui-egui/src/shell/app/sync.rs index c8a06a7..454b809 100644 --- a/crates/ui-egui/src/shell/app/sync.rs +++ b/crates/ui-egui/src/shell/app/sync.rs @@ -24,7 +24,7 @@ impl NotesApp { if self.sync_next.is_none_or(|next| now >= next) { self.sync_next = Some(now + STATUS_PERIOD); self.ensure_sync_worker(); - if self.tabs.active_path() == Some(crate::tabs::SETTINGS_PATH) { + if self.tabs.active_path() == Some(crate::tabs::SETTINGS_PATH) || self.sync_sheet.is_some() { self.refresh_sync_status(); if self.sync_conflicts_next.is_none_or(|next| now >= next) { self.sync_conflicts_next = Some(now + CONFLICTS_PERIOD); diff --git a/crates/ui-egui/src/shell/app/window/frame.rs b/crates/ui-egui/src/shell/app/window/frame.rs index 0756e6b..2bd5838 100644 --- a/crates/ui-egui/src/shell/app/window/frame.rs +++ b/crates/ui-egui/src/shell/app/window/frame.rs @@ -7,6 +7,7 @@ use egui::Key; use serde_json::{Map, Value}; +use crate::panels::sync_sheet::Facts; use notescraft_engine::CommandInfo; use crate::app::NotesApp; @@ -236,9 +237,17 @@ impl NotesApp { let edited = response.edited; let now = ctx.input(|i| i.time); self.handle_tree(tree_response, ctx); + // The sync pill asks for the sheet at phone width and for Settings > Sync + // otherwise; the request is a same-frame egui temp the pill wrote on click. + let sheet_requested = Self::take_sync_sheet_request(ctx); if let Some(category) = response.settings_navigated { self.cue(crate::sounds::Cue::Tick); - self.open_sync_page_if(category); + if !(sheet_requested && category == crate::views::settings_view::SettingsCategory::Sync) { + self.open_sync_page_if(category); + } + } + if sheet_requested { + self.sync_sheet = Some(crate::panels::sync_sheet::SyncSheet::open(&self.session)); } self.apply_settings_changes(response.settings_changes.clone(), ctx); self.apply_settings_actions(std::mem::take(&mut response.settings_actions)); @@ -359,6 +368,37 @@ impl NotesApp { self.open_sync_page_if(crate::views::settings_view::SettingsCategory::Sync); } } + // The sync sheet over everything; the shell stores a toggle and answers the rows. + let sheet_action = if self.sync_sheet.is_some() { + let facts = Facts { + status: self.session.sync_worker().map(notescraft_engine::SyncWorker::status), + conflicts: self.preferences.view.sync.data.conflicts.len(), + options: &self.preferences.settings.sync, + now_ms: crate::panels::sync_sheet::now_millis(), + compact: crate::style::compact::is_compact(ctx), + }; + let action = self.sync_sheet.as_mut().and_then(|sheet| sheet.show(ctx, &self.tokens, &facts)); + let click = ctx.input(|input| input.pointer.interact_pos()); + let rect = ctx.read_response(egui::Id::new("notescraft-sync-sheet")).map(|response| response.rect); + println!("DIAG sheet action {action:?} pointer {click:?} rect {rect:?}"); + action + } else { + None + }; + if let Some(action) = sheet_action { + match action { + crate::panels::sync_sheet::SheetAction::Close => self.sync_sheet = None, + crate::panels::sync_sheet::SheetAction::SyncNow => self.run_command("sync.now", &Value::Null), + crate::panels::sync_sheet::SheetAction::ReviewConflicts => { + self.sync_sheet = None; + self.open_sync_page_if(crate::views::settings_view::SettingsCategory::Sync); + } + crate::panels::sync_sheet::SheetAction::Toggle(option, on) => { + let params = serde_json::json!({ "key": option.key(), "value": on }); + self.run_command("settings.set", ¶ms); + } + } + } crate::dialog::paint_closing(ctx, &self.tokens); if self.export_open_path.is_some() { egui::Area::new(egui::Id::new("export-open-action")).anchor(egui::Align2::RIGHT_BOTTOM, egui::vec2(-24.0, -24.0)).show(ctx, |ui| { @@ -373,6 +413,17 @@ impl NotesApp { self.flush_cues(now); } + /// Whether the compact sync pill asked for the sheet this frame; the request + /// is read and cleared, so one click opens one sheet. + fn take_sync_sheet_request(ctx: &egui::Context) -> bool { + let id = egui::Id::new("sync-sheet-request"); + ctx.data_mut(|data| { + let asked = data.get_temp::(id).unwrap_or(false); + data.remove::(id); + asked + }) + } + /// Runs a chosen engine/UI command; commands with required parameters toast instead. /// `ui.save` is intercepted: saving needs the session and the tab state. pub(in crate::shell::app) fn run_palette_choice(&mut self, id: &str, ctx: &egui::Context) { diff --git a/crates/ui-egui/src/text.rs b/crates/ui-egui/src/text.rs index 2966175..38ee392 100644 --- a/crates/ui-egui/src/text.rs +++ b/crates/ui-egui/src/text.rs @@ -32,7 +32,7 @@ fn lookup(key: &'static str) -> &'static str { pub mod app; mod features; -pub use features::{compact, launcher, live}; +pub use features::{compact, launcher, live, sync_sheet}; mod panels; mod settings; mod views; diff --git a/crates/ui-egui/src/text/features/mod.rs b/crates/ui-egui/src/text/features/mod.rs index 0397262..e74c08f 100644 --- a/crates/ui-egui/src/text/features/mod.rs +++ b/crates/ui-egui/src/text/features/mod.rs @@ -1,5 +1,6 @@ -//! Copy for the newer feature screens: compact layout, vault launcher, live editing. +//! Copy for the newer feature screens: compact layout, vault launcher, live editing and the sync sheet. pub mod compact; pub mod launcher; pub mod live; +pub mod sync_sheet; diff --git a/crates/ui-egui/src/text/features/sync_sheet.rs b/crates/ui-egui/src/text/features/sync_sheet.rs new file mode 100644 index 0000000..6305b30 --- /dev/null +++ b/crates/ui-egui/src/text/features/sync_sheet.rs @@ -0,0 +1,28 @@ +//! Copy for the sync sheet (design `03-sync-sheet`): status, devices, the +//! conflict row, the three options and Sync now. + +table! { + title = "ui.sync-sheet.title"; + status_up_to_date = "ui.sync-sheet.status-up-to-date"; + status_syncing = "ui.sync-sheet.status-syncing"; + status_offline = "ui.sync-sheet.status-offline"; + status_sign_in = "ui.sync-sheet.status-sign-in"; + status_error = "ui.sync-sheet.status-error"; + last_sync = "ui.sync-sheet.last-sync"; + never_synced = "ui.sync-sheet.never-synced"; + devices = "ui.sync-sheet.devices"; + this_phone = "ui.sync-sheet.this-phone"; + this_computer = "ui.sync-sheet.this-computer"; + last_seen = "ui.sync-sheet.last-seen"; + never_seen = "ui.sync-sheet.never-seen"; + chip_online = "ui.sync-sheet.chip-online"; + chip_synced = "ui.sync-sheet.chip-synced"; + chip_stale = "ui.sync-sheet.chip-stale"; + conflicts = "ui.sync-sheet.conflicts"; + wifi_only = "ui.sync-sheet.wifi-only"; + attachments = "ui.sync-sheet.attachments"; + settings_and_themes = "ui.sync-sheet.settings-and-themes"; + sync_now = "ui.sync-sheet.sync-now"; + loading = "ui.sync-sheet.loading"; + pairing_failed = "ui.sync-sheet.pairing-failed"; +} diff --git a/crates/ui-egui/tests/sync_sheet.rs b/crates/ui-egui/tests/sync_sheet.rs new file mode 100644 index 0000000..fa37c9a --- /dev/null +++ b/crates/ui-egui/tests/sync_sheet.rs @@ -0,0 +1,430 @@ +//! The sync sheet driven the way a user does: real frames at 390 x 844 over a +//! vault bound to a stub server (design `03-sync-sheet`). + +use egui::{Context, Event, Key, PointerButton, Pos2, RawInput}; +use notescraft_engine::{Session, Timings, bind_folder}; +use notescraft_ui_egui::app::NotesApp; +use std::io::{BufRead, BufReader, Read, Write}; +use std::net::{TcpListener, TcpStream}; +use std::path::Path; +use std::sync::atomic::{AtomicBool, AtomicU64, Ordering}; +use std::sync::{Arc, Mutex}; +use std::time::{Duration, Instant, SystemTime, UNIX_EPOCH}; + +static NEXT: AtomicU64 = AtomicU64::new(0); +/// The window of a phone. +const PHONE: egui::Vec2 = egui::vec2(390.0, 844.0); +/// The window of a desktop. +const DESKTOP: egui::Vec2 = egui::vec2(1280.0, 800.0); +/// The token of the bound vault; it must never reach the screen. +const TOKEN: &str = "secret-token"; +/// Longest the tests wait for a stub request or an idle worker. +const WAIT: Duration = Duration::from_secs(5); + +/// One request the stub received. +#[derive(Debug, Clone)] +struct Seen { + /// The HTTP method. + method: String, + /// The path of the request line, without the query. + path: String, +} + +/// A stub server: canned answers per path and method, and every request it saw. +struct Stub { + base: String, + seen: Arc>>, + stop: Arc, +} + +impl Stub { + /// Starts a stub on an ephemeral port. `devices` is the body of the device + /// list; `/changes` answers an empty log and `/files/a.md` answers applied, + /// so the worker's cycle lands on "Synced". + fn spawn(devices: &str) -> Self { + let listener = TcpListener::bind("127.0.0.1:0").expect("bind the stub"); + let port = listener.local_addr().expect("addr").port(); + let record = r#"{"path":"a.md","hash":"x","size":2,"mtime_ms":0,"seq":1,"deleted":false}"#; + let empty = r#"{"head":0,"records":[],"more":false}"#; + let answer = |status: &str, body: &str| format!("HTTP/1.1 {status}\r\nContent-Type: application/json\r\nContent-Length: {}\r\n\r\n{body}", body.len()); + let answers: Vec<(String, String, String)> = vec![ + ("GET".to_owned(), "/changes".to_owned(), answer("200 OK", empty)), + ("PUT".to_owned(), "/files/a.md".to_owned(), answer("201 Created", record)), + ("GET".to_owned(), "/devices".to_owned(), answer("200 OK", devices)), + ]; + let seen: Arc>> = Arc::new(Mutex::new(Vec::new())); + let stop = Arc::new(AtomicBool::new(false)); + let (thread_seen, thread_stop) = (Arc::clone(&seen), Arc::clone(&stop)); + std::thread::spawn(move || { + while !thread_stop.load(Ordering::Relaxed) { + match listener.accept() { + Ok((stream, _)) => { + // One thread per connection: the client may keep the connection + // open, and the listener must go on accepting. + let Ok(mut writer) = stream.try_clone() else { continue }; + let mut reader = BufReader::new(stream); + let answers = Arc::new(answers.clone()); + let seen = Arc::clone(&thread_seen); + std::thread::spawn(move || { + while let Some(request) = read_request(&mut reader) { + seen.lock().expect("seen").push(request.clone()); + let hit = answers.iter().find(|(method, path, _)| *method == request.method && request.path.contains(path.as_str())); + let canned = hit.map_or("HTTP/1.1 404 Not Found\r\nContent-Length: 0\r\n\r\n", |(_, _, canned)| canned.as_str()); + let _ = writer.write_all(canned.as_bytes()); + let _ = writer.flush(); + } + }); + } + Err(_) => std::thread::sleep(Duration::from_millis(2)), + } + } + }); + Self { base: format!("http://127.0.0.1:{port}"), seen, stop } + } + + /// How many times the change log was polled. + fn change_polls(&self) -> u32 { + self.seen.lock().expect("seen").iter().filter(|seen| seen.method == "GET" && seen.path.ends_with("/changes")).count().try_into().unwrap_or(u32::MAX) + } + + /// The requests the stub saw. + fn requests(&self) -> Vec { + self.seen.lock().expect("seen").clone() + } + + /// Whether the stub saw a write to `path`. + fn uploaded(&self, path: &str) -> bool { + self.requests().iter().any(|seen| seen.method == "PUT" && seen.path.ends_with(path)) + } + + /// Waits, bounded, until `wanted` holds. + fn wait_for(&self, wanted: impl Fn(&[Seen]) -> bool) { + let deadline = Instant::now() + WAIT; + while Instant::now() < deadline && !wanted(&self.requests()) { + std::thread::sleep(Duration::from_millis(5)); + } + } +} + +impl Drop for Stub { + fn drop(&mut self) { + self.stop.store(true, Ordering::Relaxed); + } +} + +/// Reads one request: method, path, and its body skipped; the request ends when +/// the client closes the connection. +fn read_request(reader: &mut BufReader) -> Option { + let mut line = String::new(); + reader.read_line(&mut line).ok()?; + let mut parts = line.split_whitespace(); + let method = parts.next()?.to_owned(); + let path = parts.next()?.to_owned(); + let mut length = 0_usize; + loop { + let mut header = String::new(); + if reader.read_line(&mut header).ok()? == 0 { + break; + } + if header.trim_end().is_empty() { + break; + } + if let Some((name, value)) = header.trim_end().split_once(':') + && name.trim().eq_ignore_ascii_case("content-length") + { + length = value.trim().parse().unwrap_or(0); + } + } + let mut body = vec![0_u8; length]; + if length > 0 { + reader.read_exact(&mut body).ok()?; + } + Some(Seen { method, path: path.split('?').next().unwrap_or_default().to_owned() }) +} + +struct Driver { + app: NotesApp, + ctx: Context, + time: f64, + size: egui::Vec2, +} + +/// A painted text: what it says, where it starts and its galley. +#[derive(Debug)] +struct Painted { + text: String, + pos: Pos2, + galley: Arc, +} + +/// An app over a fresh vault bound to `stub`, with a note open; the settings +/// dir lives beside the vault, never inside it (the vault is the sync scope). +fn driven(stub: &Stub, files: &[(&str, &str)], size: egui::Vec2) -> Driver { + let root = std::env::temp_dir().join(format!("notescraft-sheet-{}-{}", std::process::id(), NEXT.fetch_add(1, Ordering::Relaxed))); + let _ = std::fs::remove_dir_all(&root); + std::fs::create_dir_all(&root).expect("root"); + seed(&root, files); + bind_folder(&root, &stub.base, "v-1", TOKEN, "d-1", Some("u-1")).expect("bind the vault"); + app_for(&root, &settings_dir(&root), size) +} + +/// Writes `(path, content)` into `root`, creating the folders. +fn seed(root: &Path, files: &[(&str, &str)]) { + for (path, text) in files { + if let Some((folder, _)) = path.rsplit_once('/') { + std::fs::create_dir_all(root.join(folder)).expect("folder"); + } + std::fs::write(root.join(path), text).expect("seed"); + } +} + +/// The settings dir beside `root`; the vault never holds the settings (SPEC 4.1). +fn settings_dir(root: &Path) -> std::path::PathBuf { + root.parent().unwrap_or(root).join("notescraft-sheet-config") +} + +/// An app over an already seeded and bound vault at `root`. +fn app_for(root: &Path, config: &Path, size: egui::Vec2) -> Driver { + let mut session = Session::open(root).expect("session"); + session.use_settings_dir(config); + let mut app = NotesApp::new(session, notescraft_themes::builtin("catppuccin-mocha").expect("theme")); + app.state.animations = false; + app.open_note("a.md"); + Driver { app, ctx: Context::default(), time: 1.0, size } +} + +impl Driver { + /// One frame with `events`; the texts it painted. + fn frame(&mut self, events: Vec) -> Vec { + self.time += 0.05; + let input = RawInput { screen_rect: Some(egui::Rect::from_min_size(Pos2::ZERO, self.size)), time: Some(self.time), events, ..Default::default() }; + let app = &mut self.app; + let mut out = self.ctx.run_ui(input, |ui| app.ui(ui.ctx())); + out.textures_delta.clear(); + out.shapes + .iter() + .filter_map(|shape| match &shape.shape { + egui::Shape::Text(text) => Some(Painted { text: text.galley.text().to_owned(), pos: text.pos, galley: text.galley.clone() }), + _ => None, + }) + .collect() + } + + fn settle(&mut self) -> Vec { + let mut last = Vec::new(); + for _ in 0..24 { + last = self.frame(Vec::new()); + } + last + } + + fn click(&mut self, at: Pos2) -> Vec { + self.frame(vec![Event::PointerMoved(at)]); + self.frame(vec![Event::PointerButton { pos: at, button: PointerButton::Primary, pressed: true, modifiers: egui::Modifiers::NONE }]); + self.frame(vec![Event::PointerButton { pos: at, button: PointerButton::Primary, pressed: false, modifiers: egui::Modifiers::NONE }]); + self.settle() + } + + /// The centre of the pill's painted `Synced` label; the hit area holds it. + fn pill_centre(&mut self) -> Pos2 { + let texts = self.settle(); + let status = self.app.session.sync_worker().map(|worker| format!("{:?}", worker.status())); + let pill = texts + .iter() + .find(|painted| painted.text == "Synced") + .unwrap_or_else(|| panic!("no pill among {:?} -- worker {status:?}", texts.iter().map(|painted| painted.text.as_str()).collect::>())); + pill.pos + pill.galley.size() / 2.0 + } +} + +fn painted<'a>(texts: &'a [Painted], wanted: &str) -> Option<&'a Painted> { + texts.iter().find(|painted| painted.text == wanted) +} + +fn expect<'a>(texts: &'a [Painted], wanted: &str) -> &'a Painted { + painted(texts, wanted).unwrap_or_else(|| panic!("no {wanted:?} among {:?}", texts.iter().map(|painted| painted.text.as_str()).collect::>())) +} + +/// The device list the stub answers: another device last seen an hour ago, +/// this device now. +fn devices_json(now_ms: u64) -> String { + format!( + r#"[{{"id":"d-2","name":"Laptop","created_ms":0,"last_seen_ms":{},"current":false}},{{"id":"d-1","name":"Pixel 9","created_ms":0,"last_seen_ms":{},"current":true}}]"#, + now_ms - 3_600_000, + now_ms + ) +} + +/// The Unix time now, in milliseconds. +fn now_ms() -> u64 { + SystemTime::now().duration_since(UNIX_EPOCH).map_or(0, |since| u64::try_from(since.as_millis()).unwrap_or(u64::MAX)) +} + +/// The worker runs fast: the tests never wait for the real debounce. +fn fast_worker(d: &mut Driver) { + if let Some(worker) = d.app.session.sync_worker() { + worker.set_timings(Timings { + debounce: Duration::from_millis(10), + poll: Duration::from_mins(15), + backoff: Duration::from_millis(10), + backoff_max: Duration::from_secs(1), + }); + } +} + +/// Waits, bounded, until the worker is idle; the cycles land before the frames read it. +fn worker_idle(d: &mut Driver) { + let deadline = Instant::now() + WAIT; + while Instant::now() < deadline && !d.app.session.sync_worker().is_some_and(|worker| worker.status().phase == notescraft_engine::SyncPhase::Idle) { + std::thread::sleep(Duration::from_millis(5)); + } +} + +#[test] +fn the_pill_opens_the_sheet_and_lists_the_current_device_first() { + let stub = Stub::spawn(&devices_json(now_ms())); + let mut d = driven(&stub, &[("a.md", "A")], PHONE); + d.settle(); + fast_worker(&mut d); + stub.wait_for(|requests| requests.iter().any(|seen| seen.method == "PUT" && seen.path.ends_with("a.md"))); + worker_idle(&mut d); + + let from = d.pill_centre(); + let texts = d.click(from); + assert!(d.app.sync_sheet.is_some(), "the pill opened the sheet"); + let current = expect(&texts, "This phone"); + let laptop = expect(&texts, "Laptop"); + assert!(current.pos.y < laptop.pos.y, "the current device first: {current:?} vs {laptop:?}"); + assert!( + texts.iter().any(|painted| painted.text.contains("last seen")), + "the ages show: {:?}", + texts.iter().map(|painted| painted.text.as_str()).collect::>() + ); + assert!(!texts.iter().any(|painted| painted.text.contains(TOKEN)), "the token never reaches the screen"); + assert!(painted(&texts, "Everything is up to date").is_some(), "the status headline shows"); + assert!(painted(&texts, "Sync now").is_some(), "the primary button shows"); +} + +#[test] +fn the_conflicts_row_appears_only_over_a_conflict_file_on_disk() { + let stub = Stub::spawn(&devices_json(now_ms())); + let mut with = driven(&stub, &[("a.md", "A"), ("a (conflict Laptop 2026-10-20).md", "A")], PHONE); + with.settle(); + let from = with.pill_centre(); + with.click(from); + let over = with.settle(); + assert!(painted(&over, "1 conflicts to review").is_some(), "the row shows"); + drop(with); + + let mut none_driver = driven(&stub, &[("a.md", "A")], PHONE); + none_driver.settle(); + let from = none_driver.pill_centre(); + none_driver.click(from); + let none = none_driver.settle(); + assert!(painted(&none, "conflicts to review").is_none(), "nothing to review: {:?}", none.iter().map(|painted| painted.text.as_str()).collect::>()); + let _ = none_driver.app.sync_sheet.take(); +} + +#[test] +fn toggling_attachments_persists_and_the_first_cycle_skips_the_png() { + let stub = Stub::spawn(&devices_json(now_ms())); + let mut d = driven(&stub, &[("a.md", "A"), ("img/scan.png", "bytes")], PHONE); + d.settle(); + worker_idle(&mut d); + let from = d.pill_centre(); + d.click(from); + let texts = d.settle(); + let label = expect(&texts, "Sync attachments"); + let hit = label.pos + label.galley.size() / 2.0; + d.click(hit); + // The stored option goes into the settings file the next load reads; the + // vault root and its settings dir outlive the app they were opened over. + let root = d.app.session.vault().root().to_path_buf(); + let config = settings_dir(&root); + let read = |text: &str| text.lines().filter(|line| line.contains("attachments") || line.contains("[sync]")).collect::>().join(", "); + let stored = std::fs::read_to_string(config.join("config.toml")).unwrap_or_default(); + println!("DIAG toml: {:?}", read(&stored)); + assert!(stored.contains("attachments = false"), "the toggle stored: {:?}", read(&stored)); + drop(d); + + let mut again = app_for(&root, &config, PHONE); + again.settle(); + fast_worker(&mut again); + worker_idle(&mut again); + assert!(!stub.uploaded("scan.png"), "the png never left: {}", stub_uploaded_list(&stub)); + assert!(again.app.session.sync_worker().is_some_and(|worker| worker.status().phase == notescraft_engine::SyncPhase::Idle), "the cycle ran"); + drop(again); + let _ = std::fs::remove_dir_all(&root); + let _ = std::fs::remove_dir_all(&config); +} + +/// The paths the stub saw written to, for the failure messages. +fn stub_uploaded_list(stub: &Stub) -> String { + stub.requests().iter().map(|seen| format!("{}/{}", seen.method, seen.path)).collect::>().join(", ") +} + +#[test] +fn sync_now_wakes_the_worker() { + let stub = Stub::spawn(&devices_json(now_ms())); + let mut d = driven(&stub, &[("a.md", "A")], PHONE); + d.settle(); + fast_worker(&mut d); + stub.wait_for(|requests| requests.iter().any(|seen| seen.method == "PUT" && seen.path.ends_with("a.md"))); + worker_idle(&mut d); + let before = stub.change_polls(); + let from = d.pill_centre(); + d.click(from); + let texts = d.settle(); + let label = expect(&texts, "Sync now"); + let hit = label.pos + label.galley.size() / 2.0; + d.click(hit); + stub.wait_for(|requests| requests.iter().filter(|seen| seen.method == "GET" && seen.path.ends_with("/changes")).count() > before as usize); + assert!(stub.change_polls() > before, "the change log was polled again"); + assert!(d.app.sync_sheet.is_some(), "Sync now keeps the sheet open"); +} + +#[test] +fn the_sheet_closes_with_x_and_escape_and_a_tap_outside() { + let stub = Stub::spawn(&devices_json(now_ms())); + let mut d = driven(&stub, &[("a.md", "A")], PHONE); + d.settle(); + let from = d.pill_centre(); + d.click(from); + assert!(d.app.sync_sheet.is_some(), "open"); + d.settle(); + // The × sits in the title row, in the top-right of the sheet. + let sheet = d.ctx.read_response(egui::Id::new("notescraft-sync-sheet")).expect("the sheet area"); + let rect = sheet.rect; + d.click(egui::pos2(rect.right() - 16.0 - 22.0, rect.top() + 16.0 + 4.0 + 12.0 + 22.0)); + assert!(d.app.sync_sheet.is_none(), "the × closed the sheet"); + + let again_from = d.pill_centre(); + d.click(again_from); + assert!(d.app.sync_sheet.is_some(), "open again"); + let escape = Event::Key { key: Key::Escape, physical_key: None, pressed: true, repeat: false, modifiers: egui::Modifiers::NONE }; + d.frame(vec![escape]); + d.settle(); + assert!(d.app.sync_sheet.is_none(), "Escape closed the sheet"); + + let third_from = d.pill_centre(); + d.click(third_from); + assert!(d.app.sync_sheet.is_some(), "open again for the tap outside"); + d.click(egui::pos2(20.0, 20.0)); + assert!(d.app.sync_sheet.is_none(), "a tap on the dimmed area closed the sheet"); +} + +#[test] +fn at_desktop_width_the_pill_opens_the_settings_page() { + let stub = Stub::spawn(&devices_json(now_ms())); + let mut d = driven(&stub, &[("a.md", "A")], DESKTOP); + d.settle(); + let from = d.pill_centre(); + let texts = d.click(from); + assert!(d.app.sync_sheet.is_none(), "no sheet at desktop width"); + assert_eq!(d.app.tabs.active_path(), Some(notescraft_ui_egui::tabs::SETTINGS_PATH), "the settings page opened"); + assert!( + texts.iter().any(|painted| painted.text == "Server") && texts.iter().any(|painted| painted.text == "Sync now"), + "the Sync page shows: {:?}", + texts.iter().map(|painted| painted.text.as_str()).collect::>() + ); +} diff --git a/specs/attachments-and-history/spec.md b/specs/attachments-and-history/spec.md index 31d7a2b..a7a3185 100644 --- a/specs/attachments-and-history/spec.md +++ b/specs/attachments-and-history/spec.md @@ -30,7 +30,7 @@ Real-time delta sync of Markdown (files are small), cross-vault dedup in encrypt 2. Content-defined chunking (rolling hash) instead of fixed 4 MiB? Recommended: fixed first; revisit if edits to big files are common. ## Tasks -1. Server: `chunks` table and endpoints, assembled-file reads, quota accounting. +1. [x] Server: `chunks` table and endpoints, assembled-file reads, quota accounting. Done 2026-10-20 by GLM 5.3 Flash via Pi. 2. [x] Server: `file_revisions`, history, restore, retention GC. Done 2026-10-20 by GLM 5.3 Flash via Pi (MiniMax was out of quota): `src/history/`, `store/revisions.rs`, endpoints `files-history`, `files/{path}?seq=`, `files-restore`, per-vault `retention_days` (default 30, newest 20 always kept), quota counts retained revisions; verified by the orchestrator (all server suites green, clippy clean). Engine and UI still open (task 4). -3. Engine: chunked upload/download in the driver with resume and progress events. +3. [x] Engine: chunked upload in the driver with resume and progress events (`sync/chunks.rs`, `Remote::{missing_chunks, put_chunk, put_chunked}`, `SyncStatus.transfer`). Done 2026-10-20 by MiniMax M3.1 Flash through the Claude Code harness; verified against the real server. Big-file download still goes through the existing `get` (capped by what fits in memory). 4. UI: transfer progress in the status pill and Settings > Sync; Versions tab shows server revisions. diff --git a/specs/mobile-and-sync/tasks.md b/specs/mobile-and-sync/tasks.md index 5defecb..59b7b14 100644 --- a/specs/mobile-and-sync/tasks.md +++ b/specs/mobile-and-sync/tasks.md @@ -53,7 +53,7 @@ Rules: each task is independently testable; order is execution order. `[x]` done - [ ] 25. Quick capture sheet and share-target receive (`note.new_from_share` command). - [ ] 26. Quick switcher and command palette on phone, over the engine registry. - [ ] 27. Background sync via WorkManager and `BGTaskScheduler` (the worker from task 5 runs headless). -- [ ] 28. Conflict resolution UI plus an offline queue view on phone (builds on task 6). +- [~] 28. Conflict resolution UI plus an offline queue view on phone (builds on task 6). The phone sync sheet with devices, conflicts row, options and Sync now is done (2026-10-20, GLM 5.3 Flash, 6 driven tests); the offline queue view is open. - [ ] 29. Mobile toolbar above the keyboard and long-press context menus (design 02). ## Related spec sets