Skip to main content

max / goingson

10.1 KB · 229 lines History Blame Raw
1 //! The `SyncStore` integration (Groups p4 / GO -> SyncStore, M2).
2 //!
3 //! GoingsOn hand-rolled its cloud sync in `sync_service/` against the raw
4 //! `SyncKitClient`. To gain group sync (and shed the glue), it migrates to the
5 //! engine's `SyncStore`, which owns the changelog, triggers, HLC, apply, blobs,
6 //! and scheduler. This module assembles that store from GO's pieces:
7 //! - the [`goingson_schema`] manifest (declared in M1),
8 //! - [`AttachmentBlobs`], GO's blob policy,
9 //! - [`GoSyncObserver`], which forwards engine events to the frontend,
10 //! - and [`run_cutover`], the one-time local migration from GO's bookkeeping to
11 //! the engine's.
12 //!
13 //! The engine opens its own rusqlite connection (WAL) to the same `goingson.db`
14 //! file the sqlx pool uses; the two coexist by design.
15
16 mod blobs;
17 pub(crate) mod manifest;
18 mod observer;
19 pub(crate) mod sync_state;
20
21 use std::path::{Path, PathBuf};
22 use std::sync::Arc;
23
24 use rusqlite::{Connection, OptionalExtension};
25 use synckit_client::store::install_policy;
26 use synckit_client::{DbSource, SyncKitClient, SyncStore};
27 use tauri::{AppHandle, Manager};
28
29 use crate::config_key::CONFIG;
30
31 use crate::state::AppState;
32 use blobs::AttachmentBlobs;
33 use manifest::goingson_schema;
34 use observer::GoSyncObserver;
35
36 /// Build GoingsOn's `SyncStore` over the app's SQLite file. The device name and
37 /// platform default to the host's (the engine's default), matching GO's prior
38 /// registration behaviour.
39 pub(crate) fn build_store(
40 db_path: PathBuf,
41 data_dir: PathBuf,
42 client: SyncKitClient,
43 ) -> Arc<SyncStore<SyncKitClient>> {
44 let store = SyncStore::builder(DbSource::path(db_path), client, goingson_schema())
45 .blob_policy(Arc::new(AttachmentBlobs::new(data_dir)))
46 .build();
47 Arc::new(store)
48 }
49
50 /// Spawn the background cloud-sync scheduler for the managed `AppState`'s
51 /// `SyncStore`, if one is configured. The engine's loop races a timer against the
52 /// server's SSE change stream, applies the auth / key-loaded / auto-sync-enabled /
53 /// backoff gate chain, and reports status through [`GoSyncObserver`].
54 ///
55 /// Must be called from within the async runtime (the engine spawns the loop with
56 /// `tokio::spawn`). The returned handle is intentionally dropped: the loop runs
57 /// for the app's lifetime (dropping the handle does not stop it), which matches
58 /// GoingsOn's prior always-on scheduler. A no-op when sync is unconfigured.
59 pub fn spawn_scheduler(app: &AppHandle) {
60 let Some(state) = app.try_state::<Arc<AppState>>() else {
61 return;
62 };
63 let Some(store) = state.sync_store.clone() else {
64 return;
65 };
66 let _handle = store.spawn_scheduler(Arc::new(GoSyncObserver::new(app.clone())));
67 }
68
69 /// One-time migration of a device's local sync bookkeeping from GoingsOn's
70 /// hand-rolled triggers/changelog to the engine's. Idempotent and guarded by the
71 /// `syncstore_migrated` flag, so it is a cheap no-op on every launch after the
72 /// first. Runs synchronously (rusqlite) at app init, after the sqlx migrations
73 /// and before the scheduler spawns.
74 ///
75 /// Steps:
76 /// 1. Open a configured connection; this runs the engine's `ensure_scope_schema`,
77 /// which adds `sync_changelog.scope` to a legacy changelog and seeds
78 /// `sync_scope_cursor('')` from the old single `pull_cursor`.
79 /// 2. Drop GO's legacy capture triggers (two naming conventions; see
80 /// [`legacy_sync_triggers`]).
81 /// 3. Run the engine's generated DDL: bookkeeping tables (`CREATE IF NOT EXISTS`)
82 /// and one trigger set per table (each with its own `DROP IF EXISTS`).
83 /// 4. Re-baseline. The server changelog is authoritative, so reset the personal
84 /// pull cursor to 0 (re-pull and re-apply idempotently) and clear
85 /// `initial_snapshot_done` (re-snapshot every local row for the first push, so
86 /// any local-only data survives the cutover). This is the one-time cost the
87 /// migration plan accepts.
88 ///
89 /// The whole rewrite runs in one transaction; a failure leaves the legacy
90 /// triggers in place and the `syncstore_migrated` flag unset, so the next launch
91 /// retries from a clean state.
92 pub(crate) fn run_cutover(db_path: &Path) -> Result<(), String> {
93 let conn = DbSource::path(db_path).open().map_err(str_err)?;
94
95 if flag_set(&conn, "syncstore_migrated")? {
96 return Ok(());
97 }
98
99 let legacy = legacy_sync_triggers(&conn)?;
100
101 let tx = conn.unchecked_transaction().map_err(str_err)?;
102 for name in &legacy {
103 tx.execute_batch(&format!("DROP TRIGGER IF EXISTS \"{name}\";"))
104 .map_err(str_err)?;
105 }
106 // The manifest is group-scoped (M3): its generated triggers read
107 // NEW.group_id. The column comes from migration 059 (which runs before this
108 // at app init), so it is guaranteed present here.
109 tx.execute_batch(&goingson_schema().migration_sql())
110 .map_err(str_err)?;
111 tx.execute_batch(
112 "UPDATE sync_scope_cursor SET cursor = 0 WHERE scope = '';\n\
113 UPDATE sync_state SET value = '0' WHERE key = 'pull_cursor';\n\
114 UPDATE sync_state SET value = '0' WHERE key = 'initial_snapshot_done';\n\
115 INSERT INTO sync_state (key, value) VALUES ('syncstore_migrated', '1')\n\
116 ON CONFLICT(key) DO UPDATE SET value = '1';\n\
117 INSERT INTO sync_state (key, value) VALUES ('syncstore_groups_migrated', '1')\n\
118 ON CONFLICT(key) DO UPDATE SET value = '1';",
119 )
120 .map_err(str_err)?;
121 tx.commit().map_err(str_err)?;
122 Ok(())
123 }
124
125 /// M3 group migration for a device that already ran the M2 cutover: regenerate
126 /// the engine triggers so the shareable tables become group-scoped (they stamp
127 /// the changelog scope from `group_id`). Idempotent, guarded by
128 /// `syncstore_groups_migrated`, transactional. The `group_id` columns come from
129 /// migration 059; a fresh install already generated group-scoped triggers inside
130 /// [`run_cutover`] and set the flag, so this is a no-op there; only an
131 /// M2-migrated device (personal-only triggers) reaches the body.
132 pub(crate) fn run_group_migration(db_path: &Path) -> Result<(), String> {
133 let conn = DbSource::path(db_path).open().map_err(str_err)?;
134 if flag_set(&conn, "syncstore_groups_migrated")? {
135 return Ok(());
136 }
137 let tx = conn.unchecked_transaction().map_err(str_err)?;
138 // Regenerate every trigger; the shareable tables' triggers now stamp the
139 // changelog scope from group_id. Each generated trigger has its own DROP.
140 tx.execute_batch(&goingson_schema().migration_sql())
141 .map_err(str_err)?;
142 tx.execute_batch(
143 "INSERT INTO sync_state (key, value) VALUES ('syncstore_groups_migrated', '1')\n\
144 ON CONFLICT(key) DO UPDATE SET value = '1';",
145 )
146 .map_err(str_err)?;
147 tx.commit().map_err(str_err)?;
148 Ok(())
149 }
150
151 /// Seed `config_key_policy` from the config spec ([`CONFIG`]).
152 ///
153 /// Runs every launch and is idempotent ([`install_policy`] clears and reinserts
154 /// the closed set), so the spec stays the single source of truth for which
155 /// `user_config` keys sync. **Must run before any
156 /// `goingson_schema().migration_sql()`** — the cutover, the group migration, the
157 /// config migration, and the store build all regenerate the engine's triggers,
158 /// and the generated config triggers join this table, so it has to exist and be
159 /// seeded first.
160 pub(crate) fn seed_config_policy(db_path: &Path) -> Result<(), String> {
161 let mut conn = DbSource::path(db_path).open().map_err(str_err)?;
162 install_policy(&mut conn, &CONFIG).map_err(str_err)?;
163 Ok(())
164 }
165
166 /// One-time: regenerate the engine triggers so the `user_config` config table
167 /// gains its export/import triggers on a device that already ran the M2/M3
168 /// cutover (both are guarded and no-op now, so neither would create them).
169 /// Idempotent, guarded by `syncstore_config_migrated`, transactional. A fresh
170 /// install already generated the config triggers inside [`run_cutover`] (the
171 /// manifest now carries the config table) and never reaches the body.
172 ///
173 /// [`seed_config_policy`] must have run first this launch, since the regenerated
174 /// config triggers join `config_key_policy`.
175 pub(crate) fn run_config_migration(db_path: &Path) -> Result<(), String> {
176 let conn = DbSource::path(db_path).open().map_err(str_err)?;
177 if flag_set(&conn, "syncstore_config_migrated")? {
178 return Ok(());
179 }
180 let tx = conn.unchecked_transaction().map_err(str_err)?;
181 // Each generated trigger has its own DROP IF EXISTS, so regenerating the full
182 // set is safe; it adds the config table's triggers to an already-migrated
183 // device.
184 tx.execute_batch(&goingson_schema().migration_sql())
185 .map_err(str_err)?;
186 tx.execute_batch(
187 "INSERT INTO sync_state (key, value) VALUES ('syncstore_config_migrated', '1')\n\
188 ON CONFLICT(key) DO UPDATE SET value = '1';",
189 )
190 .map_err(str_err)?;
191 tx.commit().map_err(str_err)?;
192 Ok(())
193 }
194
195 fn flag_set(conn: &Connection, key: &str) -> Result<bool, String> {
196 let v: Option<String> = conn
197 .query_row("SELECT value FROM sync_state WHERE key = ?1", [key], |r| {
198 r.get(0)
199 })
200 .optional()
201 .map_err(str_err)?;
202 Ok(v.as_deref() == Some("1"))
203 }
204
205 fn str_err<E: std::fmt::Display>(e: E) -> String {
206 e.to_string()
207 }
208
209 /// GO's hand-written capture triggers use two naming conventions: `sync_trg_*`
210 /// (migration 030 and later) and `*_changelog` (migration 037's `sync_accounts`).
211 /// Both write to `sync_changelog` and must be dropped before the engine's own
212 /// triggers replace them. The engine's generated triggers are named
213 /// `sync_<table>_<op>` (no `sync_trg_` prefix, no `_changelog` suffix), so this
214 /// query never matches them. The FTS triggers on `emails` (`emails_ai/ad/au`)
215 /// also never match, so email full-text search is untouched.
216 fn legacy_sync_triggers(conn: &Connection) -> Result<Vec<String>, String> {
217 let mut stmt = conn
218 .prepare(
219 "SELECT name FROM sqlite_master WHERE type = 'trigger' \
220 AND (name LIKE 'sync\\_trg\\_%' ESCAPE '\\' \
221 OR name LIKE '%\\_changelog' ESCAPE '\\')",
222 )
223 .map_err(str_err)?;
224 let names = stmt
225 .query_map([], |r| r.get::<_, String>(0))
226 .map_err(str_err)?;
227 names.collect::<rusqlite::Result<Vec<_>>>().map_err(str_err)
228 }
229