diff --git a/desktop/src-tauri/src/lib.rs b/desktop/src-tauri/src/lib.rs index dfe1352..3b176cc 100644 --- a/desktop/src-tauri/src/lib.rs +++ b/desktop/src-tauri/src/lib.rs @@ -93,6 +93,7 @@ pub fn run() { sync::commands::sync_link, sync::commands::sync_unlink, sync::commands::sync_status, + sync::commands::sync_pull, ]) .run(tauri::generate_context!()) .expect("error while running the ThoughtSync desktop app"); diff --git a/desktop/src-tauri/src/sync/client.rs b/desktop/src-tauri/src/sync/client.rs index b75fc6c..bf68a84 100644 --- a/desktop/src-tauri/src/sync/client.rs +++ b/desktop/src-tauri/src/sync/client.rs @@ -14,6 +14,7 @@ use reqwest::{RequestBuilder, StatusCode}; use serde::{Deserialize, Serialize}; use super::compat::{self, Compatibility, ServerInfo}; +use super::wire; /// Timeout for the short request/response calls in this module. Kept tight because a /// user is watching a button while they run, and the most common mistake — a wrong @@ -21,6 +22,15 @@ use super::compat::{self, Compatibility, ServerInfo}; /// just look frozen. The sync engine's bulk transfers will need their own, longer one. const REQUEST_TIMEOUT: Duration = Duration::from_secs(10); +/// Bulk transfers get much longer: a first full sync can be thousands of notes, and +/// failing one at ten seconds would make a large store impossible to ever pull. +const SYNC_TIMEOUT: Duration = Duration::from_secs(120); + +/// Shared by every call that presents a token, so a revoked one reads the same way +/// wherever it surfaces. +const TOKEN_REJECTED: &str = "This server rejected the device token — it may have been \ + revoked. Unlink and link again to issue a new one."; + /// What the link UI needs after a handshake: where we ended up (the normalized URL, /// which may differ from what was typed), who answered, and whether we can work /// with them. @@ -48,13 +58,17 @@ struct DeviceLoginResponse { user: Identity, } -fn http() -> Result { +fn http_with(timeout: Duration) -> Result { reqwest::Client::builder() - .timeout(REQUEST_TIMEOUT) + .timeout(timeout) .build() .map_err(|e| format!("Could not start the network client: {e}")) } +fn http() -> Result { + http_with(REQUEST_TIMEOUT) +} + /// Attach the client-identity headers every request carries, plus a bearer token /// when we hold one. fn prepare(builder: RequestBuilder, token: Option<&str>) -> RequestBuilder { @@ -179,6 +193,36 @@ pub async fn fetch_identity(base_url: &str, token: &str) -> Result Result { + let url = format!("{base_url}/api/sync/changes?since={since}"); + let request = prepare(http_with(SYNC_TIMEOUT)?.get(url), Some(token)); + let response = request + .send() + .await + .map_err(|e| describe_transport_error(base_url, &e))?; + + let status = response.status(); + if status == StatusCode::UNAUTHORIZED { + return Err(TOKEN_REJECTED.to_string()); + } + if !status.is_success() { + return Err(unexpected_status(base_url, status)); + } + + response + .json() + .await + .map_err(|e| format!("Couldn't read the change feed from {base_url}: {e}")) +} + /// The public, unauthenticated endpoint carrying the handshake. fn config_url(base_url: &str) -> String { format!("{base_url}/api/config") @@ -196,10 +240,12 @@ fn me_url(base_url: &str) -> String { /// Display is accurate but reads like a stack trace. fn describe_transport_error(base_url: &str, err: &reqwest::Error) -> String { if err.is_timeout() { + // No specific duration here: these calls run under two different budgets + // (interactive vs bulk sync), and naming the wrong one is worse than naming + // none. format!( - "{base_url} didn't respond within {} seconds. It may be offline, or \ - unreachable from this network.", - REQUEST_TIMEOUT.as_secs() + "{base_url} didn't respond in time. It may be offline, or unreachable \ + from this network." ) } else if err.is_connect() { format!( diff --git a/desktop/src-tauri/src/sync/commands.rs b/desktop/src-tauri/src/sync/commands.rs index 1d7a0c9..b7cbf2f 100644 --- a/desktop/src-tauri/src/sync/commands.rs +++ b/desktop/src-tauri/src/sync/commands.rs @@ -9,6 +9,7 @@ use tauri::State; use crate::local::Db; use crate::sync::client::{self, Identity, ProbeResult}; use crate::sync::compat::Compatibility; +use crate::sync::pull; use crate::sync::state; /// Ask a server who it is, without committing to anything. The UI calls this as the @@ -122,3 +123,26 @@ pub fn sync_status(db: State<'_, Db>) -> Result { let conn = db.0.lock().map_err(|e| e.to_string())?; state::status(&conn).map_err(|e| e.to_string()) } + +/// The server URL + token, or a plain "not linked" error. Every networked sync +/// command needs exactly this, and none of them may hold the lock past it. +fn credentials(db: &State<'_, Db>) -> Result<(String, String), String> { + let conn = db.0.lock().map_err(|e| e.to_string())?; + let current = state::read(&conn).map_err(|e| e.to_string())?; + match (current.server_url, current.device_token) { + (Some(url), Some(token)) => Ok((url, token)), + _ => Err("This app isn't linked to a server yet.".to_string()), + } +} + +/// Pull the server's changes into the local store. +/// +/// Standalone for now; M10.7c wraps push-then-pull into a single `sync_now`, which +/// is the ordering the protocol assumes. Run alone against unpushed local edits, the +/// server's version lands on top of them — the returned `clobbered_dirty` count +/// reports that rather than hiding it. +#[tauri::command] +pub async fn sync_pull(db: State<'_, Db>) -> Result { + let (base_url, token) = credentials(&db)?; + pull::run(db.inner(), &base_url, &token).await +} diff --git a/desktop/src-tauri/src/sync/mod.rs b/desktop/src-tauri/src/sync/mod.rs index a7d403b..b95d473 100644 --- a/desktop/src-tauri/src/sync/mod.rs +++ b/desktop/src-tauri/src/sync/mod.rs @@ -15,4 +15,6 @@ pub mod client; pub mod commands; pub mod compat; +pub mod pull; pub mod state; +pub mod wire; diff --git a/desktop/src-tauri/src/sync/pull.rs b/desktop/src-tauri/src/sync/pull.rs new file mode 100644 index 0000000..f6c026a --- /dev/null +++ b/desktop/src-tauri/src/sync/pull.rs @@ -0,0 +1,699 @@ +//! Pull: bring a server's changes into the local store (M10.7b). +//! +//! The feed is a single monotonic sequence shared by notes and labels, so one +//! integer cursor is a total-order watermark over both (docs/sync.md). We loop pages +//! until the server says there are no more, persisting the cursor **in the same +//! transaction** as the page it describes — a cursor committed ahead of its data +//! would silently skip those rows forever, which reads as a clean sync. + +use chrono::{SecondsFormat, Utc}; +use rusqlite::{params, Connection, OptionalExtension}; +use serde::Serialize; + +use super::client; +use super::state; +use super::wire; +use crate::local::Db; + +/// Backstop against a server that never stops saying `has_more`. At the server's +/// 1000-row page cap this is 10M rows — far past any real store, so hitting it means +/// something is wrong, not that someone has a lot of notes. +const MAX_PAGES: usize = 10_000; + +/// What a pull did — for the UI, and for the log when something looks off. +#[derive(Debug, Clone, Default, Serialize, PartialEq, Eq)] +pub struct PullSummary { + pub pages: usize, + pub notes_applied: usize, + pub notes_deleted: usize, + pub labels_applied: usize, + pub labels_deleted: usize, + pub cursor: i64, + /// Rows that still held unpushed local edits when the server's version landed on + /// top. Should be 0 in the normal cycle, because push runs first; anything higher + /// means local work was overwritten, which is worth saying out loud. + pub clobbered_dirty: usize, +} + +impl PullSummary { + fn absorb(&mut self, other: PullSummary) { + self.pages += other.pages; + self.notes_applied += other.notes_applied; + self.notes_deleted += other.notes_deleted; + self.labels_applied += other.labels_applied; + self.labels_deleted += other.labels_deleted; + self.clobbered_dirty += other.clobbered_dirty; + self.cursor = other.cursor; + } +} + +fn now() -> String { + Utc::now().to_rfc3339_opts(SecondsFormat::Millis, true) +} + +/// Apply one page and advance the cursor, atomically. +/// +/// Labels are applied before notes so a membership never references a label row that +/// doesn't exist yet. +pub fn apply_page(conn: &Connection, page: &wire::ChangesPage) -> rusqlite::Result { + let tx = conn.unchecked_transaction()?; + let mut summary = PullSummary { + pages: 1, + cursor: page.cursor, + ..Default::default() + }; + + for label in &page.labels { + if label.is_tombstone() { + tx.execute("DELETE FROM labels WHERE id = ?1", params![label.id])?; + summary.labels_deleted += 1; + } else { + upsert_label(&tx, label)?; + summary.labels_applied += 1; + } + } + + for note in &page.notes { + if note.is_tombstone() { + // A purge tombstone carries no content — its only job is to say "delete + // your copy". Children go with it via ON DELETE CASCADE. + tx.execute("DELETE FROM notes WHERE id = ?1", params![note.id])?; + summary.notes_deleted += 1; + continue; + } + if is_dirty(&tx, ¬e.id)? { + summary.clobbered_dirty += 1; + } + upsert_note(&tx, note)?; + summary.notes_applied += 1; + } + + state::set_cursor(&tx, page.cursor)?; + tx.commit()?; + Ok(summary) +} + +fn is_dirty(conn: &Connection, note_id: &str) -> rusqlite::Result { + let dirty: Option = conn + .query_row( + "SELECT dirty FROM notes WHERE id = ?1", + params![note_id], + |r| r.get(0), + ) + .optional()?; + Ok(dirty == Some(1)) +} + +fn upsert_label(conn: &Connection, label: &wire::Label) -> rusqlite::Result<()> { + // One label per name is enforced on both sides (locally a UNIQUE index on + // lower(name); on the server, per owner). A label created offline can therefore + // collide with one the server already had under a different id — "work" typed on + // this machine and "work" that already existed. + // + // The server's row wins, but its MEMBERSHIPS have to survive the swap. Just + // deleting the local duplicate would cascade its note_labels away, stripping the + // label off notes that this pull never even mentions — silent loss that no later + // page would repair. So: free the name, insert the server's row, re-point the + // memberships onto it, then drop the husk. + let duplicates: Vec = { + let mut stmt = + conn.prepare("SELECT id FROM labels WHERE lower(name) = lower(?1) AND id <> ?2")?; + let rows = stmt.query_map(params![label.name, label.id], |r| r.get::<_, String>(0))?; + rows.collect::>>()? + }; + // Renaming first is what makes the insert possible at all — the unique index + // would otherwise reject the server's row before anything could be merged. + for old in &duplicates { + conn.execute( + "UPDATE labels SET name = name || ' (superseded ' || id || ')' WHERE id = ?1", + params![old], + )?; + } + + let created = label.created_at.clone().unwrap_or_else(now); + conn.execute( + "INSERT INTO labels (id, name, color, created_at, updated_at, sync_revision, dirty) + VALUES (?1, ?2, ?3, ?4, ?4, ?5, 0) + ON CONFLICT(id) DO UPDATE SET + name = excluded.name, + color = excluded.color, + sync_revision = excluded.sync_revision, + dirty = 0", + params![ + label.id, + label.name, + label.color, + created, + label.sync_revision + ], + )?; + + for old in &duplicates { + // OR IGNORE guards a (note_id, label_id) collision. Today the unique index on + // lower(name) makes that unreachable — two same-name labels can't coexist + // locally — so this is belt-and-braces against that index changing, not a + // case we've seen. Anything it skips cascades away with the husk below, which + // is correct: those are duplicates of a membership that now exists. + conn.execute( + "UPDATE OR IGNORE note_labels SET label_id = ?1 WHERE label_id = ?2", + params![label.id, old], + )?; + conn.execute("DELETE FROM labels WHERE id = ?1", params![old])?; + } + Ok(()) +} + +fn upsert_note(conn: &Connection, note: &wire::Note) -> rusqlite::Result<()> { + let created = note.created_at.clone().unwrap_or_else(now); + let updated = note.updated_at.clone().unwrap_or_else(|| created.clone()); + // `created_at` is deliberately absent from the UPDATE clause: a note's birth time + // never changes, and the server's copy is the same value anyway. + conn.execute( + "INSERT INTO notes (id, title, body, color, kind, position, pinned, archived, + trashed, remind_at, recurrence, created_at, updated_at, + sync_revision, dirty) + VALUES (?1, ?2, ?3, ?4, ?5, ?6, ?7, ?8, ?9, ?10, ?11, ?12, ?13, ?14, 0) + ON CONFLICT(id) DO UPDATE SET + title = excluded.title, + body = excluded.body, + color = excluded.color, + kind = excluded.kind, + position = excluded.position, + pinned = excluded.pinned, + archived = excluded.archived, + trashed = excluded.trashed, + remind_at = excluded.remind_at, + recurrence = excluded.recurrence, + updated_at = excluded.updated_at, + sync_revision = excluded.sync_revision, + dirty = 0", + params![ + note.id, + note.title, + note.body, + note.color, + note.kind, + note.position, + note.pinned, + note.archived, + note.trashed, + note.remind_at, + note.recurrence, + created, + updated, + note.sync_revision, + ], + )?; + + // Children are replaced wholesale: a delta carries the note's FULL current state, + // so "what the server sent" IS the complete set. Diffing would be more code and + // could leave behind a row the server no longer has. + replace_items(conn, note)?; + replace_attachments(conn, note)?; + replace_previews(conn, note)?; + replace_labels(conn, note)?; + Ok(()) +} + +fn replace_items(conn: &Connection, note: &wire::Note) -> rusqlite::Result<()> { + conn.execute( + "DELETE FROM checklist_items WHERE note_id = ?1", + params![note.id], + )?; + for (index, item) in note.items.iter().enumerate() { + conn.execute( + "INSERT INTO checklist_items (id, note_id, text, checked, position) + VALUES (?1, ?2, ?3, ?4, ?5)", + params![ + item.id, + note.id, + item.text, + item.checked, + position_of(item.position, index) + ], + )?; + } + Ok(()) +} + +fn replace_attachments(conn: &Connection, note: &wire::Note) -> rusqlite::Result<()> { + conn.execute( + "DELETE FROM attachments WHERE note_id = ?1", + params![note.id], + )?; + for (index, att) in note.attachments.iter().enumerate() { + // The feed carries no explicit position for attachments — they arrive in + // creation order, so the index preserves it. + conn.execute( + "INSERT INTO attachments (id, note_id, url, filename, mime, size, sha256, position) + VALUES (?1, ?2, ?3, ?4, ?5, ?6, ?7, ?8)", + params![ + att.id, + note.id, + att.url, + att.filename, + att.mime, + att.size, + att.sha256, + index as i64 + ], + )?; + } + Ok(()) +} + +fn replace_previews(conn: &Connection, note: &wire::Note) -> rusqlite::Result<()> { + conn.execute( + "DELETE FROM link_previews WHERE note_id = ?1", + params![note.id], + )?; + for (index, preview) in note.previews.iter().enumerate() { + conn.execute( + "INSERT INTO link_previews (id, note_id, url, title, description, image_url, + site_name, position) + VALUES (?1, ?2, ?3, ?4, ?5, ?6, ?7, ?8)", + params![ + preview.id, + note.id, + preview.url, + preview.title, + preview.description, + preview.image_url, + preview.site_name, + index as i64 + ], + )?; + } + Ok(()) +} + +fn replace_labels(conn: &Connection, note: &wire::Note) -> rusqlite::Result<()> { + conn.execute( + "DELETE FROM note_labels WHERE note_id = ?1", + params![note.id], + )?; + for label in ¬e.labels { + ensure_label_stub(conn, label)?; + // `via_tag` is applied verbatim rather than re-derived from the body. The + // server already reconciled tags when it saved the note, and re-deriving here + // would call the local find-or-create path, which marks new labels dirty and + // would push them straight back — sync churn out of nothing. + conn.execute( + "INSERT OR IGNORE INTO note_labels (note_id, label_id, via_tag) + VALUES (?1, ?2, ?3)", + params![note.id, label.id, label.via_tag], + )?; + } + Ok(()) +} + +/// Materialize a label referenced by a note, if we don't have it yet. +/// +/// Notes and labels page from one shared sequence, so a note can reference a label +/// whose own delta landed in an earlier page — or, right at a page boundary, hasn't +/// landed. The note carries enough of the label to create it, so a membership never +/// fails on a missing row. `OR IGNORE` because the label's real delta (later in this +/// page or a future one) is the authority on its name and color. +fn ensure_label_stub(conn: &Connection, label: &wire::NoteLabel) -> rusqlite::Result<()> { + let ts = now(); + conn.execute( + "INSERT OR IGNORE INTO labels (id, name, color, created_at, updated_at, dirty) + VALUES (?1, ?2, ?3, ?4, ?4, 0)", + params![label.id, label.name, label.color, ts], + )?; + Ok(()) +} + +/// Trust an explicit position; fall back to arrival order when the server sent 0 for +/// everything (which is what an unordered list looks like on the wire). +fn position_of(explicit: i64, index: usize) -> i64 { + if explicit > 0 { + explicit + } else { + index as i64 + } +} + +/// Loop the feed to exhaustion, starting from the persisted cursor. +/// +/// NOTE ON ORDERING: the full cycle is push-then-pull (docs/sync.md). Running this +/// against a store with unpushed edits lets the server's version land on top of them +/// — counted as `clobbered_dirty` and logged, rather than hidden. +pub async fn run(db: &Db, base_url: &str, token: &str) -> Result { + let mut total = PullSummary::default(); + + loop { + let since = { + let conn = db.0.lock().map_err(|e| e.to_string())?; + state::read(&conn).map_err(|e| e.to_string())?.last_cursor + }; + + let page = client::fetch_changes(base_url, token, since).await?; + + // Trust the data over the flag: a server that claims more pages without + // advancing the cursor would spin this loop forever. + if page.has_more && page.cursor <= since { + return Err(format!( + "The server reported more changes but its cursor didn't advance past \ + {since}. Stopping rather than looping forever." + )); + } + + let has_more = page.has_more; + let applied = { + let conn = db.0.lock().map_err(|e| e.to_string())?; + apply_page(&conn, &page).map_err(|e| e.to_string())? + }; + total.absorb(applied); + + if !has_more { + break; + } + if total.pages >= MAX_PAGES { + return Err(format!( + "Stopped after {MAX_PAGES} pages without reaching the end of the \ + server's changes. Something is wrong with the feed." + )); + } + } + + if total.clobbered_dirty > 0 { + log::warn!( + "pull overwrote {} note(s) that still had unpushed local edits", + total.clobbered_dirty + ); + } + log::info!( + "pull complete: {} page(s), {} note(s) applied, {} deleted, {} label(s) applied, cursor {}", + total.pages, + total.notes_applied, + total.notes_deleted, + total.labels_applied, + total.cursor + ); + Ok(total) +} + +#[cfg(test)] +mod tests { + use super::*; + use crate::local::schema; + + fn db() -> Connection { + let conn = Connection::open_in_memory().expect("in-memory db"); + schema::migrate(&conn).expect("migrate"); + conn + } + + fn note(id: &str, revision: i64) -> wire::Note { + wire::Note { + id: id.to_string(), + title: Some("Title".into()), + body: "Body".into(), + color: "default".into(), + kind: "text".into(), + position: 0, + pinned: false, + archived: false, + trashed: false, + remind_at: None, + recurrence: None, + created_at: Some("2026-07-26T00:00:00.000Z".into()), + updated_at: Some("2026-07-26T00:00:00.000Z".into()), + sync_revision: revision, + purged_at: None, + labels: vec![], + items: vec![], + attachments: vec![], + previews: vec![], + } + } + + fn page(notes: Vec, labels: Vec, cursor: i64) -> wire::ChangesPage { + wire::ChangesPage { + notes, + labels, + cursor, + has_more: false, + } + } + + fn count(conn: &Connection, sql: &str) -> i64 { + conn.query_row(sql, [], |r| r.get(0)).expect("count") + } + + #[test] + fn applies_a_note_and_advances_the_cursor() { + let conn = db(); + let summary = apply_page(&conn, &page(vec![note("n1", 7)], vec![], 7)).expect("apply"); + assert_eq!(summary.notes_applied, 1); + assert_eq!(count(&conn, "SELECT COUNT(*) FROM notes"), 1); + assert_eq!(state::read(&conn).expect("state").last_cursor, 7); + } + + #[test] + fn pulled_rows_are_not_dirty() { + // They came FROM the server, so pushing them back would be pure churn. + let conn = db(); + apply_page(&conn, &page(vec![note("n1", 1)], vec![], 1)).expect("apply"); + assert_eq!(count(&conn, "SELECT dirty FROM notes WHERE id = 'n1'"), 0); + } + + #[test] + fn tombstone_deletes_the_local_note() { + let conn = db(); + apply_page(&conn, &page(vec![note("n1", 1)], vec![], 1)).expect("apply"); + let mut dead = note("n1", 2); + dead.purged_at = Some("2026-07-26T01:00:00.000Z".into()); + let summary = apply_page(&conn, &page(vec![dead], vec![], 2)).expect("apply"); + assert_eq!(summary.notes_deleted, 1); + assert_eq!(count(&conn, "SELECT COUNT(*) FROM notes"), 0); + } + + #[test] + fn trashed_is_not_a_tombstone() { + // `trashed` is ordinary state that keeps syncing; only `purged_at` deletes. + let conn = db(); + let mut trashed = note("n1", 1); + trashed.trashed = true; + apply_page(&conn, &page(vec![trashed], vec![], 1)).expect("apply"); + assert_eq!(count(&conn, "SELECT COUNT(*) FROM notes"), 1); + assert_eq!(count(&conn, "SELECT trashed FROM notes WHERE id = 'n1'"), 1); + } + + #[test] + fn children_are_replaced_not_merged() { + let conn = db(); + let mut first = note("n1", 1); + first.items = vec![ + wire::Item { + id: "i1".into(), + text: "one".into(), + checked: false, + position: 0, + }, + wire::Item { + id: "i2".into(), + text: "two".into(), + checked: false, + position: 1, + }, + ]; + apply_page(&conn, &page(vec![first], vec![], 1)).expect("apply"); + assert_eq!(count(&conn, "SELECT COUNT(*) FROM checklist_items"), 2); + + // The server dropped an item; the local copy must drop it too. + let mut second = note("n1", 2); + second.items = vec![wire::Item { + id: "i1".into(), + text: "one".into(), + checked: true, + position: 0, + }]; + apply_page(&conn, &page(vec![second], vec![], 2)).expect("apply"); + assert_eq!(count(&conn, "SELECT COUNT(*) FROM checklist_items"), 1); + } + + #[test] + fn note_label_membership_materializes_a_missing_label() { + // The label's own delta may have landed in an earlier page, or not yet. + let conn = db(); + let mut n = note("n1", 1); + n.labels = vec![wire::NoteLabel { + id: "l1".into(), + name: "work".into(), + color: "blue".into(), + via_tag: true, + }]; + apply_page(&conn, &page(vec![n], vec![], 1)).expect("apply"); + assert_eq!(count(&conn, "SELECT COUNT(*) FROM labels"), 1); + assert_eq!( + count(&conn, "SELECT via_tag FROM note_labels WHERE note_id = 'n1'"), + 1, + "via_tag is applied verbatim, not re-derived" + ); + } + + #[test] + fn server_label_replaces_a_local_duplicate_by_name() { + let conn = db(); + conn.execute( + "INSERT INTO labels (id, name, color, created_at, updated_at, dirty) + VALUES ('local-id', 'Work', 'default', '2026-01-01', '2026-01-01', 1)", + [], + ) + .expect("seed local label"); + + let server = wire::Label { + id: "server-id".into(), + name: "work".into(), + color: "blue".into(), + sync_revision: 5, + purged_at: None, + created_at: Some("2026-07-26T00:00:00.000Z".into()), + }; + apply_page(&conn, &page(vec![], vec![server], 5)).expect("apply"); + + assert_eq!(count(&conn, "SELECT COUNT(*) FROM labels"), 1); + let id: String = conn + .query_row("SELECT id FROM labels", [], |r| r.get(0)) + .expect("label"); + assert_eq!(id, "server-id", "the server's row wins on pull"); + } + + #[test] + fn merging_a_duplicate_label_keeps_its_note_memberships() { + // The notes carrying the local label may not be in this page at all, so a + // plain delete would strip the label off them with nothing to repair it. + let conn = db(); + apply_page(&conn, &page(vec![note("n1", 1)], vec![], 1)).expect("seed note"); + conn.execute( + "INSERT INTO labels (id, name, color, created_at, updated_at, dirty) + VALUES ('local-id', 'Work', 'default', '2026-01-01', '2026-01-01', 1)", + [], + ) + .expect("seed local label"); + conn.execute( + "INSERT INTO note_labels (note_id, label_id, via_tag) + VALUES ('n1', 'local-id', 0)", + [], + ) + .expect("seed membership"); + + let server = wire::Label { + id: "server-id".into(), + name: "work".into(), + color: "blue".into(), + sync_revision: 5, + purged_at: None, + created_at: None, + }; + apply_page(&conn, &page(vec![], vec![server], 5)).expect("apply"); + + assert_eq!(count(&conn, "SELECT COUNT(*) FROM labels"), 1); + let label_id: String = conn + .query_row("SELECT label_id FROM note_labels WHERE note_id = 'n1'", [], |r| r.get(0)) + .expect("membership survived"); + assert_eq!(label_id, "server-id", "membership re-pointed, not dropped"); + } + + + #[test] + fn label_tombstone_deletes_and_cascades_memberships() { + let conn = db(); + let mut n = note("n1", 1); + n.labels = vec![wire::NoteLabel { + id: "l1".into(), + name: "work".into(), + color: "blue".into(), + via_tag: false, + }]; + apply_page(&conn, &page(vec![n], vec![], 1)).expect("apply"); + assert_eq!(count(&conn, "SELECT COUNT(*) FROM note_labels"), 1); + + let dead = wire::Label { + id: "l1".into(), + name: "work".into(), + color: "blue".into(), + sync_revision: 2, + purged_at: Some("2026-07-26T01:00:00.000Z".into()), + created_at: None, + }; + apply_page(&conn, &page(vec![], vec![dead], 2)).expect("apply"); + assert_eq!(count(&conn, "SELECT COUNT(*) FROM labels"), 0); + assert_eq!( + count(&conn, "SELECT COUNT(*) FROM note_labels"), + 0, + "membership should cascade with the label" + ); + } + + #[test] + fn overwriting_a_dirty_note_is_counted() { + let conn = db(); + conn.execute( + "INSERT INTO notes (id, body, created_at, updated_at, dirty) + VALUES ('n1', 'local edit', '2026-01-01', '2026-01-01', 1)", + [], + ) + .expect("seed dirty note"); + let summary = apply_page(&conn, &page(vec![note("n1", 9)], vec![], 9)).expect("apply"); + assert_eq!(summary.clobbered_dirty, 1); + } + + #[test] + fn applying_a_fresh_note_reports_no_clobber() { + let conn = db(); + let summary = apply_page(&conn, &page(vec![note("n1", 1)], vec![], 1)).expect("apply"); + assert_eq!(summary.clobbered_dirty, 0); + } + + #[test] + fn empty_page_still_advances_the_cursor() { + // The server can page past rows that were trimmed to the shared watermark. + let conn = db(); + apply_page(&conn, &page(vec![], vec![], 42)).expect("apply"); + assert_eq!(state::read(&conn).expect("state").last_cursor, 42); + } + + #[test] + fn note_upsert_preserves_the_original_created_at() { + let conn = db(); + apply_page(&conn, &page(vec![note("n1", 1)], vec![], 1)).expect("apply"); + let mut later = note("n1", 2); + later.created_at = Some("2099-01-01T00:00:00.000Z".into()); + apply_page(&conn, &page(vec![later], vec![], 2)).expect("apply"); + let created: String = conn + .query_row("SELECT created_at FROM notes WHERE id = 'n1'", [], |r| { + r.get(0) + }) + .expect("created_at"); + assert_eq!(created, "2026-07-26T00:00:00.000Z"); + } + + #[test] + fn a_page_that_fails_leaves_the_cursor_untouched() { + // Atomicity is the whole resumability story: a cursor committed ahead of its + // data would skip those rows forever. Force a failure with a duplicate + // checklist-item id inside one page. + let conn = db(); + let mut n = note("n1", 3); + n.items = vec![ + wire::Item { + id: "dup".into(), + text: "one".into(), + checked: false, + position: 0, + }, + wire::Item { + id: "dup".into(), + text: "two".into(), + checked: false, + position: 1, + }, + ]; + assert!(apply_page(&conn, &page(vec![n], vec![], 3)).is_err()); + assert_eq!(state::read(&conn).expect("state").last_cursor, 0); + assert_eq!(count(&conn, "SELECT COUNT(*) FROM notes"), 0); + } +} diff --git a/desktop/src-tauri/src/sync/wire.rs b/desktop/src-tauri/src/sync/wire.rs new file mode 100644 index 0000000..9ee8ad8 --- /dev/null +++ b/desktop/src-tauri/src/sync/wire.rs @@ -0,0 +1,161 @@ +//! The delta-feed JSON shapes, exactly as `GET /api/sync/changes` sends them. +//! +//! Mirrors the server's serializers (`notes/serialize.py` + `serialize.py`) — see +//! `docs/sync.md` for the contract. Every field is `#[serde(default)]` or `Option` +//! so a NEWER server adding fields, or an older one omitting one, degrades to a +//! partial note rather than failing the whole page. Losing one attribute is +//! recoverable; refusing a page stalls sync permanently at that cursor. + +use serde::Deserialize; + +#[derive(Debug, Clone, Deserialize, Default)] +pub struct ChangesPage { + #[serde(default)] + pub notes: Vec, + #[serde(default)] + pub labels: Vec