From b5f7dc2635916d2dcf12fcda0440cae823f7b188 Mon Sep 17 00:00:00 2001 From: Bryan Van Deusen Date: Sun, 26 Jul 2026 00:20:09 -0400 Subject: [PATCH] M10.7c: push + the full sync cycle (task 2106) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Local -> server, then push-then-pull as the only ordering the UI can invoke. LOCAL TOMBSTONES (schema v2). Found while writing push: delete_forever and remove_label just DROPPED the row, leaving no record it existed. Offline that means the delete can never be pushed — and the next pull faithfully resurrects the note from the server. A deletion that undoes itself is about the worst thing sync can do, so deletes now record into pending_deletes until the server acknowledges them. merge_labels had the same hole. merge_labels also moved memberships without marking the affected notes dirty. A note's label set only reaches the server via the note itself, so a merge looked done locally and never synced. Now marked before the delete cascades the rows away. Result handling, per status: created/applied -> clear dirty, store the returned sync_revision noop -> clear dirty, drop the tombstone (a row the server never saw, created and deleted entirely offline) kept -> clear dirty WITHOUT touching content. Re-pushing would lose the same last-write-wins comparison forever; the following pull adopts the server's version. rejected -> stay dirty and surface the reason. A duplicate label name is the realistic case and only a human can resolve it. The subtle one is `kept` plus a skewed clock. Normally the server's kept revision sits above our cursor, so the next pull fetches it anyway. If the clock makes a genuinely later local edit look older, that revision can be BELOW the cursor — the pull skips it and the stale local copy stays on screen with nothing marking it wrong. So a kept result at or below the cursor rewinds the cursor to re-fetch that note. Both directions tested. label_ids carries MANUAL memberships only. Tag-sourced ones are re-derived server-side from the body; sending them would convert them into manual assignments that no longer disappear when the #tag is deleted from the text. engine::run_cycle is push-then-pull, and a failed push ABORTS before the pull — pulling anyway would overwrite the exact rows we just failed to save, turning a recoverable network error into lost work. sync_pull is removed from the command surface accordingly: offering a bare pull would hand the UI a way to discard unsent edits. sync_now and sync_has_pending replace it. Both loops have anti-spin guards: push stops when a batch clears nothing, pull stops when the cursor doesn't advance. 15 push tests against an in-memory database. Co-Authored-By: Claude Opus 5 (1M context) Claude-Session: https://claude.ai/code/session_01SreJkbxB4gx8pPsu8QbLPi --- desktop/src-tauri/src/lib.rs | 3 +- desktop/src-tauri/src/local/schema.rs | 20 + desktop/src-tauri/src/local/store.rs | 26 + desktop/src-tauri/src/sync/client.rs | 32 ++ desktop/src-tauri/src/sync/commands.rs | 24 +- desktop/src-tauri/src/sync/engine.rs | 43 ++ desktop/src-tauri/src/sync/mod.rs | 2 + desktop/src-tauri/src/sync/push.rs | 751 +++++++++++++++++++++++++ 8 files changed, 892 insertions(+), 9 deletions(-) create mode 100644 desktop/src-tauri/src/sync/engine.rs create mode 100644 desktop/src-tauri/src/sync/push.rs diff --git a/desktop/src-tauri/src/lib.rs b/desktop/src-tauri/src/lib.rs index 3b176cc..8a31636 100644 --- a/desktop/src-tauri/src/lib.rs +++ b/desktop/src-tauri/src/lib.rs @@ -93,7 +93,8 @@ pub fn run() { sync::commands::sync_link, sync::commands::sync_unlink, sync::commands::sync_status, - sync::commands::sync_pull, + sync::commands::sync_now, + sync::commands::sync_has_pending, ]) .run(tauri::generate_context!()) .expect("error while running the ThoughtSync desktop app"); diff --git a/desktop/src-tauri/src/local/schema.rs b/desktop/src-tauri/src/local/schema.rs index 63eb7d6..190e26b 100644 --- a/desktop/src-tauri/src/local/schema.rs +++ b/desktop/src-tauri/src/local/schema.rs @@ -105,6 +105,22 @@ CREATE TABLE sync_state ( INSERT INTO sync_state (id) VALUES (1); "#; +// v2 (M10.7c): local tombstones. +// +// A permanent delete previously just dropped the row, which left NO record that it +// ever existed. Offline, that means the delete can never be pushed — and the next +// pull would faithfully resurrect the note from the server. A deletion that undoes +// itself is about the worst outcome sync can produce, so deletes are now recorded +// here until they've been acknowledged by the server and cleared. +const SCHEMA_V2: &str = r#" +CREATE TABLE pending_deletes ( + entity TEXT NOT NULL, -- 'note' | 'label' + id TEXT NOT NULL, + deleted_at TEXT NOT NULL, + PRIMARY KEY (entity, id) +); +"#; + /// Bring the database up to the latest schema. Idempotent. pub fn migrate(conn: &Connection) -> rusqlite::Result<()> { conn.execute_batch("PRAGMA foreign_keys = ON;")?; @@ -113,5 +129,9 @@ pub fn migrate(conn: &Connection) -> rusqlite::Result<()> { conn.execute_batch(SCHEMA_V1)?; conn.execute_batch("PRAGMA user_version = 1;")?; } + if version < 2 { + conn.execute_batch(SCHEMA_V2)?; + conn.execute_batch("PRAGMA user_version = 2;")?; + } Ok(()) } diff --git a/desktop/src-tauri/src/local/store.rs b/desktop/src-tauri/src/local/store.rs index 7c6533f..74b0448 100644 --- a/desktop/src-tauri/src/local/store.rs +++ b/desktop/src-tauri/src/local/store.rs @@ -633,10 +633,25 @@ pub fn restore(conn: &Connection, id: &str) -> rusqlite::Result { } pub fn delete_forever(conn: &Connection, id: &str) -> rusqlite::Result<()> { + record_pending_delete(conn, "note", id)?; conn.execute("DELETE FROM notes WHERE id = ?1", [id])?; Ok(()) } +/// Remember that a row was permanently deleted, so the sync engine can tell the +/// server. Without this the deleted row leaves no trace at all, and the next pull +/// would resurrect it — a delete that quietly undoes itself. +/// +/// Harmless when the app is unlinked: the row is simply never read, and a later push +/// gets a `noop` for an id the server never had. +pub fn record_pending_delete(conn: &Connection, entity: &str, id: &str) -> rusqlite::Result<()> { + conn.execute( + "INSERT OR REPLACE INTO pending_deletes (entity, id, deleted_at) VALUES (?1, ?2, ?3)", + params![entity, id, now()], + )?; + Ok(()) +} + pub fn revisions(conn: &Connection, id: &str) -> rusqlite::Result> { let mut stmt = conn .prepare("SELECT id, title, body, created_at FROM note_revisions WHERE note_id = ?1 ORDER BY created_at DESC")?; @@ -727,6 +742,7 @@ pub fn set_label_color(conn: &Connection, id: &str, color: &str) -> rusqlite::Re } pub fn remove_label(conn: &Connection, id: &str) -> rusqlite::Result<()> { + record_pending_delete(conn, "label", id)?; conn.execute("DELETE FROM labels WHERE id = ?1", [id])?; Ok(()) } @@ -741,6 +757,16 @@ pub fn merge_labels( SELECT note_id, ?2, 0 FROM note_labels WHERE label_id = ?1", params![source_id, target_id], )?; + // The notes that carried the source now have a different label set, and that set + // only reaches the server via the note itself (push sends label_ids per note). + // Without this the merge would look done locally and never sync. Marked BEFORE + // the delete, which cascades the membership rows away. + conn.execute( + "UPDATE notes SET dirty = 1 + WHERE id IN (SELECT note_id FROM note_labels WHERE label_id = ?1)", + [source_id], + )?; + record_pending_delete(conn, "label", source_id)?; conn.execute("DELETE FROM labels WHERE id = ?1", [source_id])?; load_label(conn, target_id) } diff --git a/desktop/src-tauri/src/sync/client.rs b/desktop/src-tauri/src/sync/client.rs index bf68a84..17f6d9a 100644 --- a/desktop/src-tauri/src/sync/client.rs +++ b/desktop/src-tauri/src/sync/client.rs @@ -223,6 +223,38 @@ pub async fn fetch_changes( .map_err(|e| format!("Couldn't read the change feed from {base_url}: {e}")) } +/// Send a batch of changes and hand back the raw reply. +/// +/// Returns text rather than parsed results so this module stays pure transport — +/// `push::parse_results` owns the result shapes, and keeping them there is what lets +/// the parsing be unit-tested without a server. +pub async fn push_changes( + base_url: &str, + token: &str, + changes: &[T], +) -> Result { + let body = serde_json::json!({ "changes": changes }); + let url = format!("{base_url}/api/sync/push"); + let request = prepare(http_with(SYNC_TIMEOUT)?.post(url), Some(token)).json(&body); + 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 + .text() + .await + .map_err(|e| format!("Couldn't read the push reply from {base_url}: {e}")) +} + /// The public, unauthenticated endpoint carrying the handshake. fn config_url(base_url: &str) -> String { format!("{base_url}/api/config") diff --git a/desktop/src-tauri/src/sync/commands.rs b/desktop/src-tauri/src/sync/commands.rs index b7cbf2f..976f0ef 100644 --- a/desktop/src-tauri/src/sync/commands.rs +++ b/desktop/src-tauri/src/sync/commands.rs @@ -9,7 +9,8 @@ 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::engine; +use crate::sync::push; use crate::sync::state; /// Ask a server who it is, without committing to anything. The UI calls this as the @@ -135,14 +136,21 @@ fn credentials(db: &State<'_, Db>) -> Result<(String, String), String> { } } -/// Pull the server's changes into the local store. +/// Run one full sync: push local changes, then pull the server's. /// -/// 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. +/// The only sync entry point exposed to the UI, on purpose. Push and pull exist +/// separately inside the crate, but offering a bare "pull" would let the UI overwrite +/// unsent local edits — the ordering isn't a suggestion, it's what keeps them. #[tauri::command] -pub async fn sync_pull(db: State<'_, Db>) -> Result { +pub async fn sync_now(db: State<'_, Db>) -> Result { let (base_url, token) = credentials(&db)?; - pull::run(db.inner(), &base_url, &token).await + engine::run_cycle(db.inner(), &base_url, &token).await +} + +/// Whether anything is waiting to be sent. Lets the UI show an honest "unsynced +/// changes" state without running a sync to find out. +#[tauri::command] +pub fn sync_has_pending(db: State<'_, Db>) -> Result { + let conn = db.0.lock().map_err(|e| e.to_string())?; + push::has_pending(&conn).map_err(|e| e.to_string()) } diff --git a/desktop/src-tauri/src/sync/engine.rs b/desktop/src-tauri/src/sync/engine.rs new file mode 100644 index 0000000..b73911e --- /dev/null +++ b/desktop/src-tauri/src/sync/engine.rs @@ -0,0 +1,43 @@ +//! The sync cycle (M10.7c). +//! +//! Deliberately the ONLY way the UI can sync. Push and pull are each usable on their +//! own inside this crate, but exposing them separately would let a caller pull +//! without pushing, which quietly overwrites unsent local edits. + +use serde::Serialize; + +use super::pull; +use super::push; +use crate::local::Db; + +#[derive(Debug, Serialize)] +pub struct SyncOutcome { + pub push: push::PushSummary, + pub pull: pull::PullSummary, +} + +/// Push, then pull — in that order, always. +/// +/// Pull writes the server's version straight over the local row, so anything not yet +/// sent would be lost to it. Pushing first is what puts the local edit in front of +/// the server's last-write-wins comparison, and it's the reason +/// `PullSummary::clobbered_dirty` should be zero on every healthy cycle. +/// +/// A failed push aborts before the pull. Pulling anyway would take the exact rows we +/// just failed to save and overwrite them — turning a recoverable network error into +/// lost work. +pub async fn run_cycle(db: &Db, base_url: &str, token: &str) -> Result { + let push = push::run(db, base_url, token).await?; + let pull = pull::run(db, base_url, token).await?; + + if pull.clobbered_dirty > 0 { + // Push ran first and reported success, so nothing should still have been + // dirty. Reaching here means something wrote to the store mid-cycle, or a + // change never got collected — worth a loud line either way. + log::warn!( + "sync cycle overwrote {} locally-edited note(s) despite pushing first", + pull.clobbered_dirty + ); + } + Ok(SyncOutcome { push, pull }) +} diff --git a/desktop/src-tauri/src/sync/mod.rs b/desktop/src-tauri/src/sync/mod.rs index b95d473..53b19ae 100644 --- a/desktop/src-tauri/src/sync/mod.rs +++ b/desktop/src-tauri/src/sync/mod.rs @@ -15,6 +15,8 @@ pub mod client; pub mod commands; pub mod compat; +pub mod engine; pub mod pull; +pub mod push; pub mod state; pub mod wire; diff --git a/desktop/src-tauri/src/sync/push.rs b/desktop/src-tauri/src/sync/push.rs new file mode 100644 index 0000000..678d87f --- /dev/null +++ b/desktop/src-tauri/src/sync/push.rs @@ -0,0 +1,751 @@ +//! Push: send local changes to the server and apply what it says (M10.7c). +//! +//! Two sources feed a push: rows flagged `dirty` (created or edited locally) and rows +//! in `pending_deletes` (permanently deleted locally — see `local::schema` v2 for why +//! a delete needs its own record). +//! +//! Sync is **whole-note**: an upsert carries the client's full current state, not a +//! patch (docs/sync.md). The server resolves conflicts last-write-wins by the client's +//! `edited_at`, snapshotting anything it overwrites into the note's version history. + +use rusqlite::{params, Connection, OptionalExtension}; +use serde::{Deserialize, Serialize}; + +use super::client; +use super::state; +use crate::local::Db; + +/// The server rejects a batch larger than this (`MAX_PUSH` in `sync.py`). +const BATCH: usize = 500; + +/// Backstop: a batch whose results never clear `dirty` would loop forever. +const MAX_BATCHES: usize = 10_000; + +#[derive(Debug, Clone, Default, Serialize, PartialEq, Eq)] +pub struct PushSummary { + pub batches: usize, + pub sent: usize, + pub created: usize, + pub applied: usize, + /// The server had a newer edit and kept it. Not a failure — the local row stops + /// being dirty and the following pull adopts the server's version. + pub kept: usize, + pub noop: usize, + /// Still dirty, and surfaced: these need a human (a duplicate label name is the + /// realistic case). Silently retrying forever would be the wrong shape. + pub rejected: usize, + pub errors: Vec, +} + +impl PushSummary { + fn absorb(&mut self, other: PushSummary) { + self.batches += other.batches; + self.sent += other.sent; + self.created += other.created; + self.applied += other.applied; + self.kept += other.kept; + self.noop += other.noop; + self.rejected += other.rejected; + self.errors.extend(other.errors); + } +} + +// --- outgoing shapes --------------------------------------------------------- + +/// One entry in the `changes` array. Notes and labels share the envelope; serde skips +/// the fields that don't apply, so the server sees exactly the shape docs/sync.md +/// describes for each entity. +#[derive(Debug, Serialize)] +pub struct Change { + pub entity: &'static str, + pub id: String, + pub op: &'static str, + pub edited_at: String, + + #[serde(skip_serializing_if = "Option::is_none")] + pub title: Option, + #[serde(skip_serializing_if = "Option::is_none")] + pub body: Option, + #[serde(skip_serializing_if = "Option::is_none")] + pub color: Option, + #[serde(skip_serializing_if = "Option::is_none")] + pub kind: Option, + #[serde(skip_serializing_if = "Option::is_none")] + pub pinned: Option, + #[serde(skip_serializing_if = "Option::is_none")] + pub archived: Option, + #[serde(skip_serializing_if = "Option::is_none")] + pub trashed: Option, + #[serde(skip_serializing_if = "Option::is_none")] + pub remind_at: Option, + #[serde(skip_serializing_if = "Option::is_none")] + pub recurrence: Option, + #[serde(skip_serializing_if = "Option::is_none")] + pub position: Option, + #[serde(skip_serializing_if = "Option::is_none")] + pub items: Option>, + #[serde(skip_serializing_if = "Option::is_none")] + pub label_ids: Option>, + #[serde(skip_serializing_if = "Option::is_none")] + pub created_at: Option, + #[serde(skip_serializing_if = "Option::is_none")] + pub name: Option, +} + +impl Change { + fn delete(entity: &'static str, id: String, edited_at: String) -> Self { + Change { + entity, + id, + op: "delete", + edited_at, + title: None, + body: None, + color: None, + kind: None, + pinned: None, + archived: None, + trashed: None, + remind_at: None, + recurrence: None, + position: None, + items: None, + label_ids: None, + created_at: None, + name: None, + } + } +} + +#[derive(Debug, Serialize)] +pub struct ItemOut { + pub text: String, + pub checked: bool, +} + +// --- incoming results -------------------------------------------------------- + +#[derive(Debug, Deserialize)] +struct PushResponse { + #[serde(default)] + results: Vec, +} + +#[derive(Debug, Clone, Deserialize)] +pub struct PushResult { + #[serde(default)] + pub id: Option, + #[serde(default)] + pub entity: Option, + #[serde(default)] + pub status: String, + #[serde(default)] + pub sync_revision: Option, + #[serde(default)] + pub error: Option, +} + +// --- collecting -------------------------------------------------------------- + +/// Everything waiting to go up, oldest edit first so a truncated batch still makes +/// forward progress in a sensible order. +pub fn collect(conn: &Connection, limit: usize) -> rusqlite::Result> { + let mut out = Vec::new(); + collect_deletes(conn, &mut out, limit)?; + if out.len() < limit { + collect_labels(conn, &mut out, limit)?; + } + if out.len() < limit { + collect_notes(conn, &mut out, limit)?; + } + Ok(out) +} + +fn collect_deletes( + conn: &Connection, + out: &mut Vec, + limit: usize, +) -> rusqlite::Result<()> { + let mut stmt = conn.prepare( + "SELECT entity, id, deleted_at FROM pending_deletes ORDER BY deleted_at LIMIT ?1", + )?; + let rows = stmt.query_map(params![limit as i64], |r| { + Ok(( + r.get::<_, String>(0)?, + r.get::<_, String>(1)?, + r.get::<_, String>(2)?, + )) + })?; + for row in rows { + let (entity, id, deleted_at) = row?; + // Only 'note' and 'label' exist on the wire; anything else is a bug in a + // writer, and shipping it would earn a blanket rejection for the batch. + let entity: &'static str = match entity.as_str() { + "note" => "note", + "label" => "label", + _ => continue, + }; + out.push(Change::delete(entity, id, deleted_at)); + } + Ok(()) +} + +fn collect_labels(conn: &Connection, out: &mut Vec, limit: usize) -> rusqlite::Result<()> { + let remaining = limit.saturating_sub(out.len()); + let mut stmt = conn.prepare( + "SELECT id, name, color, updated_at FROM labels + WHERE dirty = 1 ORDER BY updated_at LIMIT ?1", + )?; + let rows = stmt.query_map(params![remaining as i64], |r| { + Ok(Change { + entity: "label", + id: r.get(0)?, + op: "upsert", + name: Some(r.get(1)?), + color: Some(r.get(2)?), + edited_at: r.get(3)?, + title: None, + body: None, + kind: None, + pinned: None, + archived: None, + trashed: None, + remind_at: None, + recurrence: None, + position: None, + items: None, + label_ids: None, + created_at: None, + }) + })?; + for row in rows { + out.push(row?); + } + Ok(()) +} + +fn collect_notes(conn: &Connection, out: &mut Vec, limit: usize) -> rusqlite::Result<()> { + let remaining = limit.saturating_sub(out.len()); + let ids: Vec = { + let mut stmt = conn.prepare( + "SELECT id FROM notes WHERE dirty = 1 ORDER BY updated_at LIMIT ?1", + )?; + let rows = stmt.query_map(params![remaining as i64], |r| r.get::<_, String>(0))?; + rows.collect::>>()? + }; + for id in ids { + out.push(note_change(conn, &id)?); + } + Ok(()) +} + +/// The note's own columns. A named struct rather than a twelve-wide tuple so the +/// field-to-column mapping stays readable at the call site. +struct NoteRow { + title: Option, + body: String, + color: String, + kind: String, + position: i64, + pinned: bool, + archived: bool, + trashed: bool, + remind_at: Option, + recurrence: Option, + created_at: String, + updated_at: String, +} + +fn note_row(conn: &Connection, id: &str) -> rusqlite::Result { + conn.query_row( + "SELECT title, body, color, kind, position, pinned, archived, trashed, + remind_at, recurrence, created_at, updated_at + FROM notes WHERE id = ?1", + params![id], + |r| { + Ok(NoteRow { + title: r.get(0)?, + body: r.get(1)?, + color: r.get(2)?, + kind: r.get(3)?, + position: r.get(4)?, + pinned: r.get::<_, i64>(5)? != 0, + archived: r.get::<_, i64>(6)? != 0, + trashed: r.get::<_, i64>(7)? != 0, + remind_at: r.get(8)?, + recurrence: r.get(9)?, + created_at: r.get(10)?, + updated_at: r.get(11)?, + }) + }, + ) +} + +fn note_change(conn: &Connection, id: &str) -> rusqlite::Result { + let row = note_row(conn, id)?; + + let items = { + let mut stmt = conn.prepare( + "SELECT text, checked FROM checklist_items WHERE note_id = ?1 ORDER BY position", + )?; + let rows = stmt.query_map(params![id], |r| { + Ok(ItemOut { + text: r.get(0)?, + checked: r.get::<_, i64>(1)? != 0, + }) + })?; + rows.collect::>>()? + }; + + // MANUAL memberships only. Tag-sourced ones (`via_tag = 1`) are re-derived by the + // server from the body; sending them as label_ids would convert them into manual + // assignments that no longer disappear when the #tag is removed from the text. + let label_ids = { + let mut stmt = conn.prepare( + "SELECT label_id FROM note_labels WHERE note_id = ?1 AND via_tag = 0", + )?; + let rows = stmt.query_map(params![id], |r| r.get::<_, String>(0))?; + rows.collect::>>()? + }; + + Ok(Change { + entity: "note", + id: id.to_string(), + op: "upsert", + // The local `updated_at` IS the client's edit time, which is what the + // server's last-write-wins comparison runs against. + edited_at: row.updated_at, + title: row.title, + body: Some(row.body), + color: Some(row.color), + kind: Some(row.kind), + pinned: Some(row.pinned), + archived: Some(row.archived), + trashed: Some(row.trashed), + remind_at: row.remind_at, + recurrence: row.recurrence, + position: Some(row.position), + items: Some(items), + label_ids: Some(label_ids), + created_at: Some(row.created_at), + name: None, + }) +} + +// --- applying results -------------------------------------------------------- + +/// Fold one batch's results back into the local store, atomically. +pub fn apply_results( + conn: &Connection, + sent: &[Change], + results: &[PushResult], +) -> rusqlite::Result { + let tx = conn.unchecked_transaction()?; + let mut summary = PushSummary { + batches: 1, + sent: sent.len(), + ..Default::default() + }; + // The server answers positionally, one result per change. Zip rather than trust + // the echoed id: a rejected malformed entry may carry no id at all. + let mut lowest_kept: Option = None; + + for (change, result) in sent.iter().zip(results.iter()) { + match result.status.as_str() { + "created" | "applied" => { + clear_dirty(&tx, change, result.sync_revision)?; + if result.status == "created" { + summary.created += 1; + } else { + summary.applied += 1; + } + if change.op == "delete" { + forget_pending_delete(&tx, change)?; + } + } + "noop" => { + // The server had nothing to do — typically a delete for a row it + // never saw (created and deleted while offline). + clear_dirty(&tx, change, result.sync_revision)?; + forget_pending_delete(&tx, change)?; + summary.noop += 1; + } + "kept" => { + // The server's version is newer. Stop being dirty — re-pushing would + // lose to the same comparison forever — and let the next pull bring + // the server's copy down. + clear_dirty(&tx, change, None)?; + if change.op == "delete" { + // Our delete lost to a newer server edit; the note lives on, and + // the pull will restore it locally. Drop the tombstone so we + // don't keep trying to delete a note the user has since edited. + forget_pending_delete(&tx, change)?; + } + if let Some(revision) = result.sync_revision { + lowest_kept = Some(lowest_kept.map_or(revision, |c: i64| c.min(revision))); + } + summary.kept += 1; + } + _ => { + // "rejected" and anything unrecognized: leave the row dirty so it is + // retried, and surface the reason. A duplicate label name is the + // realistic case and only a human can resolve it. + summary.rejected += 1; + let reason = result.error.clone().unwrap_or_else(|| result.status.clone()); + summary.errors.push(format!("{} {}: {reason}", change.entity, change.id)); + } + } + } + + // A `kept` result means the server holds a version we have not seen. Normally its + // revision is above our cursor and the next pull fetches it anyway. If it is NOT + // — which happens when a skewed clock makes a genuinely later local edit look + // older — rewind so that note is re-fetched. Without this the local edit is + // dropped from sync and the stale copy stays on screen with nothing marking it. + if let Some(revision) = lowest_kept { + let current = state::read(&tx)?.last_cursor; + if revision <= current { + state::set_cursor(&tx, (revision - 1).max(0))?; + } + } + + tx.commit()?; + Ok(summary) +} + +fn clear_dirty( + conn: &Connection, + change: &Change, + revision: Option, +) -> rusqlite::Result<()> { + // A delete has no local row left to update. + if change.op == "delete" { + return Ok(()); + } + let table = match change.entity { + "label" => "labels", + _ => "notes", + }; + match revision { + Some(rev) => conn.execute( + &format!("UPDATE {table} SET dirty = 0, sync_revision = ?2 WHERE id = ?1"), + params![change.id, rev], + )?, + None => conn.execute( + &format!("UPDATE {table} SET dirty = 0 WHERE id = ?1"), + params![change.id], + )?, + }; + Ok(()) +} + +fn forget_pending_delete(conn: &Connection, change: &Change) -> rusqlite::Result<()> { + if change.op != "delete" { + return Ok(()); + } + conn.execute( + "DELETE FROM pending_deletes WHERE entity = ?1 AND id = ?2", + params![change.entity, change.id], + )?; + Ok(()) +} + +/// True when anything is waiting to go up. Cheap enough to call before a cycle. +pub fn has_pending(conn: &Connection) -> rusqlite::Result { + let pending: Option = conn + .query_row( + "SELECT 1 FROM notes WHERE dirty = 1 + UNION ALL SELECT 1 FROM labels WHERE dirty = 1 + UNION ALL SELECT 1 FROM pending_deletes LIMIT 1", + [], + |r| r.get(0), + ) + .optional()?; + Ok(pending.is_some()) +} + +/// Send everything pending, in batches, applying each batch's results before the +/// next is collected. +pub async fn run(db: &Db, base_url: &str, token: &str) -> Result { + let mut total = PushSummary::default(); + + loop { + let batch = { + let conn = db.0.lock().map_err(|e| e.to_string())?; + collect(&conn, BATCH).map_err(|e| e.to_string())? + }; + if batch.is_empty() { + break; + } + + let raw = client::push_changes(base_url, token, &batch).await?; + let results = parse_results(&raw)?; + + let applied = { + let conn = db.0.lock().map_err(|e| e.to_string())?; + apply_results(&conn, &batch, &results).map_err(|e| e.to_string())? + }; + // Everything rejected clears nothing, so the same batch would be collected + // again forever. Stop and report instead. + let progressed = applied.rejected < applied.sent; + total.absorb(applied); + + if !progressed { + break; + } + if total.batches >= MAX_BATCHES { + return Err(format!( + "Stopped after {MAX_BATCHES} push batches without draining the queue." + )); + } + } + + if total.rejected > 0 { + log::warn!( + "push: {} change(s) rejected by the server: {}", + total.rejected, + total.errors.join("; ") + ); + } + log::info!( + "push complete: {} sent ({} created, {} applied, {} kept, {} noop, {} rejected)", + total.sent, + total.created, + total.applied, + total.kept, + total.noop, + total.rejected + ); + Ok(total) +} + +/// Parse the server's reply. Kept next to the shapes it produces. +pub fn parse_results(raw: &str) -> Result, String> { + let parsed: PushResponse = + serde_json::from_str(raw).map_err(|e| format!("Couldn't read the push response: {e}"))?; + Ok(parsed.results) +} + +#[cfg(test)] +mod tests { + use super::*; + use crate::local::schema; + use crate::local::store; + + fn db() -> Connection { + let conn = Connection::open_in_memory().expect("in-memory db"); + schema::migrate(&conn).expect("migrate"); + conn + } + + fn seed_note(conn: &Connection, id: &str, dirty: i64) { + conn.execute( + "INSERT INTO notes (id, title, body, color, kind, position, pinned, archived, + trashed, created_at, updated_at, sync_revision, dirty) + VALUES (?1, 'T', 'B', 'default', 'text', 0, 0, 0, 0, + '2026-07-26T00:00:00.000Z', '2026-07-26T00:00:00.000Z', 3, ?2)", + params![id, dirty], + ) + .expect("seed note"); + } + + fn ok(status: &str, revision: Option) -> PushResult { + PushResult { + id: None, + entity: None, + status: status.to_string(), + sync_revision: revision, + error: None, + } + } + + fn dirty_count(conn: &Connection) -> i64 { + conn.query_row("SELECT COUNT(*) FROM notes WHERE dirty = 1", [], |r| r.get(0)) + .expect("count") + } + + #[test] + fn collects_only_dirty_notes() { + let conn = db(); + seed_note(&conn, "clean", 0); + seed_note(&conn, "dirty", 1); + let batch = collect(&conn, 100).expect("collect"); + assert_eq!(batch.len(), 1); + assert_eq!(batch[0].id, "dirty"); + assert_eq!(batch[0].op, "upsert"); + } + + #[test] + fn sends_only_manual_label_memberships() { + // Tag-sourced labels are re-derived server-side. Sending them as label_ids + // would convert them to manual assignments that survive removing the #tag. + let conn = db(); + seed_note(&conn, "n1", 1); + for (id, name, via_tag) in [("manual", "Manual", 0), ("tagged", "Tagged", 1)] { + conn.execute( + "INSERT INTO labels (id, name, color, created_at, updated_at, dirty) + VALUES (?1, ?2, 'default', '2026-01-01', '2026-01-01', 0)", + params![id, name], + ) + .expect("seed label"); + conn.execute( + "INSERT INTO note_labels (note_id, label_id, via_tag) VALUES ('n1', ?1, ?2)", + params![id, via_tag], + ) + .expect("seed membership"); + } + let batch = collect(&conn, 100).expect("collect"); + let note = batch.iter().find(|c| c.entity == "note").expect("note"); + assert_eq!(note.label_ids.as_deref(), Some(&["manual".to_string()][..])); + } + + #[test] + fn a_local_delete_becomes_a_delete_change() { + let conn = db(); + seed_note(&conn, "n1", 0); + store::delete_forever(&conn, "n1").expect("delete"); + let batch = collect(&conn, 100).expect("collect"); + assert_eq!(batch.len(), 1); + assert_eq!(batch[0].op, "delete"); + assert_eq!(batch[0].entity, "note"); + assert_eq!(batch[0].id, "n1"); + } + + #[test] + fn applied_clears_dirty_and_records_the_revision() { + let conn = db(); + seed_note(&conn, "n1", 1); + let batch = collect(&conn, 100).expect("collect"); + apply_results(&conn, &batch, &[ok("applied", Some(42))]).expect("apply"); + assert_eq!(dirty_count(&conn), 0); + let rev: i64 = conn + .query_row("SELECT sync_revision FROM notes WHERE id = 'n1'", [], |r| { + r.get(0) + }) + .expect("revision"); + assert_eq!(rev, 42); + } + + #[test] + fn kept_clears_dirty_so_it_is_not_pushed_forever() { + // The server has a newer edit. Re-pushing would lose the same comparison + // every time; the following pull adopts the server's version instead. + let conn = db(); + seed_note(&conn, "n1", 1); + let batch = collect(&conn, 100).expect("collect"); + let summary = apply_results(&conn, &batch, &[ok("kept", Some(99))]).expect("apply"); + assert_eq!(summary.kept, 1); + assert_eq!(dirty_count(&conn), 0); + } + + #[test] + fn kept_rewinds_the_cursor_when_the_server_version_is_already_behind_it() { + // Clock skew: a genuinely later local edit can look older, so the server + // keeps its copy at a revision we have ALREADY consumed. Without a rewind the + // next pull skips it and the stale local copy stays on screen silently. + let conn = db(); + seed_note(&conn, "n1", 1); + state::set_cursor(&conn, 100).expect("cursor"); + let batch = collect(&conn, 100).expect("collect"); + apply_results(&conn, &batch, &[ok("kept", Some(40))]).expect("apply"); + assert_eq!(state::read(&conn).expect("state").last_cursor, 39); + } + + #[test] + fn kept_leaves_the_cursor_alone_when_the_server_version_is_ahead() { + let conn = db(); + seed_note(&conn, "n1", 1); + state::set_cursor(&conn, 10).expect("cursor"); + let batch = collect(&conn, 100).expect("collect"); + apply_results(&conn, &batch, &[ok("kept", Some(40))]).expect("apply"); + assert_eq!( + state::read(&conn).expect("state").last_cursor, + 10, + "the pending pull already covers it" + ); + } + + #[test] + fn rejected_stays_dirty_and_is_reported() { + let conn = db(); + seed_note(&conn, "n1", 1); + let batch = collect(&conn, 100).expect("collect"); + let mut bad = ok("rejected", None); + bad.error = Some("name in use".into()); + let summary = apply_results(&conn, &batch, &[bad]).expect("apply"); + assert_eq!(summary.rejected, 1); + assert_eq!(dirty_count(&conn), 1, "a rejected change must be retried"); + assert!(summary.errors[0].contains("name in use")); + } + + #[test] + fn an_acknowledged_delete_drops_its_tombstone() { + let conn = db(); + seed_note(&conn, "n1", 0); + store::delete_forever(&conn, "n1").expect("delete"); + let batch = collect(&conn, 100).expect("collect"); + apply_results(&conn, &batch, &[ok("applied", Some(7))]).expect("apply"); + assert!(!has_pending(&conn).expect("pending")); + } + + #[test] + fn a_noop_delete_also_drops_its_tombstone() { + // Created and deleted entirely offline: the server never saw it. + let conn = db(); + seed_note(&conn, "n1", 1); + store::delete_forever(&conn, "n1").expect("delete"); + let batch = collect(&conn, 100).expect("collect"); + apply_results(&conn, &batch, &[ok("noop", None)]).expect("apply"); + assert!(!has_pending(&conn).expect("pending")); + } + + #[test] + fn merging_labels_marks_the_affected_notes_dirty() { + // The membership change only reaches the server through the note itself. + let conn = db(); + seed_note(&conn, "n1", 0); + for (id, name) in [("src", "Source"), ("dst", "Target")] { + conn.execute( + "INSERT INTO labels (id, name, color, created_at, updated_at, dirty) + VALUES (?1, ?2, 'default', '2026-01-01', '2026-01-01', 0)", + params![id, name], + ) + .expect("seed label"); + } + conn.execute( + "INSERT INTO note_labels (note_id, label_id, via_tag) VALUES ('n1', 'src', 0)", + [], + ) + .expect("seed membership"); + + store::merge_labels(&conn, "src", "dst").expect("merge"); + assert_eq!(dirty_count(&conn), 1, "the note's label set changed"); + } + + #[test] + fn has_pending_is_false_on_a_clean_store() { + let conn = db(); + seed_note(&conn, "n1", 0); + assert!(!has_pending(&conn).expect("pending")); + } + + #[test] + fn parse_results_reads_the_documented_shape() { + let results = parse_results( + r#"{"results":[{"id":"a","entity":"note","status":"created","sync_revision":44}, + {"id":"b","entity":"label","status":"rejected","error":"name in use"}]}"#, + ) + .expect("parse"); + assert_eq!(results.len(), 2); + assert_eq!(results[0].status, "created"); + assert_eq!(results[1].error.as_deref(), Some("name in use")); + } + + #[test] + fn a_delete_change_serializes_without_note_fields() { + let change = Change::delete("note", "n1".into(), "2026-07-26T00:00:00.000Z".into()); + let json = serde_json::to_string(&change).expect("serialize"); + assert!(json.contains("\"op\":\"delete\""), "got {json}"); + assert!(!json.contains("body"), "a delete carries no content: {json}"); + } +}