Skip to main content

max / makenotwork

26.2 KB · 677 lines History Blame Raw
1 //! SyncKit blob storage: presigned upload/download URLs and upload confirmation.
2
3 use axum::{
4 Json,
5 extract::State,
6 http::StatusCode,
7 response::{IntoResponse, Response},
8 };
9 use serde_json::json;
10
11 use sqlx::PgPool;
12
13 use crate::{
14 constants,
15 db::{self, synckit_billing},
16 error::{AppError, Result, ResultExt},
17 synckit_auth::SyncUser,
18 validation,
19 };
20
21 use super::{
22 BlobConfirmRequest, BlobDownloadUrlRequest, BlobDownloadUrlResponse, BlobMultipartAbortRequest,
23 BlobMultipartCompleteRequest, BlobMultipartPartUrl, BlobMultipartPartsRequest,
24 BlobMultipartPartsResponse, BlobMultipartStartRequest, BlobMultipartStartResponse,
25 BlobUploadUrlRequest, BlobUploadUrlResponse,
26 };
27
28 /// Request a pre-signed S3 upload URL for a blob.
29 ///
30 /// Content-addressed by hash: if a blob with the same hash already exists
31 /// for this user/app, returns `already_exists: true` and an empty URL,
32 /// skipping the upload.
33 #[utoipa::path(post, path = "/api/v1/sync/blobs/upload", tag = "SyncKit",
34 request_body = BlobUploadUrlRequest,
35 responses((status = 200, description = "Pre-signed upload URL", body = BlobUploadUrlResponse)),
36 security(("bearer" = [])),
37 )]
38 #[tracing::instrument(skip_all, name = "synckit::blob_upload_url")]
39 pub(super) async fn blob_upload_url(
40 State(db): State<PgPool>,
41 State(storage): State<crate::AppStorage>,
42 sync_user: SyncUser,
43 Json(req): Json<BlobUploadUrlRequest>,
44 ) -> Result<Response> {
45 let synckit_s3 = storage.synckit_s3.as_ref().ok_or_else(|| {
46 AppError::ServiceUnavailable("SyncKit blob storage is not configured".to_string())
47 })?;
48
49 // The one-shot ceiling, not the multipart one: a single PUT is one
50 // unresumable request, so anything larger belongs on the multipart session.
51 if req.size_bytes <= 0 || req.size_bytes > constants::SYNCKIT_MAX_BLOB_SIZE_BYTES {
52 return Err(AppError::BadRequest(format!(
53 "Blob size must be between 1 and {} bytes; larger blobs upload through the multipart session",
54 constants::SYNCKIT_MAX_BLOB_SIZE_BYTES
55 )));
56 }
57
58 // Paid-only gate + content-address dedup. The gate refuses an unsubscribed
59 // first-party user before any upload credential exists, so they can't stage
60 // orphan S3 objects; storage-cap enforcement still happens atomically at
61 // confirm time. Non-internal (developer-billed) apps always pass here.
62 match blob_upload_gate(&db, &sync_user, &req.hash).await? {
63 Some(BlobUploadStop::NoSubscription) => return Ok(no_subscription_response()),
64 Some(BlobUploadStop::AlreadyExists) => {
65 return Ok(Json(BlobUploadUrlResponse {
66 upload_url: String::new(),
67 already_exists: true,
68 })
69 .into_response());
70 }
71 None => {}
72 }
73
74 let s3_key = crate::storage::S3Client::generate_synckit_blob_key(
75 sync_user.app_id,
76 sync_user.user_id,
77 &req.hash,
78 );
79
80 // Track the pending upload so the reaper can clean it up if never confirmed
81 db::pending_uploads::record_pending_upload(&db, sync_user.user_id, &s3_key, "synckit").await?;
82
83 let upload_url = synckit_s3
84 .presign_upload(
85 &s3_key,
86 "application/octet-stream",
87 Some(constants::SYNCKIT_BLOB_PRESIGN_EXPIRY_SECS),
88 None,
89 // Bind Content-Length at the S3 layer so the client can't upload
90 // more than it declared. Confirm reads the actual object size as
91 // the authoritative figure regardless.
92 Some(req.size_bytes),
93 )
94 .await
95 .context("presign upload for sync blob")?;
96
97 Ok(Json(BlobUploadUrlResponse {
98 upload_url,
99 already_exists: false,
100 })
101 .into_response())
102 }
103
104 // --- Multipart blob session (large blobs) ---
105 //
106 // The chunked counterpart to `blob_upload_url`, and the only route past
107 // `SYNCKIT_MAX_BLOB_SIZE_BYTES`: a one-shot presigned PUT is a single
108 // unresumable request, so it keeps the modest ceiling while multipart carries
109 // the large blobs. These endpoints replace the *transport* only, the client
110 // still finishes at `/blobs/confirm`, which reads the authoritative object size
111 // from S3 and does all the quota/billing work unchanged.
112 //
113 // Authorization needs no key lookup here (unlike the creator-media multipart
114 // session, which signs owner-less `staging/{uuid}` keys): a synckit blob key is
115 // `{app_id}/{user_id}/{hash}`, derived server-side from the caller's JWT, so a
116 // caller can only ever address their own blob.
117
118 /// Largest window of presigned part URLs one `parts` call will mint. The client
119 /// pulls them as it progresses rather than holding hundreds of live
120 /// credentials for an upload that may never finish.
121 const BLOB_MULTIPART_PART_URL_WINDOW: u32 = 100;
122
123 /// Why a blob upload must not open. Both transports run the same gate and
124 /// render it in their own response shape.
125 enum BlobUploadStop {
126 /// First-party app, unsubscribed user: don't hand out any upload
127 /// credential, so they cannot stage orphan S3 objects.
128 NoSubscription,
129 /// This content address is already stored for this user/app.
130 AlreadyExists,
131 }
132
133 /// Shared pre-upload checks: hash shape, the paid-only gate, and content-address
134 /// dedup. `Ok(None)` means the caller may open an upload. Size validation stays
135 /// at the call site, since the two transports have different ceilings.
136 async fn blob_upload_gate(
137 db: &PgPool,
138 sync_user: &SyncUser,
139 hash: &str,
140 ) -> Result<Option<BlobUploadStop>> {
141 validation::validate_sync_blob_hash(hash)?;
142
143 if !db::synckit::internal_write_allowed(db, sync_user.app_id, sync_user.user_id).await? {
144 return Ok(Some(BlobUploadStop::NoSubscription));
145 }
146
147 if db::synckit::get_sync_blob_by_hash(db, sync_user.app_id, sync_user.user_id, hash)
148 .await?
149 .is_some()
150 {
151 return Ok(Some(BlobUploadStop::AlreadyExists));
152 }
153
154 Ok(None)
155 }
156
157 /// The 402 both transports return for an unsubscribed first-party user.
158 fn no_subscription_response() -> Response {
159 (
160 StatusCode::PAYMENT_REQUIRED,
161 Json(json!({ "reason": "no_subscription" })),
162 )
163 .into_response()
164 }
165
166 /// Open a multipart upload session for a large blob.
167 ///
168 /// `size_bytes` is the ciphertext length, which the client derives from the
169 /// plaintext length alone (`blob_encrypted_len`) before sealing anything. The
170 /// part geometry is pure arithmetic over it, so both sides compute identical
171 /// boundaries without a round trip.
172 #[utoipa::path(post, path = "/api/v1/sync/blobs/multipart/start", tag = "SyncKit",
173 request_body = BlobMultipartStartRequest,
174 responses((status = 200, description = "Multipart session opened", body = BlobMultipartStartResponse)),
175 security(("bearer" = [])),
176 )]
177 #[tracing::instrument(skip_all, name = "synckit::blob_multipart_start")]
178 pub(super) async fn blob_multipart_start(
179 State(db): State<PgPool>,
180 State(storage): State<crate::AppStorage>,
181 sync_user: SyncUser,
182 Json(req): Json<BlobMultipartStartRequest>,
183 ) -> Result<Response> {
184 let synckit_s3 = storage.synckit_s3.as_ref().ok_or_else(|| {
185 AppError::ServiceUnavailable("SyncKit blob storage is not configured".to_string())
186 })?;
187
188 if req.size_bytes <= 0 || req.size_bytes > constants::SYNCKIT_MAX_MULTIPART_BLOB_SIZE_BYTES {
189 return Err(AppError::BadRequest(format!(
190 "Blob size must be between 1 and {} bytes",
191 constants::SYNCKIT_MAX_MULTIPART_BLOB_SIZE_BYTES
192 )));
193 }
194
195 match blob_upload_gate(&db, &sync_user, &req.hash).await? {
196 Some(BlobUploadStop::NoSubscription) => return Ok(no_subscription_response()),
197 Some(BlobUploadStop::AlreadyExists) => {
198 return Ok(Json(BlobMultipartStartResponse {
199 upload_id: String::new(),
200 part_size: 0,
201 part_count: 0,
202 already_exists: true,
203 })
204 .into_response());
205 }
206 None => {}
207 }
208
209 let plan =
210 s3_storage::MultipartPlan::auto(req.size_bytes as u64).map_err(AppError::BadRequest)?;
211
212 let s3_key = crate::storage::S3Client::generate_synckit_blob_key(
213 sync_user.app_id,
214 sync_user.user_id,
215 &req.hash,
216 );
217
218 // Track the session so the orphan reaper aborts it if the client vanishes.
219 // An abandoned multipart upload leaves no object at all, only billed parts,
220 // which the reaper recovers by listing sessions for this key.
221 db::pending_uploads::record_pending_upload(&db, sync_user.user_id, &s3_key, "synckit").await?;
222
223 let upload_id = synckit_s3
224 .create_multipart_upload(&s3_key, "application/octet-stream")
225 .await?;
226
227 tracing::info!(
228 app = %sync_user.app_id, user = %sync_user.user_id,
229 size = req.size_bytes, parts = plan.part_count,
230 "SyncKit multipart blob upload started"
231 );
232
233 Ok(Json(BlobMultipartStartResponse {
234 upload_id,
235 part_size: plan.part_size,
236 part_count: plan.part_count,
237 already_exists: false,
238 })
239 .into_response())
240 }
241
242 /// Mint a bounded window of presigned `UploadPart` URLs, each carrying its exact
243 /// signed `Content-Length`, the same defense-in-depth the one-shot presign
244 /// applies.
245 #[utoipa::path(post, path = "/api/v1/sync/blobs/multipart/parts", tag = "SyncKit",
246 request_body = BlobMultipartPartsRequest,
247 responses((status = 200, description = "Presigned part URLs", body = BlobMultipartPartsResponse)),
248 security(("bearer" = [])),
249 )]
250 #[tracing::instrument(skip_all, name = "synckit::blob_multipart_parts")]
251 pub(super) async fn blob_multipart_parts(
252 State(storage): State<crate::AppStorage>,
253 sync_user: SyncUser,
254 Json(req): Json<BlobMultipartPartsRequest>,
255 ) -> Result<Response> {
256 let synckit_s3 = storage.synckit_s3.as_ref().ok_or_else(|| {
257 AppError::ServiceUnavailable("SyncKit blob storage is not configured".to_string())
258 })?;
259
260 validation::validate_sync_blob_hash(&req.hash)?;
261 if req.size_bytes <= 0 || req.size_bytes > constants::SYNCKIT_MAX_MULTIPART_BLOB_SIZE_BYTES {
262 return Err(AppError::BadRequest(
263 "Blob size is out of range".to_string(),
264 ));
265 }
266
267 let plan =
268 s3_storage::MultipartPlan::auto(req.size_bytes as u64).map_err(AppError::BadRequest)?;
269
270 if req.count == 0 || req.count > BLOB_MULTIPART_PART_URL_WINDOW {
271 return Err(AppError::BadRequest(format!(
272 "count must be between 1 and {BLOB_MULTIPART_PART_URL_WINDOW}"
273 )));
274 }
275 if req.first_part == 0 || req.first_part > plan.part_count {
276 return Err(AppError::BadRequest(format!(
277 "first_part must be between 1 and {}",
278 plan.part_count
279 )));
280 }
281
282 // Checksums are positional, so a short or long list would silently bind the
283 // wrong digest to a part, reject it rather than guess the alignment.
284 if let Some(checksums) = &req.checksums {
285 if checksums.len() != req.count as usize {
286 return Err(AppError::BadRequest(format!(
287 "checksums must have exactly {} entries, one per requested part",
288 req.count
289 )));
290 }
291 for c in checksums {
292 validation::validate_sha256_base64(c)?;
293 }
294 }
295
296 let s3_key = crate::storage::S3Client::generate_synckit_blob_key(
297 sync_user.app_id,
298 sync_user.user_id,
299 &req.hash,
300 );
301
302 let expires_in = constants::SYNCKIT_BLOB_PRESIGN_EXPIRY_SECS;
303 let last = (req.first_part + req.count - 1).min(plan.part_count);
304 let mut parts = Vec::with_capacity((last - req.first_part + 1) as usize);
305 for part_number in req.first_part..=last {
306 let content_length = plan.part_len(part_number);
307 let checksum = req
308 .checksums
309 .as_ref()
310 .and_then(|c| c.get((part_number - req.first_part) as usize))
311 .map(String::as_str);
312 let url = synckit_s3
313 .presign_upload_part(
314 &s3_key,
315 &req.upload_id,
316 part_number as i32,
317 Some(expires_in),
318 Some(content_length as i64),
319 checksum,
320 )
321 .await
322 .context("presign upload part for sync blob")?;
323 parts.push(BlobMultipartPartUrl {
324 part_number: part_number as i32,
325 content_length,
326 url,
327 });
328 }
329
330 Ok(Json(BlobMultipartPartsResponse { parts, expires_in }).into_response())
331 }
332
333 /// Assemble the uploaded parts into the blob object.
334 ///
335 /// Transport only: the client then calls `/blobs/confirm`, which reads the real
336 /// object size from S3 and applies every quota and billing rule.
337 #[utoipa::path(post, path = "/api/v1/sync/blobs/multipart/complete", tag = "SyncKit",
338 request_body = BlobMultipartCompleteRequest,
339 responses((status = 204, description = "Parts assembled")),
340 security(("bearer" = [])),
341 )]
342 #[tracing::instrument(skip_all, name = "synckit::blob_multipart_complete")]
343 pub(super) async fn blob_multipart_complete(
344 State(storage): State<crate::AppStorage>,
345 sync_user: SyncUser,
346 Json(req): Json<BlobMultipartCompleteRequest>,
347 ) -> Result<Response> {
348 let synckit_s3 = storage.synckit_s3.as_ref().ok_or_else(|| {
349 AppError::ServiceUnavailable("SyncKit blob storage is not configured".to_string())
350 })?;
351
352 validation::validate_sync_blob_hash(&req.hash)?;
353 if req.parts.is_empty() {
354 return Err(AppError::BadRequest("No parts to complete".to_string()));
355 }
356
357 let s3_key = crate::storage::S3Client::generate_synckit_blob_key(
358 sync_user.app_id,
359 sync_user.user_id,
360 &req.hash,
361 );
362
363 let parts: Vec<(i32, String)> = req
364 .parts
365 .into_iter()
366 .map(|p| (p.part_number, p.etag))
367 .collect();
368
369 synckit_s3
370 .complete_multipart_upload(&s3_key, &req.upload_id, &parts)
371 .await?;
372
373 tracing::info!(
374 app = %sync_user.app_id, user = %sync_user.user_id, parts = parts.len(),
375 "SyncKit multipart blob upload completed"
376 );
377
378 Ok(StatusCode::NO_CONTENT.into_response())
379 }
380
381 /// Release the parts of an abandoned session (client cancel).
382 ///
383 /// Incomplete multipart uploads bill for their parts until aborted, so a client
384 /// that cleans up on cancel is the cheapest fix; the orphan reaper is the
385 /// backstop for clients that vanish.
386 #[utoipa::path(post, path = "/api/v1/sync/blobs/multipart/abort", tag = "SyncKit",
387 request_body = BlobMultipartAbortRequest,
388 responses((status = 204, description = "Session aborted")),
389 security(("bearer" = [])),
390 )]
391 #[tracing::instrument(skip_all, name = "synckit::blob_multipart_abort")]
392 pub(super) async fn blob_multipart_abort(
393 State(db): State<PgPool>,
394 State(storage): State<crate::AppStorage>,
395 sync_user: SyncUser,
396 Json(req): Json<BlobMultipartAbortRequest>,
397 ) -> Result<Response> {
398 let synckit_s3 = storage.synckit_s3.as_ref().ok_or_else(|| {
399 AppError::ServiceUnavailable("SyncKit blob storage is not configured".to_string())
400 })?;
401
402 validation::validate_sync_blob_hash(&req.hash)?;
403
404 let s3_key = crate::storage::S3Client::generate_synckit_blob_key(
405 sync_user.app_id,
406 sync_user.user_id,
407 &req.hash,
408 );
409
410 synckit_s3
411 .abort_multipart_upload(&s3_key, &req.upload_id)
412 .await?;
413 // The session is gone, so the reaper has nothing left to find; drop the
414 // tracking row rather than leaving it to age out.
415 db::pending_uploads::remove_pending_upload(&db, sync_user.user_id, &s3_key, "synckit").await?;
416
417 tracing::info!(
418 app = %sync_user.app_id, user = %sync_user.user_id,
419 "SyncKit multipart blob upload aborted"
420 );
421
422 Ok(StatusCode::NO_CONTENT.into_response())
423 }
424
425 /// Confirm that a blob upload to S3 completed successfully.
426 ///
427 /// Verifies the object exists in S3, then records it in the database.
428 /// Idempotent: returns success without creating a duplicate.
429 ///
430 /// Content-addressing trust model (ultra-fuzz Run 4 Storage NOTE, decision
431 /// 2026-06-23; revised 2026-07-21): the blob `hash` is treated as a
432 /// content-address LABEL, confirm reads the authoritative `object_size` from S3
433 /// but does not re-hash the bytes to prove they match `hash`. The blast radius
434 /// is per-user only: the key is `{app_id}/{user_id}/{hash}` and storage is
435 /// `UNIQUE(app_id, user_id, hash)`, so a client that stores mismatched bytes can
436 /// poison only its OWN dedup namespace, no cross-user effect, no data exposure.
437 ///
438 /// This note used to say the A+ fix was binding `x-amz-checksum-sha256` into the
439 /// presigned PUT so S3 rejects a mismatched upload at write time. That reasoning
440 /// does not hold for these blobs, and the correction is worth keeping: the stored
441 /// object is E2E *ciphertext* sealed with random per-chunk nonces, while `hash`
442 /// is the SHA-256 of the *plaintext*. The server never sees plaintext, so it
443 /// cannot derive the expected ciphertext digest at presign time, any checksum it
444 /// binds has to come from the client, i.e. the party whose honesty was in
445 /// question. Checksum binding (which the multipart path now does per part) buys
446 /// transport integrity, not content-address enforcement.
447 ///
448 /// What actually binds the bytes to the address is the AEAD: each chunk is sealed
449 /// with `(hash, chunk_index, chunk_count)` as associated data, so ciphertext that
450 /// opens under `hash` is cryptographically tied to it, and the client re-hashes
451 /// the plaintext after decrypting. A client storing mismatched bytes breaks only
452 /// its own blob. Server-side re-hashing would cost a full object download per
453 /// confirm to defend a client against itself, which is why it is not done.
454 #[utoipa::path(post, path = "/api/v1/sync/blobs/confirm", tag = "SyncKit",
455 request_body = BlobConfirmRequest,
456 responses((status = 204, description = "Upload confirmed")),
457 security(("bearer" = [])),
458 )]
459 #[tracing::instrument(skip_all, name = "synckit::blob_confirm_upload")]
460 pub(super) async fn blob_confirm_upload(
461 State(db): State<PgPool>,
462 State(storage): State<crate::AppStorage>,
463 sync_user: SyncUser,
464 Json(req): Json<BlobConfirmRequest>,
465 ) -> Result<Response> {
466 let synckit_s3 = storage.synckit_s3.as_ref().ok_or_else(|| {
467 AppError::ServiceUnavailable("SyncKit blob storage is not configured".to_string())
468 })?;
469
470 validation::validate_sync_blob_hash(&req.hash)?;
471
472 let billing = synckit_billing::get_app_with_billing(&db, sync_user.app_id)
473 .await?
474 .ok_or(AppError::NotFound)?;
475
476 let s3_key = crate::storage::S3Client::generate_synckit_blob_key(
477 sync_user.app_id,
478 sync_user.user_id,
479 &req.hash,
480 );
481
482 // The authoritative size is the actual S3 object, never the client's
483 // claim. `object_size` doubles as the existence check (None = not there).
484 let actual_size = synckit_s3.object_size(&s3_key).await?.ok_or_else(|| {
485 AppError::BadRequest("Blob not found in storage, upload before confirming".to_string())
486 })?;
487 // Bounded by the multipart ceiling, not the one-shot one: confirm cannot
488 // tell which transport wrote the object, and the one-shot route is already
489 // bounded to its own ceiling by the signed `Content-Length` at presign time.
490 if actual_size <= 0 || actual_size > constants::SYNCKIT_MAX_MULTIPART_BLOB_SIZE_BYTES {
491 return Err(AppError::BadRequest(format!(
492 "Stored blob size {actual_size} is out of range"
493 )));
494 }
495
496 // Record the blob and enforce the cap atomically. Internal apps gate on the
497 // user's paid subscription + per-user cap; developer apps on the app/per-key
498 // counters. Both are single-transaction (lock, check, insert, count).
499 let outcome = if billing.is_internal {
500 db::synckit::confirm_internal_blob(
501 &db,
502 sync_user.app_id,
503 sync_user.user_id,
504 &req.hash,
505 actual_size,
506 &s3_key,
507 &sync_user.key,
508 )
509 .await?
510 } else {
511 if billing.billing_status != crate::db::SyncBillingStatus::Active {
512 return Ok((
513 StatusCode::PAYMENT_REQUIRED,
514 Json(json!({ "reason": "billing_inactive" })),
515 )
516 .into_response());
517 }
518 db::synckit::confirm_developer_blob(
519 &db,
520 sync_user.app_id,
521 sync_user.user_id,
522 &req.hash,
523 actual_size,
524 &s3_key,
525 &sync_user.key,
526 billing.enforcement_mode,
527 billing.storage_gb_cap,
528 billing.key_cap,
529 billing.gb_per_key,
530 )
531 .await?
532 };
533
534 match outcome {
535 db::synckit::BlobConfirm::Stored | db::synckit::BlobConfirm::AlreadyStored => {
536 // Only now that the blob is durably recorded do we drop the pending
537 // row. Every refusal path below leaves it in place so the orphan reaper
538 // (cleanup_orphaned_uploads) reclaims the unreferenced object; clearing
539 // it before the quota gate stranded the object permanently and uncharged.
540 db::pending_uploads::remove_pending_upload(&db, sync_user.user_id, &s3_key, "synckit")
541 .await?;
542 Ok(StatusCode::NO_CONTENT.into_response())
543 }
544 db::synckit::BlobConfirm::NoSubscription => Ok((
545 StatusCode::PAYMENT_REQUIRED,
546 Json(json!({ "reason": "no_subscription" })),
547 )
548 .into_response()),
549 db::synckit::BlobConfirm::QuotaExceeded {
550 dimension,
551 used,
552 limit,
553 key,
554 } => {
555 let mut body = json!({
556 "reason": "storage_limit_reached",
557 "dimension": dimension,
558 "used": used,
559 "limit": limit,
560 });
561 if let Some(k) = key {
562 body["key"] = json!(k);
563 }
564 Ok((StatusCode::PAYMENT_REQUIRED, Json(body)).into_response())
565 }
566 }
567 }
568
569 /// Request a pre-signed S3 download URL for a blob by hash.
570 #[utoipa::path(post, path = "/api/v1/sync/blobs/download", tag = "SyncKit",
571 request_body = BlobDownloadUrlRequest,
572 responses((status = 200, description = "Pre-signed download URL", body = BlobDownloadUrlResponse), (status = 404, description = "Blob not found")),
573 security(("bearer" = [])),
574 )]
575 #[tracing::instrument(skip_all, name = "synckit::blob_download_url")]
576 pub(super) async fn blob_download_url(
577 State(db): State<PgPool>,
578 State(storage): State<crate::AppStorage>,
579 State(bg): State<crate::background::BackgroundTx>,
580 sync_user: SyncUser,
581 Json(req): Json<BlobDownloadUrlRequest>,
582 ) -> Result<Response> {
583 let synckit_s3 = storage.synckit_s3.as_ref().ok_or_else(|| {
584 AppError::ServiceUnavailable("SyncKit blob storage is not configured".to_string())
585 })?;
586
587 validation::validate_sync_blob_hash(&req.hash)?;
588
589 let blob =
590 db::synckit::get_sync_blob_by_hash(&db, sync_user.app_id, sync_user.user_id, &req.hash)
591 .await?
592 .ok_or(AppError::NotFound)?;
593
594 // Billing check (internal apps bypass). Egress is NOT enforced, it's a
595 // free metric for the developer's dashboard, absorbed in the storage rate
596 // margin. We still count it at presign time so devs see the stat.
597 let billing = synckit_billing::get_app_with_billing(&db, sync_user.app_id)
598 .await?
599 .ok_or(AppError::NotFound)?;
600 if !billing.is_internal {
601 if billing.billing_status != crate::db::SyncBillingStatus::Active {
602 return Ok((
603 StatusCode::PAYMENT_REQUIRED,
604 Json(json!({ "reason": "billing_inactive" })),
605 )
606 .into_response());
607 }
608 // Count egress optimistically at presign time. The client may not
609 // actually download (retries that hit dedup-cached content, for
610 // example), so this overcounts slightly. Acceptable for a free
611 // dashboard metric. Deferred onto the bounded background pool: it's a
612 // single hot-row UPDATE that the download path shouldn't wait on or
613 // contend its lock against (Perf P4).
614 let db = db.clone();
615 let app_id = sync_user.app_id;
616 let egress_bytes = blob.size_bytes;
617 bg.spawn("synckit-egress-bump", async move {
618 if let Err(e) = synckit_billing::add_bytes_egress(&db, app_id, egress_bytes).await {
619 tracing::error!(error = ?e, app_id = %app_id, "failed to bump bytes_egress_period");
620 }
621 });
622 }
623
624 let download_url = synckit_s3
625 .presign_download(
626 &crate::storage::S3Key::from_stored(&blob.s3_key),
627 Some(constants::SYNCKIT_BLOB_PRESIGN_EXPIRY_SECS),
628 )
629 .await
630 .context("presign download for sync blob")?;
631
632 Ok(Json(BlobDownloadUrlResponse { download_url }).into_response())
633 }
634
635 /// Delete a blob by hash for the authenticated user.
636 ///
637 /// Frees storage immediately: the row, the S3 object, and the usage counters
638 /// are released in one atomic step (see `db::synckit::delete_sync_blob`). This
639 /// is the live shrink path that the weekly drift job used to be the only source
640 /// of. Allowed regardless of billing status, a user must always be able to
641 /// reclaim space, even on a lapsed subscription. Idempotent: deleting a hash
642 /// that isn't stored returns 204.
643 #[utoipa::path(delete, path = "/api/v1/sync/blobs/{hash}", tag = "SyncKit",
644 params(("hash" = String, Path, description = "Content hash of the blob to delete")),
645 responses((status = 204, description = "Blob deleted (or already absent)")),
646 security(("bearer" = [])),
647 )]
648 #[tracing::instrument(skip_all, name = "synckit::blob_delete")]
649 pub(super) async fn blob_delete(
650 State(db): State<PgPool>,
651 sync_user: SyncUser,
652 axum::extract::Path(hash): axum::extract::Path<String>,
653 ) -> Result<Response> {
654 validation::validate_sync_blob_hash(&hash)?;
655 db::synckit::delete_sync_blob(&db, sync_user.app_id, sync_user.user_id, &hash).await?;
656 Ok(StatusCode::NO_CONTENT.into_response())
657 }
658
659 #[cfg(test)]
660 mod tests {
661 //! The unsubscribed-user response. Both blob transports return it, and a
662 //! client distinguishes "pay us" from "you are broken" by the status alone,
663 //! so the code and the reason string are a wire contract.
664
665 use super::*;
666
667 #[test]
668 fn an_unsubscribed_user_gets_402_and_not_403() {
669 let resp = no_subscription_response();
670 assert_eq!(
671 resp.status(),
672 StatusCode::PAYMENT_REQUIRED,
673 "402 tells the client to offer a subscription; 403 tells it to give up"
674 );
675 }
676 }
677