max / goingson

Cut GoingsOn cloud sync over to the SyncStore engine (Groups p4 M2) Replaces GoingsOn's hand-rolled sync engine with synckit-client's SyncStore, the multi-scope engine that carries the Groups work. Behaviour is preserved; the ~1400 LOC of push/pull/apply/hlc/snapshot/blob/trigger glue is retired in favour of the M1 manifest plus three small policy pieces. - state.rs: AppState holds Option<Arc<SyncStore<SyncKitClient>>> instead of the SyncClientCell; the store owns the client (reached via store.client() for the auth/subscription/tier commands). Built over goingson.db with its own WAL rusqlite connection, coexisting with the sqlx pool. - run_cutover runs once at init (guarded, transactional): drop GO's legacy capture triggers, run the engine's generated DDL, and re-baseline (reset the personal cursor + re-snapshot) so the first engine sync reconciles idempotently against the authoritative server changelog. Local-only rows are re-snapshotted, so nothing is lost. - commands/sync.rs: sync_now drives store.sync_now(); disconnect clears the in-memory session; status/settings read the kept sqlx sync_state helpers. - lib.rs/main.rs: the hand-rolled sync_scheduler is replaced by the engine's store.spawn_scheduler(GoSyncObserver) (its own timer + SSE race + gate chain), forwarding to the same frontend events. - Retires sync_service/ (engine + 43 tests that covered the retired code); moves the manifest to syncstore/ with self-contained invariant tests; keeps the sync_state KV helpers. Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>

Author: Max Johnson <me@maxj.phd> · 2026-07-23 21:46 UTC

Commit: 47b904c92274074e25cc43538c9c8db1b8728f65

Parent: 5ecb71a

18 files changed, +585 insertions, -2881 deletions

M src-tauri/src/commands/sync.rs +44 -36

			@@ -15,10 +15,18 @@ use crate::oauth::callback_server::OAuthCallbackServer;
15	15		use crate::oauth::credentials::CredentialStore;
16	16		use crate::oauth::provider::{generate_code_challenge, generate_code_verifier, generate_state};
17	17		use crate::state::AppState;
18		-	use crate::sync_service;
	18	+	use crate::syncstore::sync_state;
19	19
20	20		// ============ Types ============
21	21
	22	+	/// Result of a manual sync, surfaced to the frontend.
	23	+	#[derive(Debug, Serialize, Deserialize)]
	24	+	#[serde(rename_all = "camelCase")]
	25	+	pub struct SyncResult {
	26	+	pub pushed: i64,
	27	+	pub pulled: i64,
	28	+	}
	29	+
