Skip to main content

max / synckit

17.8 KB · 486 lines History Blame Raw
1 //! The engine's owned SQLite connection and bookkeeping-state primitives.
2 //!
3 //! `SyncStore` owns a private `rusqlite` connection to the app's database file
4 //! rather than borrowing the app's async pool, so the app's own persistence layer
5 //! is untouched. Under WAL, which the consuming apps already use, the engine's
6 //! short read/apply transactions coexist with the app's connections on the same
7 //! file. This module provides the low-level pieces the rest of the engine builds
8 //! on: opening/configuring that connection, the `sync_state` key-value store, the
9 //! crash-safe `applying_remote` echo-suppression transaction, and device
10 //! registration.
11
12 use std::path::PathBuf;
13 use std::sync::Arc;
14
15 use rusqlite::{Connection, OptionalExtension, TransactionBehavior, functions::FunctionFlags};
16 use sha2::{Digest, Sha256};
17 use uuid::Uuid;
18
19 use crate::error::{Result, SyncKitError};
20 use crate::ids::DeviceId;
21
22 /// Where the engine gets its SQLite connection.
23 ///
24 /// [`DbSource::path`] is the common case, the engine opens the file itself.
25 /// [`DbSource::factory`] lets an app hand back a connection it configured (custom
26 /// pragmas, shared cache, a VFS); the engine still applies its own required setup
27 /// (WAL, `foreign_keys`, the `hash_row_id` function) on top.
28 #[derive(Clone)]
29 pub enum DbSource {
30 /// Open `Connection::open(path)` on demand.
31 Path(PathBuf),
32 /// Call an app-provided factory to obtain a raw connection.
33 Factory(Arc<dyn Fn() -> rusqlite::Result<Connection> + Send + Sync>),
34 }
35
36 impl std::fmt::Debug for DbSource {
37 fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
38 match self {
39 Self::Path(p) => f.debug_tuple("Path").field(p).finish(),
40 Self::Factory(_) => f.write_str("Factory(..)"),
41 }
42 }
43 }
44
45 impl DbSource {
46 /// A source that opens the SQLite file at `path`.
47 pub fn path(path: impl Into<PathBuf>) -> Self {
48 Self::Path(path.into())
49 }
50
51 /// A source that calls `factory` for each connection.
52 pub fn factory(
53 factory: impl Fn() -> rusqlite::Result<Connection> + Send + Sync + 'static,
54 ) -> Self {
55 Self::Factory(Arc::new(factory))
56 }
57
58 /// Open and fully configure a connection: WAL journaling, foreign-key
59 /// enforcement, and the `hash_row_id` function the generated triggers call.
60 ///
61 /// This is synchronous SQLite work; callers driving it from async code should
62 /// run it on a blocking thread (the engine's push/pull loops do).
63 pub fn open(&self) -> Result<Connection> {
64 let conn = match self {
65 Self::Path(p) => Connection::open(p)?,
66 Self::Factory(f) => f()?,
67 };
68 configure_connection(&conn)?;
69 Ok(conn)
70 }
71 }
72
73 /// Apply the engine's required per-connection setup. Idempotent.
74 pub(crate) fn configure_connection(conn: &Connection) -> Result<()> {
75 // WAL lets the engine's connection coexist with the app's; foreign_keys ON is
76 // the default the apply path relaxes only for `references_unsynced` tables.
77 conn.execute_batch("PRAGMA journal_mode=WAL; PRAGMA foreign_keys=ON;")?;
78 register_hash_row_id(conn)?;
79 ensure_scope_schema(conn)?;
80 Ok(())
81 }
82
83 /// Idempotently bring a pre-Groups bookkeeping schema up to the multi-scope
84 /// shape: add `sync_changelog.scope` if the table exists without it, and ensure
85 /// `sync_scope_cursor` exists (seeded from the old single `pull_cursor`).
86 ///
87 /// `CREATE TABLE IF NOT EXISTS` in [`SyncSchema::migration_sql`] handles a fresh
88 /// DB, but it cannot *add a column* to an existing changelog and SQLite has no
89 /// `ADD COLUMN IF NOT EXISTS`, so the ALTER is guarded on `pragma_table_info`
90 /// here. On a fresh DB (no bookkeeping yet) this is a no-op, `migration_sql`
91 /// creates everything with the column. Runs on every connection open; the guards
92 /// make it a cheap no-op once applied.
93 pub(crate) fn ensure_scope_schema(conn: &Connection) -> Result<()> {
94 let bookkeeping_exists = conn
95 .query_row(
96 "SELECT 1 FROM sqlite_master WHERE type = 'table' AND name = 'sync_changelog'",
97 [],
98 |_| Ok(()),
99 )
100 .optional()?
101 .is_some();
102 if !bookkeeping_exists {
103 return Ok(());
104 }
105
106 let has_scope = conn
107 .query_row(
108 "SELECT 1 FROM pragma_table_info('sync_changelog') WHERE name = 'scope'",
109 [],
110 |_| Ok(()),
111 )
112 .optional()?
113 .is_some();
114 if !has_scope {
115 conn.execute_batch(
116 "ALTER TABLE sync_changelog ADD COLUMN scope TEXT NOT NULL DEFAULT '';",
117 )?;
118 }
119
120 conn.execute_batch(
121 "CREATE TABLE IF NOT EXISTS sync_scope_cursor (
122 scope TEXT PRIMARY KEY NOT NULL,
123 cursor INTEGER NOT NULL
124 );
125 INSERT OR IGNORE INTO sync_scope_cursor (scope, cursor)
126 SELECT '', CAST(value AS INTEGER) FROM sync_state WHERE key = 'pull_cursor';",
127 )?;
128 Ok(())
129 }
130
131 /// Read a scope's pull cursor (`''` = personal), defaulting to `0` when absent.
132 pub fn get_scope_cursor(conn: &Connection, scope: &str) -> Result<i64> {
133 conn.query_row(
134 "SELECT cursor FROM sync_scope_cursor WHERE scope = ?1",
135 [scope],
136 |r| r.get(0),
137 )
138 .optional()
139 .map(|o| o.unwrap_or(0))
140 .map_err(Into::into)
141 }
142
143 /// Persist a scope's pull cursor.
144 pub fn set_scope_cursor(conn: &Connection, scope: &str, cursor: i64) -> Result<()> {
145 conn.execute(
146 "INSERT INTO sync_scope_cursor (scope, cursor) VALUES (?1, ?2) \
147 ON CONFLICT(scope) DO UPDATE SET cursor = excluded.cursor",
148 (scope, cursor),
149 )?;
150 Ok(())
151 }
152
153 /// Register `hash_row_id(salt, key) -> TEXT`: lowercase hex of
154 /// `SHA-256(salt || ":" || key)`. Deterministic, keyed by the per-vault
155 /// `row_id_salt`, so the wire row-id of a content-bearing table never carries
156 /// cleartext. Matches the audiofiles definition byte-for-byte so its existing
157 /// hashed row-ids stay stable across the port.
158 fn register_hash_row_id(conn: &Connection) -> Result<()> {
159 conn.create_scalar_function(
160 "hash_row_id",
161 2,
162 FunctionFlags::SQLITE_DETERMINISTIC | FunctionFlags::SQLITE_UTF8,
163 |ctx| {
164 let salt: String = ctx.get(0)?;
165 let key: String = ctx.get(1)?;
166 Ok(hash_row_id(&salt, &key))
167 },
168 )?;
169 Ok(())
170 }
171
172 /// The `hash_row_id` computation, exposed for the apply path (which must derive
173 /// the same wire row-id in Rust when it needs to address a hashed row).
174 pub(crate) fn hash_row_id(salt: &str, key: &str) -> String {
175 let mut hasher = Sha256::new();
176 hasher.update(salt.as_bytes());
177 hasher.update(b":");
178 hasher.update(key.as_bytes());
179 let digest = hasher.finalize();
180 let mut hex = String::with_capacity(64);
181 for byte in digest {
182 use std::fmt::Write;
183 let _ = write!(hex, "{byte:02x}");
184 }
185 hex
186 }
187
188 // sync_state key-value store
189
190 /// Read a `sync_state` value, or `None` if the key is absent.
191 pub fn get_sync_state(conn: &Connection, key: &str) -> Result<Option<String>> {
192 conn.query_row("SELECT value FROM sync_state WHERE key = ?1", [key], |r| {
193 r.get::<_, String>(0)
194 })
195 .optional()
196 .map_err(Into::into)
197 }
198
199 /// Read a `sync_state` value, falling back to `default` when absent.
200 pub fn get_sync_state_or(conn: &Connection, key: &str, default: &str) -> Result<String> {
201 Ok(get_sync_state(conn, key)?.unwrap_or_else(|| default.to_string()))
202 }
203
204 /// Write a `sync_state` value, inserting or updating.
205 pub fn set_sync_state(conn: &Connection, key: &str, value: &str) -> Result<()> {
206 conn.execute(
207 "INSERT INTO sync_state (key, value) VALUES (?1, ?2) \
208 ON CONFLICT(key) DO UPDATE SET value = excluded.value",
209 (key, value),
210 )?;
211 Ok(())
212 }
213
214 /// Count unpushed changelog entries.
215 pub fn count_pending_changes(conn: &Connection) -> Result<i64> {
216 conn.query_row(
217 "SELECT COUNT(*) FROM sync_changelog WHERE pushed = 0",
218 [],
219 |r| r.get(0),
220 )
221 .map_err(Into::into)
222 }
223
224 // applying_remote echo-suppression
225
226 /// Run `apply` with the `applying_remote` flag raised, inside a single
227 /// `IMMEDIATE` transaction.
228 ///
229 /// The flag is set to `'1'` and back to `'0'` inside the same transaction as the
230 /// writes `apply` performs. The changelog triggers, evaluating on this same
231 /// connection, see the uncommitted `'1'` and suppress the echo; on commit the
232 /// persisted value is `'0'` again. If `apply` returns `Err` (or the process dies
233 /// mid-transaction) the whole transaction, including the flag flip, rolls back,
234 /// so the flag can never be stranded at `'1'` and silently swallow later local
235 /// edits. That stranding is exactly the failure the apps defended against with a
236 /// start-of-sync reset; here it is structurally impossible.
237 pub fn with_applying_remote<T>(
238 conn: &mut Connection,
239 apply: impl FnOnce(&rusqlite::Transaction<'_>) -> Result<T>,
240 ) -> Result<T> {
241 let tx = conn.transaction_with_behavior(TransactionBehavior::Immediate)?;
242 set_sync_state(&tx, "applying_remote", "1")?;
243 let out = apply(&tx)?;
244 set_sync_state(&tx, "applying_remote", "0")?;
245 tx.commit()?;
246 Ok(out)
247 }
248
249 /// Defensively clear a stuck `applying_remote` flag at start of sync. With
250 /// [`with_applying_remote`] the flag cannot actually strand, but a database
251 /// written by an older, non-transactional client might carry a stale `'1'`.
252 pub fn clear_applying_remote(conn: &Connection) -> Result<()> {
253 set_sync_state(conn, "applying_remote", "0")
254 }
255
256 // device identity
257
258 /// The registered device id, or `None` if this install has not registered yet.
259 pub fn device_id(conn: &Connection) -> Result<Option<DeviceId>> {
260 match get_sync_state(conn, "device_id")? {
261 Some(s) if !s.is_empty() => {
262 let uuid = Uuid::parse_str(&s)
263 .map_err(|e| SyncKitError::Database(format!("malformed device_id: {e}")))?;
264 Ok(Some(DeviceId::new(uuid)))
265 }
266 _ => Ok(None),
267 }
268 }
269
270 /// Persist the registered device id.
271 pub fn set_device_id(conn: &Connection, id: DeviceId) -> Result<()> {
272 set_sync_state(conn, "device_id", &id.as_uuid().to_string())
273 }
274
275 #[cfg(test)]
276 mod tests {
277 use super::super::schema::{SyncSchema, SyncTable};
278 use super::*;
279
280 /// An in-memory DB configured like the engine would, with a one-table
281 /// migration applied so triggers exist to exercise the flag.
282 fn db() -> Connection {
283 let conn = Connection::open_in_memory().unwrap();
284 configure_connection(&conn).unwrap();
285 conn.execute_batch("CREATE TABLE note (id TEXT PRIMARY KEY, name TEXT);")
286 .unwrap();
287 let schema = SyncSchema::new(vec![SyncTable::full("note", &["id", "name"])]);
288 conn.execute_batch(&schema.migration_sql()).unwrap();
289 conn
290 }
291
292 fn changelog_count(conn: &Connection) -> i64 {
293 conn.query_row("SELECT COUNT(*) FROM sync_changelog", [], |r| r.get(0))
294 .unwrap()
295 }
296
297 #[test]
298 fn hash_row_id_matches_audiofiles_formula() {
299 // sha256("salt:key") lowercase hex, the exact audiofiles definition.
300 let mut h = Sha256::new();
301 h.update(b"salt:key");
302 let expected = hex::encode(h.finalize());
303 assert_eq!(hash_row_id("salt", "key"), expected);
304 // Pinned against an independent SHA-256 as well, so the row id stays a
305 // fixed wire value rather than one that merely agrees with whatever
306 // hasher this crate happens to link.
307 assert_eq!(
308 hash_row_id("salt", "key"),
309 "c82b2df7a36eec83a4b5a9d83093a209006ae9029b6cb13323500da263d9fd06"
310 );
311 assert_eq!(hash_row_id("salt", "key").len(), 64);
312 // Salt actually keys the hash (not just concatenated blindly).
313 assert_ne!(hash_row_id("a", "bc"), hash_row_id("ab", "c"));
314 }
315
316 #[test]
317 fn hash_row_id_is_registered_and_callable_in_sql() {
318 let conn = db();
319 let via_sql: String = conn
320 .query_row("SELECT hash_row_id('s', 'k')", [], |r| r.get(0))
321 .unwrap();
322 assert_eq!(via_sql, hash_row_id("s", "k"));
323 }
324
325 #[test]
326 fn foreign_keys_enforced_by_default() {
327 let conn = db();
328 let fk: i64 = conn
329 .query_row("PRAGMA foreign_keys", [], |r| r.get(0))
330 .unwrap();
331 assert_eq!(fk, 1);
332 }
333
334 #[test]
335 fn sync_state_get_set_and_absent() {
336 let conn = db();
337 assert_eq!(get_sync_state(&conn, "nope").unwrap(), None);
338 assert_eq!(
339 get_sync_state_or(&conn, "nope", "fallback").unwrap(),
340 "fallback"
341 );
342 set_sync_state(&conn, "k", "v1").unwrap();
343 assert_eq!(get_sync_state(&conn, "k").unwrap().as_deref(), Some("v1"));
344 set_sync_state(&conn, "k", "v2").unwrap(); // upsert
345 assert_eq!(get_sync_state(&conn, "k").unwrap().as_deref(), Some("v2"));
346 }
347
348 #[test]
349 fn applying_remote_suppresses_capture_and_commits_flag_low() {
350 let mut conn = db();
351 with_applying_remote(&mut conn, |tx| {
352 tx.execute("INSERT INTO note (id, name) VALUES ('n1', 'x')", [])?;
353 Ok(())
354 })
355 .unwrap();
356 // The write landed, but its trigger was suppressed (echo not captured)...
357 let n: i64 = conn
358 .query_row("SELECT COUNT(*) FROM note", [], |r| r.get(0))
359 .unwrap();
360 assert_eq!(n, 1);
361 assert_eq!(changelog_count(&conn), 0);
362 // ...and the persisted flag is back to '0'.
363 assert_eq!(
364 get_sync_state(&conn, "applying_remote").unwrap().as_deref(),
365 Some("0")
366 );
367 }
368
369 #[test]
370 fn applying_remote_rolls_back_writes_and_flag_on_error() {
371 let mut conn = db();
372 let result: Result<()> = with_applying_remote(&mut conn, |tx| {
373 tx.execute("INSERT INTO note (id, name) VALUES ('n2', 'y')", [])?;
374 Err(SyncKitError::Internal("boom mid-apply".into())) // simulate a crash
375 });
376 assert!(result.is_err());
377 // The write inside the failed transaction rolled back...
378 let n: i64 = conn
379 .query_row("SELECT COUNT(*) FROM note", [], |r| r.get(0))
380 .unwrap();
381 assert_eq!(n, 0);
382 // ...and the flag was never stranded at '1'.
383 assert_eq!(
384 get_sync_state(&conn, "applying_remote").unwrap().as_deref(),
385 Some("0")
386 );
387 }
388
389 #[test]
390 fn normal_write_is_captured_outside_applying_remote() {
391 let conn = db();
392 conn.execute("INSERT INTO note (id, name) VALUES ('n3', 'z')", [])
393 .unwrap();
394 assert_eq!(changelog_count(&conn), 1);
395 }
396
397 #[test]
398 fn device_id_roundtrip() {
399 let conn = db();
400 assert_eq!(device_id(&conn).unwrap(), None); // seeded as '' → None
401 let id = DeviceId::new(Uuid::from_u128(0x42));
402 set_device_id(&conn, id).unwrap();
403 assert_eq!(device_id(&conn).unwrap(), Some(id));
404 }
405
406 #[test]
407 fn malformed_device_id_is_an_error_not_a_panic() {
408 let conn = db();
409 set_sync_state(&conn, "device_id", "not-a-uuid").unwrap();
410 assert!(device_id(&conn).is_err());
411 }
412
413 #[test]
414 fn ensure_scope_schema_upgrades_a_legacy_changelog() {
415 // Simulate a pre-Groups database: a sync_changelog WITHOUT the scope
416 // column, and a pull_cursor mid-stream at 42.
417 let conn = Connection::open_in_memory().unwrap();
418 conn.execute_batch(
419 "CREATE TABLE sync_state (key TEXT PRIMARY KEY NOT NULL, value TEXT NOT NULL);
420 INSERT INTO sync_state (key, value) VALUES ('pull_cursor', '42');
421 CREATE TABLE sync_changelog (
422 id INTEGER PRIMARY KEY AUTOINCREMENT,
423 table_name TEXT NOT NULL, op TEXT NOT NULL, row_id TEXT NOT NULL,
424 data TEXT, pushed INTEGER NOT NULL DEFAULT 0
425 );",
426 )
427 .unwrap();
428 // No scope column yet.
429 let has_scope_before: bool = conn
430 .query_row(
431 "SELECT 1 FROM pragma_table_info('sync_changelog') WHERE name = 'scope'",
432 [],
433 |_| Ok(true),
434 )
435 .optional()
436 .unwrap()
437 .unwrap_or(false);
438 assert!(!has_scope_before);
439
440 // Idempotent: run it twice.
441 ensure_scope_schema(&conn).unwrap();
442 ensure_scope_schema(&conn).unwrap();
443
444 // The scope column now exists and defaults to personal.
445 conn.execute(
446 "INSERT INTO sync_changelog (table_name, op, row_id) VALUES ('t','INSERT','r')",
447 [],
448 )
449 .unwrap();
450 let scope: String = conn
451 .query_row(
452 "SELECT scope FROM sync_changelog WHERE row_id = 'r'",
453 [],
454 |r| r.get(0),
455 )
456 .unwrap();
457 assert_eq!(scope, "");
458
459 // The personal cursor was migrated from the legacy pull_cursor value.
460 assert_eq!(get_scope_cursor(&conn, "").unwrap(), 42);
461 assert_eq!(get_scope_cursor(&conn, "grp-1").unwrap(), 0); // absent -> 0
462 set_scope_cursor(&conn, "grp-1", 7).unwrap();
463 assert_eq!(get_scope_cursor(&conn, "grp-1").unwrap(), 7);
464 }
465
466 #[test]
467 fn salt_is_provisioned_once_and_stable_across_reruns() {
468 let conn = Connection::open_in_memory().unwrap();
469 configure_connection(&conn).unwrap();
470 conn.execute_batch("CREATE TABLE samp (hash TEXT PRIMARY KEY, name TEXT);")
471 .unwrap();
472 let schema = SyncSchema::new(vec![
473 SyncTable::full("samp", &["hash", "name"])
474 .pk(&["hash"])
475 .hashed(),
476 ]);
477 conn.execute_batch(&schema.migration_sql()).unwrap();
478 let salt1 = get_sync_state(&conn, "row_id_salt").unwrap().unwrap();
479 assert_eq!(salt1.len(), 64); // hex of 32 random bytes
480 // Re-running the migration must not rotate the salt (INSERT OR IGNORE).
481 conn.execute_batch(&schema.migration_sql()).unwrap();
482 let salt2 = get_sync_state(&conn, "row_id_salt").unwrap().unwrap();
483 assert_eq!(salt1, salt2);
484 }
485 }
486