Skip to main content

max / synckit

21.6 KB · 608 lines History Blame Raw
1 //! Migration/trigger generation from a [`SyncSchema`].
2 //!
3 //! [`SyncSchema::migration_sql`] emits the full sync DDL, the `sync_state` and
4 //! `sync_changelog` bookkeeping tables, the HLC ledger, and one
5 //! INSERT/UPDATE/DELETE trigger set per table, derived entirely from the
6 //! manifest. Because the trigger's `json_object(...)` projection, the row-id
7 //! expression, and (later) the apply-side binding all read the same `columns`
8 //! and `pk`, the historical trigger/snapshot/whitelist drift is designed out.
9 //!
10 //! The generated form is deliberately *uniform* rather than a byte-for-byte copy
11 //! of any one app's hand-written triggers (which differ in naming, integer-key
12 //! casts, and whether a DELETE stores `NULL` or the key): every DELETE stores the
13 //! primary key in `data` so the pull side reconstructs the WHERE clause the same
14 //! way regardless of `RowIdScheme`. The golden tests assert *behaviour*, run the
15 //! generated triggers, mutate, inspect `sync_changelog`, not string identity.
16
17 use super::schema::{DeleteMode, RowIdScheme, SyncMode, SyncSchema, SyncTable};
18
19 /// Bookkeeping tables shared by every schema.
20 const BASE_TABLES: &str = "\
21 CREATE TABLE IF NOT EXISTS sync_state (
22 key TEXT PRIMARY KEY NOT NULL,
23 value TEXT NOT NULL
24 );
25 INSERT OR IGNORE INTO sync_state (key, value) VALUES
26 ('device_id', ''),
27 ('pull_cursor', '0'),
28 ('applying_remote', '0'),
29 ('auto_sync_enabled', '1'),
30 ('sync_interval_minutes', '15'),
31 ('last_sync_at', ''),
32 ('initial_snapshot_done', '0');
33
34 CREATE TABLE IF NOT EXISTS sync_changelog (
35 id INTEGER PRIMARY KEY AUTOINCREMENT,
36 table_name TEXT NOT NULL,
37 op TEXT NOT NULL,
38 row_id TEXT NOT NULL,
39 timestamp TEXT NOT NULL DEFAULT (strftime('%Y-%m-%dT%H:%M:%fZ', 'now')),
40 data TEXT,
41 pushed INTEGER NOT NULL DEFAULT 0,
42 hlc_wall INTEGER,
43 hlc_counter INTEGER,
44 -- SyncKit Groups scope: '' = personal, otherwise the group id this change
45 -- belongs to. Set by the triggers from the row's group_scoped column. An
46 -- existing (pre-Groups) changelog is migrated to add this column by
47 -- `ensure_scope_schema` (db.rs), since CREATE IF NOT EXISTS cannot alter it.
48 scope TEXT NOT NULL DEFAULT ''
49 );
50 CREATE INDEX IF NOT EXISTS idx_sync_changelog_unpushed
51 ON sync_changelog(pushed) WHERE pushed = 0;
52
53 -- Per-scope pull cursor. '' holds the personal cursor, migrated once from the
54 -- pre-Groups single `pull_cursor` sync_state value; each group id gets its own.
55 CREATE TABLE IF NOT EXISTS sync_scope_cursor (
56 scope TEXT PRIMARY KEY NOT NULL,
57 cursor INTEGER NOT NULL
58 );
59 INSERT OR IGNORE INTO sync_scope_cursor (scope, cursor)
60 VALUES ('', CAST((SELECT value FROM sync_state WHERE key = 'pull_cursor') AS INTEGER));
61
62 -- Committed HLC per row (conflict gating).
63 CREATE TABLE IF NOT EXISTS sync_committed_hlc (
64 table_name TEXT NOT NULL,
65 row_id TEXT NOT NULL,
66 hlc_wall INTEGER NOT NULL,
67 hlc_counter INTEGER NOT NULL,
68 hlc_node TEXT NOT NULL,
69 PRIMARY KEY (table_name, row_id)
70 ) WITHOUT ROWID;
71 ";
72
73 /// Provisioned once, never synced. Only emitted when a table hashes its row-id.
74 const SALT_SEED: &str = "\nINSERT OR IGNORE INTO sync_state (key, value) VALUES
75 ('row_id_salt', lower(hex(randomblob(32))));\n";
76
77 /// The changelog echo-suppression guard shared by every trigger.
78 const APPLYING_GUARD: &str = "(SELECT value FROM sync_state WHERE key = 'applying_remote') != '1'";
79
80 impl SyncSchema {
81 /// Emit the complete sync DDL for this schema: bookkeeping tables, the HLC
82 /// ledger, an optional `row_id_salt`, and every table's triggers.
83 pub fn migration_sql(&self) -> String {
84 let mut out = String::from(BASE_TABLES);
85 if self.any_hashed() {
86 out.push_str(SALT_SEED);
87 }
88 for table in &self.tables {
89 out.push('\n');
90 out.push_str(&triggers_for(table));
91 }
92 out
93 }
94 }
95
96 /// The wire row-id expression for a table under the given row alias (`NEW`/`OLD`
97 /// in a trigger, or the table name in the initial-snapshot SELECT). Shared with
98 /// [`super::sync::create_initial_snapshot`] so the snapshot produces byte-identical
99 /// row-ids to the triggers.
100 pub(crate) fn row_id_expr(table: &SyncTable, alias: &str) -> String {
101 let concat = table
102 .pk
103 .iter()
104 .map(|c| format!("CAST({alias}.{c} AS TEXT)"))
105 .collect::<Vec<_>>()
106 .join(" || ':' || ");
107 match table.row_id {
108 RowIdScheme::PrimaryKey => concat,
109 RowIdScheme::Hashed => format!(
110 "hash_row_id((SELECT value FROM sync_state WHERE key = 'row_id_salt'), {concat})"
111 ),
112 }
113 }
114
115 /// A `json_object('c', ALIAS.c, ...)` projection over `cols`. Shared with the
116 /// initial-snapshot SELECT so snapshot payloads match trigger payloads exactly.
117 pub(crate) fn json_object(alias: &str, cols: &[&str]) -> String {
118 let pairs = cols
119 .iter()
120 .map(|c| format!("'{c}', {alias}.{c}"))
121 .collect::<Vec<_>>()
122 .join(", ");
123 format!("json_object({pairs})")
124 }
125
126 /// The `WHEN` clause: the applying-remote guard, plus this table's exclusion
127 /// predicate (with `{row}` bound to `alias`) if any.
128 fn when_clause(table: &SyncTable, alias: &str) -> String {
129 match table.exclude_where {
130 Some(pred) => format!("{APPLYING_GUARD} AND ({})", pred.replace("{row}", alias)),
131 None => APPLYING_GUARD.to_string(),
132 }
133 }
134
135 /// One `CREATE TRIGGER` statement (with a preceding idempotent DROP).
136 fn trigger(name: &str, timing: &str, table: &str, when: &str, values: &str) -> String {
137 format!(
138 "DROP TRIGGER IF EXISTS {name};\n\
139 CREATE TRIGGER {name}\n\
140 AFTER {timing} ON {table}\n\
141 WHEN {when}\n\
142 BEGIN\n\
143 INSERT INTO sync_changelog (table_name, op, row_id, data, scope)\n\
144 VALUES {values};\n\
145 END;\n"
146 )
147 }
148
149 /// The changelog `scope` expression for a table under the given row alias. A
150 /// `group_scoped` table reads its provenance column (`NULL` -> personal `''`);
151 /// every other table is personal-only, so its scope is the literal `''`.
152 fn scope_expr(table: &SyncTable, alias: &str) -> String {
153 match table.group_scope {
154 Some(col) => format!("COALESCE({alias}.{col}, '')"),
155 None => "''".to_string(),
156 }
157 }
158
159 /// The INSERT/UPDATE/DELETE trigger set for one table.
160 fn triggers_for(table: &SyncTable) -> String {
161 let name = table.name;
162 let emitted = table.emitted_columns();
163 let mut out = String::new();
164
165 // INSERT, only for Full-mode tables. A PartialUpdate table never inserts
166 // (it only reflects state changes on rows that already exist locally).
167 if matches!(table.mode, SyncMode::Full) {
168 let values = format!(
169 "('{name}', 'INSERT', {}, {}, {})",
170 row_id_expr(table, "NEW"),
171 json_object("NEW", &emitted),
172 scope_expr(table, "NEW"),
173 );
174 out.push_str(&trigger(
175 &format!("sync_{name}_insert"),
176 "INSERT",
177 name,
178 &when_clause(table, "NEW"),
179 &values,
180 ));
181 }
182
183 // UPDATE, always. PartialUpdate additionally guards on one of its `set`
184 // columns actually changing (null-safe `IS NOT`).
185 let update_when = match &table.mode {
186 SyncMode::Full => when_clause(table, "NEW"),
187 SyncMode::PartialUpdate { set } => {
188 let changed = set
189 .iter()
190 .map(|c| format!("OLD.{c} IS NOT NEW.{c}"))
191 .collect::<Vec<_>>()
192 .join(" OR ");
193 format!("{} AND ({changed})", when_clause(table, "NEW"))
194 }
195 };
196 let update_values = format!(
197 "('{name}', 'UPDATE', {}, {}, {})",
198 row_id_expr(table, "NEW"),
199 json_object("NEW", &emitted),
200 scope_expr(table, "NEW"),
201 );
202 out.push_str(&trigger(
203 &format!("sync_{name}_update"),
204 "UPDATE",
205 name,
206 &update_when,
207 &update_values,
208 ));
209
210 // DELETE, generated unless deletes are ignored. Tombstone tables still emit
211 // a normal export trigger (a *local* delete must propagate); the tombstone is
212 // an apply-side concern. The payload carries the primary key so the pull side
213 // reconstructs the WHERE clause without the (possibly hashed) row-id.
214 if !matches!(table.deletes, DeleteMode::Ignore) {
215 let delete_values = format!(
216 "('{name}', 'DELETE', {}, {}, {})",
217 row_id_expr(table, "OLD"),
218 json_object("OLD", table.pk),
219 scope_expr(table, "OLD"),
220 );
221 out.push_str(&trigger(
222 &format!("sync_{name}_delete"),
223 "DELETE",
224 name,
225 &when_clause(table, "OLD"),
226 &delete_values,
227 ));
228 }
229
230 out
231 }
232
233 #[cfg(test)]
234 mod tests {
235 use super::super::schema::{SyncSchema, SyncTable};
236 use rusqlite::Connection;
237 use sha2::{Digest, Sha256};
238
239 /// A row observed in `sync_changelog`.
240 #[derive(Debug, PartialEq, Eq)]
241 struct Entry {
242 table: String,
243 op: String,
244 row_id: String,
245 data: Option<String>,
246 }
247
248 /// Build an in-memory DB with the domain tables the tests mutate, register a
249 /// deterministic `hash_row_id`, then run the generated migration.
250 fn setup(schema: &SyncSchema) -> Connection {
251 let conn = Connection::open_in_memory().unwrap();
252 // Deterministic stand-in for the engine's real hash_row_id (B2). Any
253 // stable non-identity function proves the row-id was hashed, not leaked.
254 conn.create_scalar_function(
255 "hash_row_id",
256 2,
257 rusqlite::functions::FunctionFlags::SQLITE_DETERMINISTIC,
258 |ctx| {
259 let salt: String = ctx.get(0)?;
260 let key: String = ctx.get(1)?;
261 let mut h = Sha256::new();
262 h.update(salt.as_bytes());
263 h.update(key.as_bytes());
264 Ok(hex::encode(h.finalize()))
265 },
266 )
267 .unwrap();
268
269 conn.execute_batch(
270 "
271 CREATE TABLE note (id TEXT PRIMARY KEY, name TEXT);
272 CREATE TABLE samp (hash TEXT PRIMARY KEY, name TEXT, deleted_at INTEGER);
273 CREATE TABLE tags (sample_hash TEXT, tag TEXT, PRIMARY KEY (sample_hash, tag));
274 CREATE TABLE items (id TEXT PRIMARY KEY, is_read INTEGER, is_starred INTEGER, title TEXT);
275 CREATE TABLE cfg (key TEXT PRIMARY KEY, value TEXT);
276 ",
277 )
278 .unwrap();
279
280 conn.execute_batch(&schema.migration_sql()).unwrap();
281 conn
282 }
283
284 /// The representative manifest: one table per trigger-affecting policy.
285 fn schema() -> SyncSchema {
286 SyncSchema::new(vec![
287 SyncTable::full("note", &["id", "name"]),
288 SyncTable::full("samp", &["hash", "name", "deleted_at"])
289 .pk(&["hash"])
290 .hashed()
291 .tombstone("deleted_at"),
292 SyncTable::full("tags", &["sample_hash", "tag"])
293 .pk(&["sample_hash", "tag"])
294 .hashed(),
295 SyncTable::full("items", &["id", "is_read", "is_starred"])
296 .partial_update(&["is_read", "is_starred"])
297 .ignore_deletes(),
298 SyncTable::full("cfg", &["key", "value"])
299 .pk(&["key"])
300 .exclude_where("{row}.key NOT LIKE 'sync\\_%' ESCAPE '\\'"),
301 ])
302 }
303
304 fn changelog(conn: &Connection) -> Vec<Entry> {
305 let mut stmt = conn
306 .prepare("SELECT table_name, op, row_id, data FROM sync_changelog ORDER BY id")
307 .unwrap();
308 let rows = stmt
309 .query_map([], |r| {
310 Ok(Entry {
311 table: r.get(0)?,
312 op: r.get(1)?,
313 row_id: r.get(2)?,
314 data: r.get(3)?,
315 })
316 })
317 .unwrap();
318 rows.map(|r| r.unwrap()).collect()
319 }
320
321 #[test]
322 fn full_cleartext_roundtrips_insert_update_delete() {
323 let conn = setup(&schema());
324 conn.execute("INSERT INTO note (id, name) VALUES ('n1', 'hello')", [])
325 .unwrap();
326 conn.execute("UPDATE note SET name = 'bye' WHERE id = 'n1'", [])
327 .unwrap();
328 conn.execute("DELETE FROM note WHERE id = 'n1'", [])
329 .unwrap();
330
331 let log = changelog(&conn);
332 assert_eq!(log.len(), 3);
333 assert_eq!(
334 (log[0].op.as_str(), log[0].row_id.as_str()),
335 ("INSERT", "n1")
336 );
337 assert!(
338 log[0]
339 .data
340 .as_deref()
341 .unwrap()
342 .contains("\"name\":\"hello\"")
343 );
344 assert_eq!(log[1].op, "UPDATE");
345 // Cleartext delete: row_id is the key, data carries the PK (not NULL).
346 assert_eq!(
347 (log[2].op.as_str(), log[2].row_id.as_str()),
348 ("DELETE", "n1")
349 );
350 assert_eq!(log[2].data.as_deref(), Some("{\"id\":\"n1\"}"));
351 }
352
353 #[test]
354 fn applying_remote_flag_suppresses_capture() {
355 let conn = setup(&schema());
356 conn.execute(
357 "UPDATE sync_state SET value = '1' WHERE key = 'applying_remote'",
358 [],
359 )
360 .unwrap();
361 conn.execute("INSERT INTO note (id, name) VALUES ('n2', 'x')", [])
362 .unwrap();
363 assert!(changelog(&conn).is_empty(), "echo must not be captured");
364 }
365
366 #[test]
367 fn hashed_rowid_hides_key_and_delete_carries_pk() {
368 let conn = setup(&schema());
369 conn.execute("INSERT INTO samp (hash, name) VALUES ('deadbeef', 's')", [])
370 .unwrap();
371 conn.execute("DELETE FROM samp WHERE hash = 'deadbeef'", [])
372 .unwrap();
373
374 let log = changelog(&conn);
375 assert_eq!(log.len(), 2);
376 // The wire row-id must not be the cleartext content hash.
377 assert_ne!(log[0].row_id, "deadbeef");
378 assert_eq!(log[0].row_id.len(), 64, "sha256 hex");
379 assert!(
380 log[0]
381 .data
382 .as_deref()
383 .unwrap()
384 .contains("\"hash\":\"deadbeef\"")
385 );
386 // Tombstone tables STILL emit a normal export DELETE; row-id hashed, PK in data.
387 assert_eq!(log[1].op, "DELETE");
388 assert_eq!(log[1].row_id, log[0].row_id);
389 assert_eq!(log[1].data.as_deref(), Some("{\"hash\":\"deadbeef\"}"));
390 }
391
392 #[test]
393 fn hashed_composite_key_concatenates_before_hashing() {
394 let conn = setup(&schema());
395 conn.execute(
396 "INSERT INTO tags (sample_hash, tag) VALUES ('abc', 'kick')",
397 [],
398 )
399 .unwrap();
400 let log = changelog(&conn);
401 assert_eq!(log.len(), 1);
402 // Composite key "abc:kick" is concatenated then hashed; both parts
403 // survive in data. Salt is random per run, so assert shape not value.
404 assert_eq!(log[0].row_id.len(), 64);
405 assert!(
406 log[0]
407 .data
408 .as_deref()
409 .unwrap()
410 .contains("\"sample_hash\":\"abc\"")
411 );
412 assert!(log[0].data.as_deref().unwrap().contains("\"tag\":\"kick\""));
413 }
414
415 #[test]
416 fn partial_update_is_update_only_and_change_gated() {
417 let conn = setup(&schema());
418 // Insert never captured (no INSERT trigger for PartialUpdate).
419 conn.execute(
420 "INSERT INTO items (id, is_read, is_starred, title) VALUES ('i1', 0, 0, 't')",
421 [],
422 )
423 .unwrap();
424 assert!(changelog(&conn).is_empty());
425
426 // Updating a non-set column does not fire (change guard on is_read/is_starred).
427 conn.execute("UPDATE items SET title = 't2' WHERE id = 'i1'", [])
428 .unwrap();
429 assert!(changelog(&conn).is_empty());
430
431 // Updating a set column fires exactly one UPDATE carrying pk + set cols.
432 conn.execute("UPDATE items SET is_read = 1 WHERE id = 'i1'", [])
433 .unwrap();
434 let log = changelog(&conn);
435 assert_eq!(log.len(), 1);
436 assert_eq!(
437 (log[0].op.as_str(), log[0].row_id.as_str()),
438 ("UPDATE", "i1")
439 );
440 let data = log[0].data.as_deref().unwrap();
441 assert!(data.contains("\"is_read\":1") && data.contains("\"is_starred\":0"));
442 }
443
444 #[test]
445 fn ignore_deletes_emits_no_delete_row() {
446 let conn = setup(&schema());
447 conn.execute(
448 "INSERT INTO items (id, is_read, is_starred) VALUES ('i2', 1, 0)",
449 [],
450 )
451 .unwrap();
452 conn.execute("UPDATE items SET is_read = 0 WHERE id = 'i2'", [])
453 .unwrap(); // baseline capture
454 let before = changelog(&conn).len();
455 conn.execute("DELETE FROM items WHERE id = 'i2'", [])
456 .unwrap();
457 assert_eq!(
458 changelog(&conn).len(),
459 before,
460 "ignored delete must not capture"
461 );
462 }
463
464 #[test]
465 fn exclude_where_filters_symmetrically_on_export() {
466 let conn = setup(&schema());
467 conn.execute(
468 "INSERT INTO cfg (key, value) VALUES ('sync_cursor', '5')",
469 [],
470 )
471 .unwrap(); // excluded
472 conn.execute("INSERT INTO cfg (key, value) VALUES ('theme', 'dark')", [])
473 .unwrap(); // included
474 let log = changelog(&conn);
475 assert_eq!(log.len(), 1);
476 assert_eq!(log[0].row_id, "theme");
477 }
478
479 /// A row observed in `sync_changelog` including its scope.
480 #[derive(Debug, PartialEq, Eq)]
481 struct ScopedEntry {
482 op: String,
483 row_id: String,
484 scope: String,
485 }
486
487 fn scoped_changelog(conn: &Connection) -> Vec<ScopedEntry> {
488 let mut stmt = conn
489 .prepare("SELECT op, row_id, scope FROM sync_changelog ORDER BY id")
490 .unwrap();
491 let rows = stmt
492 .query_map([], |r| {
493 Ok(ScopedEntry {
494 op: r.get(0)?,
495 row_id: r.get(1)?,
496 scope: r.get(2)?,
497 })
498 })
499 .unwrap();
500 rows.map(|r| r.unwrap()).collect()
501 }
502
503 #[test]
504 fn group_scoped_table_stamps_scope_from_group_id() {
505 let conn = Connection::open_in_memory().unwrap();
506 conn.execute_batch(
507 "CREATE TABLE task (id TEXT PRIMARY KEY, name TEXT, group_id TEXT);
508 CREATE TABLE cfg (key TEXT PRIMARY KEY, value TEXT);",
509 )
510 .unwrap();
511 let schema = SyncSchema::new(vec![
512 SyncTable::full("task", &["id", "name", "group_id"]).group_scoped("group_id"),
513 SyncTable::full("cfg", &["key", "value"]).pk(&["key"]),
514 ]);
515 conn.execute_batch(&schema.migration_sql()).unwrap();
516
517 // A personal task row (group_id NULL) -> scope ''.
518 conn.execute(
519 "INSERT INTO task (id, name, group_id) VALUES ('t1', 'a', NULL)",
520 [],
521 )
522 .unwrap();
523 // A group task row -> scope = the group id.
524 conn.execute(
525 "INSERT INTO task (id, name, group_id) VALUES ('t2', 'b', 'grp-9')",
526 [],
527 )
528 .unwrap();
529 // Moving a task into a group is an UPDATE that re-scopes it.
530 conn.execute("UPDATE task SET group_id = 'grp-9' WHERE id = 't1'", [])
531 .unwrap();
532 // A personal-only (non-group_scoped) table is always scope ''.
533 conn.execute("INSERT INTO cfg (key, value) VALUES ('theme', 'dark')", [])
534 .unwrap();
535 // A delete carries the group scope of the deleted row (OLD).
536 conn.execute("DELETE FROM task WHERE id = 't2'", [])
537 .unwrap();
538
539 let log = scoped_changelog(&conn);
540 assert_eq!(
541 log,
542 vec![
543 ScopedEntry {
544 op: "INSERT".into(),
545 row_id: "t1".into(),
546 scope: String::new()
547 },
548 ScopedEntry {
549 op: "INSERT".into(),
550 row_id: "t2".into(),
551 scope: "grp-9".into()
552 },
553 ScopedEntry {
554 op: "UPDATE".into(),
555 row_id: "t1".into(),
556 scope: "grp-9".into()
557 },
558 ScopedEntry {
559 op: "INSERT".into(),
560 row_id: "theme".into(),
561 scope: String::new()
562 },
563 ScopedEntry {
564 op: "DELETE".into(),
565 row_id: "t2".into(),
566 scope: "grp-9".into()
567 },
568 ]
569 );
570 }
571
572 #[test]
573 fn scope_cursor_seeded_from_legacy_pull_cursor() {
574 let conn = Connection::open_in_memory().unwrap();
575 let schema = SyncSchema::new(vec![SyncTable::full("note", &["id", "name"])]);
576 conn.execute_batch("CREATE TABLE note (id TEXT PRIMARY KEY, name TEXT);")
577 .unwrap();
578 conn.execute_batch(&schema.migration_sql()).unwrap();
579 // Personal scope cursor exists, seeded from pull_cursor's default '0'.
580 let personal: i64 = conn
581 .query_row(
582 "SELECT cursor FROM sync_scope_cursor WHERE scope = ''",
583 [],
584 |r| r.get(0),
585 )
586 .unwrap();
587 assert_eq!(personal, 0);
588 }
589
590 #[test]
591 fn salt_seeded_only_when_a_table_is_hashed() {
592 // schema() has hashed tables → salt present.
593 let with = setup(&schema());
594 let salt: i64 = with
595 .query_row(
596 "SELECT COUNT(*) FROM sync_state WHERE key = 'row_id_salt'",
597 [],
598 |r| r.get(0),
599 )
600 .unwrap();
601 assert_eq!(salt, 1);
602
603 // A schema with no hashed table must not seed a salt.
604 let plain = SyncSchema::new(vec![SyncTable::full("note", &["id", "name"])]);
605 assert!(!plain.migration_sql().contains("row_id_salt"));
606 }
607 }
608