Skip to main content

max / goingson

11.8 KB · 298 lines History Blame Raw
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 ("projects", &[
36 "id", "name", "description", "project_type", "status", "created_at", "user_id",
37 ]),
38 ("tasks", &[
39 "id", "project_id", "description", "status", "priority", "due", "tags", "urgency",
40 "recurrence", "recurrence_rule", "created_at", "user_id", "recurrence_parent_id",
41 "source_email_id", "snoozed_until", "waiting_for_response", "waiting_since",
42 "expected_response_date", "scheduled_start", "scheduled_duration", "is_focus",
43 "focus_set_at", "contact_id", "milestone_id", "completed_at", "estimated_minutes",
44 "actual_minutes",
45 ]),
46 ("events", &[
47 "id", "project_id", "title", "description", "start_time", "end_time", "location",
48 "user_id", "linked_task_id", "recurrence", "recurrence_parent_id", "recurrence_rule",
49 "contact_id", "block_type", "external_source", "external_id", "is_read_only",
50 "snoozed_until", "reminder_offsets_seconds",
51 ]),
52 ("contacts", &[
53 "id", "user_id", "display_name", "nickname", "company", "title", "notes", "tags",
54 "birthday", "timezone", "external_source", "external_id", "created_at", "updated_at",
55 ]),
56 ("contact_emails", &["id", "contact_id", "address", "label", "is_primary"]),
57 ("contact_phones", &["id", "contact_id", "number", "label", "is_primary"]),
58 ("contact_social_handles", &["id", "contact_id", "platform", "handle", "url"]),
59 ("contact_custom_fields", &["id", "contact_id", "label", "value", "url"]),
60 ("annotations", &["id", "task_id", "timestamp", "note"]),
61 ("subtasks", &[
62 "id", "task_id", "text", "is_completed", "position", "created_at", "linked_task_id",
63 ]),
64 ("task_status_tokens", &[
65 "id", "task_id", "kind", "reference", "state", "is_primary", "position", "created_at",
66 ]),
67 ("milestones", &[
68 "id", "user_id", "project_id", "name", "description", "position", "target_date",
69 "status", "created_at",
70 ]),
71 ("time_sessions", &[
72 "id", "task_id", "user_id", "started_at", "ended_at", "duration_minutes", "created_at",
73 ]),
74 ("attachments", &[
75 "id", "user_id", "task_id", "project_id", "filename", "file_size", "mime_type",
76 "blob_hash", "source_email_id", "created_at",
77 ]),
78 ("sync_accounts", &[
79 "id", "user_id", "provider", "account_name", "email", "sync_calendars", "sync_contacts",
80 "calendar_ids", "sync_interval_minutes", "enabled", "created_at",
81 ]),
82 ("email_accounts", EMAIL_ACCOUNT_SYNC_COLS),
83 ("daily_notes", &[
84 "id", "user_id", "note_date", "went_well", "could_improve", "is_reviewed", "reviewed_at",
85 "created_at", "updated_at",
86 ]),
87 ("saved_views", &[
88 "id", "user_id", "name", "view_type", "filters", "sort_by", "sort_order", "is_pinned",
89 "position", "created_at", "updated_at",
90 ]),
91 ("weekly_reviews", &[
92 "id", "user_id", "week_start_date", "completed_at", "notes", "vacation_days",
93 ]),
94 ("monthly_goals", &[
95 "id", "user_id", "month", "text", "status", "position", "created_at", "updated_at",
96 ]),
97 ("monthly_reflections", &[
98 "id", "user_id", "month", "highlight_text", "change_text", "completed_at",
99 ]),
100 ];
101
102 /// Return the syncable column whitelist for a given table, or `None` if unknown.
103 pub(crate) fn table_columns(table: &str) -> Option<&'static [&'static str]> {
104 SYNCED_COLUMNS
105 .iter()
106 .find(|(t, _)| *t == table)
107 .map(|(_, cols)| *cols)
108 }
109
110 /// Apply an upsert for a remote change.
111 ///
112 /// Uses `INSERT ... ON CONFLICT(id) DO UPDATE` rather than `INSERT OR REPLACE`.
113 /// `INSERT OR REPLACE` deletes the conflicting row before reinserting it, which
114 /// fires `ON DELETE CASCADE` on children (annotations, subtasks, contact_*) and
115 /// would wipe them on any parent re-upsert. `ON CONFLICT DO UPDATE` mutates the
116 /// row in place, so children survive and referential integrity holds regardless
117 /// of FK enforcement. Every syncable table has `id` as its primary key.
118 #[tracing::instrument(skip_all)]
119 pub(crate) async fn apply_upsert(
120 conn: &mut SqliteConnection,
121 table: &str,
122 _row_id: &str,
123 data: &serde_json::Value,
124 ) -> Result<(), CoreError> {
125 // Email accounts use a bespoke ON CONFLICT that preserves local credentials
126 if table == "email_accounts" {
127 return apply_email_account_upsert(conn, data).await;
128 }
129
130 let columns = table_columns(table)
131 .ok_or_else(|| CoreError::bad_request(format!("unknown syncable table: {}", table)))?;
132
133 // A remote null for a NOT NULL column is invalid input the changelog never
134 // emits. INSERT OR REPLACE silently coerced it to the column default; to
135 // keep that tolerance under ON CONFLICT, omit such columns: a new row then
136 // takes the schema default and an existing row keeps its current value.
137 // Nullable columns keep null (so legitimate clears — e.g. un-snoozing —
138 // still propagate). NOT NULL columns are derived from the live schema, not
139 // a hand-maintained list, so they can't drift.
140 let not_null = not_null_columns(conn, table).await?;
141 let included: Vec<&str> = columns
142 .iter()
143 .copied()
144 .filter(|c| !(data[*c].is_null() && not_null.contains(*c)))
145 .collect();
146
147 let col_list = included.join(", ");
148 let placeholders = included.iter().map(|_| "?").collect::<Vec<_>>().join(", ");
149 let update_set = included
150 .iter()
151 .filter(|c| **c != "id")
152 .map(|c| format!("{} = excluded.{}", c, c))
153 .collect::<Vec<_>>()
154 .join(", ");
155 // If only `id` survived, there is nothing to update on conflict.
156 let conflict = if update_set.is_empty() {
157 "DO NOTHING".to_string()
158 } else {
159 format!("DO UPDATE SET {}", update_set)
160 };
161 let sql = format!(
162 "INSERT INTO {} ({}) VALUES ({}) ON CONFLICT(id) {}",
163 table, col_list, placeholders, conflict
164 );
165
166 let mut query = sqlx::query(&sql);
167
168 for col in &included {
169 query = bind_json_value(query, &data[*col]);
170 }
171
172 query
173 .execute(&mut *conn)
174 .await
175 .map_err(CoreError::database)?;
176
177 Ok(())
178 }
179
180 /// Names of the NOT NULL columns of a table, read from the live schema.
181 ///
182 /// Derived via `PRAGMA table_info` rather than a static list so it can never
183 /// drift from the migrations. `table` is already validated against the column
184 /// whitelist by the caller, so interpolating it is safe.
185 async fn not_null_columns(
186 conn: &mut SqliteConnection,
187 table: &str,
188 ) -> Result<std::collections::HashSet<String>, CoreError> {
189 let rows = sqlx::query(&format!("PRAGMA table_info({})", table))
190 .fetch_all(&mut *conn)
191 .await
192 .map_err(CoreError::database)?;
193
194 let mut set = std::collections::HashSet::new();
195 for row in rows {
196 let notnull: i64 = row.try_get("notnull").map_err(CoreError::database)?;
197 if notnull != 0 {
198 let name: String = row.try_get("name").map_err(CoreError::database)?;
199 set.insert(name);
200 }
201 }
202 Ok(set)
203 }
204
205 /// Apply an upsert for email_accounts that preserves local credentials.
206 ///
207 /// Uses INSERT ... ON CONFLICT(id) DO UPDATE to only touch the 16 config columns,
208 /// leaving `password`, `oauth2_access_token`, `oauth2_refresh_token`, and
209 /// `oauth2_token_expires_at` untouched on existing rows. New rows get `password = ''`
210 /// to satisfy the NOT NULL constraint.
211 #[tracing::instrument(skip_all)]
212 async fn apply_email_account_upsert(
213 conn: &mut SqliteConnection,
214 data: &serde_json::Value,
215 ) -> Result<(), CoreError> {
216 let cols = EMAIL_ACCOUNT_SYNC_COLS;
217
218 // INSERT columns: 16 sync cols + password (hardcoded to '')
219 let mut insert_cols: Vec<&str> = cols.to_vec();
220 insert_cols.push("password");
221
222 let col_list = insert_cols.join(", ");
223 let placeholders = insert_cols.iter().map(|_| "?").collect::<Vec<_>>().join(", ");
224
225 // ON CONFLICT: only update the 16 sync columns
226 let update_set = cols
227 .iter()
228 .filter(|c| **c != "id")
229 .map(|c| format!("{} = excluded.{}", c, c))
230 .collect::<Vec<_>>()
231 .join(", ");
232
233 let sql = format!(
234 "INSERT INTO email_accounts ({}) VALUES ({}) ON CONFLICT(id) DO UPDATE SET {}",
235 col_list, placeholders, update_set
236 );
237
238 let mut query = sqlx::query(&sql);
239
240 // Bind the 16 sync columns from data
241 for col in cols {
242 query = bind_json_value(query, &data[*col]);
243 }
244
245 // Bind password = '' for the INSERT
246 query = query.bind("");
247
248 query
249 .execute(&mut *conn)
250 .await
251 .map_err(CoreError::database)?;
252
253 Ok(())
254 }
255
256 /// Bind a JSON value to a sqlx query.
257 pub(crate) fn bind_json_value<'q>(
258 query: sqlx::query::Query<'q, sqlx::Sqlite, sqlx::sqlite::SqliteArguments<'q>>,
259 val: &'q serde_json::Value,
260 ) -> sqlx::query::Query<'q, sqlx::Sqlite, sqlx::sqlite::SqliteArguments<'q>> {
261 match val {
262 serde_json::Value::String(s) => query.bind(s.as_str()),
263 serde_json::Value::Number(n) => {
264 if let Some(i) = n.as_i64() {
265 query.bind(i)
266 } else if let Some(f) = n.as_f64() {
267 query.bind(f)
268 } else {
269 query.bind(None::<String>)
270 }
271 }
272 serde_json::Value::Bool(b) => query.bind(if *b { 1i32 } else { 0i32 }),
273 serde_json::Value::Null => query.bind(None::<String>),
274 _ => {
275 // Arrays/objects: serialize as JSON string -- need owned String
276 query.bind(val.to_string())
277 }
278 }
279 }
280
281 /// Apply a DELETE for a remote change.
282 #[tracing::instrument(skip_all)]
283 pub(crate) async fn apply_delete(conn: &mut SqliteConnection, table: &str, row_id: &str) -> Result<(), CoreError> {
284 // Validate table name is in our whitelist
285 if table_columns(table).is_none() {
286 return Err(CoreError::bad_request(format!("unknown syncable table: {}", table)));
287 }
288
289 let sql = format!("DELETE FROM {} WHERE id = ?", table);
290 sqlx::query(&sql)
291 .bind(row_id)
292 .execute(&mut *conn)
293 .await
294 .map_err(CoreError::database)?;
295
296 Ok(())
297 }
298