Skip to main content

max / makenotwork

27.0 KB · 695 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(db): State<PgPool>,
253 State(storage): State<crate::AppStorage>,
254 sync_user: SyncUser,
255 Json(req): Json<BlobMultipartPartsRequest>,
256 ) -> Result<Response> {
257 let synckit_s3 = storage.synckit_s3.as_ref().ok_or_else(|| {
258 AppError::ServiceUnavailable("SyncKit blob storage is not configured".to_string())
259 })?;
260
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 // The same gate `start` runs, and it belongs here too: a client that skips
268 // `start` reaches this route directly, and without the gate an unsubscribed
269 // first-party user was handed presigned part URLs. The gate also validates
270 // the hash, which is why the standalone check above it is gone.
271 match blob_upload_gate(&db, &sync_user, &req.hash).await? {
272 Some(BlobUploadStop::NoSubscription) => return Ok(no_subscription_response()),
273 // Mirrors `start`: the content address is already stored for this
274 // user and app, so there is nothing to upload and no URL to mint.
275 Some(BlobUploadStop::AlreadyExists) => {
276 return Ok(Json(BlobMultipartPartsResponse {
277 parts: Vec::new(),
278 expires_in: 0,
279 })
280 .into_response());
281 }
282 None => {}
283 }
284
285 let plan =
286 s3_storage::MultipartPlan::auto(req.size_bytes as u64).map_err(AppError::BadRequest)?;
287
288 if req.count == 0 || req.count > BLOB_MULTIPART_PART_URL_WINDOW {
289 return Err(AppError::BadRequest(format!(
290 "count must be between 1 and {BLOB_MULTIPART_PART_URL_WINDOW}"
291 )));
292 }
293 if req.first_part == 0 || req.first_part > plan.part_count {
294 return Err(AppError::BadRequest(format!(
295 "first_part must be between 1 and {}",
296 plan.part_count
297 )));
298 }
299
300 // Checksums are positional, so a short or long list would silently bind the
301 // wrong digest to a part, reject it rather than guess the alignment.
302 if let Some(checksums) = &req.checksums {
303 if checksums.len() != req.count as usize {
304 return Err(AppError::BadRequest(format!(
305 "checksums must have exactly {} entries, one per requested part",
306 req.count
307 )));
308 }
309 for c in checksums {
310 validation::validate_sha256_base64(c)?;
311 }
312 }
313
314 let s3_key = crate::storage::S3Client::generate_synckit_blob_key(
315 sync_user.app_id,
316 sync_user.user_id,
317 &req.hash,
318 );
319
320 let expires_in = constants::SYNCKIT_BLOB_PRESIGN_EXPIRY_SECS;
321 let last = (req.first_part + req.count - 1).min(plan.part_count);
322 let mut parts = Vec::with_capacity((last - req.first_part + 1) as usize);
323 for part_number in req.first_part..=last {
324 let content_length = plan.part_len(part_number);
325 let checksum = req
326 .checksums
327 .as_ref()
328 .and_then(|c| c.get((part_number - req.first_part) as usize))
329 .map(String::as_str);
330 let url = synckit_s3
331 .presign_upload_part(
332 &s3_key,
333 &req.upload_id,
334 part_number as i32,
335 Some(expires_in),
336 Some(content_length as i64),
337 checksum,
338 )
339 .await
340 .context("presign upload part for sync blob")?;
341 parts.push(BlobMultipartPartUrl {
342 part_number: part_number as i32,
343 content_length,
344 url,
345 });
346 }
347
348 Ok(Json(BlobMultipartPartsResponse { parts, expires_in }).into_response())
349 }
350
351 /// Assemble the uploaded parts into the blob object.
352 ///
353 /// Transport only: the client then calls `/blobs/confirm`, which reads the real
354 /// object size from S3 and applies every quota and billing rule.
355 #[utoipa::path(post, path = "/api/v1/sync/blobs/multipart/complete", tag = "SyncKit",
356 request_body = BlobMultipartCompleteRequest,
357 responses((status = 204, description = "Parts assembled")),
358 security(("bearer" = [])),
359 )]
360 #[tracing::instrument(skip_all, name = "synckit::blob_multipart_complete")]
361 pub(super) async fn blob_multipart_complete(
362 State(storage): State<crate::AppStorage>,
363 sync_user: SyncUser,
364 Json(req): Json<BlobMultipartCompleteRequest>,
365 ) -> Result<Response> {
366 let synckit_s3 = storage.synckit_s3.as_ref().ok_or_else(|| {
367 AppError::ServiceUnavailable("SyncKit blob storage is not configured".to_string())
368 })?;
369
370 validation::validate_sync_blob_hash(&req.hash)?;
371 if req.parts.is_empty() {
372 return Err(AppError::BadRequest("No parts to complete".to_string()));
373 }
374
375 let s3_key = crate::storage::S3Client::generate_synckit_blob_key(
376 sync_user.app_id,
377 sync_user.user_id,
378 &req.hash,
379 );
380
381 let parts: Vec<(i32, String)> = req
382 .parts
383 .into_iter()
384 .map(|p| (p.part_number, p.etag))
385 .collect();
386
387 synckit_s3
388 .complete_multipart_upload(&s3_key, &req.upload_id, &parts)
389 .await?;
390
391 tracing::info!(
392 app = %sync_user.app_id, user = %sync_user.user_id, parts = parts.len(),
393 "SyncKit multipart blob upload completed"
394 );
395
396 Ok(StatusCode::NO_CONTENT.into_response())
397 }
398
399 /// Release the parts of an abandoned session (client cancel).
400 ///
401 /// Incomplete multipart uploads bill for their parts until aborted, so a client
402 /// that cleans up on cancel is the cheapest fix; the orphan reaper is the
403 /// backstop for clients that vanish.
404 #[utoipa::path(post, path = "/api/v1/sync/blobs/multipart/abort", tag = "SyncKit",
405 request_body = BlobMultipartAbortRequest,
406 responses((status = 204, description = "Session aborted")),
407 security(("bearer" = [])),
408 )]
409 #[tracing::instrument(skip_all, name = "synckit::blob_multipart_abort")]
410 pub(super) async fn blob_multipart_abort(
411 State(db): State<PgPool>,
412 State(storage): State<crate::AppStorage>,
413 sync_user: SyncUser,
414 Json(req): Json<BlobMultipartAbortRequest>,
415 ) -> Result<Response> {
416 let synckit_s3 = storage.synckit_s3.as_ref().ok_or_else(|| {
417 AppError::ServiceUnavailable("SyncKit blob storage is not configured".to_string())
418 })?;
419
420 validation::validate_sync_blob_hash(&req.hash)?;
421
422 let s3_key = crate::storage::S3Client::generate_synckit_blob_key(
423 sync_user.app_id,
424 sync_user.user_id,
425 &req.hash,
426 );
427
428 synckit_s3
429 .abort_multipart_upload(&s3_key, &req.upload_id)
430 .await?;
431 // The session is gone, so the reaper has nothing left to find; drop the
432 // tracking row rather than leaving it to age out.
433 db::pending_uploads::remove_pending_upload(&db, sync_user.user_id, &s3_key, "synckit").await?;
434
435 tracing::info!(
436 app = %sync_user.app_id, user = %sync_user.user_id,
437 "SyncKit multipart blob upload aborted"
438 );
439
440 Ok(StatusCode::NO_CONTENT.into_response())
441 }
442
443 /// Confirm that a blob upload to S3 completed successfully.
444 ///
445 /// Verifies the object exists in S3, then records it in the database.
446 /// Idempotent: returns success without creating a duplicate.
447 ///
448 /// Content-addressing trust model (ultra-fuzz Run 4 Storage NOTE, decision
449 /// 2026-06-23; revised 2026-07-21): the blob `hash` is treated as a
450 /// content-address LABEL, confirm reads the authoritative `object_size` from S3
451 /// but does not re-hash the bytes to prove they match `hash`. The blast radius
452 /// is per-user only: the key is `{app_id}/{user_id}/{hash}` and storage is
453 /// `UNIQUE(app_id, user_id, hash)`, so a client that stores mismatched bytes can
454 /// poison only its OWN dedup namespace, no cross-user effect, no data exposure.
455 ///
456 /// This note used to say the A+ fix was binding `x-amz-checksum-sha256` into the
457 /// presigned PUT so S3 rejects a mismatched upload at write time. That reasoning
458 /// does not hold for these blobs, and the correction is worth keeping: the stored
459 /// object is E2E *ciphertext* sealed with random per-chunk nonces, while `hash`
460 /// is the SHA-256 of the *plaintext*. The server never sees plaintext, so it
461 /// cannot derive the expected ciphertext digest at presign time, any checksum it
462 /// binds has to come from the client, i.e. the party whose honesty was in
463 /// question. Checksum binding (which the multipart path now does per part) buys
464 /// transport integrity, not content-address enforcement.
465 ///
466 /// What actually binds the bytes to the address is the AEAD: each chunk is sealed
467 /// with `(hash, chunk_index, chunk_count)` as associated data, so ciphertext that
468 /// opens under `hash` is cryptographically tied to it, and the client re-hashes
469 /// the plaintext after decrypting. A client storing mismatched bytes breaks only
470 /// its own blob. Server-side re-hashing would cost a full object download per
471 /// confirm to defend a client against itself, which is why it is not done.
472 #[utoipa::path(post, path = "/api/v1/sync/blobs/confirm", tag = "SyncKit",
473 request_body = BlobConfirmRequest,
474 responses((status = 204, description = "Upload confirmed")),
475 security(("bearer" = [])),
476 )]
477 #[tracing::instrument(skip_all, name = "synckit::blob_confirm_upload")]
478 pub(super) async fn blob_confirm_upload(
479 State(db): State<PgPool>,
480 State(storage): State<crate::AppStorage>,
481 sync_user: SyncUser,
482 Json(req): Json<BlobConfirmRequest>,
483 ) -> Result<Response> {
484 let synckit_s3 = storage.synckit_s3.as_ref().ok_or_else(|| {
485 AppError::ServiceUnavailable("SyncKit blob storage is not configured".to_string())
486 })?;
487
488 validation::validate_sync_blob_hash(&req.hash)?;
489
490 let billing = synckit_billing::get_app_with_billing(&db, sync_user.app_id)
491 .await?
492 .ok_or(AppError::NotFound)?;
493
494 let s3_key = crate::storage::S3Client::generate_synckit_blob_key(
495 sync_user.app_id,
496 sync_user.user_id,
497 &req.hash,
498 );
499
500 // The authoritative size is the actual S3 object, never the client's
501 // claim. `object_size` doubles as the existence check (None = not there).
502 let actual_size = synckit_s3.object_size(&s3_key).await?.ok_or_else(|| {
503 AppError::BadRequest("Blob not found in storage, upload before confirming".to_string())
504 })?;
505 // Bounded by the multipart ceiling, not the one-shot one: confirm cannot
506 // tell which transport wrote the object, and the one-shot route is already
507 // bounded to its own ceiling by the signed `Content-Length` at presign time.
508 if actual_size <= 0 || actual_size > constants::SYNCKIT_MAX_MULTIPART_BLOB_SIZE_BYTES {
509 return Err(AppError::BadRequest(format!(
510 "Stored blob size {actual_size} is out of range"
511 )));
512 }
513
514 // Record the blob and enforce the cap atomically. Internal apps gate on the
515 // user's paid subscription + per-user cap; developer apps on the app/per-key
516 // counters. Both are single-transaction (lock, check, insert, count).
517 let outcome = if billing.is_internal {
518 db::synckit::confirm_internal_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 )
527 .await?
528 } else {
529 if billing.billing_status != crate::db::SyncBillingStatus::Active {
530 return Ok((
531 StatusCode::PAYMENT_REQUIRED,
532 Json(json!({ "reason": "billing_inactive" })),
533 )
534 .into_response());
535 }
536 db::synckit::confirm_developer_blob(
537 &db,
538 sync_user.app_id,
539 sync_user.user_id,
540 &req.hash,
541 actual_size,
542 &s3_key,
543 &sync_user.key,
544 billing.enforcement_mode,
545 billing.storage_gb_cap,
546 billing.key_cap,
547 billing.gb_per_key,
548 )
549 .await?
550 };
551
552 match outcome {
553 db::synckit::BlobConfirm::Stored | db::synckit::BlobConfirm::AlreadyStored => {
554 // Only now that the blob is durably recorded do we drop the pending
555 // row. Every refusal path below leaves it in place so the orphan reaper
556 // (cleanup_orphaned_uploads) reclaims the unreferenced object; clearing
557 // it before the quota gate stranded the object permanently and uncharged.
558 db::pending_uploads::remove_pending_upload(&db, sync_user.user_id, &s3_key, "synckit")
559 .await?;
560 Ok(StatusCode::NO_CONTENT.into_response())
561 }
562 db::synckit::BlobConfirm::NoSubscription => Ok((
563 StatusCode::PAYMENT_REQUIRED,
564 Json(json!({ "reason": "no_subscription" })),
565 )
566 .into_response()),
567 db::synckit::BlobConfirm::QuotaExceeded {
568 dimension,
569 used,
570 limit,
571 key,
572 } => {
573 let mut body = json!({
574 "reason": "storage_limit_reached",
575 "dimension": dimension,
576 "used": used,
577 "limit": limit,
578 });
579 if let Some(k) = key {
580 body["key"] = json!(k);
581 }
582 Ok((StatusCode::PAYMENT_REQUIRED, Json(body)).into_response())
583 }
584 }
585 }
586
587 /// Request a pre-signed S3 download URL for a blob by hash.
588 #[utoipa::path(post, path = "/api/v1/sync/blobs/download", tag = "SyncKit",
589 request_body = BlobDownloadUrlRequest,
590 responses((status = 200, description = "Pre-signed download URL", body = BlobDownloadUrlResponse), (status = 404, description = "Blob not found")),
591 security(("bearer" = [])),
592 )]
593 #[tracing::instrument(skip_all, name = "synckit::blob_download_url")]
594 pub(super) async fn blob_download_url(
595 State(db): State<PgPool>,
596 State(storage): State<crate::AppStorage>,
597 State(bg): State<crate::background::BackgroundTx>,
598 sync_user: SyncUser,
599 Json(req): Json<BlobDownloadUrlRequest>,
600 ) -> Result<Response> {
601 let synckit_s3 = storage.synckit_s3.as_ref().ok_or_else(|| {
602 AppError::ServiceUnavailable("SyncKit blob storage is not configured".to_string())
603 })?;
604
605 validation::validate_sync_blob_hash(&req.hash)?;
606
607 let blob =
608 db::synckit::get_sync_blob_by_hash(&db, sync_user.app_id, sync_user.user_id, &req.hash)
609 .await?
610 .ok_or(AppError::NotFound)?;
611
612 // Billing check (internal apps bypass). Egress is NOT enforced, it's a
613 // free metric for the developer's dashboard, absorbed in the storage rate
614 // margin. We still count it at presign time so devs see the stat.
615 let billing = synckit_billing::get_app_with_billing(&db, sync_user.app_id)
616 .await?
617 .ok_or(AppError::NotFound)?;
618 if !billing.is_internal {
619 if billing.billing_status != crate::db::SyncBillingStatus::Active {
620 return Ok((
621 StatusCode::PAYMENT_REQUIRED,
622 Json(json!({ "reason": "billing_inactive" })),
623 )
624 .into_response());
625 }
626 // Count egress optimistically at presign time. The client may not
627 // actually download (retries that hit dedup-cached content, for
628 // example), so this overcounts slightly. Acceptable for a free
629 // dashboard metric. Deferred onto the bounded background pool: it's a
630 // single hot-row UPDATE that the download path shouldn't wait on or
631 // contend its lock against (Perf P4).
632 let db = db.clone();
633 let app_id = sync_user.app_id;
634 let egress_bytes = blob.size_bytes;
635 bg.spawn("synckit-egress-bump", async move {
636 if let Err(e) = synckit_billing::add_bytes_egress(&db, app_id, egress_bytes).await {
637 tracing::error!(error = ?e, app_id = %app_id, "failed to bump bytes_egress_period");
638 }
639 });
640 }
641
642 let download_url = synckit_s3
643 .presign_download(
644 &crate::storage::S3Key::from_stored(&blob.s3_key),
645 Some(constants::SYNCKIT_BLOB_PRESIGN_EXPIRY_SECS),
646 )
647 .await
648 .context("presign download for sync blob")?;
649
650 Ok(Json(BlobDownloadUrlResponse { download_url }).into_response())
651 }
652
653 /// Delete a blob by hash for the authenticated user.
654 ///
655 /// Frees storage immediately: the row, the S3 object, and the usage counters
656 /// are released in one atomic step (see `db::synckit::delete_sync_blob`). This
657 /// is the live shrink path that the weekly drift job used to be the only source
658 /// of. Allowed regardless of billing status, a user must always be able to
659 /// reclaim space, even on a lapsed subscription. Idempotent: deleting a hash
660 /// that isn't stored returns 204.
661 #[utoipa::path(delete, path = "/api/v1/sync/blobs/{hash}", tag = "SyncKit",
662 params(("hash" = String, Path, description = "Content hash of the blob to delete")),
663 responses((status = 204, description = "Blob deleted (or already absent)")),
664 security(("bearer" = [])),
665 )]
666 #[tracing::instrument(skip_all, name = "synckit::blob_delete")]
667 pub(super) async fn blob_delete(
668 State(db): State<PgPool>,
669 sync_user: SyncUser,
670 axum::extract::Path(hash): axum::extract::Path<String>,
671 ) -> Result<Response> {
672 validation::validate_sync_blob_hash(&hash)?;
673 db::synckit::delete_sync_blob(&db, sync_user.app_id, sync_user.user_id, &hash).await?;
674 Ok(StatusCode::NO_CONTENT.into_response())
675 }
676
677 #[cfg(test)]
678 mod tests {
679 //! The unsubscribed-user response. Both blob transports return it, and a
680 //! client distinguishes "pay us" from "you are broken" by the status alone,
681 //! so the code and the reason string are a wire contract.
682
683 use super::*;
684
685 #[test]
686 fn an_unsubscribed_user_gets_402_and_not_403() {
687 let resp = no_subscription_response();
688 assert_eq!(
689 resp.status(),
690 StatusCode::PAYMENT_REQUIRED,
691 "402 tells the client to offer a subscription; 403 tells it to give up"
692 );
693 }
694 }
695