Skip to main content

max / makenotwork

11.2 KB · 337 lines History Blame Raw
1 //! SyncKit change log: the push/pull sync protocol core, append a device's
2 //! changes and pull the changes since a cursor (optionally filtered), the basis
3 //! of the multi-device convergence protocol.
4
5 use serde_json::Value as JsonValue;
6 use sqlx::PgPool;
7
8 use crate::db::models::{DbSyncGroupLogEntry, DbSyncLogEntry};
9 use crate::db::{SyncAppId, SyncDeviceId, SyncGroupId, UserId};
10 use crate::error::Result;
11
12 // ── Sync Log ──
13
14 /// Push a batch of changes to the sync log. Returns the highest seq assigned.
15 ///
16 /// `batch_id` is a client-generated UUID for idempotent push. If a batch with
17 /// the same ID has already been committed for this app+user, the existing max
18 /// seq is returned without re-inserting (at-most-once semantics).
19 ///
20 /// All changes are inserted within a single transaction for atomicity and
21 /// performance (one fsync instead of N).
22 #[allow(clippy::type_complexity)]
23 #[tracing::instrument(skip_all)]
24 pub async fn push_sync_changes(
25 pool: &PgPool,
26 app_id: SyncAppId,
27 user_id: UserId,
28 device_id: SyncDeviceId,
29 batch_id: uuid::Uuid,
30 changes: &[(
31 String,
32 String,
33 String,
34 chrono::DateTime<chrono::Utc>,
35 Option<JsonValue>,
36 )],
37 ) -> Result<i64> {
38 if changes.is_empty() {
39 return Ok(0);
40 }
41
42 // Decompose into parallel Vecs for UNNEST-based batch INSERT
43 let mut table_names: Vec<String> = Vec::with_capacity(changes.len());
44 let mut operations: Vec<String> = Vec::with_capacity(changes.len());
45 let mut row_ids: Vec<String> = Vec::with_capacity(changes.len());
46 let mut client_timestamps: Vec<chrono::DateTime<chrono::Utc>> =
47 Vec::with_capacity(changes.len());
48 let mut data_values: Vec<JsonValue> = Vec::with_capacity(changes.len());
49
50 for (table_name, operation, row_id, client_timestamp, data) in changes {
51 table_names.push(table_name.clone());
52 operations.push(operation.clone());
53 row_ids.push(row_id.clone());
54 client_timestamps.push(*client_timestamp);
55 data_values.push(data.clone().unwrap_or(JsonValue::Null));
56 }
57
58 let mut tx = pool.begin().await?;
59
60 // Serialize concurrent pushes of the SAME batch. A batch legitimately
61 // inserts many sync_log rows sharing one batch_id, so migration 086 had to
62 // drop the unique index that used to backstop idempotency, which left the
63 // MAX(seq) check-then-insert below open to a TOCTOU: two concurrent
64 // redeliveries of the same batch both read MAX = NULL and both insert,
65 // duplicating the client's changes. This per-(app,user,batch) transaction
66 // advisory lock makes the check-and-insert atomic; the loser blocks, then
67 // sees the winner's rows and returns the existing cursor. Same
68 // `hashtextextended('ns:' || ...)` pattern as the Stripe webhook lock.
69 sqlx::query(
70 "SELECT pg_advisory_xact_lock(hashtextextended('synckit_push:' || $1::text || ':' || $2::text || ':' || $3::text, 0))",
71 )
72 .bind(app_id)
73 .bind(user_id)
74 .bind(batch_id)
75 .execute(&mut *tx)
76 .await?;
77
78 // Idempotency check inside the transaction, race-free under the lock above.
79 let existing: (Option<i64>,) = sqlx::query_as(
80 "SELECT MAX(seq) FROM sync_log WHERE app_id = $1 AND user_id = $2 AND batch_id = $3",
81 )
82 .bind(app_id)
83 .bind(user_id)
84 .bind(batch_id)
85 .fetch_one(&mut *tx)
86 .await?;
87
88 if let Some(max_seq) = existing.0 {
89 tracing::debug!(batch_id = %batch_id, cursor = max_seq, "Push batch already committed, returning existing cursor");
90 tx.rollback().await.ok();
91 return Ok(max_seq);
92 }
93
94 // Read the current key_id from sync_keys so pushed entries are stamped
95 // with the active encryption key. Falls back to NULL if no key exists yet
96 // (pre-encryption setup, entries have no encrypted data anyway).
97 let current_key_id: Option<i32> =
98 sqlx::query_scalar("SELECT key_id FROM sync_keys WHERE app_id = $1 AND user_id = $2")
99 .bind(app_id)
100 .bind(user_id)
101 .fetch_optional(&mut *tx)
102 .await?;
103
104 let seqs: Vec<i64> = sqlx::query_scalar(
105 r"
106 INSERT INTO sync_log (app_id, user_id, device_id, batch_id, table_name, operation, row_id, client_timestamp, data, key_id)
107 SELECT $1, $2, $3, $4, t.*, $10
108 FROM UNNEST($5::text[], $6::text[], $7::text[], $8::timestamptz[], $9::jsonb[]) AS t
109 RETURNING seq
110 ",
111 )
112 .bind(app_id)
113 .bind(user_id)
114 .bind(device_id)
115 .bind(batch_id)
116 .bind(&table_names)
117 .bind(&operations)
118 .bind(&row_ids)
119 .bind(&client_timestamps)
120 .bind(&data_values)
121 .bind(current_key_id)
122 .fetch_all(&mut *tx)
123 .await?;
124
125 tx.commit().await?;
126
127 let max_seq = seqs.iter().copied().max().unwrap_or(0);
128
129 Ok(max_seq)
130 }
131
132 /// Push a batch of changes to a **group's** shared changelog (`sync_group_log`).
133 /// Returns the highest seq assigned. The group scope (not the pushing user) owns
134 /// the entries, so every member's pull sees them.
135 ///
136 /// Mirrors [`push_sync_changes`] but writes the separate `sync_group_log` table
137 /// and scopes idempotency by `(app_id, group_id, batch_id)`. Group entries carry
138 /// no `key_id`: their key generation is the group's `gck_version`, stamped onto
139 /// each row at insert so a rotation does not orphan what came before it. The
140 /// dedicated table is what keeps group rows out of every personal-scope query
141 /// with no `group_id IS NULL` guard to remember. Membership is enforced by the
142 /// caller.
143 #[allow(clippy::type_complexity)]
144 #[tracing::instrument(skip_all)]
145 pub async fn push_group_changes(
146 pool: &PgPool,
147 app_id: SyncAppId,
148 group_id: SyncGroupId,
149 user_id: UserId,
150 device_id: SyncDeviceId,
151 batch_id: uuid::Uuid,
152 changes: &[(
153 String,
154 String,
155 String,
156 chrono::DateTime<chrono::Utc>,
157 Option<JsonValue>,
158 )],
159 ) -> Result<i64> {
160 if changes.is_empty() {
161 return Ok(0);
162 }
163
164 let mut table_names: Vec<String> = Vec::with_capacity(changes.len());
165 let mut operations: Vec<String> = Vec::with_capacity(changes.len());
166 let mut row_ids: Vec<String> = Vec::with_capacity(changes.len());
167 let mut client_timestamps: Vec<chrono::DateTime<chrono::Utc>> =
168 Vec::with_capacity(changes.len());
169 let mut data_values: Vec<JsonValue> = Vec::with_capacity(changes.len());
170
171 for (table_name, operation, row_id, client_timestamp, data) in changes {
172 table_names.push(table_name.clone());
173 operations.push(operation.clone());
174 row_ids.push(row_id.clone());
175 client_timestamps.push(*client_timestamp);
176 data_values.push(data.clone().unwrap_or(JsonValue::Null));
177 }
178
179 let mut tx = pool.begin().await?;
180
181 // Serialize concurrent redeliveries of the same batch within this group, so
182 // the MAX(seq) idempotency check below is race-free (same pattern as the
183 // per-user push lock, keyed by group instead of user).
184 sqlx::query(
185 "SELECT pg_advisory_xact_lock(hashtextextended('synckit_group_push:' || $1::text || ':' || $2::text, 0))",
186 )
187 .bind(group_id)
188 .bind(batch_id)
189 .execute(&mut *tx)
190 .await?;
191
192 let existing: (Option<i64>,) = sqlx::query_as(
193 "SELECT MAX(seq) FROM sync_group_log WHERE app_id = $1 AND group_id = $2 AND batch_id = $3",
194 )
195 .bind(app_id)
196 .bind(group_id)
197 .bind(batch_id)
198 .fetch_one(&mut *tx)
199 .await?;
200
201 if let Some(max_seq) = existing.0 {
202 tx.rollback().await.ok();
203 return Ok(max_seq);
204 }
205
206 // Stamp the generation the pusher's ciphertext is sealed under, read inside
207 // this transaction so a rotation committing concurrently cannot leave a row
208 // labelled with a generation it was not encrypted under.
209 let seqs: Vec<i64> = sqlx::query_scalar(
210 r"
211 INSERT INTO sync_group_log (app_id, user_id, device_id, group_id, batch_id, table_name, operation, row_id, client_timestamp, data, gck_version)
212 SELECT $1, $2, $3, $4, $5, t.*, (SELECT gck_version FROM sync_groups WHERE id = $4)
213 FROM UNNEST($6::text[], $7::text[], $8::text[], $9::timestamptz[], $10::jsonb[]) AS t
214 RETURNING seq
215 ",
216 )
217 .bind(app_id)
218 .bind(user_id)
219 .bind(device_id)
220 .bind(group_id)
221 .bind(batch_id)
222 .bind(&table_names)
223 .bind(&operations)
224 .bind(&row_ids)
225 .bind(&client_timestamps)
226 .bind(&data_values)
227 .fetch_all(&mut *tx)
228 .await?;
229
230 tx.commit().await?;
231
232 Ok(seqs.iter().copied().max().unwrap_or(0))
233 }
234
235 /// Pull a group's changes since a cursor from `sync_group_log`, with the same
236 /// optional table/timestamp filters as [`pull_sync_changes_filtered`], scoped by
237 /// `(app_id, group_id)`. Membership is enforced by the caller.
238 #[tracing::instrument(skip_all)]
239 pub async fn pull_group_changes_filtered(
240 pool: &PgPool,
241 app_id: SyncAppId,
242 group_id: SyncGroupId,
243 cursor: i64,
244 limit: i64,
245 tables: Option<&[String]>,
246 since: Option<chrono::DateTime<chrono::Utc>>,
247 ) -> Result<Vec<DbSyncGroupLogEntry>> {
248 let entries = sqlx::query_as::<_, DbSyncGroupLogEntry>(
249 r"
250 SELECT * FROM sync_group_log
251 WHERE app_id = $1 AND group_id = $2 AND seq > $3
252 AND ($5::text[] IS NULL OR table_name = ANY($5))
253 AND ($6::timestamptz IS NULL OR client_timestamp >= $6)
254 ORDER BY seq ASC
255 LIMIT $4
256 ",
257 )
258 .bind(app_id)
259 .bind(group_id)
260 .bind(cursor)
261 .bind(limit)
262 .bind(tables)
263 .bind(since)
264 .fetch_all(pool)
265 .await?;
266
267 Ok(entries)
268 }
269
270 /// Pull changes since a cursor for a user within an app.
271 ///
272 /// Prefer `pull_sync_changes_filtered` for new code; it supports optional
273 /// table and timestamp filters. This function is kept for backward compatibility.
274 #[allow(dead_code)]
275 #[tracing::instrument(skip_all)]
276 pub async fn pull_sync_changes(
277 pool: &PgPool,
278 app_id: SyncAppId,
279 user_id: UserId,
280 cursor: i64,
281 limit: i64,
282 ) -> Result<Vec<DbSyncLogEntry>> {
283 let entries = sqlx::query_as::<_, DbSyncLogEntry>(
284 r"
285 SELECT * FROM sync_log
286 WHERE app_id = $1 AND user_id = $2 AND seq > $3
287 ORDER BY seq ASC
288 LIMIT $4
289 ",
290 )
291 .bind(app_id)
292 .bind(user_id)
293 .bind(cursor)
294 .bind(limit)
295 .fetch_all(pool)
296 .await?;
297
298 Ok(entries)
299 }
300
301 /// Pull changes since a cursor with optional table and timestamp filters.
302 ///
303 /// When `tables` is `Some`, only entries whose `table_name` is in the list are returned.
304 /// When `since` is `Some`, only entries with `client_timestamp >= since` are returned.
305 /// Both filters compose (AND). Passing `None` for both is identical to `pull_sync_changes`.
306 #[tracing::instrument(skip_all)]
307 pub async fn pull_sync_changes_filtered(
308 pool: &PgPool,
309 app_id: SyncAppId,
310 user_id: UserId,
311 cursor: i64,
312 limit: i64,
313 tables: Option<&[String]>,
314 since: Option<chrono::DateTime<chrono::Utc>>,
315 ) -> Result<Vec<DbSyncLogEntry>> {
316 let entries = sqlx::query_as::<_, DbSyncLogEntry>(
317 r"
318 SELECT * FROM sync_log
319 WHERE app_id = $1 AND user_id = $2 AND seq > $3
320 AND ($5::text[] IS NULL OR table_name = ANY($5))
321 AND ($6::timestamptz IS NULL OR client_timestamp >= $6)
322 ORDER BY seq ASC
323 LIMIT $4
324 ",
325 )
326 .bind(app_id)
327 .bind(user_id)
328 .bind(cursor)
329 .bind(limit)
330 .bind(tables)
331 .bind(since)
332 .fetch_all(pool)
333 .await?;
334
335 Ok(entries)
336 }
337