Skip to main content

max / goingson

Add SyncStore integration foundation (Groups p4 M2, additive) Introduces the pieces the SyncStore cutover assembles, compiling alongside the existing hand-rolled sync path with no behaviour change yet: - syncstore::blobs — AttachmentBlobs, GO's BlobPolicy (disk-existence presence, no-op hooks). - syncstore::observer — GoSyncObserver, forwarding engine scheduler events to the frontend's sync:status-changed / sync:changes-applied / sync:subscription-required. - syncstore::build_store — assembles SyncStore over goingson.db (WAL coexistence with the sqlx pool) from the M1 manifest + blob policy. - syncstore::run_cutover — the one-time, flag-guarded, transactional local migration: drop GO's legacy triggers, run the engine's generated DDL, re-baseline (reset personal cursor + re-snapshot) so the first engine sync reconciles idempotently against the authoritative server changelog. Adds rusqlite 0.39 (shares libsqlite3-sys 0.37 with sqlx) so the blob policy and cutover can name rusqlite::Connection. Not yet wired into AppState / lib.rs / commands — that is the live-path flip. Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
Author: Max Johnson <me@maxj.phd> · 2026-07-23 21:34 UTC
Commit: 5ecb71acbcac0f7365c52b0c37da316b5c4cfdb2
Parent: a33aa73
7 files changed, +257 insertions, -1 deletion
M Cargo.lock +1
@@ -2281,6 +2281,7 @@ dependencies = [
2281 2281 "rand 0.10.2",
2282 2282 "reqwest",
2283 2283 "rfd",
2284 + "rusqlite",
2284 2285 "rustls",
2285 2286 "rustls-platform-verifier",
2286 2287 "serde",
M src-tauri/Cargo.toml +5
@@ -30,6 +30,11 @@ tokio = { workspace = true }
30 30
31 31 # Database
32 32 sqlx = { workspace = true, features = ["sqlite"] }
33 + # The SyncStore engine (synckit-client) owns a private rusqlite connection to the
34 + # same SQLite file the sqlx pool uses; naming rusqlite::Connection in the blob
35 + # policy / cutover needs it as a direct dep. Same version + libsqlite3-sys 0.37 as
36 + # the engine, so both bindings share one bundled SQLite.
37 + rusqlite = { version = "0.39", features = ["bundled"] }
33 38
34 39 # Serialization
35 40 serde = { workspace = true }
M src-tauri/src/lib.rs +1
@@ -16,6 +16,7 @@ pub mod oauth;
16 16 pub mod state;
17 17 pub mod sync_scheduler;
18 18 pub mod sync_service;
19 + pub mod syncstore;
19 20 pub mod tz;
20 21
21 22 // Desktop-only: file watching for external DB changes
M src-tauri/src/sync_service/mod.rs +1 -1
@@ -22,7 +22,7 @@
22 22 mod apply;
23 23 pub(crate) mod blob_sync;
24 24 mod hlc;
25 - mod manifest;
25 + pub(crate) mod manifest;
26 26 mod pull;
27 27 mod push;
28 28 mod state;
A src-tauri/src/syncstore/blobs.rs +68
@@ -0,0 +1,68 @@
1 + //! GoingsOn's [`BlobPolicy`]: attachment blobs are content-addressed files under
2 + //! the app data dir, tracked purely by disk existence (no `cloud_only` flag), so
3 + //! the three presence hooks stay no-ops. This mirrors the retiring
4 + //! `sync_service/blob_sync.rs`, but the engine now owns the invariant machinery:
5 + //! dedup-aware upload, integrity-verified download, and atomic writes.
6 +
7 + use std::path::PathBuf;
8 +
9 + use rusqlite::Connection;
10 + use synckit_client::{BlobPolicy, BlobRef, Result, SyncKitError};
11 +
12 + use crate::state::DESKTOP_USER_ID;
13 +
14 + /// Attachment blob policy for the single desktop user.
15 + pub(crate) struct AttachmentBlobs {
16 + data_dir: PathBuf,
17 + user_id: String,
18 + }
19 +
20 + impl AttachmentBlobs {
21 + pub(crate) fn new(data_dir: PathBuf) -> Self {
22 + Self {
23 + data_dir,
24 + user_id: DESKTOP_USER_ID.to_string(),
25 + }
26 + }
27 +
28 + /// Every distinct attachment blob for the desktop user. The engine filters
29 + /// this candidate set per pass: uploads skip blobs with no local file,
30 + /// downloads skip blobs already present on disk (then SHA-256-verify).
31 + fn all_blobs(&self, conn: &Connection) -> Result<Vec<BlobRef>> {
32 + let mut stmt = conn
33 + .prepare("SELECT DISTINCT blob_hash, file_size FROM attachments WHERE user_id = ?1")
34 + .map_err(map_err)?;
35 + let rows = stmt
36 + .query_map([&self.user_id], |r| {
37 + let hash: String = r.get(0)?;
38 + let size: i64 = r.get(1)?;
39 + Ok(BlobRef {
40 + hash,
41 + ext: String::new(),
42 + size: size.max(0) as u64,
43 + })
44 + })
45 + .map_err(map_err)?;
46 + rows.collect::<rusqlite::Result<Vec<_>>>().map_err(map_err)
47 + }
48 + }
49 +
50 + fn map_err(e: rusqlite::Error) -> SyncKitError {
51 + SyncKitError::Internal(format!("attachment blob query: {e}"))
52 + }
53 +
54 + impl BlobPolicy for AttachmentBlobs {
55 + fn pending_uploads(&self, conn: &Connection) -> Result<Vec<BlobRef>> {
56 + self.all_blobs(conn)
57 + }
58 +
59 + fn pending_downloads(&self, conn: &Connection) -> Result<Vec<BlobRef>> {
60 + self.all_blobs(conn)
61 + }
62 +
63 + fn local_path(&self, blob: &BlobRef) -> PathBuf {
64 + crate::commands::attachment::blob_path(&self.data_dir, &blob.hash)
65 + }
66 + // reconcile / on_uploaded / on_downloaded: inherited no-ops (presence is disk
67 + // existence, so there is no flag to reconcile).
68 + }
A src-tauri/src/syncstore/mod.rs +135
@@ -0,0 +1,135 @@
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 + // Assembled here; the M2 cutover wires build_store/run_cutover/GoSyncObserver into
17 + // AppState, lib.rs setup, and commands/sync.rs. Until that flip lands they are
18 + // only referenced across the module, so silence the not-yet-wired lint.
19 + #![allow(dead_code)]
20 +
21 + mod blobs;
22 + mod observer;
23 +
24 + use std::path::{Path, PathBuf};
25 + use std::sync::Arc;
26 +
27 + use rusqlite::{Connection, OptionalExtension};
28 + use synckit_client::{DbSource, SyncKitClient, SyncStore};
29 +
30 + use crate::sync_service::manifest::goingson_schema;
31 + use blobs::AttachmentBlobs;
32 + #[allow(unused_imports)] // wired into lib.rs setup when the M2 flip lands.
33 + pub(crate) use observer::GoSyncObserver;
34 +
35 + /// Build GoingsOn's `SyncStore` over the app's SQLite file. The device name and
36 + /// platform default to the host's (the engine's default), matching GO's prior
37 + /// registration behaviour.
38 + pub(crate) fn build_store(
39 + db_path: PathBuf,
40 + data_dir: PathBuf,
41 + client: SyncKitClient,
42 + ) -> Arc<SyncStore<SyncKitClient>> {
43 + let store = SyncStore::builder(DbSource::path(db_path), client, goingson_schema())
44 + .blob_policy(Arc::new(AttachmentBlobs::new(data_dir)))
45 + .build();
46 + Arc::new(store)
47 + }
48 +
49 + /// One-time migration of a device's local sync bookkeeping from GoingsOn's
50 + /// hand-rolled triggers/changelog to the engine's. Idempotent and guarded by the
51 + /// `syncstore_migrated` flag, so it is a cheap no-op on every launch after the
52 + /// first. Runs synchronously (rusqlite) at app init, after the sqlx migrations
53 + /// and before the scheduler spawns.
54 + ///
55 + /// Steps:
56 + /// 1. Open a configured connection — this runs the engine's `ensure_scope_schema`,
57 + /// which adds `sync_changelog.scope` to a legacy changelog and seeds
58 + /// `sync_scope_cursor('')` from the old single `pull_cursor`.
59 + /// 2. Drop GO's legacy capture triggers (two naming conventions; see
60 + /// [`legacy_sync_triggers`]).
61 + /// 3. Run the engine's generated DDL: bookkeeping tables (`CREATE IF NOT EXISTS`)
62 + /// and one trigger set per table (each with its own `DROP IF EXISTS`).
63 + /// 4. Re-baseline. The server changelog is authoritative, so reset the personal
64 + /// pull cursor to 0 (re-pull and re-apply idempotently) and clear
65 + /// `initial_snapshot_done` (re-snapshot every local row for the first push, so
66 + /// any local-only data survives the cutover). This is the one-time cost the
67 + /// migration plan accepts.
68 + ///
69 + /// The whole rewrite runs in one transaction; a failure leaves the legacy
70 + /// triggers in place and the `syncstore_migrated` flag unset, so the next launch
71 + /// retries from a clean state.
72 + pub(crate) fn run_cutover(db_path: &Path) -> Result<(), String> {
73 + let conn = DbSource::path(db_path).open().map_err(str_err)?;
74 +
75 + if already_migrated(&conn)? {
76 + return Ok(());
77 + }
78 +
79 + let legacy = legacy_sync_triggers(&conn)?;
80 +
81 + let tx = conn.unchecked_transaction().map_err(str_err)?;
82 + for name in &legacy {
83 + tx.execute_batch(&format!("DROP TRIGGER IF EXISTS \"{name}\";"))
84 + .map_err(str_err)?;
85 + }
86 + tx.execute_batch(&goingson_schema().migration_sql())
87 + .map_err(str_err)?;
88 + tx.execute_batch(
89 + "UPDATE sync_scope_cursor SET cursor = 0 WHERE scope = '';\n\
90 + UPDATE sync_state SET value = '0' WHERE key = 'pull_cursor';\n\
91 + UPDATE sync_state SET value = '0' WHERE key = 'initial_snapshot_done';\n\
92 + INSERT INTO sync_state (key, value) VALUES ('syncstore_migrated', '1')\n\
93 + ON CONFLICT(key) DO UPDATE SET value = '1';",
94 + )
95 + .map_err(str_err)?;
96 + tx.commit().map_err(str_err)?;
97 + Ok(())
98 + }
99 +
100 + fn str_err<E: std::fmt::Display>(e: E) -> String {
101 + e.to_string()
102 + }
103 +
104 + fn already_migrated(conn: &Connection) -> Result<bool, String> {
105 + let v: Option<String> = conn
106 + .query_row(
107 + "SELECT value FROM sync_state WHERE key = 'syncstore_migrated'",
108 + [],
109 + |r| r.get(0),
110 + )
111 + .optional()
112 + .map_err(str_err)?;
113 + Ok(v.as_deref() == Some("1"))
114 + }
115 +
116 + /// GO's hand-written capture triggers use two naming conventions: `sync_trg_*`
117 + /// (migration 030 and later) and `*_changelog` (migration 037's `sync_accounts`).
118 + /// Both write to `sync_changelog` and must be dropped before the engine's own
119 + /// triggers replace them. The engine's generated triggers are named
120 + /// `sync_<table>_<op>` — no `sync_trg_` prefix, no `_changelog` suffix — so this
121 + /// query never matches them. The FTS triggers on `emails` (`emails_ai/ad/au`)
122 + /// also never match, so email full-text search is untouched.
123 + fn legacy_sync_triggers(conn: &Connection) -> Result<Vec<String>, String> {
124 + let mut stmt = conn
125 + .prepare(
126 + "SELECT name FROM sqlite_master WHERE type = 'trigger' \
127 + AND (name LIKE 'sync\\_trg\\_%' ESCAPE '\\' \
128 + OR name LIKE '%\\_changelog' ESCAPE '\\')",
129 + )
130 + .map_err(str_err)?;
131 + let names = stmt
132 + .query_map([], |r| r.get::<_, String>(0))
133 + .map_err(str_err)?;
134 + names.collect::<rusqlite::Result<Vec<_>>>().map_err(str_err)
135 + }
A src-tauri/src/syncstore/observer.rs +46
@@ -0,0 +1,46 @@
1 + //! GoingsOn's [`SyncObserver`]: maps the engine scheduler's callbacks onto the
2 + //! Tauri events the frontend already listens for. The frontend (`js/app.js`)
3 + //! consumes:
4 + //! - `sync:status-changed` — payload `"syncing"` / `"error"` / anything else = idle
5 + //! - `sync:changes-applied` — payload = the list of DB tables the pull touched
6 + //! (drives selective cache invalidation)
7 + //! - `sync:subscription-required` — no payload (shows the subscribe toast)
8 +
9 + use std::collections::HashSet;
10 +
11 + use synckit_client::{SyncObserver, SyncState};
12 + use tauri::{AppHandle, Emitter};
13 +
14 + /// Emits GoingsOn's sync events from engine scheduler callbacks.
15 + pub(crate) struct GoSyncObserver {
16 + app: AppHandle,
17 + }
18 +
19 + impl GoSyncObserver {
20 + pub(crate) fn new(app: AppHandle) -> Self {
21 + Self { app }
22 + }
23 + }
24 +
25 + impl SyncObserver for GoSyncObserver {
26 + fn on_status(&self, state: SyncState) {
27 + // The frontend's dot only distinguishes syncing / error / idle. Idle,
28 + // LoggedOut, and SubscriptionRequired all rest it to connected; the last
29 + // additionally surfaces through on_subscription_required below.
30 + let payload = match state {
31 + SyncState::Syncing => "syncing",
32 + SyncState::Error(_) => "error",
33 + _ => "idle",
34 + };
35 + let _ = self.app.emit("sync:status-changed", payload);
36 + }
37 +
38 + fn on_changes_applied(&self, tables: &HashSet<String>) {
39 + let tables: Vec<&String> = tables.iter().collect();
40 + let _ = self.app.emit("sync:changes-applied", tables);
41 + }
42 +
43 + fn on_subscription_required(&self) {
44 + let _ = self.app.emit("sync:subscription-required", ());
45 + }
46 + }