Skip to main content

max / makenotwork

10.9 KB · 333 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`, not the
139 /// per-user `sync_keys.key_id`. The dedicated table is what keeps group rows out
140 /// of every personal-scope query with no `group_id IS NULL` guard to remember.
141 /// Membership is enforced by the caller.
142 #[allow(clippy::type_complexity)]
143 #[tracing::instrument(skip_all)]
144 pub async fn push_group_changes(
145 pool: &PgPool,
146 app_id: SyncAppId,
147 group_id: SyncGroupId,
148 user_id: UserId,
149 device_id: SyncDeviceId,
150 batch_id: uuid::Uuid,
151 changes: &[(
152 String,
153 String,
154 String,
155 chrono::DateTime<chrono::Utc>,
156 Option<JsonValue>,
157 )],
158 ) -> Result<i64> {
159 if changes.is_empty() {
160 return Ok(0);
161 }
162
163 let mut table_names: Vec<String> = Vec::with_capacity(changes.len());
164 let mut operations: Vec<String> = Vec::with_capacity(changes.len());
165 let mut row_ids: Vec<String> = Vec::with_capacity(changes.len());
166 let mut client_timestamps: Vec<chrono::DateTime<chrono::Utc>> =
167 Vec::with_capacity(changes.len());
168 let mut data_values: Vec<JsonValue> = Vec::with_capacity(changes.len());
169
170 for (table_name, operation, row_id, client_timestamp, data) in changes {
171 table_names.push(table_name.clone());
172 operations.push(operation.clone());
173 row_ids.push(row_id.clone());
174 client_timestamps.push(*client_timestamp);
175 data_values.push(data.clone().unwrap_or(JsonValue::Null));
176 }
177
178 let mut tx = pool.begin().await?;
179
180 // Serialize concurrent redeliveries of the same batch within this group, so
181 // the MAX(seq) idempotency check below is race-free (same pattern as the
182 // per-user push lock, keyed by group instead of user).
183 sqlx::query(
184 "SELECT pg_advisory_xact_lock(hashtextextended('synckit_group_push:' || $1::text || ':' || $2::text, 0))",
185 )
186 .bind(group_id)
187 .bind(batch_id)
188 .execute(&mut *tx)
189 .await?;
190
191 let existing: (Option<i64>,) = sqlx::query_as(
192 "SELECT MAX(seq) FROM sync_group_log WHERE app_id = $1 AND group_id = $2 AND batch_id = $3",
193 )
194 .bind(app_id)
195 .bind(group_id)
196 .bind(batch_id)
197 .fetch_one(&mut *tx)
198 .await?;
199
200 if let Some(max_seq) = existing.0 {
201 tx.rollback().await.ok();
202 return Ok(max_seq);
203 }
204
205 let seqs: Vec<i64> = sqlx::query_scalar(
206 r"
207 INSERT INTO sync_group_log (app_id, user_id, device_id, group_id, batch_id, table_name, operation, row_id, client_timestamp, data)
208 SELECT $1, $2, $3, $4, $5, t.*
209 FROM UNNEST($6::text[], $7::text[], $8::text[], $9::timestamptz[], $10::jsonb[]) AS t
210 RETURNING seq
211 ",
212 )
213 .bind(app_id)
214 .bind(user_id)
215 .bind(device_id)
216 .bind(group_id)
217 .bind(batch_id)
218 .bind(&table_names)
219 .bind(&operations)
220 .bind(&row_ids)
221 .bind(&client_timestamps)
222 .bind(&data_values)
223 .fetch_all(&mut *tx)
224 .await?;
225
226 tx.commit().await?;
227
228 Ok(seqs.iter().copied().max().unwrap_or(0))
229 }
230
231 /// Pull a group's changes since a cursor from `sync_group_log`, with the same
232 /// optional table/timestamp filters as [`pull_sync_changes_filtered`], scoped by
233 /// `(app_id, group_id)`. Membership is enforced by the caller.
234 #[tracing::instrument(skip_all)]
235 pub async fn pull_group_changes_filtered(
236 pool: &PgPool,
237 app_id: SyncAppId,
238 group_id: SyncGroupId,
239 cursor: i64,
240 limit: i64,
241 tables: Option<&[String]>,
242 since: Option<chrono::DateTime<chrono::Utc>>,
243 ) -> Result<Vec<DbSyncGroupLogEntry>> {
244 let entries = sqlx::query_as::<_, DbSyncGroupLogEntry>(
245 r"
246 SELECT * FROM sync_group_log
247 WHERE app_id = $1 AND group_id = $2 AND seq > $3
248 AND ($5::text[] IS NULL OR table_name = ANY($5))
249 AND ($6::timestamptz IS NULL OR client_timestamp >= $6)
250 ORDER BY seq ASC
251 LIMIT $4
252 ",
253 )
254 .bind(app_id)
255 .bind(group_id)
256 .bind(cursor)
257 .bind(limit)
258 .bind(tables)
259 .bind(since)
260 .fetch_all(pool)
261 .await?;
262
263 Ok(entries)
264 }
265
266 /// Pull changes since a cursor for a user within an app.
267 ///
268 /// Prefer `pull_sync_changes_filtered` for new code; it supports optional
269 /// table and timestamp filters. This function is kept for backward compatibility.
270 #[allow(dead_code)]
271 #[tracing::instrument(skip_all)]
272 pub async fn pull_sync_changes(
273 pool: &PgPool,
274 app_id: SyncAppId,
275 user_id: UserId,
276 cursor: i64,
277 limit: i64,
278 ) -> Result<Vec<DbSyncLogEntry>> {
279 let entries = sqlx::query_as::<_, DbSyncLogEntry>(
280 r"
281 SELECT * FROM sync_log
282 WHERE app_id = $1 AND user_id = $2 AND seq > $3
283 ORDER BY seq ASC
284 LIMIT $4
285 ",
286 )
287 .bind(app_id)
288 .bind(user_id)
289 .bind(cursor)
290 .bind(limit)
291 .fetch_all(pool)
292 .await?;
293
294 Ok(entries)
295 }
296
297 /// Pull changes since a cursor with optional table and timestamp filters.
298 ///
299 /// When `tables` is `Some`, only entries whose `table_name` is in the list are returned.
300 /// When `since` is `Some`, only entries with `client_timestamp >= since` are returned.
301 /// Both filters compose (AND). Passing `None` for both is identical to `pull_sync_changes`.
302 #[tracing::instrument(skip_all)]
303 pub async fn pull_sync_changes_filtered(
304 pool: &PgPool,
305 app_id: SyncAppId,
306 user_id: UserId,
307 cursor: i64,
308 limit: i64,
309 tables: Option<&[String]>,
310 since: Option<chrono::DateTime<chrono::Utc>>,
311 ) -> Result<Vec<DbSyncLogEntry>> {
312 let entries = sqlx::query_as::<_, DbSyncLogEntry>(
313 r"
314 SELECT * FROM sync_log
315 WHERE app_id = $1 AND user_id = $2 AND seq > $3
316 AND ($5::text[] IS NULL OR table_name = ANY($5))
317 AND ($6::timestamptz IS NULL OR client_timestamp >= $6)
318 ORDER BY seq ASC
319 LIMIT $4
320 ",
321 )
322 .bind(app_id)
323 .bind(user_id)
324 .bind(cursor)
325 .bind(limit)
326 .bind(tables)
327 .bind(since)
328 .fetch_all(pool)
329 .await?;
330
331 Ok(entries)
332 }
333