Skip to main content

max / makenotwork

13.6 KB · 399 lines History Blame Raw
1 //! SyncKit binary blobs: resolve a blob by content hash, confirm an uploaded
2 //! blob (internal vs developer paths) once its S3 object lands, and delete a
3 //! blob. Content-addressed, so a repeat hash dedups rather than re-storing.
4
5 use sqlx::PgPool;
6
7 use crate::db::models::DbSyncBlob;
8 use crate::db::{SyncAppId, UserId};
9 use crate::error::Result;
10
11 // ── Sync Blobs ──
12
13 /// Get a blob by content hash for a user within an app.
14 #[tracing::instrument(skip_all)]
15 pub async fn get_sync_blob_by_hash(
16 pool: &PgPool,
17 app_id: SyncAppId,
18 user_id: UserId,
19 hash: &str,
20 ) -> Result<Option<DbSyncBlob>> {
21 let blob = sqlx::query_as::<_, DbSyncBlob>(
22 "SELECT * FROM sync_blobs WHERE app_id = $1 AND user_id = $2 AND hash = $3",
23 )
24 .bind(app_id)
25 .bind(user_id)
26 .bind(hash)
27 .fetch_optional(pool)
28 .await?;
29
30 Ok(blob)
31 }
32
33 /// Sum the bytes a user currently stores for an app. `sync_blobs` is the
34 /// authoritative record, so this is the same number the quota gate enforces
35 /// against; there is no counter to drift. Generic over the executor so the
36 /// status route can ask the pool and `confirm_internal_blob` can ask inside
37 /// its own locked transaction.
38 async fn sum_blob_bytes<'e, E>(executor: E, app_id: SyncAppId, user_id: UserId) -> Result<i64>
39 where
40 E: sqlx::PgExecutor<'e>,
41 {
42 let used: i64 = sqlx::query_scalar(
43 "SELECT COALESCE(SUM(size_bytes), 0)::BIGINT FROM sync_blobs
44 WHERE app_id = $1 AND user_id = $2",
45 )
46 .bind(app_id)
47 .bind(user_id)
48 .fetch_one(executor)
49 .await?;
50 Ok(used)
51 }
52
53 /// Bytes a user currently stores for an app, for the subscription-status
54 /// route. Summed live rather than cached: the same query already runs on every
55 /// blob confirm, and `sync_blobs` is indexed on `(app_id, user_id)`.
56 #[tracing::instrument(skip_all)]
57 pub async fn storage_used_bytes(pool: &PgPool, app_id: SyncAppId, user_id: UserId) -> Result<i64> {
58 sum_blob_bytes(pool, app_id, user_id).await
59 }
60
61 /// Result of an atomic blob-confirm. Every variant is terminal for one confirm
62 /// call; the route handler maps it to a 204 or a 402 with the right reason.
63 #[derive(Debug, Clone, PartialEq, Eq)]
64 pub enum BlobConfirm {
65 /// Newly recorded; usage counters were incremented.
66 Stored,
67 /// The blob (same hash) was already recorded, idempotent re-confirm, no
68 /// double counting.
69 AlreadyStored,
70 /// First-party app and the user has no `active` subscription (paid-only).
71 NoSubscription,
72 /// Storing this blob would exceed the applicable cap. Nothing was written.
73 QuotaExceeded {
74 dimension: &'static str,
75 used: i64,
76 limit: i64,
77 key: Option<String>,
78 },
79 }
80
81 /// Insert a blob row inside an open transaction. `ON CONFLICT DO UPDATE` keeps
82 /// `size_bytes` consistent with the actual S3 object; the idempotency check in
83 /// the callers means we only reach this for a genuinely new `(app,user,hash)`.
84 async fn insert_blob_tx(
85 tx: &mut sqlx::Transaction<'_, sqlx::Postgres>,
86 app_id: SyncAppId,
87 user_id: UserId,
88 hash: &str,
89 size_bytes: i64,
90 s3_key: &str,
91 key: &str,
92 ) -> Result<()> {
93 sqlx::query(
94 r"
95 INSERT INTO sync_blobs (app_id, user_id, hash, size_bytes, s3_key, key)
96 VALUES ($1, $2, $3, $4, $5, $6)
97 ON CONFLICT (app_id, user_id, hash)
98 DO UPDATE SET size_bytes = EXCLUDED.size_bytes
99 ",
100 )
101 .bind(app_id)
102 .bind(user_id)
103 .bind(hash)
104 .bind(size_bytes)
105 .bind(s3_key)
106 .bind(key)
107 .execute(&mut **tx)
108 .await?;
109 Ok(())
110 }
111
112 /// Confirm a blob for a first-party (`is_internal`) app under the paid-only
113 /// end-user model. Atomic: locks the user's subscription row, enforces an
114 /// `active` status and the per-user `storage_limit_bytes`, then inserts, all
115 /// in one transaction, so concurrent confirms for the same user can't overshoot
116 /// the cap (the prior read-then-add gate could). Usage is summed from the
117 /// authoritative `sync_blobs` table, so there is no counter to drift.
118 #[tracing::instrument(skip_all)]
119 pub async fn confirm_internal_blob(
120 pool: &PgPool,
121 app_id: SyncAppId,
122 user_id: UserId,
123 hash: &str,
124 size_bytes: i64,
125 s3_key: &str,
126 key: &str,
127 ) -> Result<BlobConfirm> {
128 let mut tx = pool.begin().await?;
129
130 // Lock the subscription row for this user+app. Serializes this user's
131 // confirms; different users don't contend.
132 let sub: Option<(String, Option<i64>)> = sqlx::query_as(
133 "SELECT status, storage_limit_bytes FROM app_sync_subscriptions
134 WHERE app_id = $1 AND user_id = $2 FOR UPDATE",
135 )
136 .bind(app_id)
137 .bind(user_id)
138 .fetch_optional(&mut *tx)
139 .await?;
140 let Some((status, limit)) = sub else {
141 return Ok(BlobConfirm::NoSubscription);
142 };
143 if status != "active" {
144 return Ok(BlobConfirm::NoSubscription);
145 }
146
147 // Idempotent re-confirm: already recorded, don't recount.
148 let exists: Option<i32> = sqlx::query_scalar(
149 "SELECT 1 FROM sync_blobs WHERE app_id = $1 AND user_id = $2 AND hash = $3",
150 )
151 .bind(app_id)
152 .bind(user_id)
153 .bind(hash)
154 .fetch_optional(&mut *tx)
155 .await?;
156 if exists.is_some() {
157 tx.commit().await?;
158 return Ok(BlobConfirm::AlreadyStored);
159 }
160
161 let used = sum_blob_bytes(&mut *tx, app_id, user_id).await?;
162 let limit = limit.unwrap_or(0);
163 if used.saturating_add(size_bytes) > limit {
164 return Ok(BlobConfirm::QuotaExceeded {
165 dimension: "storage",
166 used,
167 limit,
168 key: None,
169 });
170 }
171
172 insert_blob_tx(&mut tx, app_id, user_id, hash, size_bytes, s3_key, key).await?;
173 tx.commit().await?;
174 Ok(BlobConfirm::Stored)
175 }
176
177 /// Confirm a blob for a developer-billed (non-internal) app. Atomic: locks the
178 /// app usage row (and the per-key row in `per_key` mode), re-checks the cap
179 /// under the lock, inserts, and increments the counters, folding the old
180 /// `would_exceed_storage` (read) + `add_bytes_stored` (add) pair into one
181 /// transaction so concurrent uploads can't slip past the cap. Re-confirms are
182 /// idempotent and never double-count.
183 #[tracing::instrument(skip_all)]
184 #[allow(clippy::too_many_arguments)]
185 pub async fn confirm_developer_blob(
186 pool: &PgPool,
187 app_id: SyncAppId,
188 user_id: UserId,
189 hash: &str,
190 size_bytes: i64,
191 s3_key: &str,
192 key: &str,
193 enforcement_mode: crate::db::SyncEnforcementMode,
194 storage_gb_cap: Option<i32>,
195 key_cap: Option<i32>,
196 gb_per_key: Option<i32>,
197 ) -> Result<BlobConfirm> {
198 let mut tx = pool.begin().await?;
199
200 // Lock the app usage row; this is the serialization point for the app-wide
201 // counter. Returns current app-level bytes_stored.
202 let app_used: i64 = sqlx::query_scalar(
203 "SELECT bytes_stored FROM sync_app_usage_current WHERE app_id = $1 FOR UPDATE",
204 )
205 .bind(app_id)
206 .fetch_one(&mut *tx)
207 .await?;
208
209 // Idempotent re-confirm.
210 let exists: Option<i32> = sqlx::query_scalar(
211 "SELECT 1 FROM sync_blobs WHERE app_id = $1 AND user_id = $2 AND hash = $3",
212 )
213 .bind(app_id)
214 .bind(user_id)
215 .bind(hash)
216 .fetch_optional(&mut *tx)
217 .await?;
218 if exists.is_some() {
219 tx.commit().await?;
220 return Ok(BlobConfirm::AlreadyStored);
221 }
222
223 match enforcement_mode {
224 crate::db::SyncEnforcementMode::Bulk => {
225 if let Some(gb) = storage_gb_cap {
226 let limit = crate::synckit_billing::storage_cap_bytes(gb as u32);
227 if app_used.saturating_add(size_bytes) > limit {
228 return Ok(BlobConfirm::QuotaExceeded {
229 dimension: "storage",
230 used: app_used,
231 limit,
232 key: None,
233 });
234 }
235 }
236 }
237 crate::db::SyncEnforcementMode::PerKey => {
238 if let (Some(kc), Some(g)) = (key_cap, gb_per_key) {
239 let per_key_limit = crate::synckit_billing::storage_cap_bytes(g as u32);
240 let app_limit =
241 crate::synckit_billing::storage_cap_bytes(kc.saturating_mul(g) as u32);
242 let key_used: i64 = sqlx::query_scalar(
243 "SELECT bytes_stored FROM sync_key_usage_current
244 WHERE app_id = $1 AND key = $2 FOR UPDATE",
245 )
246 .bind(app_id)
247 .bind(key)
248 .fetch_optional(&mut *tx)
249 .await?
250 .unwrap_or(0);
251 if key_used.saturating_add(size_bytes) > per_key_limit {
252 return Ok(BlobConfirm::QuotaExceeded {
253 dimension: "storage_per_key",
254 used: key_used,
255 limit: per_key_limit,
256 key: Some(key.to_string()),
257 });
258 }
259 // Defensive app-aggregate ceiling (guards counter drift).
260 if app_used.saturating_add(size_bytes) > app_limit {
261 return Ok(BlobConfirm::QuotaExceeded {
262 dimension: "storage",
263 used: app_used,
264 limit: app_limit,
265 key: None,
266 });
267 }
268 }
269 }
270 }
271
272 insert_blob_tx(&mut tx, app_id, user_id, hash, size_bytes, s3_key, key).await?;
273
274 // Increment the app-wide and per-key counters in the same transaction.
275 sqlx::query(
276 "UPDATE sync_app_usage_current
277 SET bytes_stored = GREATEST(bytes_stored + $2, 0), updated_at = NOW()
278 WHERE app_id = $1",
279 )
280 .bind(app_id)
281 .bind(size_bytes)
282 .execute(&mut *tx)
283 .await?;
284 sqlx::query(
285 "INSERT INTO sync_key_usage_current (app_id, key, bytes_stored)
286 VALUES ($1, $2, GREATEST($3, 0))
287 ON CONFLICT (app_id, key)
288 DO UPDATE SET bytes_stored = GREATEST(sync_key_usage_current.bytes_stored + $3, 0),
289 updated_at = NOW()",
290 )
291 .bind(app_id)
292 .bind(key)
293 .bind(size_bytes)
294 .execute(&mut *tx)
295 .await?;
296
297 tx.commit().await?;
298 Ok(BlobConfirm::Stored)
299 }
300
301 /// Outcome of a blob delete. Terminal for one call.
302 #[derive(Debug, Clone, PartialEq, Eq)]
303 pub enum BlobDelete {
304 /// Row removed; S3 object dead-lettered; usage counters refunded.
305 Deleted { size_bytes: i64 },
306 /// No blob with that hash for this `(app, user)`, idempotent no-op.
307 NotFound,
308 }
309
310 /// Delete a blob by hash for a user within an app. Atomic in one transaction:
311 /// removes the `sync_blobs` row, refunds the developer-billing counters, and
312 /// dead-letters the S3 object via `pending_s3_deletions`. This is the sole
313 /// shrink path for storage usage that doesn't wait on the weekly drift job,
314 /// and because the row delete, the counter refund, and the S3-delete enqueue
315 /// commit together, a delete can never leave `bytes_stored` overstating reality
316 /// or orphan the object on a mid-operation crash.
317 ///
318 /// Internal (first-party) apps keep no counter, `confirm_internal_blob` sums
319 /// usage straight from `sync_blobs`, so for them the counter UPDATEs match no
320 /// row and are harmless no-ops; removing the row is the whole refund. Idempotent:
321 /// deleting an absent blob returns `NotFound` with nothing written.
322 #[tracing::instrument(skip_all)]
323 pub async fn delete_sync_blob(
324 pool: &PgPool,
325 app_id: SyncAppId,
326 user_id: UserId,
327 hash: &str,
328 ) -> Result<BlobDelete> {
329 let mut tx = pool.begin().await?;
330
331 // Remove the row, capturing what we need to refund counters and dead-letter
332 // the object. SyncKit blob keys are per-`(app, user, hash)` (no cross-user
333 // dedup), so deleting this row's object frees only this user's copy.
334 let row: Option<(i64, String, String)> = sqlx::query_as(
335 "DELETE FROM sync_blobs WHERE app_id = $1 AND user_id = $2 AND hash = $3
336 RETURNING size_bytes, s3_key, key",
337 )
338 .bind(app_id)
339 .bind(user_id)
340 .bind(hash)
341 .fetch_optional(&mut *tx)
342 .await?;
343
344 let Some((size_bytes, s3_key, key)) = row else {
345 tx.commit().await?;
346 return Ok(BlobDelete::NotFound);
347 };
348
349 // Refund the developer-billing counters. `GREATEST(_, 0)` floors at zero so
350 // a drifted counter can never go negative; internal apps have no usage row
351 // and these UPDATEs touch nothing.
352 sqlx::query(
353 "UPDATE sync_app_usage_current
354 SET bytes_stored = GREATEST(bytes_stored - $2, 0), updated_at = NOW()
355 WHERE app_id = $1",
356 )
357 .bind(app_id)
358 .bind(size_bytes)
359 .execute(&mut *tx)
360 .await?;
361 sqlx::query(
362 "UPDATE sync_key_usage_current
363 SET bytes_stored = GREATEST(bytes_stored - $3, 0), updated_at = NOW()
364 WHERE app_id = $1 AND key = $2",
365 )
366 .bind(app_id)
367 .bind(&key)
368 .bind(size_bytes)
369 .execute(&mut *tx)
370 .await?;
371
372 // Dead-letter the object in the SAME tx as the row delete + refund (model:
373 // `versions::delete_version`). After commit the row is gone, so the key is
374 // non-live and the deletion worker (`retry_pending_s3_deletions`, which
375 // routes `bucket = "synckit"` to `synckit_s3`) can act on it.
376 crate::db::pending_s3_deletions::enqueue_deletions(
377 &mut *tx,
378 &[(s3_key, "synckit".to_string())],
379 "synckit_blob_delete",
380 )
381 .await?;
382
383 tx.commit().await?;
384 Ok(BlobDelete::Deleted { size_bytes })
385 }
386
387 /// Count devices registered for a user/app pair.
388 #[tracing::instrument(skip_all)]
389 pub async fn count_sync_devices(pool: &PgPool, app_id: SyncAppId, user_id: UserId) -> Result<i64> {
390 let count: i64 =
391 sqlx::query_scalar("SELECT COUNT(*) FROM sync_devices WHERE app_id = $1 AND user_id = $2")
392 .bind(app_id)
393 .bind(user_id)
394 .fetch_one(pool)
395 .await?;
396
397 Ok(count)
398 }
399