Skip to main content

max / synckit

30.4 KB · 934 lines History Blame Raw
1 //! Applying resolved remote changes to the local database.
2 //!
3 //! [`apply_remote_changes`] takes changes the conflict layer has already resolved
4 //! and writes them, in foreign-key-safe order, inside the crash-safe
5 //! `applying_remote` transaction from [`super::db`]. All per-table policy declared
6 //! in the [`SyncSchema`](super::schema::SyncSchema) resolves here: `Full` vs
7 //! `PartialUpdate` upserts, `Hard`/`Ignore`/`Tombstone` deletes, `preserve_local`
8 //! secrets, `insert_defaults`, NOT-NULL null-tolerance, the `exclude_where` import
9 //! guard, hashed-row reconstruction from the payload, and `references_unsynced`
10 //! FK relaxation.
11 //!
12 //! A single row failing a constraint is skipped and logged, never fatal, so a
13 //! poison row cannot wedge cursor advancement; a genuine (non-constraint) SQLite
14 //! error rolls the whole batch back.
15
16 use std::collections::{HashMap, HashSet};
17
18 use rusqlite::types::ToSql;
19 use rusqlite::{Connection, ErrorCode, Transaction};
20 use serde_json::Value;
21 use tracing::warn;
22
23 use super::db::with_applying_remote;
24 use super::schema::{DeleteMode, SyncMode, SyncSchema, SyncTable};
25 use crate::error::Result;
26 use crate::types::{ChangeEntry, ChangeOp};
27
28 /// A JSON null to borrow when a payload omits a whitelisted column.
29 static JSON_NULL: Value = Value::Null;
30
31 /// Summary of an apply pass.
32 #[derive(Debug, Default, PartialEq, Eq)]
33 pub struct ApplyOutcome {
34 /// Rows written (insert/update/delete/tombstone).
35 pub applied: usize,
36 /// Rows intentionally dropped or unappliable (excluded, unknown table,
37 /// missing key, or a skipped constraint violation).
38 pub skipped: usize,
39 /// Distinct tables an applied change touched, for selective UI invalidation.
40 pub changed_tables: HashSet<String>,
41 }
42
43 /// Apply resolved remote `changes` to `conn` in FK-safe order.
44 ///
45 /// Upserts run parents-first (schema declaration order); deletes run
46 /// children-first (reverse). If any change targets a `references_unsynced` table,
47 /// foreign-key enforcement is disabled for the whole apply, SQLite only honours
48 /// a `foreign_keys` pragma change *outside* a transaction, so it is toggled around
49 /// the `applying_remote` transaction, not inside it.
50 pub fn apply_remote_changes(
51 conn: &mut Connection,
52 schema: &SyncSchema,
53 changes: &[ChangeEntry],
54 scope: &str,
55 ) -> Result<ApplyOutcome> {
56 let by_name: HashMap<&str, &SyncTable> = schema.tables.iter().map(|t| (t.name, t)).collect();
57
58 let fk_off = changes.iter().any(|c| {
59 by_name
60 .get(c.table.as_str())
61 .is_some_and(|t| t.references_unsynced)
62 });
63 if fk_off {
64 conn.execute_batch("PRAGMA foreign_keys=OFF")?;
65 }
66
67 let outcome =
68 with_applying_remote(conn, |tx| apply_inner(tx, schema, &by_name, changes, scope));
69
70 if fk_off {
71 // Restore enforcement regardless of the apply result. A failed restore is
72 // self-healing: the engine re-asserts foreign_keys=ON when it next opens
73 // the connection (see db::configure_connection).
74 if let Err(e) = conn.execute_batch("PRAGMA foreign_keys=ON") {
75 warn!("failed to restore foreign_keys=ON after apply: {e}");
76 }
77 }
78
79 outcome
80 }
81
82 fn apply_inner(
83 tx: &Transaction<'_>,
84 schema: &SyncSchema,
85 by_name: &HashMap<&str, &SyncTable>,
86 changes: &[ChangeEntry],
87 scope: &str,
88 ) -> Result<ApplyOutcome> {
89 let mut out = ApplyOutcome::default();
90
91 // Changes targeting a table the schema doesn't know are dropped (defensive,
92 // a hostile or newer server could send one), not applied.
93 let unknown = changes
94 .iter()
95 .filter(|c| !by_name.contains_key(c.table.as_str()))
96 .count();
97 if unknown > 0 {
98 warn!("dropping {unknown} change(s) for unknown table(s)");
99 out.skipped += unknown;
100 }
101
102 // Precompute NOT NULL columns for each Full-mode table once, not per row.
103 let mut not_null: HashMap<&str, HashSet<String>> = HashMap::new();
104 for t in &schema.tables {
105 if matches!(t.mode, SyncMode::Full) {
106 not_null.insert(t.name, not_null_columns(tx, t.name)?);
107 }
108 }
109
110 // Upserts: parents first.
111 for table in &schema.tables {
112 for change in changes.iter().filter(|c| {
113 c.table == table.name && matches!(c.op, ChangeOp::Insert | ChangeOp::Update)
114 }) {
115 step(
116 &mut out,
117 table,
118 apply_upsert(tx, table, change, &not_null, scope),
119 )?;
120 }
121 }
122
123 // Deletes: children first.
124 for table in schema.tables.iter().rev() {
125 for change in changes
126 .iter()
127 .filter(|c| c.table == table.name && matches!(c.op, ChangeOp::Delete))
128 {
129 step(&mut out, table, apply_delete(tx, table, change))?;
130 }
131 }
132
133 Ok(out)
134 }
135
136 /// Fold one row's result into the outcome. A constraint violation is skipped and
137 /// logged; any other SQLite error is returned so `?` rolls the whole batch back
138 /// (converting to `SyncKitError` at the caller).
139 fn step(
140 out: &mut ApplyOutcome,
141 table: &SyncTable,
142 result: rusqlite::Result<bool>,
143 ) -> rusqlite::Result<()> {
144 match result {
145 Ok(true) => {
146 out.applied += 1;
147 out.changed_tables.insert(table.name.to_string());
148 Ok(())
149 }
150 Ok(false) => {
151 out.skipped += 1;
152 Ok(())
153 }
154 Err(e) if is_constraint_violation(&e) => {
155 warn!(
156 table = table.name,
157 "skipping row on constraint violation: {e}"
158 );
159 out.skipped += 1;
160 Ok(())
161 }
162 Err(e) => Err(e),
163 }
164 }
165
166 /// Names of the NOT NULL columns of `table`, read from the live schema so the
167 /// list can never drift from the migrations.
168 fn not_null_columns(tx: &Transaction<'_>, table: &str) -> rusqlite::Result<HashSet<String>> {
169 let mut stmt = tx.prepare(&format!("PRAGMA table_info({table})"))?;
170 let rows = stmt.query_map([], |r| {
171 let name: String = r.get("name")?;
172 let notnull: i64 = r.get("notnull")?;
173 Ok((name, notnull != 0))
174 })?;
175 let mut set = HashSet::new();
176 for row in rows {
177 let (name, nn) = row?;
178 if nn {
179 set.insert(name);
180 }
181 }
182 Ok(set)
183 }
184
185 fn field<'a>(data: &'a Value, col: &str) -> &'a Value {
186 data.get(col).unwrap_or(&JSON_NULL)
187 }
188
189 /// Bind a JSON value as a SQLite parameter.
190 fn json_to_sql(v: &Value) -> Box<dyn ToSql> {
191 match v {
192 Value::String(s) => Box::new(s.clone()),
193 Value::Number(n) => {
194 if let Some(i) = n.as_i64() {
195 Box::new(i)
196 } else if let Some(f) = n.as_f64() {
197 Box::new(f)
198 } else {
199 Box::new(n.to_string())
200 }
201 }
202 Value::Bool(b) => Box::new(i64::from(*b)),
203 Value::Null => Box::new(rusqlite::types::Null),
204 other => Box::new(other.to_string()),
205 }
206 }
207
208 fn is_constraint_violation(e: &rusqlite::Error) -> bool {
209 matches!(e, rusqlite::Error::SqliteFailure(err, _) if err.code == ErrorCode::ConstraintViolation)
210 }
211
212 // upsert
213
214 fn apply_upsert(
215 tx: &Transaction<'_>,
216 table: &SyncTable,
217 change: &ChangeEntry,
218 not_null: &HashMap<&str, HashSet<String>>,
219 scope: &str,
220 ) -> rusqlite::Result<bool> {
221 let Some(data) = change.data.as_ref().filter(|v| v.is_object()) else {
222 return Ok(false); // an upsert with no object payload cannot be applied
223 };
224 if is_excluded(tx, table, Some(data))? {
225 return Ok(false);
226 }
227 match &table.mode {
228 SyncMode::PartialUpdate { set } => apply_partial_update(tx, table, data, set),
229 SyncMode::Full => {
230 let nn = not_null
231 .get(table.name)
232 .expect("Full table not_null precomputed");
233 apply_full_upsert(tx, table, data, nn, scope)
234 }
235 }
236 }
237
238 fn apply_full_upsert(
239 tx: &Transaction<'_>,
240 table: &SyncTable,
241 data: &Value,
242 not_null: &HashSet<String>,
243 scope: &str,
244 ) -> rusqlite::Result<bool> {
245 // A remote null for a NOT NULL column is invalid input the changelog never
246 // emits; omit such columns so a new row takes the schema default and an
247 // existing row keeps its value. Nullable columns keep null so legitimate
248 // clears still propagate.
249 let present: Vec<&str> = table
250 .columns
251 .iter()
252 .copied()
253 .filter(|c| !(field(data, c).is_null() && not_null.contains(*c)))
254 .collect();
255
256 // INSERT columns = present whitelist columns + any insert_defaults not already
257 // present (e.g. a NOT NULL preserved secret gets its default on first insert).
258 let mut insert_cols: Vec<&str> = present.clone();
259 let mut default_vals: Vec<&str> = Vec::new();
260 for &(col, val) in table.insert_defaults {
261 if !insert_cols.contains(&col) {
262 insert_cols.push(col);
263 default_vals.push(val);
264 }
265 }
266 if insert_cols.is_empty() {
267 return Ok(false);
268 }
269
270 // Group provenance: stamp a group-scoped table's `group_id` column from the
271 // pull scope (never from the payload) so the pulled row is tagged with the
272 // group it came from. Personal scope ("") leaves group_id at its default
273 // (NULL). Injected into the INSERT columns only, never into the ON CONFLICT
274 // SET, so an existing row's group never changes.
275 let group_stamp: Option<&str> = table.group_scope.filter(|_| !scope.is_empty());
276 if let Some(gcol) = group_stamp {
277 insert_cols.push(gcol);
278 }
279
280 let pk: HashSet<&str> = table.pk.iter().copied().collect();
281 let preserve: HashSet<&str> = table.preserve_local.iter().copied().collect();
282 // ON CONFLICT update touches present columns that are neither PK nor a
283 // preserved local secret, so an existing row's secrets are never overwritten.
284 let update_cols: Vec<&str> = present
285 .iter()
286 .copied()
287 .filter(|c| !pk.contains(c) && !preserve.contains(c))
288 .collect();
289
290 let placeholders = (1..=insert_cols.len())
291 .map(|i| format!("?{i}"))
292 .collect::<Vec<_>>()
293 .join(", ");
294
295 let sql = if update_cols.is_empty() {
296 // Every column is part of the PK (or nothing to update), ignore conflicts
297 // rather than DELETE+INSERT, which would cascade to children.
298 format!(
299 "INSERT OR IGNORE INTO {} ({}) VALUES ({})",
300 table.name,
301 insert_cols.join(", "),
302 placeholders
303 )
304 } else {
305 let set = update_cols
306 .iter()
307 .map(|c| format!("{c} = excluded.{c}"))
308 .collect::<Vec<_>>()
309 .join(", ");
310 format!(
311 "INSERT INTO {} ({}) VALUES ({}) ON CONFLICT({}) DO UPDATE SET {}",
312 table.name,
313 insert_cols.join(", "),
314 placeholders,
315 table.pk.join(", "),
316 set
317 )
318 };
319
320 let mut params: Vec<Box<dyn ToSql>> = present
321 .iter()
322 .map(|c| json_to_sql(field(data, c)))
323 .collect();
324 params.extend(
325 default_vals
326 .iter()
327 .map(|v| Box::new((*v).to_string()) as Box<dyn ToSql>),
328 );
329 if group_stamp.is_some() {
330 params.push(Box::new(scope.to_string()) as Box<dyn ToSql>);
331 }
332
333 exec(tx, &sql, &params)?;
334 Ok(true)
335 }
336
337 fn apply_partial_update(
338 tx: &Transaction<'_>,
339 table: &SyncTable,
340 data: &Value,
341 set: &[&str],
342 ) -> rusqlite::Result<bool> {
343 let Some(pk) = pk_bindings(table, data, None) else {
344 return Ok(false); // no primary key to target
345 };
346 let set_present: Vec<&str> = set
347 .iter()
348 .copied()
349 .filter(|c| data.get(*c).is_some())
350 .collect();
351 if set_present.is_empty() {
352 return Ok(false);
353 }
354
355 let mut params: Vec<Box<dyn ToSql>> = Vec::new();
356 let mut idx = 1;
357 let set_sql = set_present
358 .iter()
359 .map(|c| {
360 let s = format!("{c} = ?{idx}");
361 idx += 1;
362 params.push(json_to_sql(field(data, c)));
363 s
364 })
365 .collect::<Vec<_>>()
366 .join(", ");
367 let where_sql = pk
368 .into_iter()
369 .map(|(c, v)| {
370 let s = format!("{c} = ?{idx}");
371 idx += 1;
372 params.push(v);
373 s
374 })
375 .collect::<Vec<_>>()
376 .join(" AND ");
377
378 let sql = format!("UPDATE {} SET {set_sql} WHERE {where_sql}", table.name);
379 exec(tx, &sql, &params)?;
380 Ok(true)
381 }
382
383 // delete
384
385 fn apply_delete(
386 tx: &Transaction<'_>,
387 table: &SyncTable,
388 change: &ChangeEntry,
389 ) -> rusqlite::Result<bool> {
390 if matches!(table.deletes, DeleteMode::Ignore) {
391 return Ok(false);
392 }
393 if is_excluded(tx, table, change.data.as_ref())? {
394 return Ok(false);
395 }
396 let Some(pk) = pk_bindings(
397 table,
398 change.data.as_ref().unwrap_or(&JSON_NULL),
399 Some(&change.row_id),
400 ) else {
401 return Ok(false); // can't reconstruct the key, skip
402 };
403
404 let mut params: Vec<Box<dyn ToSql>> = Vec::new();
405 let mut idx = 1;
406 let where_sql = pk
407 .into_iter()
408 .map(|(c, v)| {
409 let s = format!("{c} = ?{idx}");
410 idx += 1;
411 params.push(v);
412 s
413 })
414 .collect::<Vec<_>>()
415 .join(" AND ");
416
417 let sql = match &table.deletes {
418 // A remote delete on a tombstone table must not hard-DELETE: a local
419 // ON DELETE CASCADE would wipe organizational state the remote never
420 // meant to touch. COALESCE keeps the earliest tombstone instant.
421 DeleteMode::Tombstone { column } => format!(
422 "UPDATE {} SET {column} = COALESCE({column}, unixepoch()) WHERE {where_sql}",
423 table.name
424 ),
425 DeleteMode::Hard => format!("DELETE FROM {} WHERE {where_sql}", table.name),
426 DeleteMode::Ignore => unreachable!("returned above"),
427 };
428 exec(tx, &sql, &params)?;
429 Ok(true)
430 }
431
432 /// Reconstruct the primary-key bindings for a row.
433 ///
434 /// Prefers the payload (our generated triggers always carry the PK in `data`,
435 /// even for hashed rows whose wire row-id is opaque). Falls back to the wire
436 /// `row_id` for single-key tables written by an older client that stored the key
437 /// there; a composite key with no payload cannot be reconstructed and yields
438 /// `None`.
439 fn pk_bindings(
440 table: &SyncTable,
441 data: &Value,
442 row_id: Option<&str>,
443 ) -> Option<Vec<(&'static str, Box<dyn ToSql>)>> {
444 if let Some(obj) = data.as_object() {
445 let mut out = Vec::with_capacity(table.pk.len());
446 let mut complete = true;
447 for pk in table.pk {
448 match obj.get(*pk) {
449 Some(v) if !v.is_null() => out.push((*pk, json_to_sql(v))),
450 _ => {
451 complete = false;
452 break;
453 }
454 }
455 }
456 if complete && !out.is_empty() {
457 return Some(out);
458 }
459 }
460 // Payload didn't carry the key, fall back to the wire row_id (single-PK only).
461 match (row_id, table.pk) {
462 (Some(id), [pk]) => Some(vec![(*pk, Box::new(id.to_string()) as Box<dyn ToSql>)]),
463 _ => None,
464 }
465 }
466
467 // exclusion
468
469 /// Whether `table`'s `exclude_where` predicate excludes this row. The predicate
470 /// is the *include* condition; a row is excluded when it does not hold. Evaluated
471 /// in SQLite against a one-row derived table so the same predicate text drives
472 /// both the export trigger and this import guard. No predicate, or no payload to
473 /// test, means not excluded.
474 fn is_excluded(
475 tx: &Transaction<'_>,
476 table: &SyncTable,
477 data: Option<&Value>,
478 ) -> rusqlite::Result<bool> {
479 let Some(pred) = table.exclude_where else {
480 return Ok(false);
481 };
482 let Some(obj) = data.and_then(|v| v.as_object()) else {
483 return Ok(false);
484 };
485
486 let cols = table.columns;
487 let derived = cols
488 .iter()
489 .enumerate()
490 .map(|(i, c)| format!("?{} AS {c}", i + 1))
491 .collect::<Vec<_>>()
492 .join(", ");
493 // COALESCE(..., 0): a NULL predicate (e.g. NULL key) is treated as included.
494 let sql = format!(
495 "SELECT COALESCE(NOT ({}), 0) FROM (SELECT {derived}) AS x",
496 pred.replace("{row}", "x")
497 );
498 let params: Vec<Box<dyn ToSql>> = cols
499 .iter()
500 .map(|c| json_to_sql(field_opt(obj, c)))
501 .collect();
502 let excluded: i64 = tx.query_row(&sql, params_slice(&params).as_slice(), |r| r.get(0))?;
503 Ok(excluded != 0)
504 }
505
506 fn field_opt<'a>(obj: &'a serde_json::Map<String, Value>, col: &str) -> &'a Value {
507 obj.get(col).unwrap_or(&JSON_NULL)
508 }
509
510 // exec helpers
511
512 fn params_slice(params: &[Box<dyn ToSql>]) -> Vec<&dyn ToSql> {
513 params.iter().map(std::convert::AsRef::as_ref).collect()
514 }
515
516 fn exec(tx: &Transaction<'_>, sql: &str, params: &[Box<dyn ToSql>]) -> rusqlite::Result<()> {
517 tx.execute(sql, params_slice(params).as_slice())?;
518 Ok(())
519 }
520
521 #[cfg(test)]
522 mod tests {
523 use super::*;
524 use crate::types::{ChangeEntry, ChangeOp, hlc_legacy_floor};
525 use rusqlite::Connection;
526 use serde_json::json;
527
528 use super::super::db::configure_connection;
529 use super::super::schema::{SyncSchema, SyncTable};
530
531 fn upsert(table: &str, row_id: &str, data: Value) -> ChangeEntry {
532 ChangeEntry {
533 table: table.into(),
534 op: ChangeOp::Insert,
535 row_id: row_id.into(),
536 timestamp: chrono::Utc::now(),
537 hlc: hlc_legacy_floor(),
538 data: Some(data),
539 extra: serde_json::Map::default(),
540 }
541 }
542
543 fn delete(table: &str, row_id: &str, data: Value) -> ChangeEntry {
544 ChangeEntry {
545 op: ChangeOp::Delete,
546 ..upsert(table, row_id, data)
547 }
548 }
549
550 fn schema() -> SyncSchema {
551 SyncSchema::new(vec![
552 SyncTable::full("parent", &["id", "name"]),
553 SyncTable::full("child", &["id", "parent_id", "note"]),
554 SyncTable::full("acct", &["id", "name"])
555 .preserve_local(&["secret"])
556 .insert_defaults(&[("secret", "")]),
557 SyncTable::full("tagpair", &["a", "b"]).pk(&["a", "b"]),
558 SyncTable::full("items", &["id", "is_read", "is_starred"])
559 .partial_update(&["is_read", "is_starred"])
560 .ignore_deletes(),
561 SyncTable::full("samp", &["hash", "name", "deleted_at"])
562 .pk(&["hash"])
563 .hashed()
564 .tombstone("deleted_at"),
565 SyncTable::full("cfg", &["key", "value"])
566 .pk(&["key"])
567 .exclude_where("{row}.key NOT LIKE 'sync\\_%' ESCAPE '\\'"),
568 SyncTable::full("reffer", &["id", "ext_id"]).references_unsynced(),
569 ])
570 }
571
572 fn db() -> Connection {
573 let conn = Connection::open_in_memory().unwrap();
574 configure_connection(&conn).unwrap();
575 conn.execute_batch(
576 "
577 CREATE TABLE parent (id TEXT PRIMARY KEY, name TEXT);
578 CREATE TABLE child (id TEXT PRIMARY KEY, parent_id TEXT NOT NULL REFERENCES parent(id) ON DELETE CASCADE, note TEXT);
579 CREATE TABLE acct (id TEXT PRIMARY KEY, name TEXT, secret TEXT NOT NULL);
580 CREATE TABLE tagpair (a TEXT, b TEXT, PRIMARY KEY (a, b));
581 CREATE TABLE items (id TEXT PRIMARY KEY, is_read INTEGER, is_starred INTEGER, title TEXT);
582 CREATE TABLE ghost (id INTEGER PRIMARY KEY);
583 CREATE TABLE samp (hash TEXT PRIMARY KEY, name TEXT, deleted_at INTEGER);
584 CREATE TABLE cfg (key TEXT PRIMARY KEY, value TEXT);
585 CREATE TABLE reffer (id TEXT PRIMARY KEY, ext_id INTEGER NOT NULL REFERENCES ghost(id));
586 ",
587 )
588 .unwrap();
589 let s = schema();
590 conn.execute_batch(&s.migration_sql()).unwrap();
591 conn
592 }
593
594 fn apply(conn: &mut Connection, changes: &[ChangeEntry]) -> ApplyOutcome {
595 apply_remote_changes(conn, &schema(), changes, "").unwrap()
596 }
597
598 #[test]
599 fn full_insert_then_update_via_on_conflict() {
600 let mut conn = db();
601 let o = apply(
602 &mut conn,
603 &[upsert("parent", "p1", json!({"id":"p1","name":"a"}))],
604 );
605 assert_eq!(o.applied, 1);
606 assert!(o.changed_tables.contains("parent"));
607 apply(
608 &mut conn,
609 &[upsert("parent", "p1", json!({"id":"p1","name":"b"}))],
610 );
611 let name: String = conn
612 .query_row("SELECT name FROM parent WHERE id='p1'", [], |r| r.get(0))
613 .unwrap();
614 assert_eq!(name, "b");
615 }
616
617 #[test]
618 fn on_conflict_update_does_not_cascade_to_children() {
619 let mut conn = db();
620 apply(
621 &mut conn,
622 &[
623 upsert("parent", "p1", json!({"id":"p1","name":"a"})),
624 upsert(
625 "child",
626 "c1",
627 json!({"id":"c1","parent_id":"p1","note":"n"}),
628 ),
629 ],
630 );
631 // Re-upsert the parent; the child must survive (ON CONFLICT DO UPDATE, not REPLACE).
632 apply(
633 &mut conn,
634 &[upsert("parent", "p1", json!({"id":"p1","name":"a2"}))],
635 );
636 let kids: i64 = conn
637 .query_row("SELECT COUNT(*) FROM child", [], |r| r.get(0))
638 .unwrap();
639 assert_eq!(kids, 1);
640 }
641
642 #[test]
643 fn preserve_local_and_insert_defaults() {
644 let mut conn = db();
645 // First insert: secret defaults to '' (satisfies NOT NULL); payload never carries it.
646 apply(
647 &mut conn,
648 &[upsert("acct", "a1", json!({"id":"a1","name":"n1"}))],
649 );
650 // Locally the user sets a real secret.
651 conn.execute("UPDATE acct SET secret='hunter2' WHERE id='a1'", [])
652 .unwrap();
653 // A remote update to config columns must NOT clobber the local secret.
654 apply(
655 &mut conn,
656 &[upsert("acct", "a1", json!({"id":"a1","name":"n2"}))],
657 );
658 let (name, secret): (String, String) = conn
659 .query_row("SELECT name, secret FROM acct WHERE id='a1'", [], |r| {
660 Ok((r.get(0)?, r.get(1)?))
661 })
662 .unwrap();
663 assert_eq!(name, "n2");
664 assert_eq!(
665 secret, "hunter2",
666 "preserved secret must survive a remote update"
667 );
668 }
669
670 #[test]
671 fn null_tolerance_omits_not_null_but_keeps_nullable_null() {
672 let mut conn = db();
673 apply(
674 &mut conn,
675 &[upsert("parent", "p1", json!({"id":"p1","name":"start"}))],
676 );
677 // name is nullable → an explicit null clears it.
678 apply(
679 &mut conn,
680 &[upsert("parent", "p1", json!({"id":"p1","name":null}))],
681 );
682 let name: Option<String> = conn
683 .query_row("SELECT name FROM parent WHERE id='p1'", [], |r| r.get(0))
684 .unwrap();
685 assert_eq!(name, None);
686 // A null for a NOT NULL column (child.parent_id) is omitted, so an insert
687 // takes no value for it → constraint violation → skipped, not fatal.
688 let o = apply(
689 &mut conn,
690 &[upsert(
691 "child",
692 "c1",
693 json!({"id":"c1","parent_id":null,"note":"x"}),
694 )],
695 );
696 assert_eq!(o.applied, 0);
697 assert_eq!(o.skipped, 1);
698 }
699
700 #[test]
701 fn all_pk_table_uses_insert_or_ignore() {
702 let mut conn = db();
703 let o = apply(
704 &mut conn,
705 &[upsert("tagpair", "x", json!({"a":"x","b":"y"}))],
706 );
707 assert_eq!(o.applied, 1);
708 // Re-applying the same all-PK row is a no-op, not an error.
709 let o2 = apply(
710 &mut conn,
711 &[upsert("tagpair", "x", json!({"a":"x","b":"y"}))],
712 );
713 assert_eq!(o2.applied, 1); // executed, 0 rows changed, still Ok
714 let n: i64 = conn
715 .query_row("SELECT COUNT(*) FROM tagpair", [], |r| r.get(0))
716 .unwrap();
717 assert_eq!(n, 1);
718 }
719
720 #[test]
721 fn partial_update_touches_only_set_columns() {
722 let mut conn = db();
723 conn.execute(
724 "INSERT INTO items (id, is_read, is_starred, title) VALUES ('i1', 0, 0, 'keep')",
725 [],
726 )
727 .unwrap();
728 apply(
729 &mut conn,
730 &[ChangeEntry {
731 op: ChangeOp::Update,
732 ..upsert("items", "i1", json!({"id":"i1","is_read":1,"is_starred":0}))
733 }],
734 );
735 let (read, title): (i64, String) = conn
736 .query_row("SELECT is_read, title FROM items WHERE id='i1'", [], |r| {
737 Ok((r.get(0)?, r.get(1)?))
738 })
739 .unwrap();
740 assert_eq!(read, 1);
741 assert_eq!(
742 title, "keep",
743 "partial update must not touch non-set columns"
744 );
745 }
746
747 #[test]
748 fn hard_delete_and_ignore_delete() {
749 let mut conn = db();
750 apply(
751 &mut conn,
752 &[upsert("parent", "p1", json!({"id":"p1","name":"a"}))],
753 );
754 let o = apply(&mut conn, &[delete("parent", "p1", json!({"id":"p1"}))]);
755 assert_eq!(o.applied, 1);
756 assert_eq!(
757 conn.query_row("SELECT COUNT(*) FROM parent", [], |r| r.get::<_, i64>(0))
758 .unwrap(),
759 0
760 );
761
762 // items ignore deletes.
763 conn.execute(
764 "INSERT INTO items (id, is_read, is_starred) VALUES ('i1', 1, 0)",
765 [],
766 )
767 .unwrap();
768 let o2 = apply(&mut conn, &[delete("items", "i1", json!({"id":"i1"}))]);
769 assert_eq!(o2.applied, 0);
770 assert_eq!(
771 conn.query_row("SELECT COUNT(*) FROM items", [], |r| r.get::<_, i64>(0))
772 .unwrap(),
773 1
774 );
775 }
776
777 #[test]
778 fn tombstone_delete_sets_column_and_keeps_earliest() {
779 let mut conn = db();
780 conn.execute("INSERT INTO samp (hash, name) VALUES ('h1', 's')", [])
781 .unwrap();
782 // A hashed table's delete carries the PK in data; the opaque row_id is ignored.
783 apply(
784 &mut conn,
785 &[delete("samp", "OPAQUE_HASH", json!({"hash":"h1"}))],
786 );
787 let (present, del): (i64, Option<i64>) = conn
788 .query_row(
789 "SELECT COUNT(*), MAX(deleted_at) FROM samp WHERE hash='h1'",
790 [],
791 |r| Ok((r.get(0)?, r.get(1)?)),
792 )
793 .unwrap();
794 assert_eq!(present, 1, "tombstone keeps the row");
795 let first = del.unwrap();
796 // Re-deleting keeps the earliest instant (COALESCE).
797 apply(
798 &mut conn,
799 &[delete("samp", "OPAQUE_HASH", json!({"hash":"h1"}))],
800 );
801 let second: i64 = conn
802 .query_row("SELECT deleted_at FROM samp WHERE hash='h1'", [], |r| {
803 r.get(0)
804 })
805 .unwrap();
806 assert_eq!(first, second);
807 }
808
809 #[test]
810 fn exclude_where_guards_import_both_ways() {
811 let mut conn = db();
812 let o = apply(
813 &mut conn,
814 &[
815 upsert(
816 "cfg",
817 "sync_cursor",
818 json!({"key":"sync_cursor","value":"9"}),
819 ), // excluded
820 upsert("cfg", "theme", json!({"key":"theme","value":"dark"})), // included
821 ],
822 );
823 assert_eq!(o.applied, 1);
824 let keys: Vec<String> = {
825 let mut s = conn.prepare("SELECT key FROM cfg ORDER BY key").unwrap();
826 s.query_map([], |r| r.get(0))
827 .unwrap()
828 .map(|r| r.unwrap())
829 .collect()
830 };
831 assert_eq!(keys, vec!["theme".to_string()]);
832 // A hostile delete of an excluded key is also dropped.
833 conn.execute(
834 "INSERT INTO cfg (key, value) VALUES ('sync_secret', 'x')",
835 [],
836 )
837 .unwrap();
838 let o2 = apply(
839 &mut conn,
840 &[delete("cfg", "sync_secret", json!({"key":"sync_secret"}))],
841 );
842 assert_eq!(o2.applied, 0);
843 assert_eq!(
844 conn.query_row(
845 "SELECT COUNT(*) FROM cfg WHERE key='sync_secret'",
846 [],
847 |r| r.get::<_, i64>(0)
848 )
849 .unwrap(),
850 1
851 );
852 }
853
854 #[test]
855 fn fk_ordering_parents_before_children_children_before_parents() {
856 let mut conn = db();
857 // Child listed before parent in the batch, but FK enforced, must still apply
858 // because the engine orders upserts parents-first.
859 let o = apply(
860 &mut conn,
861 &[
862 upsert(
863 "child",
864 "c1",
865 json!({"id":"c1","parent_id":"p1","note":"n"}),
866 ),
867 upsert("parent", "p1", json!({"id":"p1","name":"a"})),
868 ],
869 );
870 assert_eq!(o.applied, 2);
871 // Delete both; children-first ordering means the child goes before the parent.
872 let o2 = apply(
873 &mut conn,
874 &[
875 delete("parent", "p1", json!({"id":"p1"})),
876 delete("child", "c1", json!({"id":"c1"})),
877 ],
878 );
879 assert_eq!(o2.applied, 2);
880 }
881
882 #[test]
883 fn references_unsynced_relaxes_fk() {
884 let mut conn = db();
885 // reffer.ext_id points at a ghost row that does not exist and is not synced.
886 // Without FK relaxation this would be a constraint violation.
887 let o = apply(
888 &mut conn,
889 &[upsert("reffer", "r1", json!({"id":"r1","ext_id":999}))],
890 );
891 assert_eq!(
892 o.applied, 1,
893 "references_unsynced disables FK for the apply"
894 );
895 // FK enforcement is restored afterward.
896 let fk: i64 = conn
897 .query_row("PRAGMA foreign_keys", [], |r| r.get(0))
898 .unwrap();
899 assert_eq!(fk, 1);
900 }
901
902 #[test]
903 fn constraint_violation_is_skipped_not_fatal() {
904 let mut conn = db();
905 // First row violates FK (no parent p9); second is valid. Batch must not abort.
906 let o = apply(
907 &mut conn,
908 &[
909 upsert(
910 "child",
911 "bad",
912 json!({"id":"bad","parent_id":"p9","note":"x"}),
913 ),
914 upsert("parent", "p1", json!({"id":"p1","name":"ok"})),
915 ],
916 );
917 assert_eq!(o.applied, 1);
918 assert_eq!(o.skipped, 1);
919 assert_eq!(
920 conn.query_row("SELECT COUNT(*) FROM parent", [], |r| r.get::<_, i64>(0))
921 .unwrap(),
922 1
923 );
924 }
925
926 #[test]
927 fn unknown_table_change_is_skipped() {
928 let mut conn = db();
929 let o = apply(&mut conn, &[upsert("nonexistent", "x", json!({"id":"x"}))]);
930 assert_eq!(o.applied, 0);
931 assert_eq!(o.skipped, 1);
932 }
933 }
934