Skip to main content

max / makenotwork

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