max / goingson

7.2 KB · 225 lines History Blame Raw

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