M10.7c: push + the full sync cycle (task 2106)
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) <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01SreJkbxB4gx8pPsu8QbLPi
This commit is contained in:
@@ -93,7 +93,8 @@ pub fn run() {
|
|||||||
sync::commands::sync_link,
|
sync::commands::sync_link,
|
||||||
sync::commands::sync_unlink,
|
sync::commands::sync_unlink,
|
||||||
sync::commands::sync_status,
|
sync::commands::sync_status,
|
||||||
sync::commands::sync_pull,
|
sync::commands::sync_now,
|
||||||
|
sync::commands::sync_has_pending,
|
||||||
])
|
])
|
||||||
.run(tauri::generate_context!())
|
.run(tauri::generate_context!())
|
||||||
.expect("error while running the ThoughtSync desktop app");
|
.expect("error while running the ThoughtSync desktop app");
|
||||||
|
|||||||
@@ -105,6 +105,22 @@ CREATE TABLE sync_state (
|
|||||||
INSERT INTO sync_state (id) VALUES (1);
|
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.
|
/// Bring the database up to the latest schema. Idempotent.
|
||||||
pub fn migrate(conn: &Connection) -> rusqlite::Result<()> {
|
pub fn migrate(conn: &Connection) -> rusqlite::Result<()> {
|
||||||
conn.execute_batch("PRAGMA foreign_keys = ON;")?;
|
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(SCHEMA_V1)?;
|
||||||
conn.execute_batch("PRAGMA user_version = 1;")?;
|
conn.execute_batch("PRAGMA user_version = 1;")?;
|
||||||
}
|
}
|
||||||
|
if version < 2 {
|
||||||
|
conn.execute_batch(SCHEMA_V2)?;
|
||||||
|
conn.execute_batch("PRAGMA user_version = 2;")?;
|
||||||
|
}
|
||||||
Ok(())
|
Ok(())
|
||||||
}
|
}
|
||||||
|
|||||||
@@ -633,10 +633,25 @@ pub fn restore(conn: &Connection, id: &str) -> rusqlite::Result<Note> {
|
|||||||
}
|
}
|
||||||
|
|
||||||
pub fn delete_forever(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])?;
|
conn.execute("DELETE FROM notes WHERE id = ?1", [id])?;
|
||||||
Ok(())
|
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<Vec<NoteRevision>> {
|
pub fn revisions(conn: &Connection, id: &str) -> rusqlite::Result<Vec<NoteRevision>> {
|
||||||
let mut stmt = conn
|
let mut stmt = conn
|
||||||
.prepare("SELECT id, title, body, created_at FROM note_revisions WHERE note_id = ?1 ORDER BY created_at DESC")?;
|
.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<()> {
|
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])?;
|
conn.execute("DELETE FROM labels WHERE id = ?1", [id])?;
|
||||||
Ok(())
|
Ok(())
|
||||||
}
|
}
|
||||||
@@ -741,6 +757,16 @@ pub fn merge_labels(
|
|||||||
SELECT note_id, ?2, 0 FROM note_labels WHERE label_id = ?1",
|
SELECT note_id, ?2, 0 FROM note_labels WHERE label_id = ?1",
|
||||||
params![source_id, target_id],
|
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])?;
|
conn.execute("DELETE FROM labels WHERE id = ?1", [source_id])?;
|
||||||
load_label(conn, target_id)
|
load_label(conn, target_id)
|
||||||
}
|
}
|
||||||
|
|||||||
@@ -223,6 +223,38 @@ pub async fn fetch_changes(
|
|||||||
.map_err(|e| format!("Couldn't read the change feed from {base_url}: {e}"))
|
.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<T: Serialize>(
|
||||||
|
base_url: &str,
|
||||||
|
token: &str,
|
||||||
|
changes: &[T],
|
||||||
|
) -> Result<String, String> {
|
||||||
|
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.
|
/// The public, unauthenticated endpoint carrying the handshake.
|
||||||
fn config_url(base_url: &str) -> String {
|
fn config_url(base_url: &str) -> String {
|
||||||
format!("{base_url}/api/config")
|
format!("{base_url}/api/config")
|
||||||
|
|||||||
@@ -9,7 +9,8 @@ use tauri::State;
|
|||||||
use crate::local::Db;
|
use crate::local::Db;
|
||||||
use crate::sync::client::{self, Identity, ProbeResult};
|
use crate::sync::client::{self, Identity, ProbeResult};
|
||||||
use crate::sync::compat::Compatibility;
|
use crate::sync::compat::Compatibility;
|
||||||
use crate::sync::pull;
|
use crate::sync::engine;
|
||||||
|
use crate::sync::push;
|
||||||
use crate::sync::state;
|
use crate::sync::state;
|
||||||
|
|
||||||
/// Ask a server who it is, without committing to anything. The UI calls this as the
|
/// 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
|
/// The only sync entry point exposed to the UI, on purpose. Push and pull exist
|
||||||
/// is the ordering the protocol assumes. Run alone against unpushed local edits, the
|
/// separately inside the crate, but offering a bare "pull" would let the UI overwrite
|
||||||
/// server's version lands on top of them — the returned `clobbered_dirty` count
|
/// unsent local edits — the ordering isn't a suggestion, it's what keeps them.
|
||||||
/// reports that rather than hiding it.
|
|
||||||
#[tauri::command]
|
#[tauri::command]
|
||||||
pub async fn sync_pull(db: State<'_, Db>) -> Result<pull::PullSummary, String> {
|
pub async fn sync_now(db: State<'_, Db>) -> Result<engine::SyncOutcome, String> {
|
||||||
let (base_url, token) = credentials(&db)?;
|
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<bool, String> {
|
||||||
|
let conn = db.0.lock().map_err(|e| e.to_string())?;
|
||||||
|
push::has_pending(&conn).map_err(|e| e.to_string())
|
||||||
}
|
}
|
||||||
|
|||||||
@@ -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<SyncOutcome, String> {
|
||||||
|
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 })
|
||||||
|
}
|
||||||
@@ -15,6 +15,8 @@
|
|||||||
pub mod client;
|
pub mod client;
|
||||||
pub mod commands;
|
pub mod commands;
|
||||||
pub mod compat;
|
pub mod compat;
|
||||||
|
pub mod engine;
|
||||||
pub mod pull;
|
pub mod pull;
|
||||||
|
pub mod push;
|
||||||
pub mod state;
|
pub mod state;
|
||||||
pub mod wire;
|
pub mod wire;
|
||||||
|
|||||||
@@ -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<String>,
|
||||||
|
}
|
||||||
|
|
||||||
|
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<String>,
|
||||||
|
#[serde(skip_serializing_if = "Option::is_none")]
|
||||||
|
pub body: Option<String>,
|
||||||
|
#[serde(skip_serializing_if = "Option::is_none")]
|
||||||
|
pub color: Option<String>,
|
||||||
|
#[serde(skip_serializing_if = "Option::is_none")]
|
||||||
|
pub kind: Option<String>,
|
||||||
|
#[serde(skip_serializing_if = "Option::is_none")]
|
||||||
|
pub pinned: Option<bool>,
|
||||||
|
#[serde(skip_serializing_if = "Option::is_none")]
|
||||||
|
pub archived: Option<bool>,
|
||||||
|
#[serde(skip_serializing_if = "Option::is_none")]
|
||||||
|
pub trashed: Option<bool>,
|
||||||
|
#[serde(skip_serializing_if = "Option::is_none")]
|
||||||
|
pub remind_at: Option<String>,
|
||||||
|
#[serde(skip_serializing_if = "Option::is_none")]
|
||||||
|
pub recurrence: Option<String>,
|
||||||
|
#[serde(skip_serializing_if = "Option::is_none")]
|
||||||
|
pub position: Option<i64>,
|
||||||
|
#[serde(skip_serializing_if = "Option::is_none")]
|
||||||
|
pub items: Option<Vec<ItemOut>>,
|
||||||
|
#[serde(skip_serializing_if = "Option::is_none")]
|
||||||
|
pub label_ids: Option<Vec<String>>,
|
||||||
|
#[serde(skip_serializing_if = "Option::is_none")]
|
||||||
|
pub created_at: Option<String>,
|
||||||
|
#[serde(skip_serializing_if = "Option::is_none")]
|
||||||
|
pub name: Option<String>,
|
||||||
|
}
|
||||||
|
|
||||||
|
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<PushResult>,
|
||||||
|
}
|
||||||
|
|
||||||
|
#[derive(Debug, Clone, Deserialize)]
|
||||||
|
pub struct PushResult {
|
||||||
|
#[serde(default)]
|
||||||
|
pub id: Option<String>,
|
||||||
|
#[serde(default)]
|
||||||
|
pub entity: Option<String>,
|
||||||
|
#[serde(default)]
|
||||||
|
pub status: String,
|
||||||
|
#[serde(default)]
|
||||||
|
pub sync_revision: Option<i64>,
|
||||||
|
#[serde(default)]
|
||||||
|
pub error: Option<String>,
|
||||||
|
}
|
||||||
|
|
||||||
|
// --- 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<Vec<Change>> {
|
||||||
|
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<Change>,
|
||||||
|
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<Change>, 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<Change>, limit: usize) -> rusqlite::Result<()> {
|
||||||
|
let remaining = limit.saturating_sub(out.len());
|
||||||
|
let ids: Vec<String> = {
|
||||||
|
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::<rusqlite::Result<Vec<String>>>()?
|
||||||
|
};
|
||||||
|
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<String>,
|
||||||
|
body: String,
|
||||||
|
color: String,
|
||||||
|
kind: String,
|
||||||
|
position: i64,
|
||||||
|
pinned: bool,
|
||||||
|
archived: bool,
|
||||||
|
trashed: bool,
|
||||||
|
remind_at: Option<String>,
|
||||||
|
recurrence: Option<String>,
|
||||||
|
created_at: String,
|
||||||
|
updated_at: String,
|
||||||
|
}
|
||||||
|
|
||||||
|
fn note_row(conn: &Connection, id: &str) -> rusqlite::Result<NoteRow> {
|
||||||
|
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<Change> {
|
||||||
|
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::<rusqlite::Result<Vec<ItemOut>>>()?
|
||||||
|
};
|
||||||
|
|
||||||
|
// 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::<rusqlite::Result<Vec<String>>>()?
|
||||||
|
};
|
||||||
|
|
||||||
|
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<PushSummary> {
|
||||||
|
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<i64> = 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<i64>,
|
||||||
|
) -> 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<bool> {
|
||||||
|
let pending: Option<i64> = 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<PushSummary, String> {
|
||||||
|
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<Vec<PushResult>, 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<i64>) -> 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}");
|
||||||
|
}
|
||||||
|
}
|
||||||
Reference in New Issue
Block a user