22	30		/// Response for sync_status command.
23	31		#[derive(Debug, Serialize)]
24	32		#[serde(rename_all = "camelCase")]
			@@ -127,7 +135,7 @@ pub async fn sync_status(state: State<'_, Arc<AppState>>) -> Result<SyncStatusRe
127	135
128	136		// Read sync state from DB — batch query + pending count in parallel
129	137		let (states_result, pending_changes) = tokio::join!(
130		-	sync_service::get_sync_states_batch(
	138	+	sync_state::get_sync_states_batch(
131	139		&state.pool,
132	140		&[
133	141		"device_id",
			@@ -136,7 +144,7 @@ pub async fn sync_status(state: State<'_, Arc<AppState>>) -> Result<SyncStatusRe
136	144		"last_sync_at"
137	145		],
138	146		),
139		-	sync_service::count_pending_changes(&state.pool),
	147	+	sync_state::count_pending_changes(&state.pool),
140	148		);
141	149
142	150		let states = states_result.unwrap_or_default();
			@@ -275,9 +283,14 @@ pub async fn sync_complete_auth(
275	283		/// Disconnects from the sync service by clearing stored credentials.
276	284		#[tauri::command]
277	285		#[instrument(skip_all)]
278		-	pub async fn sync_disconnect(_state: State<'_, Arc<AppState>>) -> Result<bool, ApiError> {
	286	+	pub async fn sync_disconnect(state: State<'_, Arc<AppState>>) -> Result<bool, ApiError> {
279	287		CredentialStore::delete_sync_token()
280	288		.map_api_err("Failed to delete sync token", ApiError::internal)?;
	289	+	// Clear the in-memory session/master key too, so sync stops immediately
	290	+	// (the scheduler's gate chain sees the logged-out client on its next tick).
	291	+	if let Some(store) = &state.sync_store {
	292	+	store.disconnect();
	293	+	}
281	294		Ok(true)
282	295		}
283	296
			@@ -287,8 +300,12 @@ pub async fn sync_disconnect(_state: State<'_, Arc<AppState>>) -> Result<bool, A
287	300		pub async fn sync_now(
288	301		state: State<'_, Arc<AppState>>,
289	302		app: tauri::AppHandle,
290		-	) -> Result<sync_service::SyncResult, ApiError> {
291		-	let client = require_sync_client(&state)?;
	303	+	) -> Result<SyncResult, ApiError> {
	304	+	let store = state
	305	+	.sync_store
	306	+	.as_ref()
	307	+	.ok_or_else(\|\| ApiError::bad_request("Sync is not configured"))?;
	308	+	let client = store.client();
292	309
293	310		if client.session_info().is_none() {
294	311		return Err(ApiError::bad_request("Not authenticated"));
			@@ -298,41 +315,32 @@ pub async fn sync_now(
298	315		return Err(ApiError::bad_request("Encryption not set up"));
299	316		}
300	317
	318	+	// Serialize manual syncs against each other and the background scheduler.
301	319		let _sync_guard = state.sync_lock.lock().await;
302	320
303		-	// Create initial snapshot if needed (must be inside sync_lock to avoid TOCTOU race)
304		-	let snapshot_done = sync_service::get_sync_state(&state.pool, "initial_snapshot_done")
305		-	.await
306		-	.unwrap_or_default();
307		-	if snapshot_done != "1" {
308		-	sync_service::create_initial_snapshot(&state.pool)
309		-	.await
310		-	.map_api_err("Failed to create initial snapshot", ApiError::internal)?;
311		-	}
	321	+	// The engine owns the initial snapshot, push/pull, blob pass, and cleanup.
312	322		let _ = app.emit("sync:status-changed", "syncing");
313		-	let result =
314		-	match sync_service::perform_sync_with_blobs(&state.pool, &client, Some(&state.data_dir))
315		-	.await
316		-	{
317		-	Ok(r) => {
318		-	let _ = app.emit("sync:status-changed", "idle");
319		-	r
320		-	}
321		-	Err(e) => {
322		-	let _ = app.emit("sync:status-changed", "error");
323		-	return Err(ApiError::external_service(format!("Sync failed: {e}")));
324		-	}
325		-	};
	323	+	let outcome = match store.sync_now().await {
	324	+	Ok(o) => {
	325	+	let _ = app.emit("sync:status-changed", "idle");
	326	+	o
	327	+	}
	328	+	Err(e) => {
	329	+	let _ = app.emit("sync:status-changed", "error");
	330	+	return Err(ApiError::external_service(format!("Sync failed: {e}")));
	331	+	}
	332	+	};
326	333
327		-	if result.pulled > 0 {
	334	+	if outcome.pulled > 0 {
328	335		// Carry the changed tables so the UI invalidates selectively.
329		-	let _ = app.emit("sync:changes-applied", &result.pulled_tables);
	336	+	let tables: Vec<&String> = outcome.changed_tables.iter().collect();
	337	+	let _ = app.emit("sync:changes-applied", tables);
330	338		}
331	339
332		-	// Cleanup after manual sync too
333		-	let _ = sync_service::cleanup_changelog(&state.pool).await;
334		-
335		-	Ok(result)
	340	+	Ok(SyncResult {
	341	+	pushed: outcome.pushed as i64,
	342	+	pulled: outcome.pulled as i64,
	343	+	})
336	344		}
337	345
338	346		/// First device: generate a new master key, encrypt with password, push to server.
			@@ -377,7 +385,7 @@ pub async fn sync_update_settings(
377	385		input: SyncSettingsInput,
378	386		) -> Result<bool, ApiError> {
379	387		if let Some(enabled) = input.auto_sync_enabled {
380		-	sync_service::set_sync_state(
	388	+	sync_state::set_sync_state(
381	389		&state.pool,
382	390		"auto_sync_enabled",
383	391		if enabled { "1" } else { "0" },
			@@ -387,7 +395,7 @@ pub async fn sync_update_settings(
387	395		}
388	396
389	397		if let Some(minutes) = input.sync_interval_minutes {
390		-	sync_service::set_sync_state(&state.pool, "sync_interval_minutes", &minutes.to_string())
	398	+	sync_state::set_sync_state(&state.pool, "sync_interval_minutes", &minutes.to_string())
391	399		.await
392	400		.map_api_err("Failed to update setting", ApiError::internal)?;
393	401		}

M src-tauri/src/lib.rs +3 -4

			@@ -14,8 +14,6 @@ pub mod external_sync;
14	14		pub mod jmap;
15	15		pub mod oauth;
16	16		pub mod state;
17		-	pub mod sync_scheduler;
18		-	pub mod sync_service;
19	17		pub mod syncstore;
20	18		pub mod tz;
21	19
			@@ -344,10 +342,11 @@ pub fn build_mobile_app() -> tauri::Builder<tauri::Wry> {
344	342		email_sync_scheduler::start_email_sync_scheduler(sync_handle, email_cancel).await;
345	343		});
346	344
	345	+	// Cloud sync runs through the SyncStore engine's own scheduler now
	346	+	// (its own timer + SSE race + gate chain), reporting via GoSyncObserver.
347	347		let cloud_sync_handle = app.handle().clone();
348		-	let cloud_cancel = cancel_token;
349	348		tauri::async_runtime::spawn(async move {
350		-	sync_scheduler::start_sync_scheduler(cloud_sync_handle, cloud_cancel).await;
	349	+	syncstore::spawn_scheduler(&cloud_sync_handle);
351	350		});
352	351
353	352		// Clean up stale email preview + attachment temp files from previous sessions

M src-tauri/src/main.rs +4 -4

			@@ -8,7 +8,7 @@ use goingson_desktop::blob_gc;
8	8		use goingson_desktop::commands;
9	9		use goingson_desktop::email_sync_scheduler;
10	10		use goingson_desktop::state::AppState;
11		-	use goingson_desktop::sync_scheduler;
	11	+	use goingson_desktop::syncstore;
12	12		use std::sync::Arc;
13	13		use std::sync::atomic::{AtomicBool, Ordering};
14	14		use tauri::menu::{Menu, MenuItem, PredefinedMenuItem, Submenu};
			@@ -388,11 +388,11 @@ fn main() {
388	388		email_sync_scheduler::start_email_sync_scheduler(sync_handle, email_cancel).await;
389	389		});
390	390
391		-	// Start background cloud sync scheduler
	391	+	// Cloud sync runs through the SyncStore engine's own scheduler (its
	392	+	// own timer + SSE race + gate chain), reporting via GoSyncObserver.
392	393		let cloud_sync_handle = app.handle().clone();
393		-	let cloud_cancel = cancel_token;
394	394		tauri::async_runtime::spawn(async move {
395		-	sync_scheduler::start_sync_scheduler(cloud_sync_handle, cloud_cancel).await;
	395	+	syncstore::spawn_scheduler(&cloud_sync_handle);
396	396		});
397	397
398	398		// Clean up stale email preview + attachment temp files from previous sessions

M src-tauri/src/state.rs +22 -32

			@@ -16,8 +16,8 @@ use goingson_db_sqlite::{
16	16		};
17	17		use sqlx::SqlitePool;
18	18		use std::path::PathBuf;
19		-	use std::sync::{Arc, Mutex, RwLock};
20		-	use synckit_client::{SyncKitClient, SyncKitConfig};
	19	+	use std::sync::{Arc, Mutex};
	20	+	use synckit_client::{SyncKitClient, SyncKitConfig, SyncStore};
21	21		use tauri::{AppHandle, Manager};
22	22		use tokio::sync::Mutex as TokioMutex;
23	23		use tracing::{debug, info, instrument, warn};
			@@ -29,28 +29,6 @@ pub const SYNC_SERVER_URL: &str = "https://makenot.work";
29	29		/// The API key is a public client identifier, not a secret.
30	30		const SYNCKIT_TOML: &str = include_str!("../../synckit.toml");
31	31
32		-	/// Holds the optional sync client behind a poison-recovering lock.
33		-	///
34		-	/// The inner `RwLock` is private to this newtype, so no caller can write
35		-	/// `.read().expect("poisoned")` and crash a background loop on a poisoned lock.
36		-	/// The only access is [`SyncClientCell::get`], which recovers instead of
37		-	/// panicking — the constructive fix for the recurring sync-scheduler poison nit.
38		-	#[derive(Default)]
39		-	pub struct SyncClientCell(RwLock<Option<Arc<SyncKitClient>>>);
40		-
41		-	impl SyncClientCell {
42		-	/// Wrap an initial (possibly absent) client.
43		-	pub(crate) fn new(client: Option<Arc<SyncKitClient>>) -> Self {
44		-	Self(RwLock::new(client))
45		-	}
46		-
47		-	/// Clone out the current client, recovering from a poisoned lock rather
48		-	/// than panicking.
49		-	pub(crate) fn get(&self) -> Option<Arc<SyncKitClient>> {
50		-	self.0.read().unwrap_or_else(\|e\| e.into_inner()).clone()
51		-	}
52		-	}
53		-
54	32		/// Application state holding database connections and repositories
55	33		pub struct AppState {
56	34		pub pool: SqlitePool,
			@@ -70,7 +48,10 @@ pub struct AppState {
70	48		pub monthly_reviews: Arc<dyn MonthlyReviewRepository>,
71	49		pub backup_settings: Arc<dyn BackupSettingsRepository>,
72	50		pub sync_accounts: Arc<dyn SyncAccountRepository>,
73		-	pub sync_client: SyncClientCell,
	51	+	/// The SyncStore engine over `goingson.db`, present when a sync client is
	52	+	/// configured. It owns the underlying `SyncKitClient` (reached via
	53	+	/// `store.client()` for the auth/subscription commands).
	54	+	pub sync_store: Option<Arc<SyncStore<SyncKitClient>>>,
74	55		pub sync_lock: Arc<TokioMutex<()>>,
75	56		/// Per-account email sync locks to prevent concurrent syncs on the same account.
76	57		pub email_sync_locks: Arc<Mutex<std::collections::HashSet<goingson_core::EmailAccountId>>>,
			@@ -168,8 +149,18 @@ impl AppState {
168	149		let backup_settings = Arc::new(SqliteBackupSettingsRepository::new(pool.clone()));
169	150		let sync_accounts = Arc::new(SqliteSyncAccountRepository::new(pool.clone()));
170	151
171		-	// Initialize SyncKit client from saved key or env vars (optional)
172		-	let sync_client = load_sync_client(&app_data_dir);
	152	+	// Groups p4 / M2: migrate this device's sync bookkeeping to the SyncStore
	153	+	// engine (idempotent, guarded by the `syncstore_migrated` flag). Non-fatal
	154	+	// and transactional — a failure leaves the legacy state intact and retries
	155	+	// on the next launch.
	156	+	if let Err(e) = crate::syncstore::run_cutover(&db_path) {
	157	+	warn!("SyncStore cutover failed (will retry next launch): {e}");
	158	+	}
	159	+
	160	+	// Build the SyncStore over goingson.db (its own WAL rusqlite connection,
	161	+	// coexisting with the sqlx pool) when a sync client is configured.
	162	+	let sync_store = load_sync_client(&app_data_dir)
	163	+	.map(\|client\| crate::syncstore::build_store(db_path, app_data_dir.clone(), client));
173	164
174	165		Ok(Self {
175	166		pool,
			@@ -189,7 +180,7 @@ impl AppState {
189	180		monthly_reviews,
190	181		backup_settings,
191	182		sync_accounts,
192		-	sync_client: SyncClientCell::new(sync_client.map(Arc::new)),
	183	+	sync_store,
193	184		sync_lock: Arc::new(TokioMutex::new(())),
194	185		email_sync_locks: Arc::new(Mutex::new(std::collections::HashSet::new())),
195	186		token_refresh_locks: Arc::new(Mutex::new(std::collections::HashMap::new())),
			@@ -199,11 +190,10 @@ impl AppState {
199	190		})
200	191		}
201	192
202		-	/// Clone out the current sync client, recovering from a poisoned lock
203		-	/// instead of panicking. Background loops must use this (never a raw
204		-	/// `.read().expect(...)`), so a writer panic can't kill sync for the session.
	193	+	/// Clone out the underlying sync client, if one is configured. The client is
	194	+	/// owned by the `SyncStore`; the auth/subscription/tier commands reach it here.
205	195		pub(crate) fn read_recovering(&self) -> Option<Arc<SyncKitClient>> {
206		-	self.sync_client.get()
	196	+	self.sync_store.as_ref().map(\|s\| s.client().clone())
207	197		}
208	198
209	199		/// Gets or creates a per-account token refresh lock.

D src-tauri/src/sync_scheduler.rs -244

		@@ -1,244 +0,0 @@
1	-	//! Background scheduler for automatic cloud sync.
2	-	//!
3	-	//! Checks every 60 seconds whether a sync is due based on the configured
4	-	//! interval. On first run, creates an initial snapshot if needed.
5	-	//! An SSE connection receives real-time push notifications from the server,
6	-	//! triggering immediate sync when another device pushes changes.
7	-
8	-	use std::sync::Arc;
9	-	use tauri::{Emitter, Manager};
10	-	use tokio::time::{Duration, interval};
11	-	use tokio_util::sync::CancellationToken;
12	-	use tracing::{debug, error, info, warn};
13	-
14	-	use crate::state::AppState;
15	-	use crate::sync_service;
16	-
17	-	/// How often the scheduler checks if sync is due (seconds).
18	-	const CHECK_INTERVAL_SECS: u64 = 60;
19	-
20	-	/// Delay before reconnecting the SSE stream after a disconnect (seconds).
21	-	const SSE_RECONNECT_DELAY_SECS: u64 = 5;
22	-
23	-	/// Starts the cloud sync scheduler background task.
24	-	pub async fn start_sync_scheduler(app: tauri::AppHandle, cancel: CancellationToken) {
25	-	let mut check_interval = interval(Duration::from_secs(CHECK_INTERVAL_SECS));
26	-
27	-	info!(
28	-	"Cloud sync scheduler started (checking every {} seconds)",
29	-	CHECK_INTERVAL_SECS
30	-	);
31	-
32	-	let mut consecutive_failures: u32 = 0;
33	-	let mut backoff_until: Option<chrono::DateTime<chrono::Utc>> = None;
34	-	// Flag set by SSE notification to trigger immediate sync
35	-	let mut sse_triggered = false;
36	-	// SSE stream handle — None until first successful connection
37	-	let mut sse_stream: Option<synckit_client::SyncNotifyStream> = None;
38	-
39	-	loop {
40	-	// Wait for either: timer tick, cancellation, or SSE notification
41	-	tokio::select! {
42	-	_ = cancel.cancelled() => {
43	-	info!("Cloud sync scheduler shutting down");
44	-	break;
45	-	}
46	-	_ = check_interval.tick() => {}
47	-	result = async {
48	-	if let Some(ref mut stream) = sse_stream {
49	-	stream.next_change().await
50	-	} else {
51	-	// No stream — sleep forever (timer or cancel will fire)
52	-	std::future::pending::<Option<()>>().await
53	-	}
54	-	} => {
55	-	match result {
56	-	Some(()) => {
57	-	debug!("SSE: received change notification, triggering immediate sync");
58	-	sse_triggered = true;
59	-	}
60	-	None => {
61	-	// Stream ended — reconnect after delay
62	-	debug!("SSE: stream disconnected, will reconnect");
63	-	sse_stream = None;
64	-	tokio::time::sleep(Duration::from_secs(SSE_RECONNECT_DELAY_SECS)).await;
65	-	}
66	-	}
67	-	}
68	-	}
69	-
70	-	let state: Arc<AppState> = match app.try_state::<Arc<AppState>>() {
71	-	Some(s) => s.inner().clone(),
72	-	None => {
73	-	debug!("Sync scheduler: app state not available yet");
74	-	continue;
75	-	}
76	-	};
77	-
78	-	let client: Arc<synckit_client::SyncKitClient> = match state.read_recovering() {
79	-	Some(c) => c,
80	-	None => continue,
81	-	};
82	-
83	-	// Must be authenticated with a non-expired token
84	-	if client.session_info().is_none() {
85	-	continue;
86	-	}
87	-	if client.is_token_expired() {
88	-	debug!("Sync scheduler: token expired, skipping sync");
89	-	client.clear_session();
90	-	let _ = app.emit("sync:status-changed", "logged_out");
91	-	sse_stream = None;
92	-	continue;
93	-	}
94	-
95	-	// Must have encryption key loaded
96	-	if !client.has_master_key() {
97	-	continue;
98	-	}
99	-
100	-	// Try to establish SSE connection if not connected
101	-	if sse_stream.is_none() {
102	-	match client.subscribe().await {
103	-	Ok(stream) => {
104	-	debug!("SSE: connected to push notification stream");
105	-	sse_stream = Some(stream);
106	-	}
107	-	Err(e) => {
108	-	debug!("SSE: failed to connect (will retry): {}", e);
109	-	}
110	-	}
111	-	}
112	-
113	-	// Check auto_sync_enabled
114	-	let enabled = match sync_service::get_sync_state(&state.pool, "auto_sync_enabled").await {
115	-	Ok(v) => v == "1",
116	-	Err(e) => {
117	-	warn!("Sync scheduler: failed to read auto_sync_enabled: {}", e);
118	-	continue;
119	-	}
120	-	};
121	-	if !enabled {
122	-	continue;
123	-	}
124	-
125	-	// Check if sync interval has elapsed (skip check if SSE-triggered)
126	-	if !sse_triggered {
127	-	let interval_minutes: u64 = match sync_service::get_sync_state(
128	-	&state.pool,
129	-	"sync_interval_minutes",
130	-	)
131	-	.await
132	-	{
133	-	Ok(v) => match v.parse() {
134	-	Ok(mins) => mins,
135	-	Err(e) => {
136	-	warn!(value = %v, error = %e, "Failed to parse sync_interval_minutes, using default 5");
137	-	5
138	-	}
139	-	},
140	-	Err(e) => {
141	-	warn!(error = %e, "Failed to read sync_interval_minutes, using default 5");
142	-	5
143	-	}
144	-	};
145	-
146	-	let last_sync = match sync_service::get_sync_state(&state.pool, "last_sync_at").await {
147	-	Ok(v) => v,
148	-	Err(e) => {
149	-	warn!(error = %e, "Failed to read last_sync_at");
150	-	String::new()
151	-	}
152	-	};
153	-
154	-	if !last_sync.is_empty()
155	-	&& let Ok(last) = chrono::DateTime::parse_from_rfc3339(&last_sync)
156	-	{
157	-	let elapsed = chrono::Utc::now() - last.with_timezone(&chrono::Utc);
158	-	if elapsed.num_minutes() < interval_minutes as i64 {
159	-	continue;
160	-	}
161	-	}
162	-	}
163	-	sse_triggered = false;
164	-
165	-	// Check backoff
166	-	if let Some(until) = backoff_until
167	-	&& chrono::Utc::now() < until
168	-	{
169	-	debug!("Sync scheduler: backing off until {}", until);
170	-	continue;
171	-	}
172	-
173	-	// Create initial snapshot on first sync
174	-	let snapshot_done = sync_service::get_sync_state(&state.pool, "initial_snapshot_done")
175	-	.await
176	-	.unwrap_or_default();
177	-
178	-	if snapshot_done != "1" {
179	-	match sync_service::create_initial_snapshot(&state.pool).await {
180	-	Ok(count) => info!("Initial sync snapshot: {} rows", count),
181	-	Err(e) => {
182	-	error!("Failed to create initial snapshot: {}", e);
183	-	continue;
184	-	}
185	-	}
186	-	}
187	-
188	-	// Perform sync (acquire lock to prevent concurrent syncs with manual sync_now)
189	-	let _sync_guard = state.sync_lock.lock().await;
190	-	let _ = app.emit("sync:status-changed", "syncing");
191	-	match sync_service::perform_sync_with_blobs(&state.pool, &client, Some(&state.data_dir))
192	-	.await
193	-	{
194	-	Ok(result) => {
195	-	consecutive_failures = 0;
196	-	backoff_until = None;
197	-	let _ = app.emit("sync:status-changed", "idle");
198	-	if result.pushed > 0 \|\| result.pulled > 0 {
199	-	info!(
200	-	"Auto-sync: pushed {}, pulled {}",
201	-	result.pushed, result.pulled
202	-	);
203	-	}
204	-	if result.pulled > 0 {
205	-	// Carry the changed tables so the UI invalidates selectively.
206	-	let _ = app.emit("sync:changes-applied", &result.pulled_tables);
207	-	}
208	-	}
209	-	Err(e) => {
210	-	// If the server returned 402 (payment required), stop retrying —
211	-	// the user needs to subscribe before sync will work.
212	-	let is_payment_required = e.to_string().contains("402");
213	-	if is_payment_required {
214	-	let _ = app.emit("sync:subscription-required", ());
215	-	let _ = app.emit("sync:status-changed", "subscription_required");
216	-	warn!("Auto-sync: subscription required, pausing scheduler");
217	-	// Back off for 1 hour — recheck after that in case user subscribes
218	-	backoff_until = Some(chrono::Utc::now() + chrono::Duration::hours(1));
219	-	} else {
220	-	consecutive_failures += 1;
221	-	// Clamp the exponent before pow: an unbounded counter (failures
222	-	// only reset on success) would overflow u64 at 64 consecutive
223	-	// failures — panic in debug, wrap-to-0 (tight retry loop) in
224	-	// release. min(4) caps backoff at 16m before the 15m clamp,
225	-	// matching email_sync_scheduler.
226	-	let backoff_minutes = std::cmp::min(2u64.pow(consecutive_failures.min(4)), 15);
227	-	backoff_until = Some(
228	-	chrono::Utc::now() + chrono::Duration::minutes(backoff_minutes as i64),
229	-	);
230	-	let _ = app.emit("sync:status-changed", "error");
231	-	warn!(
232	-	"Auto-sync failed (attempt {}, backoff {}m): {}",
233	-	consecutive_failures, backoff_minutes, e
234	-	);
235	-	}
236	-	}
237	-	}
238	-
239	-	// Cleanup old changelog entries
240	-	if let Err(e) = sync_service::cleanup_changelog(&state.pool).await {
241	-	warn!("Sync changelog cleanup failed: {}", e);
242	-	}
243	-	}
244	-	}

D src-tauri/src/sync_service/apply.rs -495

		@@ -1,495 +0,0 @@
1	-	//! Table column whitelists and row-level apply logic (upsert, delete).
2	-	//!
3	-	//! # SQL Safety: `format!()` for table and column names
4	-	//!
5	-	//! Several functions in this module use `format!()` to interpolate table and column
6	-	//! names into SQL strings (e.g., `apply_upsert`, `apply_delete`).
7	-	//! This is safe because:
8	-	//!
9	-	//! 1. Table names come from hardcoded constants (`UPSERT_ORDER`, `DELETE_ORDER`) -- never
10	-	//! from user input or remote data.
11	-	//! 2. Column names come from `table_columns()`, a compile-time whitelist of `&'static str`
12	-	//! literals -- also never from user input.
13	-	//! 3. All user-supplied values (row IDs, field data) are passed through `sqlx::query().bind()`,
14	-	//! which parameterizes them safely.
15	-	//! 4. Unknown table names are rejected before any SQL is constructed: both `apply_upsert` and
16	-	//! `apply_delete` return an error if `table_columns()` returns `None`.
17	-
18	-	use goingson_core::CoreError;
19	-	use sqlx::{Row, SqliteConnection};
20	-
21	-	use super::EMAIL_ACCOUNT_SYNC_COLS;
22	-
23	-	/// The single source of truth for which columns each table syncs.
24	-	///
25	-	/// CHRONIC-C invariant: every column a sync changelog trigger emits into
26	-	/// `sync_changelog.data` (via `json_object(...)`) must appear here, and every column
27	-	/// here must be emitted by the trigger. `apply_upsert` and `create_initial_snapshot`
28	-	/// both read their column list from this table, so the apply side has exactly one list.
29	-	/// The trigger side lives in raw migration SQL (`migrations/sqlite/*.sql`); the
30	-	/// `trigger_columns_match_whitelist` round-trip test reads the live trigger DDL and
31	-	/// asserts the emitted `json_object` keys equal this list for every table, so the two
32	-	/// can never silently diverge again. This is the bug behind UF-1: the trigger emitted
33	-	/// `recurrence_rule` but it was missing here, so it arrived NULL on every pull.
34	-	pub(crate) const SYNCED_COLUMNS: &[(&str, &[&str])] = &[
35	-	(
36	-	"projects",
37	-	&[
38	-	"id",
39	-	"name",
40	-	"description",
41	-	"project_type",
42	-	"status",
43	-	"created_at",
44	-	"user_id",
45	-	],
46	-	),
47	-	(
48	-	"tasks",
49	-	&[
50	-	"id",
51	-	"project_id",
52	-	"description",
53	-	"status",
54	-	"priority",
55	-	"due",
56	-	"tags",
57	-	"urgency",
58	-	"recurrence",
59	-	"recurrence_rule",
60	-	"created_at",
61	-	"user_id",
62	-	"recurrence_parent_id",
63	-	"source_email_id",
64	-	"snoozed_until",
65	-	"waiting_for_response",
66	-	"waiting_since",
67	-	"expected_response_date",
68	-	"scheduled_start",
69	-	"scheduled_duration",
70	-	"is_focus",
71	-	"focus_set_at",
72	-	"contact_id",
73	-	"milestone_id",
74	-	"completed_at",
75	-	"estimated_minutes",
76	-	"actual_minutes",
77	-	],
78	-	),
79	-	(
80	-	"events",
81	-	&[
82	-	"id",
83	-	"project_id",
84	-	"title",
85	-	"description",
86	-	"start_time",
87	-	"end_time",
88	-	"location",
89	-	"user_id",
90	-	"linked_task_id",
91	-	"recurrence",
92	-	"recurrence_parent_id",
93	-	"recurrence_rule",
94	-	"contact_id",
95	-	"block_type",
96	-	"external_source",
97	-	"external_id",
98	-	"is_read_only",
99	-	"snoozed_until",
100	-	"reminder_offsets_seconds",
101	-	],
102	-	),
103	-	(
104	-	"contacts",
105	-	&[
106	-	"id",
107	-	"user_id",
108	-	"display_name",
109	-	"nickname",
110	-	"company",
111	-	"title",
112	-	"notes",
113	-	"tags",
114	-	"birthday",
115	-	"timezone",
116	-	"external_source",
117	-	"external_id",
118	-	"created_at",
119	-	"updated_at",
120	-	],
121	-	),
122	-	(
123	-	"contact_emails",
124	-	&["id", "contact_id", "address", "label", "is_primary"],
125	-	),
126	-	(
127	-	"contact_phones",
128	-	&["id", "contact_id", "number", "label", "is_primary"],
129	-	),
130	-	(
131	-	"contact_social_handles",
132	-	&["id", "contact_id", "platform", "handle", "url"],
133	-	),
134	-	(
135	-	"contact_custom_fields",
136	-	&["id", "contact_id", "label", "value", "url"],
137	-	),
138	-	("annotations", &["id", "task_id", "timestamp", "note"]),
139	-	(
140	-	"subtasks",
141	-	&[
142	-	"id",
143	-	"task_id",
144	-	"text",
145	-	"is_completed",
146	-	"position",
147	-	"created_at",
148	-	"linked_task_id",
149	-	],
150	-	),
151	-	(
152	-	"task_status_tokens",
153	-	&[
154	-	"id",
155	-	"task_id",
156	-	"kind",
157	-	"reference",
158	-	"state",
159	-	"is_primary",
160	-	"position",
161	-	"created_at",
162	-	],
163	-	),
164	-	(
165	-	"milestones",
166	-	&[
167	-	"id",
168	-	"user_id",
169	-	"project_id",
170	-	"name",
171	-	"description",
172	-	"position",
173	-	"target_date",
174	-	"status",
175	-	"created_at",
176	-	],
177	-	),
178	-	(
179	-	"time_sessions",
180	-	&[
181	-	"id",
182	-	"task_id",
183	-	"user_id",
184	-	"started_at",
185	-	"ended_at",
186	-	"duration_minutes",
187	-	"created_at",
188	-	],
189	-	),
190	-	(
191	-	"attachments",
192	-	&[
193	-	"id",
194	-	"user_id",
195	-	"task_id",
196	-	"project_id",
197	-	"filename",
198	-	"file_size",
199	-	"mime_type",
200	-	"blob_hash",
201	-	"source_email_id",
202	-	"created_at",
203	-	],
204	-	),
205	-	(
206	-	"sync_accounts",
207	-	&[
208	-	"id",
209	-	"user_id",
210	-	"provider",
211	-	"account_name",
212	-	"email",
213	-	"sync_calendars",
214	-	"sync_contacts",
215	-	"calendar_ids",
216	-	"sync_interval_minutes",
217	-	"enabled",
218	-	"created_at",
219	-	],
220	-	),
221	-	("email_accounts", EMAIL_ACCOUNT_SYNC_COLS),
222	-	(
223	-	"daily_notes",
224	-	&[
225	-	"id",
226	-	"user_id",
227	-	"note_date",
228	-	"went_well",
229	-	"could_improve",
230	-	"is_reviewed",
231	-	"reviewed_at",
232	-	"created_at",
233	-	"updated_at",
234	-	],
235	-	),
236	-	(
237	-	"saved_views",
238	-	&[
239	-	"id",
240	-	"user_id",
241	-	"name",
242	-	"view_type",
243	-	"filters",
244	-	"sort_by",
245	-	"sort_order",
246	-	"is_pinned",
247	-	"position",
248	-	"created_at",
249	-	"updated_at",
250	-	],
251	-	),
252	-	(
253	-	"weekly_reviews",
254	-	&[
255	-	"id",
256	-	"user_id",
257	-	"week_start_date",
258	-	"completed_at",
259	-	"notes",
260	-	"vacation_days",
261	-	],
262	-	),
263	-	(
264	-	"monthly_goals",
265	-	&[
266	-	"id",
267	-	"user_id",
268	-	"month",
269	-	"text",
270	-	"status",
271	-	"position",
272	-	"created_at",
273	-	"updated_at",
274	-	],
275	-	),
276	-	(
277	-	"monthly_reflections",
278	-	&[
279	-	"id",
280	-	"user_id",
281	-	"month",
282	-	"highlight_text",
283	-	"change_text",
284	-	"completed_at",
285	-	],
286	-	),
287	-	];
288	-
289	-	/// Return the syncable column whitelist for a given table, or `None` if unknown.
290	-	pub(crate) fn table_columns(table: &str) -> Option<&'static [&'static str]> {
291	-	SYNCED_COLUMNS
292	-	.iter()
293	-	.find(\|(t, _)\| *t == table)
294	-	.map(\|(_, cols)\| *cols)
295	-	}
296	-
297	-	/// Apply an upsert for a remote change.
298	-	///
299	-	/// Uses `INSERT ... ON CONFLICT(id) DO UPDATE` rather than `INSERT OR REPLACE`.
300	-	/// `INSERT OR REPLACE` deletes the conflicting row before reinserting it, which
301	-	/// fires `ON DELETE CASCADE` on children (annotations, subtasks, contact_*) and
302	-	/// would wipe them on any parent re-upsert. `ON CONFLICT DO UPDATE` mutates the
303	-	/// row in place, so children survive and referential integrity holds regardless
304	-	/// of FK enforcement. Every syncable table has `id` as its primary key.
305	-	#[tracing::instrument(skip_all)]
306	-	pub(crate) async fn apply_upsert(
307	-	conn: &mut SqliteConnection,
308	-	table: &str,
309	-	_row_id: &str,
310	-	data: &serde_json::Value,
311	-	) -> Result<(), CoreError> {
312	-	// Email accounts use a bespoke ON CONFLICT that preserves local credentials
313	-	if table == "email_accounts" {
314	-	return apply_email_account_upsert(conn, data).await;
315	-	}
316	-
317	-	let columns = table_columns(table)
318	-	.ok_or_else(\|\| CoreError::bad_request(format!("unknown syncable table: {}", table)))?;
319	-
320	-	// A remote null for a NOT NULL column is invalid input the changelog never
321	-	// emits. INSERT OR REPLACE silently coerced it to the column default; to
322	-	// keep that tolerance under ON CONFLICT, omit such columns: a new row then
323	-	// takes the schema default and an existing row keeps its current value.
324	-	// Nullable columns keep null (so legitimate clears — e.g. un-snoozing —
325	-	// still propagate). NOT NULL columns are derived from the live schema, not
326	-	// a hand-maintained list, so they can't drift.
327	-	let not_null = not_null_columns(conn, table).await?;
328	-	let included: Vec<&str> = columns
329	-	.iter()
330	-	.copied()
331	-	.filter(\|c\| !(data[c].is_null() && not_null.contains(c)))
332	-	.collect();
333	-
334	-	let col_list = included.join(", ");
335	-	let placeholders = included.iter().map(\|_\| "?").collect::<Vec<_>>().join(", ");
336	-	let update_set = included
337	-	.iter()
338	-	.filter(\|c\| **c != "id")
339	-	.map(\|c\| format!("{} = excluded.{}", c, c))
340	-	.collect::<Vec<_>>()
341	-	.join(", ");
342	-	// If only `id` survived, there is nothing to update on conflict.
343	-	let conflict = if update_set.is_empty() {
344	-	"DO NOTHING".to_string()
345	-	} else {
346	-	format!("DO UPDATE SET {}", update_set)
347	-	};
348	-	let sql = format!(
349	-	"INSERT INTO {} ({}) VALUES ({}) ON CONFLICT(id) {}",
350	-	table, col_list, placeholders, conflict
351	-	);
352	-
353	-	let mut query = sqlx::query(sqlx::AssertSqlSafe(sql.as_str()));
354	-
355	-	for col in &included {
356	-	query = bind_json_value(query, &data[*col]);
357	-	}
358	-
359	-	query
360	-	.execute(&mut *conn)
361	-	.await
362	-	.map_err(CoreError::database)?;
363	-
364	-	Ok(())
365	-	}
366	-
367	-	/// Names of the NOT NULL columns of a table, read from the live schema.
368	-	///
369	-	/// Derived via `PRAGMA table_info` rather than a static list so it can never
370	-	/// drift from the migrations. `table` is already validated against the column
371	-	/// whitelist by the caller, so interpolating it is safe.
372	-	async fn not_null_columns(
373	-	conn: &mut SqliteConnection,
374	-	table: &str,
375	-	) -> Result<std::collections::HashSet<String>, CoreError> {
376	-	let rows = sqlx::query(sqlx::AssertSqlSafe(format!("PRAGMA table_info({})", table)))
377	-	.fetch_all(&mut *conn)
378	-	.await
379	-	.map_err(CoreError::database)?;
380	-
381	-	let mut set = std::collections::HashSet::new();
382	-	for row in rows {
383	-	let notnull: i64 = row.try_get("notnull").map_err(CoreError::database)?;
384	-	if notnull != 0 {
385	-	let name: String = row.try_get("name").map_err(CoreError::database)?;
386	-	set.insert(name);
387	-	}
388	-	}
389	-	Ok(set)
390	-	}
391	-
392	-	/// Apply an upsert for email_accounts that preserves local credentials.
393	-	///
394	-	/// Uses INSERT ... ON CONFLICT(id) DO UPDATE to only touch the 16 config columns,
395	-	/// leaving `password`, `oauth2_access_token`, `oauth2_refresh_token`, and
396	-	/// `oauth2_token_expires_at` untouched on existing rows. New rows get `password = ''`
397	-	/// to satisfy the NOT NULL constraint.
398	-	#[tracing::instrument(skip_all)]
399	-	async fn apply_email_account_upsert(
400	-	conn: &mut SqliteConnection,
401	-	data: &serde_json::Value,
402	-	) -> Result<(), CoreError> {
403	-	let cols = EMAIL_ACCOUNT_SYNC_COLS;
404	-
405	-	// INSERT columns: 16 sync cols + password (hardcoded to '')
406	-	let mut insert_cols: Vec<&str> = cols.to_vec();
407	-	insert_cols.push("password");
408	-
409	-	let col_list = insert_cols.join(", ");
410	-	let placeholders = insert_cols
411	-	.iter()
412	-	.map(\|_\| "?")
413	-	.collect::<Vec<_>>()
414	-	.join(", ");
415	-
416	-	// ON CONFLICT: only update the 16 sync columns
417	-	let update_set = cols
418	-	.iter()
419	-	.filter(\|c\| **c != "id")
420	-	.map(\|c\| format!("{} = excluded.{}", c, c))
421	-	.collect::<Vec<_>>()
422	-	.join(", ");
423	-
424	-	let sql = format!(
425	-	"INSERT INTO email_accounts ({}) VALUES ({}) ON CONFLICT(id) DO UPDATE SET {}",
426	-	col_list, placeholders, update_set
427	-	);
428	-
429	-	let mut query = sqlx::query(sqlx::AssertSqlSafe(sql.as_str()));
430	-
431	-	// Bind the 16 sync columns from data
432	-	for col in cols {
433	-	query = bind_json_value(query, &data[*col]);
434	-	}
435	-
436	-	// Bind password = '' for the INSERT
437	-	query = query.bind("");
438	-
439	-	query
440	-	.execute(&mut *conn)
441	-	.await
442	-	.map_err(CoreError::database)?;
443	-
444	-	Ok(())
445	-	}
446	-
447	-	/// Bind a JSON value to a sqlx query.
448	-	pub(crate) fn bind_json_value<'q>(
449	-	query: sqlx::query::Query<'q, sqlx::Sqlite, sqlx::sqlite::SqliteArguments>,
450	-	val: &'q serde_json::Value,
451	-	) -> sqlx::query::Query<'q, sqlx::Sqlite, sqlx::sqlite::SqliteArguments> {
452	-	match val {
453	-	serde_json::Value::String(s) => query.bind(s.as_str()),
454	-	serde_json::Value::Number(n) => {
455	-	if let Some(i) = n.as_i64() {
456	-	query.bind(i)
457	-	} else if let Some(f) = n.as_f64() {
458	-	query.bind(f)
459	-	} else {
460	-	query.bind(None::<String>)
461	-	}
462	-	}
463	-	serde_json::Value::Bool(b) => query.bind(if *b { 1i32 } else { 0i32 }),
464	-	serde_json::Value::Null => query.bind(None::<String>),
465	-	_ => {
466	-	// Arrays/objects: serialize as JSON string -- need owned String
467	-	query.bind(val.to_string())
468	-	}
469	-	}
470	-	}
471	-
472	-	/// Apply a DELETE for a remote change.
473	-	#[tracing::instrument(skip_all)]
474	-	pub(crate) async fn apply_delete(
475	-	conn: &mut SqliteConnection,
476	-	table: &str,
477	-	row_id: &str,
478	-	) -> Result<(), CoreError> {
479	-	// Validate table name is in our whitelist
480	-	if table_columns(table).is_none() {
481	-	return Err(CoreError::bad_request(format!(
482	-	"unknown syncable table: {}",
483	-	table
484	-	)));
485	-	}
486	-
487	-	let sql = format!("DELETE FROM {} WHERE id = ?", table);
488	-	sqlx::query(sqlx::AssertSqlSafe(sql.as_str()))
489	-	.bind(row_id)
490	-	.execute(&mut *conn)
491	-	.await
492	-	.map_err(CoreError::database)?;
493	-
494	-	Ok(())
495	-	}

D src-tauri/src/sync_service/blob_sync.rs -166

		@@ -1,166 +0,0 @@
1	-	//! Blob sync: upload local blobs to SyncKit, download missing blobs from SyncKit.
2	-
3	-	use std::path::Path;
4	-
5	-	use goingson_core::CoreError;
6	-	use sha2::{Digest, Sha256};
7	-	use sqlx::SqlitePool;
8	-	use synckit_client::{BlobUploadOutcome, SyncKitClient};
9	-	use tracing::{debug, info, warn};
10	-
11	-	use crate::commands::attachment::blob_path;
12	-	use crate::state::DESKTOP_USER_ID;
13	-
14	-	/// First 8 chars of a blob hash for log lines. Panic-safe: a `blob_hash` can arrive
15	-	/// via sync and need not be a clean 64-hex string, so a raw `&hash[..8]` could panic
16	-	/// on a short or multibyte value (ultra-fuzz Run #27 Data minor).
17	-	fn short(hash: &str) -> &str {
18	-	hash.get(..8).unwrap_or(hash)
19	-	}
20	-
21	-	/// Upload local blobs that haven't been synced to the server yet.
22	-	///
23	-	/// Queries all distinct blob hashes from attachments, checks which ones have local
24	-	/// files, and uploads them via the SyncKit blob API (presigned S3 + E2E encryption).
25	-	#[tracing::instrument(skip_all)]
26	-	pub async fn upload_pending_blobs(
27	-	pool: &SqlitePool,
28	-	data_dir: &Path,
29	-	client: &SyncKitClient,
30	-	) -> Result<i64, CoreError> {
31	-	let hashes: Vec<(String, i64)> = sqlx::query_as(
32	-	"SELECT DISTINCT a.blob_hash, a.file_size FROM attachments a WHERE a.user_id = ?",
33	-	)
34	-	.bind(DESKTOP_USER_ID.to_string())
35	-	.fetch_all(pool)
36	-	.await
37	-	.map_err(CoreError::database)?;
38	-
39	-	let mut uploaded = 0i64;
40	-
41	-	for (hash, size) in &hashes {
42	-	let path = blob_path(data_dir, hash);
43	-	if !tokio::fs::try_exists(&path).await.unwrap_or(false) {
44	-	continue; // No local blob to upload
45	-	}
46	-
47	-	// Streams the file a part at a time rather than reading the whole
48	-	// attachment into memory, and it is the only path the server accepts
49	-	// above its one-shot PUT ceiling. The server's dedup check runs inside,
50	-	// before any read, so an attachment it already holds costs one round
51	-	// trip and no disk I/O.
52	-	match client.blob_upload_streaming(hash, &path).await {
53	-	Ok(BlobUploadOutcome::AlreadyPresent) => {
54	-	debug!("Blob {} already on server, skipping", short(hash));
55	-	continue;
56	-	}
57	-	Ok(BlobUploadOutcome::Uploaded) => {}
58	-	Err(e) => {
59	-	warn!("Failed to upload blob {}: {}", short(hash), e);
60	-	continue;
61	-	}
62	-	}
63	-
64	-	// Confirm upload
65	-	if let Err(e) = client.blob_confirm(hash, *size).await {
66	-	warn!("Failed to confirm blob {}: {}", short(hash), e);
67	-	continue;
68	-	}
69	-
70	-	uploaded += 1;
71	-	debug!("Uploaded blob {}", short(hash));
72	-	}
73	-
74	-	if uploaded > 0 {
75	-	info!("Uploaded {} blobs", uploaded);
76	-	}
77	-	Ok(uploaded)
78	-	}
79	-
80	-	/// Download blobs that exist in attachment records but not on local disk.
81	-	///
82	-	/// After metadata sync pulls attachment records from other devices, this function
83	-	/// downloads the actual blob data from SyncKit (presigned S3 + E2E decryption).
84	-	#[tracing::instrument(skip_all)]
85	-	pub async fn download_missing_blobs(
86	-	pool: &SqlitePool,
87	-	data_dir: &Path,
88	-	client: &SyncKitClient,
89	-	) -> Result<i64, CoreError> {
90	-	let hashes: Vec<(String,)> =
91	-	sqlx::query_as("SELECT DISTINCT blob_hash FROM attachments WHERE user_id = ?")
92	-	.bind(DESKTOP_USER_ID.to_string())
93	-	.fetch_all(pool)
94	-	.await
95	-	.map_err(CoreError::database)?;
96	-
97	-	let blobs_dir = data_dir.join("blobs");
98	-	tokio::fs::create_dir_all(&blobs_dir)
99	-	.await
100	-	.map_err(\|e\| CoreError::internal(format!("Failed to create blobs dir: {}", e)))?;
101	-
102	-	let mut downloaded = 0i64;
103	-
104	-	for (hash,) in &hashes {
105	-	let path = blob_path(data_dir, hash);
106	-	if tokio::fs::try_exists(&path).await.unwrap_or(false) {
107	-	continue; // Already have it locally
108	-	}
109	-
110	-	// Get download URL
111	-	let download_url = match client.blob_download_url(hash).await {
112	-	Ok(url) => url,
113	-	Err(e) => {
114	-	warn!("Failed to get download URL for blob {}: {}", short(hash), e);
115	-	continue;
116	-	}
117	-	};
118	-
119	-	// Download and decrypt. The SDK now AAD-binds and re-verifies the content
120	-	// hash itself; the local check below stays as belt-and-braces before the
121	-	// file is committed under its hash name.
122	-	let data = match client.blob_download(hash, &download_url).await {
123	-	Ok(d) => d,
124	-	Err(e) => {
125	-	warn!("Failed to download blob {}: {}", short(hash), e);
126	-	continue;
127	-	}
128	-	};
129	-
130	-	// Verify content-addressed integrity before committing the file. The store's
131	-	// invariant is "the file at blobs/<hash> hashes to <hash>" — enforced on write
132	-	// (add_attachment) but previously only assumed on read. E2E AEAD stops a network
133	-	// attacker, but a server-side corruption or mis-bound ciphertext would otherwise
134	-	// land wrong bytes under a trusted name and be handed to open::that()
135	-	// (ultra-fuzz Run #28 S1). Hashing here, before the rename, keeps the store clean.
136	-	let actual = hex::encode(Sha256::digest(&data));
137	-	if actual != *hash {
138	-	warn!(
139	-	"Blob {} failed integrity check (content hashed to {}); discarding download",
140	-	short(hash),
141	-	short(&actual)
142	-	);
143	-	continue;
144	-	}
145	-
146	-	// Write to disk atomically (tmp + rename) to prevent corrupt partial files
147	-	let tmp_path = path.with_extension("tmp");
148	-	if let Err(e) = tokio::fs::write(&tmp_path, &data).await {
149	-	warn!("Failed to write blob {}: {}", short(hash), e);
150	-	continue;
151	-	}
152	-	if let Err(e) = tokio::fs::rename(&tmp_path, &path).await {
153	-	warn!("Failed to rename blob {}: {}", short(hash), e);
154	-	let _ = tokio::fs::remove_file(&tmp_path).await;
155	-	continue;
156	-	}
157	-
158	-	downloaded += 1;
159	-	debug!("Downloaded blob {}", short(hash));
160	-	}
161	-
162	-	if downloaded > 0 {
163	-	info!("Downloaded {} blobs", downloaded);
164	-	}
165	-	Ok(downloaded)
166	-	}

D src-tauri/src/sync_service/hlc.rs -198

		@@ -1,198 +0,0 @@
1	-	//! Hybrid logical clock persistence for cross-device sync conflict resolution.
2	-	//!
3	-	//! SyncKit orders conflicts by an [`Hlc`] rather than a bare wall-clock timestamp
4	-	//! (ultra-fuzz Run #28): a skewed-fast device no longer wins every conflict, and a
5	-	//! strictly-newer edit beats an older delete. This module owns the device's
6	-	//! persistent clock — `(wall_ms, counter)` in the `hlc_state` table; the HLC's node
7	-	//! component is always this device's id, so it is not stored.
8	-	//!
9	-	//! Local changelog rows are stamped lazily: when a row is first read for push or
10	-	//! conflict detection it gets the next HLC, and the stamp is written back onto the
11	-	//! row. This keeps the changelog triggers unchanged and guarantees push and
12	-	//! conflict-detection read the same HLC for a given change.
13	-
14	-	use chrono::Utc;
15	-	use goingson_core::CoreError;
16	-	use sqlx::SqlitePool;
17	-	use std::collections::{HashMap, HashSet};
18	-	use synckit_client::{DeviceId, Hlc};
19	-	use uuid::Uuid;
20	-
21	-	/// Load the persistent clock, reattaching `node` (which is not stored).
22	-	async fn load_clock(pool: &SqlitePool, node: Uuid) -> Result<Hlc, CoreError> {
23	-	let (wall_ms, counter): (i64, i64) =
24	-	sqlx::query_as("SELECT wall_ms, counter FROM hlc_state WHERE id = 1")
25	-	.fetch_one(pool)
26	-	.await
27	-	.map_err(CoreError::database)?;
28	-	Ok(Hlc {
29	-	wall_ms,
30	-	counter: counter as u32,
31	-	node: DeviceId::new(node),
32	-	})
33	-	}
34	-
35	-	/// Stamp every unpushed changelog row that lacks an HLC, in id order, advancing the
36	-	/// persistent clock once per row. Idempotent — already-stamped rows are skipped — so
37	-	/// push and conflict-detection observe identical stamps. Runs in one transaction so a
38	-	/// crash mid-stamp leaves the clock and the rows consistent (all-or-nothing).
39	-	#[tracing::instrument(skip_all)]
40	-	pub async fn assign_pending_hlcs(pool: &SqlitePool, device_id: Uuid) -> Result<(), CoreError> {
41	-	let ids: Vec<(i64, String, String)> = sqlx::query_as(
42	-	"SELECT id, table_name, row_id FROM sync_changelog \
43	-	WHERE pushed = 0 AND hlc_wall IS NULL ORDER BY id ASC",
44	-	)
45	-	.fetch_all(pool)
46	-	.await
47	-	.map_err(CoreError::database)?;
48	-
49	-	if ids.is_empty() {
50	-	return Ok(());
51	-	}
52	-
53	-	let mut tx = pool.begin().await.map_err(CoreError::database)?;
54	-
55	-	let (wall_ms, counter): (i64, i64) =
56	-	sqlx::query_as("SELECT wall_ms, counter FROM hlc_state WHERE id = 1")
57	-	.fetch_one(&mut *tx)
58	-	.await
59	-	.map_err(CoreError::database)?;
60	-	let node = DeviceId::new(device_id);
61	-	let mut clock = Hlc {
62	-	wall_ms,
63	-	counter: counter as u32,
64	-	node,
65	-	};
66	-
67	-	for (id, table_name, row_id) in ids {
68	-	clock = Hlc::tick(clock, Utc::now().timestamp_millis(), node);
69	-	sqlx::query("UPDATE sync_changelog SET hlc_wall = ?, hlc_counter = ? WHERE id = ?")
70	-	.bind(clock.wall_ms)
71	-	.bind(clock.counter as i64)
72	-	.bind(id)
73	-	.execute(&mut *tx)
74	-	.await
75	-	.map_err(CoreError::database)?;
76	-	// The row now holds this local value at `clock`; record it as the row's
77	-	// committed HLC so a later older remote edit gates out instead of clobbering it.
78	-	record_committed_hlc(&mut *tx, &table_name, &row_id, clock).await?;
79	-	}
80	-
81	-	sqlx::query("UPDATE hlc_state SET wall_ms = ?, counter = ? WHERE id = 1")
82	-	.bind(clock.wall_ms)
83	-	.bind(clock.counter as i64)
84	-	.execute(&mut *tx)
85	-	.await
86	-	.map_err(CoreError::database)?;
87	-
88	-	tx.commit().await.map_err(CoreError::database)?;
89	-	Ok(())
90	-	}
91	-
92	-	/// Advance the persistent clock past a remote HLC observed during pull, so that
93	-	/// subsequent local changes causally follow it. Monotonic: advancing past the batch
94	-	/// maximum is enough to cover every change in the batch.
95	-	#[tracing::instrument(skip_all)]
96	-	pub async fn observe_remote(
97	-	pool: &SqlitePool,
98	-	remote: Hlc,
99	-	device_id: Uuid,
100	-	) -> Result<(), CoreError> {
101	-	let clock = load_clock(pool, device_id).await?;
102	-	let advanced = Hlc::observe(
103	-	clock,
104	-	remote,
105	-	Utc::now().timestamp_millis(),
106	-	DeviceId::new(device_id),
107	-	);
108	-	sqlx::query("UPDATE hlc_state SET wall_ms = ?, counter = ? WHERE id = 1")
109	-	.bind(advanced.wall_ms)
110	-	.bind(advanced.counter as i64)
111	-	.execute(pool)
112	-	.await
113	-	.map_err(CoreError::database)?;
114	-	Ok(())
115	-	}
116	-
117	-	/// Record the committed HLC for a row, keeping the maximum.
118	-	///
119	-	/// This is the committed-clock store SyncKit's `CleanChanges` gate reads: a clean
120	-	/// remote change older-or-equal to a row's committed HLC is dropped rather than
121	-	/// clobbering a newer local value. Called on every applied change — a local stamp
122	-	/// ([`assign_pending_hlcs`]) or a remote apply (`pull::apply_changes_inner`) — so
123	-	/// the stored clock always reflects the value currently in the row. The
124	-	/// `ON CONFLICT ... WHERE` keeps the larger HLC, comparing `(wall, counter, node)`
125	-	/// as a row value (node is the canonical lowercase UUID string, whose lexicographic
126	-	/// order matches the byte order [`Hlc`] compares on).
127	-	pub(crate) async fn record_committed_hlc<'e, E>(
128	-	executor: E,
129	-	table: &str,
130	-	row_id: &str,
131	-	hlc: Hlc,
132	-	) -> Result<(), CoreError>
133	-	where
134	-	E: sqlx::Executor<'e, Database = sqlx::Sqlite>,
135	-	{
136	-	sqlx::query(
137	-	"INSERT INTO sync_committed_hlc (table_name, row_id, hlc_wall, hlc_counter, hlc_node) \
138	-	VALUES (?, ?, ?, ?, ?) \
139	-	ON CONFLICT(table_name, row_id) DO UPDATE SET \
140	-	hlc_wall = excluded.hlc_wall, \
141	-	hlc_counter = excluded.hlc_counter, \
142	-	hlc_node = excluded.hlc_node \
143	-	WHERE (excluded.hlc_wall, excluded.hlc_counter, excluded.hlc_node) > \
144	-	(sync_committed_hlc.hlc_wall, sync_committed_hlc.hlc_counter, sync_committed_hlc.hlc_node)",
145	-	)
146	-	.bind(table)
147	-	.bind(row_id)
148	-	.bind(hlc.wall_ms)
149	-	.bind(hlc.counter as i64)
150	-	.bind(hlc.node.to_string())
151	-	.execute(executor)
152	-	.await
153	-	.map_err(CoreError::database)?;
154	-	Ok(())
155	-	}
156	-
157	-	/// Load the committed HLCs for a set of `(table, row_id)` keys, for pre-fetching
158	-	/// before the `CleanChanges` gate. Queries by `row_id` (one statement) and filters
159	-	/// the exact `(table, row_id)` pairs in memory, so a row never seen locally is
160	-	/// simply absent from the map (the gate then keeps that change).
161	-	pub(crate) async fn load_committed_hlcs(
162	-	pool: &SqlitePool,
163	-	keys: &[(String, String)],
164	-	) -> Result<HashMap<(String, String), Hlc>, CoreError> {
165	-	let mut map = HashMap::new();
166	-	if keys.is_empty() {
167	-	return Ok(map);
168	-	}
169	-
170	-	let placeholders = vec!["?"; keys.len()].join(",");
171	-	let sql = format!(
172	-	"SELECT table_name, row_id, hlc_wall, hlc_counter, hlc_node \
173	-	FROM sync_committed_hlc WHERE row_id IN ({placeholders})"
174	-	);
175	-	let mut query =
176	-	sqlx::query_as::<_, (String, String, i64, i64, String)>(sqlx::AssertSqlSafe(sql.as_str()));
177	-	for (_, row_id) in keys {
178	-	query = query.bind(row_id);
179	-	}
180	-	let rows = query.fetch_all(pool).await.map_err(CoreError::database)?;
181	-
182	-	let want: HashSet<(&str, &str)> = keys.iter().map(\|(t, r)\| (t.as_str(), r.as_str())).collect();
183	-	for (table, row_id, wall, counter, node) in rows {
184	-	if want.contains(&(table.as_str(), row_id.as_str()))
185	-	&& let Ok(node) = Uuid::parse_str(&node)
186	-	{
187	-	map.insert(
188	-	(table, row_id),
189	-	Hlc {
190	-	wall_ms: wall,
191	-	counter: counter as u32,
192	-	node: DeviceId::new(node),
193	-	},
194	-	);
195	-	}
196	-	}
197	-	Ok(map)
198	-	}

D src-tauri/src/sync_service/manifest.rs -374

		@@ -1,374 +0,0 @@
1	-	//! The declarative [`SyncSchema`] manifest for GoingsOn's synced tables.
2	-	//!
3	-	//! This is the M1 slice of the GO -> `SyncStore` migration (SyncKit Groups p4):
4	-	//! it declares what GO syncs in the engine's manifest form without yet cutting
5	-	//! the live sync path over to `SyncStore`. The hand-rolled `push.rs`/`pull.rs`/
6	-	//! `apply.rs`/`hlc.rs` still drive sync until M2.
7	-	//!
8	-	//! The manifest is the single source of truth the engine will use to generate the
9	-	//! changelog triggers, the initial snapshot, and the apply-side binding. It must
10	-	//! reproduce GO's current behaviour exactly, so the column lists here are the same
11	-	//! `SYNCED_COLUMNS` whitelist `apply.rs` uses today, in the same FK-safe
12	-	//! [`UPSERT_ORDER`](super::UPSERT_ORDER) (parents first; the engine derives the
13	-	//! delete order by reversing it). The `manifest_matches_synced_columns` test below
14	-	//! is the behaviour-diff that guards against any table's sync policy silently
15	-	//! changing.
16	-	//!
17	-	//! Three tables need more than the common `t(name, cols)` default:
18	-	//! - `email_accounts` syncs config only. Passwords and OAuth tokens never leave
19	-	//! the device: they are excluded from the synced columns (so they never enter
20	-	//! the changelog), `preserve_local` keeps them untouched on a conflict-update,
21	-	//! and `insert_defaults` satisfies the `password` NOT NULL on a fresh remote
22	-	//! insert with `''`. This replaces GO's bespoke `apply_email_account_upsert`.
23	-	//! - `tasks` and `attachments` both carry `source_email_id`, a foreign key into
24	-	//! the `emails` table, which is NOT synced (emails stay per-device). They are
25	-	//! marked `references_unsynced` so FK enforcement is relaxed during a remote
26	-	//! apply — the engine's equivalent of GO's current blanket `foreign_keys=OFF`
27	-	//! over the apply transaction.
28	-	//!
29	-	//! Groups (M3) will add `.group_scoped("group_id")` to the shareable tables; the
30	-	//! personal-only default here means a config/credential table can never leak into
31	-	//! a group scope.
32	-
33	-	// M1 declares the manifest and proves it faithful (the test below); M2 wires
34	-	// `goingson_schema` into the live `SyncStore` path. Until then it is exercised
35	-	// only by its own behaviour-diff test, so suppress the non-test dead-code lint.
36	-	#![allow(dead_code)]
37	-
38	-	use synckit_client::{ConflictStrategy, SyncSchema, SyncTable};
39	-
40	-	use super::EMAIL_ACCOUNT_SYNC_COLS;
41	-
42	-	/// The common case: single `id` primary key, cleartext wire row-id, `Full`
43	-	/// upsert, `Hard` delete, HLC conflict resolution. Most of GO's tables are one
44	-	/// line each.
45	-	fn t(name: &'static str, cols: &'static [&'static str]) -> SyncTable {
46	-	SyncTable::full(name, cols)
47	-	}
48	-
49	-	/// GoingsOn's sync manifest: every table GO replicates, in FK (parents-first)
50	-	/// order matching [`super::UPSERT_ORDER`].
51	-	pub(crate) fn goingson_schema() -> SyncSchema {
52	-	SyncSchema::new(vec![
53	-	t(
54	-	"projects",
55	-	&[
56	-	"id",
57	-	"name",
58	-	"description",
59	-	"project_type",
60	-	"status",
61	-	"created_at",
62	-	"user_id",
63	-	],
64	-	),
65	-	t(
66	-	"contacts",
67	-	&[
68	-	"id",
69	-	"user_id",
70	-	"display_name",
71	-	"nickname",
72	-	"company",
73	-	"title",
74	-	"notes",
75	-	"tags",
76	-	"birthday",
77	-	"timezone",
78	-	"external_source",
79	-	"external_id",
80	-	"created_at",
81	-	"updated_at",
82	-	],
83	-	),
84	-	// Config only, never credentials. See the module docs.
85	-	SyncTable::full("email_accounts", EMAIL_ACCOUNT_SYNC_COLS)
86	-	.preserve_local(&[
87	-	"password",
88	-	"oauth2_access_token",
89	-	"oauth2_refresh_token",
90	-	"oauth2_token_expires_at",
91	-	])
92	-	.insert_defaults(&[("password", "''")]),
93	-	t(
94	-	"sync_accounts",
95	-	&[
96	-	"id",
97	-	"user_id",
98	-	"provider",
99	-	"account_name",
100	-	"email",
101	-	"sync_calendars",
102	-	"sync_contacts",
103	-	"calendar_ids",
104	-	"sync_interval_minutes",
105	-	"enabled",
106	-	"created_at",
107	-	],
108	-	),
109	-	t(
110	-	"milestones",
111	-	&[
112	-	"id",
113	-	"user_id",
114	-	"project_id",
115	-	"name",
116	-	"description",
117	-	"position",
118	-	"target_date",
119	-	"status",
120	-	"created_at",
121	-	],
122	-	),
123	-	// Carries source_email_id -> the unsynced emails table.
124	-	SyncTable::full(
125	-	"tasks",
126	-	&[
127	-	"id",
128	-	"project_id",
129	-	"description",
130	-	"status",
131	-	"priority",
132	-	"due",
133	-	"tags",
134	-	"urgency",
135	-	"recurrence",
136	-	"recurrence_rule",
137	-	"created_at",
138	-	"user_id",
139	-	"recurrence_parent_id",
140	-	"source_email_id",
141	-	"snoozed_until",
142	-	"waiting_for_response",
143	-	"waiting_since",
144	-	"expected_response_date",
145	-	"scheduled_start",
146	-	"scheduled_duration",
147	-	"is_focus",
148	-	"focus_set_at",
149	-	"contact_id",
150	-	"milestone_id",
151	-	"completed_at",
152	-	"estimated_minutes",
153	-	"actual_minutes",
154	-	],
155	-	)
156	-	.references_unsynced(),
157	-	t(
158	-	"time_sessions",
159	-	&[
160	-	"id",
161	-	"task_id",
162	-	"user_id",
163	-	"started_at",
164	-	"ended_at",
165	-	"duration_minutes",
166	-	"created_at",
167	-	],
168	-	),
169	-	// Also carries source_email_id -> the unsynced emails table.
170	-	SyncTable::full(
171	-	"attachments",
172	-	&[
173	-	"id",
174	-	"user_id",
175	-	"task_id",
176	-	"project_id",
177	-	"filename",
178	-	"file_size",
179	-	"mime_type",
180	-	"blob_hash",
181	-	"source_email_id",
182	-	"created_at",
183	-	],
184	-	)
185	-	.references_unsynced(),
186	-	t(
187	-	"events",
188	-	&[
189	-	"id",
190	-	"project_id",
191	-	"title",
192	-	"description",
193	-	"start_time",
194	-	"end_time",
195	-	"location",
196	-	"user_id",
197	-	"linked_task_id",
198	-	"recurrence",
199	-	"recurrence_parent_id",
200	-	"recurrence_rule",
201	-	"contact_id",
202	-	"block_type",
203	-	"external_source",
204	-	"external_id",
205	-	"is_read_only",
206	-	"snoozed_until",
207	-	"reminder_offsets_seconds",
208	-	],
209	-	),
210	-	t("annotations", &["id", "task_id", "timestamp", "note"]),
211	-	t(
212	-	"subtasks",
213	-	&[
214	-	"id",
215	-	"task_id",
216	-	"text",
217	-	"is_completed",
218	-	"position",
219	-	"created_at",
220	-	"linked_task_id",
221	-	],
222	-	),
223	-	t(
224	-	"task_status_tokens",
225	-	&[
226	-	"id",
227	-	"task_id",
228	-	"kind",
229	-	"reference",
230	-	"state",
231	-	"is_primary",
232	-	"position",
233	-	"created_at",
234	-	],
235	-	),
236	-	t(
237	-	"contact_emails",
238	-	&["id", "contact_id", "address", "label", "is_primary"],
239	-	),
240	-	t(
241	-	"contact_phones",
242	-	&["id", "contact_id", "number", "label", "is_primary"],
243	-	),
244	-	t(
245	-	"contact_social_handles",
246	-	&["id", "contact_id", "platform", "handle", "url"],
247	-	),
248	-	t(
249	-	"contact_custom_fields",
250	-	&["id", "contact_id", "label", "value", "url"],
251	-	),
252	-	t(
253	-	"daily_notes",
254	-	&[
255	-	"id",
256	-	"user_id",
257	-	"note_date",
258	-	"went_well",
259	-	"could_improve",
260	-	"is_reviewed",
261	-	"reviewed_at",
262	-	"created_at",
263	-	"updated_at",
264	-	],
265	-	),
266	-	// Independent roots (FK only to local `users`); order among them is
267	-	// irrelevant, but kept identical to UPSERT_ORDER for the diff test.
268	-	t(
269	-	"saved_views",
270	-	&[
271	-	"id",
272	-	"user_id",
273	-	"name",
274	-	"view_type",
275	-	"filters",
276	-	"sort_by",
277	-	"sort_order",
278	-	"is_pinned",
279	-	"position",
280	-	"created_at",
281	-	"updated_at",
282	-	],
283	-	),
284	-	t(
285	-	"weekly_reviews",
286	-	&[
287	-	"id",
288	-	"user_id",
289	-	"week_start_date",
290	-	"completed_at",
291	-	"notes",
292	-	"vacation_days",
293	-	],
294	-	),
295	-	t(
296	-	"monthly_goals",
297	-	&[
298	-	"id",
299	-	"user_id",
300	-	"month",
301	-	"text",
302	-	"status",
303	-	"position",
304	-	"created_at",
305	-	"updated_at",
306	-	],
307	-	),
308	-	t(
309	-	"monthly_reflections",
310	-	&[
311	-	"id",
312	-	"user_id",
313	-	"month",
314	-	"highlight_text",
315	-	"change_text",
316	-	"completed_at",
317	-	],
318	-	),
319	-	])
320	-	// GO's current model; also the engine default.
321	-	.conflict_strategy(ConflictStrategy::HybridLogicalClock)
322	-	}
323	-
324	-	#[cfg(test)]
325	-	mod tests {
326	-	use super::super::UPSERT_ORDER;
327	-	use super::super::apply::SYNCED_COLUMNS;
328	-	use super::goingson_schema;
329	-
330	-	/// Behaviour-diff (M1): the declared manifest must emit exactly the columns
331	-	/// GO's live triggers/apply whitelist emit today, table for table, so the
332	-	/// engine-generated changelog is byte-for-byte equivalent to GO's current
333	-	/// output and no table's sync policy silently changes at cutover (M2).
334	-	///
335	-	/// This is the check that catches drift like a table added to GO's triggers
336	-	/// (`task_status_tokens`, migration 058) but missing from the manifest.
337	-	#[test]
338	-	fn manifest_matches_synced_columns() {
339	-	let schema = goingson_schema();
340	-
341	-	// Same table set, same FK order as GO's authoritative UPSERT_ORDER.
342	-	let manifest_order: Vec<&str> = schema.tables().iter().map(\|t\| t.name()).collect();
343	-	assert_eq!(
344	-	manifest_order, UPSERT_ORDER,
345	-	"manifest table set/order diverged from UPSERT_ORDER"
346	-	);
347	-
348	-	// Same emitted columns per table as apply.rs SYNCED_COLUMNS. For
349	-	// email_accounts the preserved credential columns are apply-side policy
350	-	// and correctly absent from the emitted (changelog) columns.
351	-	for table in schema.tables() {
352	-	let expected = SYNCED_COLUMNS
353	-	.iter()
354	-	.find(\|(name, _)\| *name == table.name())
355	-	.unwrap_or_else(\|\| {
356	-	panic!("{} is in the manifest but not SYNCED_COLUMNS", table.name())
357	-	})
358	-	.1;
359	-	assert_eq!(
360	-	table.emitted_columns(),
361	-	expected,
362	-	"emitted columns for {} diverged from SYNCED_COLUMNS",
363	-	table.name()
364	-	);
365	-	}
366	-
367	-	// And no SYNCED_COLUMNS table is missing from the manifest.
368	-	assert_eq!(
369	-	schema.tables().len(),
370	-	SYNCED_COLUMNS.len(),
371	-	"manifest and SYNCED_COLUMNS have different table counts"
372	-	);
373	-	}
374	-	}

D src-tauri/src/sync_service/mod.rs -246

		@@ -1,246 +0,0 @@
1	-	//! Core sync engine: push local changes, pull remote changes, apply to DB.
2	-	//!
3	-	//! # SQL Safety: `format!()` for table and column names
4	-	//!
5	-	//! Several functions in this module use `format!()` to interpolate table and column
6	-	//! names into SQL strings (e.g., `apply_upsert`, `apply_delete`, `create_initial_snapshot`).
7	-	//! This is safe because:
8	-	//!
9	-	//! 1. Table names come from hardcoded constants (`UPSERT_ORDER`, `DELETE_ORDER`) -- never
10	-	//! from user input or remote data.
11	-	//! 2. Column names come from `table_columns()`, a compile-time whitelist of `&'static str`
12	-	//! literals -- also never from user input.
13	-	//! 3. All user-supplied values (row IDs, field data) are passed through `sqlx::query().bind()`,
14	-	//! which parameterizes them safely.
15	-	//! 4. Unknown table names are rejected before any SQL is constructed: both `apply_upsert` and
16	-	//! `apply_delete` return an error if `table_columns()` returns `None`.
17	-	//!
18	-	//! The `format!()` pattern is used here instead of a procedural macro because the table/column
19	-	//! sets are dynamic per-row (determined by the sync changelog), but the values themselves are
20	-	//! drawn exclusively from the static whitelist above.
21	-
22	-	mod apply;
23	-	pub(crate) mod blob_sync;
24	-	mod hlc;
25	-	pub(crate) mod manifest;
26	-	mod pull;
27	-	mod push;
28	-	mod state;
29	-
30	-	#[cfg(test)]
31	-	#[path = "tests.rs"]
32	-	mod tests;
33	-
34	-	use std::path::Path;
35	-
36	-	use chrono::Utc;
37	-	use goingson_core::CoreError;
38	-	use serde::{Deserialize, Serialize};
39	-	use sqlx::SqlitePool;
40	-	use synckit_client::SyncKitClient;
41	-	use tracing::{debug, info, warn};
42	-
43	-	// Re-export public API so `crate::sync_service::*` continues to work.
44	-	pub use self::pull::pull_changes;
45	-	pub use self::push::push_changes;
46	-	pub use self::state::{
47	-	ensure_device_registered, get_sync_state, get_sync_states_batch, set_sync_state,
48	-	};
49	-
50	-	/// Maximum changes to push in a single batch.
51	-	pub(crate) const PUSH_BATCH_LIMIT: i64 = 500;
52	-
53	-	/// Email account columns that sync (config only -- credentials stay per-device).
54	-	pub(crate) const EMAIL_ACCOUNT_SYNC_COLS: &[&str] = &[
55	-	"id",
56	-	"user_id",
57	-	"account_name",
58	-	"email_address",
59	-	"imap_server",
60	-	"imap_port",
61	-	"smtp_server",
62	-	"smtp_port",
63	-	"username",
64	-	"use_tls",
65	-	"created_at",
66	-	"archive_folder_name",
67	-	"auth_type",
68	-	"jmap_session_url",
69	-	"jmap_account_id",
70	-	"sync_interval_minutes",
71	-	"email_signature",
72	-	];
73	-
74	-	/// Tables in FK-safe order for upserts (parents first).
75	-	pub(crate) const UPSERT_ORDER: &[&str] = &[
76	-	"projects",
77	-	"contacts",
78	-	"email_accounts",
79	-	"sync_accounts",
80	-	"milestones",
81	-	"tasks",
82	-	"time_sessions",
83	-	"attachments",
84	-	"events",
85	-	"annotations",
86	-	"subtasks",
87	-	"task_status_tokens",
88	-	"contact_emails",
89	-	"contact_phones",
90	-	"contact_social_handles",
91	-	"contact_custom_fields",
92	-	"daily_notes",
93	-	// Independent roots (FK only to local `users`); order among them is irrelevant.
94	-	"saved_views",
95	-	"weekly_reviews",
96	-	"monthly_goals",
97	-	"monthly_reflections",
98	-	];
99	-
100	-	/// Tables in reverse FK-safe order for deletes (children first).
101	-	pub(crate) const DELETE_ORDER: &[&str] = &[
102	-	"monthly_reflections",
103	-	"monthly_goals",
104	-	"weekly_reviews",
105	-	"saved_views",
106	-	"daily_notes",
107	-	"contact_custom_fields",
108	-	"contact_social_handles",
109	-	"contact_phones",
110	-	"contact_emails",
111	-	"task_status_tokens",
112	-	"subtasks",
113	-	"annotations",
114	-	"events",
115	-	"attachments",
116	-	"time_sessions",
117	-	"tasks",
118	-	"milestones",
119	-	"sync_accounts",
120	-	"email_accounts",
121	-	"contacts",
122	-	"projects",
123	-	];
124	-
125	-	/// Result of a sync operation.
126	-	#[derive(Debug, Serialize, Deserialize)]
127	-	#[serde(rename_all = "camelCase")]
128	-	pub struct SyncResult {
129	-	pub pushed: i64,
130	-	pub pulled: i64,
131	-	/// Distinct DB tables touched by the pull, for selective UI cache invalidation.
132	-	#[serde(skip)]
133	-	pub pulled_tables: Vec<String>,
134	-	}
135	-
136	-	// -- High-level sync --
137	-
138	-	pub async fn perform_sync(
139	-	pool: &SqlitePool,
140	-	client: &SyncKitClient,
141	-	) -> Result<SyncResult, CoreError> {
142	-	perform_sync_with_blobs(pool, client, None).await
143	-	}
144	-
145	-	pub async fn perform_sync_with_blobs(
146	-	pool: &SqlitePool,
147	-	client: &SyncKitClient,
148	-	data_dir: Option<&Path>,
149	-	) -> Result<SyncResult, CoreError> {
150	-	// Clear applying_remote flag in case a previous sync crashed mid-apply.
151	-	// If the flag is stuck at "1", all local changes silently skip the changelog.
152	-	set_sync_state(pool, "applying_remote", "0").await?;
153	-
154	-	let device_id = ensure_device_registered(pool, client).await?;
155	-
156	-	// Push first, then pull
157	-	let pushed = push_changes(pool, client, device_id).await?;
158	-	let pull_outcome = pull_changes(pool, client, device_id).await?;
159	-	let pulled = pull_outcome.applied;
160	-	let mut pulled_tables: Vec<String> = pull_outcome.changed_tables.into_iter().collect();
161	-	pulled_tables.sort();
162	-
163	-	// Sync blobs after metadata (upload local, download missing)
164	-	if let Some(dir) = data_dir {
165	-	if let Err(e) = blob_sync::upload_pending_blobs(pool, dir, client).await {
166	-	warn!("Blob upload failed (non-fatal): {}", e);
167	-	}
168	-	if let Err(e) = blob_sync::download_missing_blobs(pool, dir, client).await {
169	-	warn!("Blob download failed (non-fatal): {}", e);
170	-	}
171	-	}
172	-
173	-	// Update last sync timestamp
174	-	let now = Utc::now().to_rfc3339();
175	-	set_sync_state(pool, "last_sync_at", &now).await?;
176	-
177	-	Ok(SyncResult {
178	-	pushed,
179	-	pulled,
180	-	pulled_tables,
181	-	})
182	-	}
183	-
184	-	// -- Initial snapshot --
185	-
186	-	/// One-time: snapshot all existing rows into the changelog for the first push.
187	-	pub async fn create_initial_snapshot(pool: &SqlitePool) -> Result<i64, CoreError> {
188	-	let tables_and_cols: Vec<(&str, &[&str])> = UPSERT_ORDER
189	-	.iter()
190	-	.filter_map(\|t\| apply::table_columns(t).map(\|c\| (*t, c)))
191	-	.collect();
192	-
193	-	let mut total: i64 = 0;
194	-
195	-	for (table, columns) in &tables_and_cols {
196	-	let col_list = columns
197	-	.iter()
198	-	.map(\|c\| format!("'{}', {}", c, c))
199	-	.collect::<Vec<_>>()
200	-	.join(", ");
201	-
202	-	let sql = format!(
203	-	"INSERT INTO sync_changelog (table_name, op, row_id, data) \
204	-	SELECT '{table}', 'INSERT', id, json_object({col_list}) FROM {table} \
205	-	WHERE NOT EXISTS (SELECT 1 FROM sync_changelog sc WHERE sc.table_name = '{table}' AND sc.row_id = {table}.id)",
206	-	);
207	-
208	-	let result = sqlx::query(sqlx::AssertSqlSafe(sql.as_str()))
209	-	.execute(pool)
210	-	.await
211	-	.map_err(CoreError::database)?;
212	-
213	-	total += result.rows_affected() as i64;
214	-	}
215	-
216	-	set_sync_state(pool, "initial_snapshot_done", "1").await?;
217	-	info!("Initial sync snapshot created: {} rows", total);
218	-	Ok(total)
219	-	}
220	-
221	-	// -- Changelog cleanup --
222	-
223	-	/// Prune pushed entries older than 7 days.
224	-	pub async fn cleanup_changelog(pool: &SqlitePool) -> Result<i64, CoreError> {
225	-	let result = sqlx::query(
226	-	"DELETE FROM sync_changelog WHERE pushed = 1 AND timestamp < datetime('now', '-7 days')",
227	-	)
228	-	.execute(pool)
229	-	.await
230	-	.map_err(CoreError::database)?;
231	-
232	-	let deleted = result.rows_affected() as i64;
233	-	if deleted > 0 {
234	-	debug!("Cleaned up {} old changelog entries", deleted);
235	-	}
236	-	Ok(deleted)
237	-	}
238	-
239	-	/// Count unpushed changes.
240	-	pub async fn count_pending_changes(pool: &SqlitePool) -> Result<i64, CoreError> {
241	-	let row: (i64,) = sqlx::query_as("SELECT COUNT(*) FROM sync_changelog WHERE pushed = 0")
242	-	.fetch_one(pool)
243	-	.await
244	-	.map_err(CoreError::database)?;
245	-	Ok(row.0)
246	-	}

D src-tauri/src/sync_service/pull.rs -366

		@@ -1,366 +0,0 @@
1	-	//! Pull remote changes and apply them to the local database.
2	-
3	-	use chrono::Utc;
4	-	use goingson_core::CoreError;
5	-	use sqlx::{SqliteConnection, SqlitePool};
6	-	use synckit_client::{
7	-	ChangeEntry, ChangeOp, DeviceId, Hlc, Resolution, SyncKitClient, detect_conflicts, resolve_lww,
8	-	};
9	-	use tracing::{debug, warn};
10	-	use uuid::Uuid;
11	-
12	-	use super::apply::{apply_delete, apply_upsert};
13	-	use super::hlc::{assign_pending_hlcs, load_committed_hlcs, observe_remote, record_committed_hlc};
14	-	use super::state::{get_sync_state, set_sync_state};
15	-	use super::{DELETE_ORDER, UPSERT_ORDER};
16	-
17	-	#[tracing::instrument(skip_all)]
18	-	pub async fn pull_changes(
19	-	pool: &SqlitePool,
20	-	client: &SyncKitClient,
21	-	device_id: Uuid,
22	-	) -> Result<PullOutcome, CoreError> {
23	-	let cursor_str = get_sync_state(pool, "pull_cursor").await?;
24	-	let mut cursor: i64 = cursor_str.parse().unwrap_or(0);
25	-	let mut total_applied: i64 = 0;
26	-	let mut changed_tables: std::collections::HashSet<String> = std::collections::HashSet::new();
27	-
28	-	// Read local pending changes once (unpushed changelog entries). This also
29	-	// stamps them with their HLC so conflict resolution can compare clocks.
30	-	let local_pending = read_local_pending(pool, device_id).await?;
31	-
32	-	loop {
33	-	let (changes, new_cursor, has_more) = client
34	-	.pull_rich(DeviceId::new(device_id), cursor)
35	-	.await
36	-	.map_err(\|e\| CoreError::sync(format!("pull failed: {}", e)))?;
37	-
38	-	if changes.is_empty() {
39	-	set_sync_state(pool, "pull_cursor", &new_cursor.to_string()).await?;
40	-	break;
41	-	}
42	-
43	-	// Advance our clock past everything in this batch, so future local changes
44	-	// causally follow the remote ones. Observing the batch maximum suffices.
45	-	if let Some(max_remote) = changes.iter().map(\|p\| p.entry.hlc).max() {
46	-	observe_remote(pool, max_remote, device_id).await?;
47	-	}
48	-
49	-	// Detect conflicts between remote changes and local pending
50	-	let (clean, conflicts) =
51	-	detect_conflicts(changes, &local_pending, DeviceId::new(device_id));
52	-
53	-	// Apply non-conflicting changes — but HLC-gate them first. "Clean" only
54	-	// means no local pending edit contests the row; an older remote edit for
55	-	// an already-committed row also lands here, and applying it blindly would
56	-	// clobber the newer local value. Pre-fetch the rows' committed clocks, then
57	-	// let SyncKit's CleanChanges gate drop anything older-or-equal.
58	-	let clean_count = clean.len() as i64;
59	-	if !clean.is_empty() {
60	-	let keys: Vec<(String, String)> = clean
61	-	.row_keys()
62	-	.map(\|(t, r)\| (t.to_string(), r.to_string()))
63	-	.collect();
64	-	let committed = load_committed_hlcs(pool, &keys).await?;
65	-	let clean_entries: Vec<ChangeEntry> =
66	-	clean.gated(\|t, r\| committed.get(&(t.to_string(), r.to_string())).copied());
67	-	if !clean_entries.is_empty() {
68	-	changed_tables.extend(clean_entries.iter().map(\|e\| e.table.clone()));
69	-	apply_remote_changes(pool, clean_entries).await?;
70	-	}
71	-	}
72	-
73	-	// Resolve conflicts with LWW
74	-	let mut resolved_count: i64 = 0;
75	-	if !conflicts.is_empty() {
76	-	let conflict_count = conflicts.len();
77	-	let mut resolved_entries: Vec<ChangeEntry> = Vec::new();
78	-
79	-	for pair in &conflicts {
80	-	match resolve_lww(&pair.local, &pair.remote) {
81	-	Resolution::KeepRemote => {
82	-	resolved_entries.push(pair.remote.entry.clone());
83	-	}
84	-	Resolution::KeepLocal => {
85	-	// Skip — local version will push on next cycle
86	-	}
87	-	Resolution::Merged(data) => {
88	-	// Apply merged data using the remote entry as template
89	-	let mut merged = pair.remote.entry.clone();
90	-	merged.data = Some(data);
91	-	resolved_entries.push(merged);
92	-	}
93	-	Resolution::Skip => {
94	-	// Skip both
95	-	}
96	-	}
97	-	}
98	-
99	-	resolved_count = resolved_entries.len() as i64;
100	-	if !resolved_entries.is_empty() {
101	-	changed_tables.extend(resolved_entries.iter().map(\|e\| e.table.clone()));
102	-	apply_remote_changes(pool, resolved_entries).await?;
103	-	}
104	-
105	-	debug!(
106	-	"Resolved {} conflicts ({} applied as remote wins)",
107	-	conflict_count,
108	-	conflict_count
109	-	- conflicts
110	-	.iter()
111	-	.filter(\|p\| {
112	-	matches!(
113	-	resolve_lww(&p.local, &p.remote),
114	-	Resolution::KeepLocal \| Resolution::Skip
115	-	)
116	-	})
117	-	.count(),
118	-	);
119	-	}
120	-
121	-	total_applied += clean_count + resolved_count;
122	-
123	-	// Save cursor after each batch (crash-safe)
124	-	set_sync_state(pool, "pull_cursor", &new_cursor.to_string()).await?;
125	-	cursor = new_cursor;
126	-
127	-	if !has_more {
128	-	break;
129	-	}
130	-	}
131	-
132	-	if total_applied > 0 {
133	-	debug!("Pulled and applied {} remote changes", total_applied);
134	-	}
135	-	Ok(PullOutcome {
136	-	applied: total_applied,
137	-	changed_tables,
138	-	})
139	-	}
140	-
141	-	/// Result of a pull: how many changes landed and which tables they touched.
142	-	///
143	-	/// The table set drives selective cache invalidation in the UI so a pull that
144	-	/// only touched, say, `tasks` doesn't force the compose screen to re-hydrate
145	-	/// every contact.
146	-	pub struct PullOutcome {
147	-	pub applied: i64,
148	-	pub changed_tables: std::collections::HashSet<String>,
149	-	}
150	-
151	-	/// Read unpushed changelog entries as ChangeEntry values for conflict detection.
152	-	///
153	-	/// Stamps any not-yet-stamped rows with their HLC first, so the clocks used here
154	-	/// match the ones the push path will send.
155	-	/// A pending sync_changelog row: (table_name, op, row_id, timestamp, data, hlc_wall, hlc_counter).
156	-	type PendingRow = (
157	-	String,
158	-	String,
159	-	String,
160	-	String,
161	-	Option<String>,
162	-	Option<i64>,
163	-	Option<i64>,
164	-	);
165	-
166	-	async fn read_local_pending(
167	-	pool: &SqlitePool,
168	-	device_id: Uuid,
169	-	) -> Result<Vec<ChangeEntry>, CoreError> {
170	-	assign_pending_hlcs(pool, device_id).await?;
171	-
172	-	let rows: Vec<PendingRow> = sqlx::query_as(
173	-	"SELECT table_name, op, row_id, timestamp, data, hlc_wall, hlc_counter \
174	-	FROM sync_changelog WHERE pushed = 0 ORDER BY id ASC",
175	-	)
176	-	.fetch_all(pool)
177	-	.await
178	-	.map_err(CoreError::database)?;
179	-
180	-	let entries: Vec<ChangeEntry> = rows
181	-	.into_iter()
182	-	.filter_map(
183	-	\|(table, op, row_id, timestamp, data, hlc_wall, hlc_counter)\| {
184	-	let ts = chrono::DateTime::parse_from_rfc3339(&timestamp)
185	-	.map(\|dt\| dt.with_timezone(&Utc))
186	-	.unwrap_or(chrono::DateTime::UNIX_EPOCH);
187	-
188	-	let change_op = ChangeOp::from_str_opt(&op)?;
189	-	let json_data = data.and_then(\|d\| serde_json::from_str(&d).ok());
190	-
191	-	let hlc = match (hlc_wall, hlc_counter) {
192	-	(Some(wall), Some(counter)) => Hlc {
193	-	wall_ms: wall,
194	-	counter: counter as u32,
195	-	node: DeviceId::new(device_id),
196	-	},
197	-	_ => Hlc::from_legacy(ts.timestamp_millis(), DeviceId::new(device_id)),
198	-	};
199	-
200	-	Some(ChangeEntry {
201	-	table,
202	-	op: change_op,
203	-	row_id,
204	-	timestamp: ts,
205	-	hlc,
206	-	data: json_data,
207	-	extra: Default::default(),
208	-	})
209	-	},
210	-	)
211	-	.collect();
212	-
213	-	Ok(entries)
214	-	}
215	-
216	-	/// Apply remote changes to local DB with triggers suppressed and FK enforcement off.
217	-	///
218	-	/// FK enforcement is disabled so that tasks with `source_email_id` pointing to
219	-	/// emails not yet fetched locally can be inserted without error. Uses a dedicated
220	-	/// connection (same pattern as `crates/db-sqlite/src/migrations.rs`).
221	-	///
222	-	/// The `applying_remote` flag is set inside a transaction on the dedicated connection.
223	-	/// In WAL mode, uncommitted changes are only visible to the writing connection, so
224	-	/// other connections (handling user edits) never see the flag and their triggers fire
225	-	/// normally. The flag is reset to '0' before commit so it's never globally visible as '1'.
226	-	///
227	-	/// Uses `detach()` to prevent returning the connection to the pool with FK OFF
228	-	/// if pragma restoration fails. The pool will create a fresh connection (with
229	-	/// FK ON via connect options) to replace it.
230	-	pub(crate) async fn apply_remote_changes(
231	-	pool: &SqlitePool,
232	-	changes: Vec<ChangeEntry>,
233	-	) -> Result<(), CoreError> {
234	-	// Acquire a dedicated connection for FK pragma and transaction
235	-	let mut conn = pool.acquire().await.map_err(CoreError::database)?;
236	-
237	-	// FK pragma must be set outside a transaction (SQLite requirement)
238	-	sqlx::query("PRAGMA foreign_keys = OFF")
239	-	.execute(&mut *conn)
240	-	.await
241	-	.map_err(CoreError::database)?;
242	-
243	-	// Use a transaction so applying_remote is only visible to this connection (WAL isolation).
244	-	// Other connections see the committed value ('0') and their triggers fire normally.
245	-	sqlx::query("BEGIN IMMEDIATE")
246	-	.execute(&mut *conn)
247	-	.await
248	-	.map_err(CoreError::database)?;
249	-
250	-	sqlx::query("UPDATE sync_state SET value = '1' WHERE key = 'applying_remote'")
251	-	.execute(&mut *conn)
252	-	.await
253	-	.map_err(CoreError::database)?;
254	-
255	-	let result = apply_changes_inner(&mut conn, changes).await;
256	-
257	-	// Reset flag before commit so it's never globally visible as '1'
258	-	let _ = sqlx::query("UPDATE sync_state SET value = '0' WHERE key = 'applying_remote'")
259	-	.execute(&mut *conn)
260	-	.await;
261	-
262	-	match result {
263	-	Ok(skipped) => {
264	-	if let Err(e) = sqlx::query("COMMIT").execute(&mut *conn).await {
265	-	let _ = sqlx::query("ROLLBACK").execute(&mut *conn).await;
266	-	let _ = sqlx::query("PRAGMA foreign_keys = ON")
267	-	.execute(&mut *conn)
268	-	.await;
269	-	return Err(CoreError::database(e));
270	-	}
271	-	if skipped > 0 {
272	-	warn!(
273	-	skipped,
274	-	"Skipped un-appliable remote change entries; the rest of the batch \
275	-	was applied and the pull cursor advanced (sync not wedged)."
276	-	);
277	-	}
278	-	}
279	-	Err(ref _e) => {
280	-	let _ = sqlx::query("ROLLBACK").execute(&mut *conn).await;
281	-	}
282	-	}
283	-
284	-	// Always restore FK enforcement. If this fails, detach the connection
285	-	// so it doesn't return to the pool with FK OFF.
286	-	if let Err(e) = sqlx::query("PRAGMA foreign_keys = ON")
287	-	.execute(&mut *conn)
288	-	.await
289	-	{
290	-	conn.detach();
291	-	return Err(CoreError::database(e));
292	-	}
293	-
294	-	result.map(\|_\| ())
295	-	}
296	-
297	-	/// Apply a batch of remote changes within an open transaction.
298	-	///
299	-	/// Per-entry failures are skipped and logged rather than aborting the whole batch:
300	-	/// before, one un-appliable remote row (e.g. a payload missing a NOT NULL column)
301	-	/// rolled back the transaction, `apply_remote_changes` returned Err, and the pull
302	-	/// cursor never advanced -- so every subsequent pull re-fetched the same poisoned
303	-	/// batch and failed identically, wedging sync permanently (ultra-fuzz Run #27
304	-	/// Data S-1). A failed statement does not poison the SQLite transaction, so the
305	-	/// good rows still commit and the cursor advances. Returns the count of skipped
306	-	/// entries (the caller logs a summary); only a catastrophic error returns `Err`.
307	-	pub(crate) async fn apply_changes_inner(
308	-	conn: &mut SqliteConnection,
309	-	changes: Vec<ChangeEntry>,
310	-	) -> Result<usize, CoreError> {
311	-	// Separate upserts from deletes
312	-	let mut upserts: Vec<&ChangeEntry> = Vec::new();
313	-	let mut deletes: Vec<&ChangeEntry> = Vec::new();
314	-
315	-	for change in &changes {
316	-	match change.op {
317	-	ChangeOp::Insert \| ChangeOp::Update => upserts.push(change),
318	-	ChangeOp::Delete => deletes.push(change),
319	-	}
320	-	}
321	-
322	-	let mut skipped = 0usize;
323	-
324	-	// Apply upserts in parent-first FK order
325	-	for table in UPSERT_ORDER {
326	-	for change in &upserts {
327	-	if change.table != *table {
328	-	continue;
329	-	}
330	-	let Some(ref data) = change.data else {
331	-	continue;
332	-	};
333	-	match apply_upsert(&mut *conn, table, &change.row_id, data).await {
334	-	Ok(()) => {
335	-	record_committed_hlc(&mut *conn, table, &change.row_id, change.hlc).await?;
336	-	}
337	-	Err(e) => {
338	-	warn!(table = *table, row_id = %change.row_id, error = %e,
339	-	"Skipping un-appliable remote upsert");
340	-	skipped += 1;
341	-	}
342	-	}
343	-	}
344	-	}
345	-
346	-	// Apply deletes in child-first FK order
347	-	for table in DELETE_ORDER {
348	-	for change in &deletes {
349	-	if change.table != *table {
350	-	continue;
351	-	}
352	-	match apply_delete(&mut *conn, table, &change.row_id).await {
353	-	Ok(()) => {
354	-	record_committed_hlc(&mut *conn, table, &change.row_id, change.hlc).await?;
355	-	}
356	-	Err(e) => {
357	-	warn!(table = *table, row_id = %change.row_id, error = %e,
358	-	"Skipping un-appliable remote delete");
359	-	skipped += 1;
360	-	}
361	-	}
362	-	}
363	-	}
364	-
365	-	Ok(skipped)
366	-	}

D src-tauri/src/sync_service/push.rs -128

		@@ -1,128 +0,0 @@
1	-	//! Push local changes to the remote sync server.
2	-
3	-	use chrono::Utc;
4	-	use goingson_core::CoreError;
5	-	use sqlx::SqlitePool;
6	-	use synckit_client::{ChangeEntry, ChangeOp, DeviceId, Hlc, SyncKitClient};
7	-	use tracing::{debug, warn};
8	-	use uuid::Uuid;
9	-
10	-	use super::PUSH_BATCH_LIMIT;
11	-	use super::hlc::assign_pending_hlcs;
12	-
13	-	/// A pending sync_changelog row with its id:
14	-	/// (id, table_name, op, row_id, timestamp, data, hlc_wall, hlc_counter).
15	-	type PendingRow = (
16	-	i64,
17	-	String,
18	-	String,
19	-	String,
20	-	String,
21	-	Option<String>,
22	-	Option<i64>,
23	-	Option<i64>,
24	-	);
25	-
26	-	#[tracing::instrument(skip_all)]
27	-	pub async fn push_changes(
28	-	pool: &SqlitePool,
29	-	client: &SyncKitClient,
30	-	device_id: Uuid,
31	-	) -> Result<i64, CoreError> {
32	-	// Stamp any unpushed changes with their HLC before reading them for the wire.
33	-	assign_pending_hlcs(pool, device_id).await?;
34	-
35	-	let mut total_pushed: i64 = 0;
36	-
37	-	loop {
38	-	let rows: Vec<PendingRow> = sqlx::query_as(
39	-	"SELECT id, table_name, op, row_id, timestamp, data, hlc_wall, hlc_counter \
40	-	FROM sync_changelog WHERE pushed = 0 ORDER BY id ASC LIMIT ?",
41	-	)
42	-	.bind(PUSH_BATCH_LIMIT)
43	-	.fetch_all(pool)
44	-	.await
45	-	.map_err(CoreError::database)?;
46	-
47	-	if rows.is_empty() {
48	-	break;
49	-	}
50	-
51	-	let row_count = rows.len() as i64;
52	-	let mut pushed_ids: Vec<i64> = Vec::new();
53	-	let changes: Vec<ChangeEntry> = rows
54	-	.into_iter()
55	-	.filter_map(
56	-	\|(id, table, op, row_id, timestamp, data, hlc_wall, hlc_counter)\| {
57	-	// Always mark as pushed (even unknown ops) to avoid infinite re-fetch
58	-	pushed_ids.push(id);
59	-
60	-	let ts = chrono::DateTime::parse_from_rfc3339(&timestamp)
61	-	.map(\|dt\| dt.with_timezone(&Utc))
62	-	.unwrap_or(chrono::DateTime::UNIX_EPOCH);
63	-
64	-	let json_data = data.and_then(\|d\| serde_json::from_str(&d).ok());
65	-
66	-	let change_op = match ChangeOp::from_str_opt(&op) {
67	-	Some(o) => o,
68	-	None => {
69	-	warn!("Skipping changelog entry with unknown op: {}", op);
70	-	return None;
71	-	}
72	-	};
73	-
74	-	// Stamped by assign_pending_hlcs above; fall back to a
75	-	// timestamp-derived HLC if a row was somehow left unstamped.
76	-	let hlc = match (hlc_wall, hlc_counter) {
77	-	(Some(wall), Some(counter)) => Hlc {
78	-	wall_ms: wall,
79	-	counter: counter as u32,
80	-	node: DeviceId::new(device_id),
81	-	},
82	-	_ => Hlc::from_legacy(ts.timestamp_millis(), DeviceId::new(device_id)),
83	-	};
84	-
85	-	Some(ChangeEntry {
86	-	table,
87	-	op: change_op,
88	-	row_id,
89	-	timestamp: ts,
90	-	hlc,
91	-	data: json_data,
92	-	extra: Default::default(),
93	-	})
94	-	},
95	-	)
96	-	.collect();
97	-
98	-	let count = changes.len() as i64;
99	-	let is_last_batch = row_count < PUSH_BATCH_LIMIT;
100	-
101	-	client
102	-	.push(DeviceId::new(device_id), changes)
103	-	.await
104	-	.map_err(\|e\| CoreError::sync(format!("push failed: {}", e)))?;
105	-
106	-	// Mark pushed entries in a single transaction
107	-	let mut tx = pool.begin().await.map_err(CoreError::database)?;
108	-	for id in &pushed_ids {
109	-	sqlx::query("UPDATE sync_changelog SET pushed = 1 WHERE id = ?")
110	-	.bind(id)
111	-	.execute(&mut *tx)
112	-	.await
113	-	.map_err(CoreError::database)?;
114	-	}
115	-	tx.commit().await.map_err(CoreError::database)?;
116	-
117	-	total_pushed += count;
118	-
119	-	if is_last_batch {
120	-	break;
121	-	}
122	-	}
123	-
124	-	if total_pushed > 0 {
125	-	debug!("Pushed {} changes", total_pushed);
126	-	}
127	-	Ok(total_pushed)
128	-	}

D src-tauri/src/sync_service/state.rs -79

		@@ -1,79 +0,0 @@
1	-	//! Sync state helpers: key-value store and device registration.
2	-
3	-	use goingson_core::CoreError;
4	-	use sqlx::SqlitePool;
5	-	use std::collections::HashMap;
6	-	use synckit_client::SyncKitClient;
7	-	use tracing::info;
8	-	use uuid::Uuid;
9	-
10	-	pub async fn get_sync_state(pool: &SqlitePool, key: &str) -> Result<String, CoreError> {
11	-	let row: Option<(String,)> = sqlx::query_as("SELECT value FROM sync_state WHERE key = ?")
12	-	.bind(key)
13	-	.fetch_optional(pool)
14	-	.await
15	-	.map_err(CoreError::database)?;
16	-
17	-	Ok(row.map(\|r\| r.0).unwrap_or_default())
18	-	}
19	-
20	-	pub async fn get_sync_states_batch(
21	-	pool: &SqlitePool,
22	-	keys: &[&str],
23	-	) -> Result<HashMap<String, String>, CoreError> {
24	-	if keys.is_empty() {
25	-	return Ok(HashMap::new());
26	-	}
27	-
28	-	let placeholders = keys.iter().map(\|_\| "?").collect::<Vec<_>>().join(",");
29	-	let query = format!(
30	-	"SELECT key, value FROM sync_state WHERE key IN ({})",
31	-	placeholders
32	-	);
33	-
34	-	let mut q = sqlx::query_as::<_, (String, String)>(sqlx::AssertSqlSafe(query.as_str()));
35	-	for key in keys {
36	-	q = q.bind(*key);
37	-	}
38	-
39	-	let rows = q.fetch_all(pool).await.map_err(CoreError::database)?;
40	-	Ok(rows.into_iter().collect())
41	-	}
42	-
43	-	pub async fn set_sync_state(pool: &SqlitePool, key: &str, value: &str) -> Result<(), CoreError> {
44	-	sqlx::query("INSERT OR REPLACE INTO sync_state (key, value) VALUES (?, ?)")
45	-	.bind(key)
46	-	.bind(value)
47	-	.execute(pool)
48	-	.await
49	-	.map_err(CoreError::database)?;
50	-	Ok(())
51	-	}
52	-
53	-	pub async fn ensure_device_registered(
54	-	pool: &SqlitePool,
55	-	client: &SyncKitClient,
56	-	) -> Result<Uuid, CoreError> {
57	-	let stored = get_sync_state(pool, "device_id").await?;
58	-	if !stored.is_empty() {
59	-	return stored
60	-	.parse::<Uuid>()
61	-	.map_err(\|e\| CoreError::parse(format!("invalid stored device_id: {}", e)));
62	-	}
63	-
64	-	let hostname = std::env::var("HOSTNAME")
65	-	.or_else(\|_\| std::env::var("COMPUTERNAME"))
66	-	.unwrap_or_else(\|_\| "GoingsOn Desktop".to_string());
67	-
68	-	let platform = std::env::consts::OS.to_string();
69	-
70	-	let device = client
71	-	.register_device(&hostname, &platform)
72	-	.await
73	-	.map_err(\|e\| CoreError::sync(format!("device registration failed: {}", e)))?;
74	-
75	-	set_sync_state(pool, "device_id", &device.id.to_string()).await?;
76	-	info!("Registered device: {} ({})", device.device_name, device.id);
77	-
78	-	Ok(device.id.as_uuid())
79	-	}

D src-tauri/src/sync_service/tests.rs -500

		@@ -1,2536 +0,0 @@
1	-	use crate::sync_service::*;
2	-	use crate::sync_service::{DELETE_ORDER, UPSERT_ORDER, apply, pull};
3	-	use goingson_db_sqlite::{init_pool, run_migrations};
4	-	use serde_json::json;
5	-	use sqlx::SqlitePool;
6	-
7	-	/// Format the current UTC time as a SQL-compatible timestamp string.
8	-	fn now_sql() -> String {
9	-	chrono::Utc::now().format("%Y-%m-%d %H:%M:%S").to_string()
10	-	}
11	-
12	-	/// Creates an in-memory test database with all migrations applied.
13	-	async fn setup_test_db() -> SqlitePool {
14	-	let pool = init_pool(Some(":memory:"))
15	-	.await
16	-	.expect("Failed to create in-memory pool");
17	-	run_migrations(&pool)
18	-	.await
19	-	.expect("Failed to run migrations");
20	-	pool
21	-	}
22	-
23	-	/// Creates a test user and returns their UUID string.
24	-	async fn create_test_user(pool: &SqlitePool) -> String {
25	-	let user_id = uuid::Uuid::new_v4().to_string();
26	-	let now = now_sql();
27	-	sqlx::query(
28	-	"INSERT INTO users (id, email, password_hash, display_name, created_at) VALUES (?, ?, ?, ?, ?)"
29	-	)
30	-	.bind(&user_id)
31	-	.bind(format!("test-{}@example.com", user_id))
32	-	.bind("test-password-hash")
33	-	.bind("Test User")
34	-	.bind(&now)
35	-	.execute(pool)
36	-	.await
37	-	.expect("Failed to create test user");
38	-	user_id
39	-	}
40	-
41	-	/// Helper: build a ChangeEntry for tests.
42	-	fn change(
43	-	table: &str,
44	-	op: synckit_client::ChangeOp,
45	-	row_id: &str,
46	-	data: Option<serde_json::Value>,
47	-	) -> synckit_client::ChangeEntry {
48	-	synckit_client::ChangeEntry {
49	-	table: table.to_string(),
50	-	op,
51	-	row_id: row_id.to_string(),
52	-	timestamp: chrono::Utc::now(),
53	-	hlc: synckit_client::Hlc::zero(synckit_client::DeviceId::nil()),
54	-	data,
55	-	extra: Default::default(),
56	-	}
57	-	}
58	-
59	-	// -- FK ordering tests --
60	-
61	-	#[test]
62	-	fn upsert_order_has_parents_before_children() {
63	-	let pos = \|table: &str\| UPSERT_ORDER.iter().position(\|t\| *t == table);
64	-
65	-	// projects before tasks, events, milestones
66	-	assert!(pos("projects").unwrap() < pos("tasks").unwrap());
67	-	assert!(pos("projects").unwrap() < pos("events").unwrap());
68	-	assert!(pos("projects").unwrap() < pos("milestones").unwrap());
69	-
70	-	// contacts before contact_emails, contact_phones, etc.
71	-	assert!(pos("contacts").unwrap() < pos("contact_emails").unwrap());
72	-	assert!(pos("contacts").unwrap() < pos("contact_phones").unwrap());
73	-	assert!(pos("contacts").unwrap() < pos("contact_social_handles").unwrap());
74	-	assert!(pos("contacts").unwrap() < pos("contact_custom_fields").unwrap());
75	-
76	-	// tasks before annotations, subtasks
77	-	assert!(pos("tasks").unwrap() < pos("annotations").unwrap());
78	-	assert!(pos("tasks").unwrap() < pos("subtasks").unwrap());
79	-
80	-	// milestones before tasks (tasks.milestone_id FK)
81	-	assert!(pos("milestones").unwrap() < pos("tasks").unwrap());
82	-	}
83	-
84	-	#[test]
85	-	fn delete_order_has_children_before_parents() {
86	-	let pos = \|table: &str\| DELETE_ORDER.iter().position(\|t\| *t == table);
87	-
88	-	// children before parents (reverse of upsert)
89	-	assert!(pos("tasks").unwrap() < pos("projects").unwrap());
90	-	assert!(pos("events").unwrap() < pos("projects").unwrap());
91	-	assert!(pos("milestones").unwrap() < pos("projects").unwrap());
92	-
93	-	assert!(pos("contact_emails").unwrap() < pos("contacts").unwrap());
94	-	assert!(pos("contact_phones").unwrap() < pos("contacts").unwrap());
95	-	assert!(pos("contact_social_handles").unwrap() < pos("contacts").unwrap());
96	-	assert!(pos("contact_custom_fields").unwrap() < pos("contacts").unwrap());
97	-
98	-	assert!(pos("annotations").unwrap() < pos("tasks").unwrap());
99	-	assert!(pos("subtasks").unwrap() < pos("tasks").unwrap());
100	-
101	-	assert!(pos("tasks").unwrap() < pos("milestones").unwrap());
102	-	}
103	-
104	-	#[test]
105	-	fn upsert_and_delete_orders_are_exact_reverses() {
106	-	let reversed: Vec<&str> = UPSERT_ORDER.iter().rev().copied().collect();
107	-	assert_eq!(reversed, DELETE_ORDER);
108	-	}
109	-
110	-	#[test]
111	-	fn all_upsert_tables_have_column_whitelists() {
112	-	for table in UPSERT_ORDER {
113	-	assert!(
114	-	apply::table_columns(table).is_some(),
115	-	"table_columns missing for syncable table: {}",
116	-	table
117	-	);
118	-	}
119	-	}
120	-
121	-	#[test]
122	-	fn unknown_table_returns_none() {
123	-	assert!(apply::table_columns("nonexistent").is_none());
124	-	assert!(apply::table_columns("users").is_none());
125	-	assert!(apply::table_columns("emails").is_none());
126	-	}
127	-
128	-	#[test]
129	-	fn every_table_whitelist_starts_with_id() {
130	-	for table in UPSERT_ORDER {
131	-	let cols = apply::table_columns(table).unwrap();
132	-	assert_eq!(
133	-	cols[0], "id",
134	-	"table {} column whitelist should start with 'id'",
135	-	table
136	-	);
137	-	}
138	-	}
139	-
140	-	// -- sync_state helpers --
141	-
142	-	#[tokio::test]
143	-	async fn get_sync_state_returns_empty_for_missing_key() {
144	-	let pool = setup_test_db().await;
145	-	let val = get_sync_state(&pool, "nonexistent_key").await.unwrap();
146	-	assert_eq!(val, "");
147	-	}
148	-
149	-	#[tokio::test]
150	-	async fn set_and_get_sync_state() {
151	-	let pool = setup_test_db().await;
152	-	set_sync_state(&pool, "test_key", "test_value")
153	-	.await
154	-	.unwrap();
155	-	let val = get_sync_state(&pool, "test_key").await.unwrap();
156	-	assert_eq!(val, "test_value");
157	-	}
158	-
159	-	#[tokio::test]
160	-	async fn set_sync_state_overwrites_existing() {
161	-	let pool = setup_test_db().await;
162	-	set_sync_state(&pool, "key", "first").await.unwrap();
163	-	set_sync_state(&pool, "key", "second").await.unwrap();
164	-	let val = get_sync_state(&pool, "key").await.unwrap();
165	-	assert_eq!(val, "second");
166	-	}
167	-
168	-	#[tokio::test]
169	-	async fn migration_seeds_default_sync_state() {
170	-	let pool = setup_test_db().await;
171	-
172	-	assert_eq!(get_sync_state(&pool, "device_id").await.unwrap(), "");
173	-	assert_eq!(get_sync_state(&pool, "pull_cursor").await.unwrap(), "0");
174	-	assert_eq!(get_sync_state(&pool, "applying_remote").await.unwrap(), "0");
175	-	assert_eq!(
176	-	get_sync_state(&pool, "initial_snapshot_done")
177	-	.await
178	-	.unwrap(),
179	-	"0"
180	-	);
181	-	}
182	-
183	-	// -- committed HLC store (CleanChanges gate backing) --
184	-
185	-	#[tokio::test]
186	-	async fn committed_hlc_records_keeps_max_and_loads() {
187	-	use crate::sync_service::hlc::{load_committed_hlcs, record_committed_hlc};
188	-	use synckit_client::Hlc;
189	-
190	-	let pool = setup_test_db().await;
191	-	let node = uuid::Uuid::new_v4();
192	-	let lo = Hlc {
193	-	wall_ms: 100,
194	-	counter: 0,
195	-	node: synckit_client::DeviceId::new(node),
196	-	};
197	-	let hi = Hlc {
198	-	wall_ms: 200,
199	-	counter: 0,
200	-	node: synckit_client::DeviceId::new(node),
201	-	};
202	-	let key = ("tasks".to_string(), "r1".to_string());
203	-
204	-	record_committed_hlc(&pool, "tasks", "r1", lo)
205	-	.await
206	-	.unwrap();
207	-	let m = load_committed_hlcs(&pool, std::slice::from_ref(&key))
208	-	.await
209	-	.unwrap();
210	-	assert_eq!(m.get(&key).copied(), Some(lo));
211	-
212	-	// A higher HLC overwrites; a subsequent lower one does NOT (max-keeping).
213	-	record_committed_hlc(&pool, "tasks", "r1", hi)
214	-	.await
215	-	.unwrap();
216	-	record_committed_hlc(&pool, "tasks", "r1", lo)
217	-	.await
218	-	.unwrap();
219	-	let m = load_committed_hlcs(&pool, std::slice::from_ref(&key))
220	-	.await
221	-	.unwrap();
222	-	assert_eq!(
223	-	m.get(&key).copied(),
224	-	Some(hi),
225	-	"older HLC must not overwrite newer committed clock"
226	-	);
227	-
228	-	// A row never recorded is absent (the gate then keeps that change).
229	-	let missing = ("tasks".to_string(), "nope".to_string());
230	-	let m = load_committed_hlcs(&pool, std::slice::from_ref(&missing))
231	-	.await
232	-	.unwrap();
233	-	assert!(m.is_empty());
234	-	}
235	-
236	-	// (The clean-change gate itself — detect_conflicts -> CleanChanges::gated —
237	-	// is exercised end-to-end in synckit-client's own conflict tests, which can
238	-	// construct the non-exhaustive PulledChange internally. Here we only cover
239	-	// GO's committed-clock store that backs the gate.)
240	-
241	-	// -- apply_upsert --
242	-
243	-	#[tokio::test]
244	-	async fn apply_upsert_inserts_project() {
245	-	let pool = setup_test_db().await;
246	-	let user_id = create_test_user(&pool).await;
247	-	let project_id = uuid::Uuid::new_v4().to_string();
248	-	let now = now_sql();
249	-
250	-	let data = json!({
251	-	"id": project_id,
252	-	"name": "Synced Project",
253	-	"description": "From remote",
254	-	"project_type": "SideProject",
255	-	"status": "Active",
256	-	"created_at": now,
257	-	"user_id": user_id,
258	-	});
259	-
260	-	let mut conn = pool.acquire().await.unwrap();
261	-	apply::apply_upsert(&mut conn, "projects", &project_id, &data)
262	-	.await
263	-	.unwrap();
264	-
265	-	let row: (String,) = sqlx::query_as("SELECT name FROM projects WHERE id = ?")
266	-	.bind(&project_id)
267	-	.fetch_one(&pool)
268	-	.await
269	-	.unwrap();
270	-	assert_eq!(row.0, "Synced Project");
271	-	}
272	-
273	-	#[tokio::test]
274	-	async fn apply_upsert_replaces_existing_row() {
275	-	let pool = setup_test_db().await;
276	-	let user_id = create_test_user(&pool).await;
277	-	let project_id = uuid::Uuid::new_v4().to_string();
278	-	let now = now_sql();
279	-
280	-	let data1 = json!({
281	-	"id": project_id,
282	-	"name": "Original",
283	-	"description": "",
284	-	"project_type": "Job",
285	-	"status": "Active",
286	-	"created_at": now,
287	-	"user_id": user_id,
288	-	});
289	-	let mut conn = pool.acquire().await.unwrap();
290	-	apply::apply_upsert(&mut conn, "projects", &project_id, &data1)
291	-	.await
292	-	.unwrap();
293	-
294	-	let data2 = json!({
295	-	"id": project_id,
296	-	"name": "Updated",
297	-	"description": "changed",
298	-	"project_type": "Job",
299	-	"status": "OnHold",
300	-	"created_at": now,
301	-	"user_id": user_id,
302	-	});
303	-	apply::apply_upsert(&mut conn, "projects", &project_id, &data2)
304	-	.await
305	-	.unwrap();
306	-
307	-	let row: (String, String) = sqlx::query_as("SELECT name, status FROM projects WHERE id = ?")
308	-	.bind(&project_id)
309	-	.fetch_one(&pool)
310	-	.await
311	-	.unwrap();
312	-	assert_eq!(row.0, "Updated");
313	-	assert_eq!(row.1, "OnHold");
314	-	}
315	-
316	-	#[tokio::test]
317	-	async fn apply_upsert_parent_reupsert_preserves_children() {
318	-	// Re-upserting a parent task must not cascade-delete its annotations.
319	-	// INSERT OR REPLACE would; ON CONFLICT DO UPDATE must not. Enforce FKs
320	-	// so the cascade would actually fire if the row were deleted.
321	-	let pool = setup_test_db().await;
322	-	let user_id = create_test_user(&pool).await;
323	-	let task_id = uuid::Uuid::new_v4().to_string();
324	-	let annotation_id = uuid::Uuid::new_v4().to_string();
325	-	let now = now_sql();
326	-
327	-	let task = \|status: &str\| {
328	-	json!({
329	-	"id": task_id, "project_id": null, "description": "Parent",
330	-	"status": status, "priority": "Medium", "due": null, "tags": "",
331	-	"urgency": 0.0, "recurrence": "None", "recurrence_rule": null,
332	-	"created_at": now, "user_id": user_id, "recurrence_parent_id": null,
333	-	"source_email_id": null, "snoozed_until": null, "waiting_for_response": 0,
334	-	"waiting_since": null, "expected_response_date": null, "scheduled_start": null,
335	-	"scheduled_duration": null, "is_focus": 0, "focus_set_at": null,
336	-	"contact_id": null, "milestone_id": null, "completed_at": null,
337	-	"estimated_minutes": null, "actual_minutes": null,
338	-	})
339	-	};
340	-
341	-	let mut conn = pool.acquire().await.unwrap();
342	-	sqlx::query("PRAGMA foreign_keys = ON")
343	-	.execute(&mut *conn)
344	-	.await
345	-	.unwrap();
346	-
347	-	apply::apply_upsert(&mut conn, "tasks", &task_id, &task("Active"))
348	-	.await
349	-	.unwrap();
350	-
351	-	let annotation = json!({
352	-	"id": annotation_id, "task_id": task_id, "timestamp": now, "note": "child note",
353	-	});
354	-	apply::apply_upsert(&mut conn, "annotations", &annotation_id, &annotation)
355	-	.await
356	-	.unwrap();
357	-
358	-	// Re-upsert the parent (e.g. a remote status change).
359	-	apply::apply_upsert(&mut conn, "tasks", &task_id, &task("Completed"))
360	-	.await
361	-	.unwrap();
362	-
363	-	let count: (i64,) = sqlx::query_as("SELECT COUNT(*) FROM annotations WHERE task_id = ?")
364	-	.bind(&task_id)
365	-	.fetch_one(&pool)
366	-	.await
367	-	.unwrap();
368	-	assert_eq!(count.0, 1, "annotation should survive parent re-upsert");
369	-
370	-	let status: (String,) = sqlx::query_as("SELECT status FROM tasks WHERE id = ?")
371	-	.bind(&task_id)
372	-	.fetch_one(&pool)
373	-	.await
374	-	.unwrap();
375	-	assert_eq!(status.0, "Completed");
376	-	}
377	-
378	-	#[tokio::test]
379	-	async fn apply_upsert_rejects_unknown_table() {
380	-	let pool = setup_test_db().await;
381	-	let data = json!({"id": "abc"});
382	-	let mut conn = pool.acquire().await.unwrap();
383	-	let result = apply::apply_upsert(&mut conn, "nonexistent", "abc", &data).await;
384	-	assert!(result.is_err());
385	-	let err_msg = result.unwrap_err().to_string();
386	-	assert!(err_msg.contains("unknown syncable table"));
387	-	}
388	-
389	-	#[tokio::test]
390	-	async fn apply_upsert_handles_null_fields() {
391	-	let pool = setup_test_db().await;
392	-	let user_id = create_test_user(&pool).await;
393	-	let project_id = uuid::Uuid::new_v4().to_string();
394	-	let now = now_sql();
395	-
396	-	let mut conn = pool.acquire().await.unwrap();
397	-
398	-	// Insert a project to be a FK parent for the task
399	-	let data_project = json!({
400	-	"id": project_id,
401	-	"name": "Parent",
402	-	"description": "",
403	-	"project_type": "Job",
404	-	"status": "Active",
405	-	"created_at": now,
406	-	"user_id": user_id,
407	-	});
408	-	apply::apply_upsert(&mut conn, "projects", &project_id, &data_project)
409	-	.await
410	-	.unwrap();
411	-
412	-	let task_id = uuid::Uuid::new_v4().to_string();
413	-	let data = json!({
414	-	"id": task_id,
415	-	"project_id": project_id,
416	-	"description": "Null test",
417	-	"status": "Pending",
418	-	"priority": "Medium",
419	-	"due": null,
420	-	"tags": null,
421	-	"urgency": 50,
422	-	"recurrence": "None",
423	-	"created_at": now,
424	-	"user_id": user_id,
425	-	"recurrence_parent_id": null,
426	-	"source_email_id": null,
427	-	"snoozed_until": null,
428	-	"waiting_for_response": false,
429	-	"waiting_since": null,
430	-	"expected_response_date": null,
431	-	"scheduled_start": null,
432	-	"scheduled_duration": null,
433	-	"is_focus": false,
434	-	"focus_set_at": null,
435	-	"contact_id": null,
436	-	"milestone_id": null,
437	-	});
438	-
439	-	apply::apply_upsert(&mut conn, "tasks", &task_id, &data)
440	-	.await
441	-	.unwrap();
442	-
443	-	let row: (String,) = sqlx::query_as("SELECT description FROM tasks WHERE id = ?")
444	-	.bind(&task_id)
445	-	.fetch_one(&pool)
446	-	.await
447	-	.unwrap();
448	-	assert_eq!(row.0, "Null test");
449	-	}
450	-
451	-	// -- apply_delete --
452	-
453	-	#[tokio::test]
454	-	async fn apply_delete_removes_row() {
455	-	let pool = setup_test_db().await;
456	-	let user_id = create_test_user(&pool).await;
457	-	let project_id = uuid::Uuid::new_v4().to_string();
458	-	let now = now_sql();
459	-
460	-	let mut conn = pool.acquire().await.unwrap();
461	-
462	-	let data = json!({
463	-	"id": project_id,
464	-	"name": "To Delete",
465	-	"description": "",
466	-	"project_type": "Other",
467	-	"status": "Active",
468	-	"created_at": now,
469	-	"user_id": user_id,
470	-	});
471	-	apply::apply_upsert(&mut conn, "projects", &project_id, &data)
472	-	.await
473	-	.unwrap();
474	-
475	-	apply::apply_delete(&mut conn, "projects", &project_id)
476	-	.await
477	-	.unwrap();
478	-
479	-	let count: (i64,) = sqlx::query_as("SELECT COUNT(*) FROM projects WHERE id = ?")
480	-	.bind(&project_id)
481	-	.fetch_one(&pool)
482	-	.await
483	-	.unwrap();
484	-	assert_eq!(count.0, 0);
485	-	}
486	-
487	-	#[tokio::test]
488	-	async fn apply_delete_rejects_unknown_table() {
489	-	let pool = setup_test_db().await;
490	-	let mut conn = pool.acquire().await.unwrap();
491	-	let result = apply::apply_delete(&mut conn, "not_a_table", "abc").await;
492	-	assert!(result.is_err());
493	-	let err_msg = result.unwrap_err().to_string();
494	-	assert!(err_msg.contains("unknown syncable table"));
495	-	}
496	-
497	-	#[tokio::test]
498	-	async fn apply_delete_is_idempotent() {
499	-	let pool = setup_test_db().await;
500	-	let mut conn = pool.acquire().await.unwrap();

Lines truncated

A src-tauri/src/syncstore/manifest.rs +426

		@@ -0,0 +1,426 @@
1	+	//! The declarative [`SyncSchema`] manifest for GoingsOn's synced tables.
2	+	//!
3	+	//! This is the single source of truth the engine uses to generate the changelog
4	+	//! triggers, the initial snapshot, and the apply-side binding. It reproduces the
5	+	//! behaviour of GoingsOn's original hand-written triggers exactly (proven at
6	+	//! cutover against the retired `apply.rs::SYNCED_COLUMNS`); the column lists are
7	+	//! the synced columns of each table, in FK-safe (parents-first) order — the
8	+	//! engine derives the delete order by reversing it.
9	+	//!
10	+	//! Three tables need more than the common `t(name, cols)` default:
11	+	//! - `email_accounts` syncs config only. Passwords and OAuth tokens never leave
12	+	//! the device: they are excluded from the synced columns (so they never enter
13	+	//! the changelog), `preserve_local` keeps them untouched on a conflict-update,
14	+	//! and `insert_defaults` satisfies the `password` NOT NULL on a fresh remote
15	+	//! insert with `''`.
16	+	//! - `tasks` and `attachments` both carry `source_email_id`, a foreign key into
17	+	//! the `emails` table, which is NOT synced (emails stay per-device). They are
18	+	//! marked `references_unsynced` so FK enforcement is relaxed during a remote
19	+	//! apply.
20	+	//!
21	+	//! Groups (M3) will add `.group_scoped("group_id")` to the shareable tables; the
22	+	//! personal-only default here means a config/credential table can never leak into
23	+	//! a group scope.
24	+
25	+	use synckit_client::{ConflictStrategy, SyncSchema, SyncTable};
26	+
27	+	/// `email_accounts` synced columns: config only, never credentials. The secret
28	+	/// columns (`password`, the three `oauth2_*`) are excluded here and handled by
29	+	/// `preserve_local` + `insert_defaults` in [`goingson_schema`].
30	+	const EMAIL_ACCOUNT_SYNC_COLS: &[&str] = &[
31	+	"id",
32	+	"user_id",
33	+	"account_name",
34	+	"email_address",
35	+	"imap_server",
36	+	"imap_port",
37	+	"smtp_server",
38	+	"smtp_port",
39	+	"username",
40	+	"use_tls",
41	+	"created_at",
42	+	"archive_folder_name",
43	+	"auth_type",
44	+	"jmap_session_url",
45	+	"jmap_account_id",
46	+	"sync_interval_minutes",
47	+	"email_signature",
48	+	];
49	+
50	+	/// The common case: single `id` primary key, cleartext wire row-id, `Full`
51	+	/// upsert, `Hard` delete, HLC conflict resolution. Most of GO's tables are one
52	+	/// line each.
53	+	fn t(name: &'static str, cols: &'static [&'static str]) -> SyncTable {
54	+	SyncTable::full(name, cols)
55	+	}
56	+
57	+	/// GoingsOn's sync manifest: every table GO replicates, in FK (parents-first)
58	+	/// order.
59	+	pub(crate) fn goingson_schema() -> SyncSchema {
60	+	SyncSchema::new(vec![
61	+	t(
62	+	"projects",
63	+	&[
64	+	"id",
65	+	"name",
66	+	"description",
67	+	"project_type",
68	+	"status",
69	+	"created_at",
70	+	"user_id",
71	+	],
72	+	),
73	+	t(
74	+	"contacts",
75	+	&[
76	+	"id",
77	+	"user_id",
78	+	"display_name",
79	+	"nickname",
80	+	"company",
81	+	"title",
82	+	"notes",
83	+	"tags",
84	+	"birthday",
85	+	"timezone",
86	+	"external_source",
87	+	"external_id",
88	+	"created_at",
89	+	"updated_at",
90	+	],
91	+	),
92	+	// Config only, never credentials. See the module docs.
93	+	SyncTable::full("email_accounts", EMAIL_ACCOUNT_SYNC_COLS)
94	+	.preserve_local(&[
95	+	"password",
96	+	"oauth2_access_token",
97	+	"oauth2_refresh_token",
98	+	"oauth2_token_expires_at",
99	+	])
100	+	.insert_defaults(&[("password", "''")]),
101	+	t(
102	+	"sync_accounts",
103	+	&[
104	+	"id",
105	+	"user_id",
106	+	"provider",
107	+	"account_name",
108	+	"email",
109	+	"sync_calendars",
110	+	"sync_contacts",
111	+	"calendar_ids",
112	+	"sync_interval_minutes",
113	+	"enabled",
114	+	"created_at",
115	+	],
116	+	),
117	+	t(
118	+	"milestones",
119	+	&[
120	+	"id",
121	+	"user_id",
122	+	"project_id",
123	+	"name",
124	+	"description",
125	+	"position",
126	+	"target_date",
127	+	"status",
128	+	"created_at",
129	+	],
130	+	),
131	+	// Carries source_email_id -> the unsynced emails table.
132	+	SyncTable::full(
133	+	"tasks",
134	+	&[
135	+	"id",
136	+	"project_id",
137	+	"description",
138	+	"status",
139	+	"priority",
140	+	"due",
141	+	"tags",
142	+	"urgency",
143	+	"recurrence",
144	+	"recurrence_rule",
145	+	"created_at",
146	+	"user_id",
147	+	"recurrence_parent_id",
148	+	"source_email_id",
149	+	"snoozed_until",
150	+	"waiting_for_response",
151	+	"waiting_since",
152	+	"expected_response_date",
153	+	"scheduled_start",
154	+	"scheduled_duration",
155	+	"is_focus",
156	+	"focus_set_at",
157	+	"contact_id",
158	+	"milestone_id",
159	+	"completed_at",
160	+	"estimated_minutes",
161	+	"actual_minutes",
162	+	],
163	+	)
164	+	.references_unsynced(),
165	+	t(
166	+	"time_sessions",
167	+	&[
168	+	"id",
169	+	"task_id",
170	+	"user_id",
171	+	"started_at",
172	+	"ended_at",
173	+	"duration_minutes",
174	+	"created_at",
175	+	],
176	+	),
177	+	// Also carries source_email_id -> the unsynced emails table.
178	+	SyncTable::full(
179	+	"attachments",
180	+	&[
181	+	"id",
182	+	"user_id",
183	+	"task_id",
184	+	"project_id",
185	+	"filename",
186	+	"file_size",
187	+	"mime_type",
188	+	"blob_hash",
189	+	"source_email_id",
190	+	"created_at",
191	+	],
192	+	)
193	+	.references_unsynced(),
194	+	t(
195	+	"events",
196	+	&[
197	+	"id",
198	+	"project_id",
199	+	"title",
200	+	"description",
201	+	"start_time",
202	+	"end_time",
203	+	"location",
204	+	"user_id",
205	+	"linked_task_id",
206	+	"recurrence",
207	+	"recurrence_parent_id",
208	+	"recurrence_rule",
209	+	"contact_id",
210	+	"block_type",
211	+	"external_source",
212	+	"external_id",
213	+	"is_read_only",
214	+	"snoozed_until",
215	+	"reminder_offsets_seconds",
216	+	],
217	+	),
218	+	t("annotations", &["id", "task_id", "timestamp", "note"]),
219	+	t(
220	+	"subtasks",
221	+	&[
222	+	"id",
223	+	"task_id",
224	+	"text",
225	+	"is_completed",
226	+	"position",
227	+	"created_at",
228	+	"linked_task_id",
229	+	],
230	+	),
231	+	t(
232	+	"task_status_tokens",
233	+	&[
234	+	"id",
235	+	"task_id",
236	+	"kind",
237	+	"reference",
238	+	"state",
239	+	"is_primary",
240	+	"position",
241	+	"created_at",
242	+	],
243	+	),
244	+	t(
245	+	"contact_emails",
246	+	&["id", "contact_id", "address", "label", "is_primary"],
247	+	),
248	+	t(
249	+	"contact_phones",
250	+	&["id", "contact_id", "number", "label", "is_primary"],
251	+	),
252	+	t(
253	+	"contact_social_handles",
254	+	&["id", "contact_id", "platform", "handle", "url"],
255	+	),
256	+	t(
257	+	"contact_custom_fields",
258	+	&["id", "contact_id", "label", "value", "url"],
259	+	),
260	+	t(
261	+	"daily_notes",
262	+	&[
263	+	"id",
264	+	"user_id",
265	+	"note_date",
266	+	"went_well",
267	+	"could_improve",
268	+	"is_reviewed",
269	+	"reviewed_at",
270	+	"created_at",
271	+	"updated_at",
272	+	],
273	+	),
274	+	// Independent roots (FK only to local `users`); order among them is
275	+	// irrelevant.
276	+	t(
277	+	"saved_views",
278	+	&[
279	+	"id",
280	+	"user_id",
281	+	"name",
282	+	"view_type",
283	+	"filters",
284	+	"sort_by",
285	+	"sort_order",
286	+	"is_pinned",
287	+	"position",
288	+	"created_at",
289	+	"updated_at",
290	+	],
291	+	),
292	+	t(
293	+	"weekly_reviews",
294	+	&[
295	+	"id",
296	+	"user_id",
297	+	"week_start_date",
298	+	"completed_at",
299	+	"notes",
300	+	"vacation_days",
301	+	],
302	+	),
303	+	t(
304	+	"monthly_goals",
305	+	&[
306	+	"id",
307	+	"user_id",
308	+	"month",
309	+	"text",
310	+	"status",
311	+	"position",
312	+	"created_at",
313	+	"updated_at",
314	+	],
315	+	),
316	+	t(
317	+	"monthly_reflections",
318	+	&[
319	+	"id",
320	+	"user_id",
321	+	"month",
322	+	"highlight_text",
323	+	"change_text",
324	+	"completed_at",
325	+	],
326	+	),
327	+	])
328	+	// GO's current model; also the engine default.
329	+	.conflict_strategy(ConflictStrategy::HybridLogicalClock)
330	+	}
331	+
332	+	#[cfg(test)]
333	+	mod tests {
334	+	use super::goingson_schema;
335	+
336	+	/// The frozen table set (FK parents-first). Regenerating the manifest must not
337	+	/// silently add, drop, or reorder a table — that changes sync behaviour.
338	+	const EXPECTED_TABLES: &[&str] = &[
339	+	"projects",
340	+	"contacts",
341	+	"email_accounts",
342	+	"sync_accounts",
343	+	"milestones",
344	+	"tasks",
345	+	"time_sessions",
346	+	"attachments",
347	+	"events",
348	+	"annotations",
349	+	"subtasks",
350	+	"task_status_tokens",
351	+	"contact_emails",
352	+	"contact_phones",
353	+	"contact_social_handles",
354	+	"contact_custom_fields",
355	+	"daily_notes",
356	+	"saved_views",
357	+	"weekly_reviews",
358	+	"monthly_goals",
359	+	"monthly_reflections",
360	+	];
361	+
362	+	#[test]
363	+	fn manifest_table_set_and_order_is_frozen() {
364	+	let schema = goingson_schema();
365	+	let names: Vec<&str> = schema.tables().iter().map(\|t\| t.name()).collect();
366	+	assert_eq!(names, EXPECTED_TABLES, "manifest table set/order changed");
367	+	}
368	+
369	+	#[test]
370	+	fn every_table_is_id_keyed_and_id_first() {
371	+	for table in goingson_schema().tables() {
372	+	let cols = table.emitted_columns();
373	+	assert_eq!(
374	+	cols.first(),
375	+	Some(&"id"),
376	+	"{} must emit id first",
377	+	table.name()
378	+	);
379	+	}
380	+	}
381	+
382	+	#[test]
383	+	fn email_accounts_never_emits_credentials() {
384	+	let schema = goingson_schema();
385	+	let email = schema
386	+	.tables()
387	+	.iter()
388	+	.find(\|t\| t.name() == "email_accounts")
389	+	.expect("email_accounts in manifest");
390	+	for secret in [
391	+	"password",
392	+	"oauth2_access_token",
393	+	"oauth2_refresh_token",
394	+	"oauth2_token_expires_at",
395	+	] {
396	+	assert!(
397	+	!email.emitted_columns().contains(&secret),
398	+	"email_accounts must never sync {secret}"
399	+	);
400	+	}
401	+	}
402	+
403	+	/// FK dependencies: a child must not be declared before its parent, or the
404	+	/// engine's parents-first upsert would violate referential integrity.
405	+	#[test]
406	+	fn parents_precede_children() {
407	+	let order: Vec<&str> = goingson_schema()
408	+	.tables()
409	+	.iter()
410	+	.map(\|t\| t.name())
411	+	.collect();
412	+	let pos = \|name: &str\| order.iter().position(\|n\| *n == name).unwrap();
413	+	for (parent, child) in [
414	+	("projects", "tasks"),
415	+	("projects", "milestones"),
416	+	("tasks", "subtasks"),
417	+	("tasks", "annotations"),
418	+	("tasks", "task_status_tokens"),
419	+	("tasks", "time_sessions"),
420	+	("contacts", "contact_emails"),
421	+	("contacts", "contact_phones"),
422	+	] {
423	+	assert!(pos(parent) < pos(child), "{parent} must precede {child}");
424	+	}
425	+	}
426	+	}

M src-tauri/src/syncstore/mod.rs +25 -8

			@@ -13,24 +13,22 @@
13	13		//! The engine opens its own rusqlite connection (WAL) to the same `goingson.db`
14	14		//! file the sqlx pool uses; the two coexist by design.
15	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	16		mod blobs;
	17	+	pub(crate) mod manifest;
22	18		mod observer;
	19	+	pub(crate) mod sync_state;
23	20
24	21		use std::path::{Path, PathBuf};
25	22		use std::sync::Arc;
26	23
27	24		use rusqlite::{Connection, OptionalExtension};
28	25		use synckit_client::{DbSource, SyncKitClient, SyncStore};
	26	+	use tauri::{AppHandle, Manager};
29	27
30		-	use crate::sync_service::manifest::goingson_schema;
	28	+	use crate::state::AppState;
31	29		use blobs::AttachmentBlobs;
32		-	#[allow(unused_imports)] // wired into lib.rs setup when the M2 flip lands.
33		-	pub(crate) use observer::GoSyncObserver;
	30	+	use manifest::goingson_schema;
	31	+	use observer::GoSyncObserver;
34	32
35	33		/// Build GoingsOn's `SyncStore` over the app's SQLite file. The device name and
36	34		/// platform default to the host's (the engine's default), matching GO's prior
			@@ -46,6 +44,25 @@ pub(crate) fn build_store(
46	44		Arc::new(store)
47	45		}
48	46
	47	+	/// Spawn the background cloud-sync scheduler for the managed `AppState`'s
	48	+	/// `SyncStore`, if one is configured. The engine's loop races a timer against the
	49	+	/// server's SSE change stream, applies the auth / key-loaded / auto-sync-enabled /
	50	+	/// backoff gate chain, and reports status through [`GoSyncObserver`].
	51	+	///
	52	+	/// Must be called from within the async runtime (the engine spawns the loop with
	53	+	/// `tokio::spawn`). The returned handle is intentionally dropped: the loop runs
	54	+	/// for the app's lifetime — dropping the handle does not stop it — which matches
	55	+	/// GoingsOn's prior always-on scheduler. A no-op when sync is unconfigured.
	56	+	pub fn spawn_scheduler(app: &AppHandle) {
	57	+	let Some(state) = app.try_state::<Arc<AppState>>() else {
	58	+	return;
	59	+	};
	60	+	let Some(store) = state.sync_store.clone() else {
	61	+	return;
	62	+	};
	63	+	let _handle = store.spawn_scheduler(Arc::new(GoSyncObserver::new(app.clone())));
	64	+	}
	65	+
49	66		/// One-time migration of a device's local sync bookkeeping from GoingsOn's
50	67		/// hand-rolled triggers/changelog to the engine's. Idempotent and guarded by the
51	68		/// `syncstore_migrated` flag, so it is a cheap no-op on every launch after the

A src-tauri/src/syncstore/sync_state.rs +60

		@@ -0,0 +1,60 @@
1	+	//! Thin `sync_state` key-value helpers over the sqlx pool.
2	+	//!
3	+	//! `sync_state` is a plain KV table (device_id, auto_sync_enabled,
4	+	//! sync_interval_minutes, last_sync_at, ...) that the engine and the app share.
5	+	//! Reading and writing settings is app-side and needs only the pool, so these
6	+	//! stay as sqlx helpers here rather than routing through the store. The engine
7	+	//! owns the sync semantics (cursor, changelog, apply); these touch only user
8	+	//! settings and the pending-count readout for the status UI.
9	+
10	+	use goingson_core::CoreError;
11	+	use sqlx::SqlitePool;
12	+	use std::collections::HashMap;
13	+
14	+	/// Read several `sync_state` values in one query.
15	+	pub(crate) async fn get_sync_states_batch(
16	+	pool: &SqlitePool,
17	+	keys: &[&str],
18	+	) -> Result<HashMap<String, String>, CoreError> {
19	+	if keys.is_empty() {
20	+	return Ok(HashMap::new());
21	+	}
22	+
23	+	let placeholders = keys.iter().map(\|_\| "?").collect::<Vec<_>>().join(",");
24	+	let query = format!(
25	+	"SELECT key, value FROM sync_state WHERE key IN ({})",
26	+	placeholders
27	+	);
28	+
29	+	let mut q = sqlx::query_as::<_, (String, String)>(sqlx::AssertSqlSafe(query.as_str()));
30	+	for key in keys {
31	+	q = q.bind(*key);
32	+	}
33	+
34	+	let rows = q.fetch_all(pool).await.map_err(CoreError::database)?;
35	+	Ok(rows.into_iter().collect())
36	+	}
37	+
38	+	/// Upsert one `sync_state` value.
39	+	pub(crate) async fn set_sync_state(
40	+	pool: &SqlitePool,
41	+	key: &str,
42	+	value: &str,
43	+	) -> Result<(), CoreError> {
44	+	sqlx::query("INSERT OR REPLACE INTO sync_state (key, value) VALUES (?, ?)")
45	+	.bind(key)
46	+	.bind(value)
47	+	.execute(pool)
48	+	.await
49	+	.map_err(CoreError::database)?;
50	+	Ok(())
51	+	}
52	+
53	+	/// Count local changes not yet pushed (drives the status UI's pending badge).
54	+	pub(crate) async fn count_pending_changes(pool: &SqlitePool) -> Result<i64, CoreError> {
55	+	let row: (i64,) = sqlx::query_as("SELECT COUNT(*) FROM sync_changelog WHERE pushed = 0")
56	+	.fetch_one(pool)
57	+	.await
58	+	.map_err(CoreError::database)?;
59	+	Ok(row.0)
60	+	}

M src-tauri/src/test_utils.rs +1 -1

			@@ -71,7 +71,7 @@ pub async fn setup_test_state() -> (Arc<AppState>, UserId) {
71	71		monthly_reviews: Arc::new(SqliteMonthlyReviewRepository::new(pool.clone())),
72	72		backup_settings: Arc::new(SqliteBackupSettingsRepository::new(pool.clone())),
73	73		sync_accounts: Arc::new(SqliteSyncAccountRepository::new(pool.clone())),
74		-	sync_client: crate::state::SyncClientCell::default(),
	74	+	sync_store: None,
75	75		sync_lock: Arc::new(tokio::sync::Mutex::new(())),
76	76		email_sync_locks: Arc::new(std::sync::Mutex::new(std::collections::HashSet::new())),
77	77		token_refresh_locks: Arc::new(std::sync::Mutex::new(std::collections::HashMap::new())),