Skip to main content

max / makenotwork

23.0 KB · 658 lines History Blame Raw
1 //! Presigned upload and confirm handlers for project and item images.
2
3 use axum::{Json, extract::State, response::IntoResponse};
4 use serde::{Deserialize, Serialize};
5 use sqlx::PgPool;
6
7 use crate::{
8 AppStorage, Scanning,
9 auth::AuthUser,
10 config::Config,
11 db::{self, ItemId, ProjectId},
12 error::{AppError, Result, ResultExt},
13 storage::{self, CACHE_CONTROL_IMMUTABLE, FileType, S3Client},
14 };
15
16 use super::{CommitTarget, PresignUploadResponse, commit_upload};
17
18 /// JSON input for requesting a presigned project image upload URL.
19 #[derive(Debug, Deserialize)]
20 pub(super) struct ProjectImagePresignRequest {
21 pub project_id: ProjectId,
22 pub file_name: String,
23 pub content_type: String,
24 }
25
26 /// JSON input for confirming a completed project image upload.
27 #[derive(Debug, Deserialize)]
28 pub(super) struct ProjectImageConfirmRequest {
29 pub project_id: ProjectId,
30 pub s3_key: String,
31 }
32
33 /// JSON response from a successful project image confirm.
34 #[derive(Debug, Serialize)]
35 pub(super) struct ProjectImageConfirmResponse {
36 pub success: bool,
37 pub image_url: String,
38 }
39
40 /// JSON input for requesting a presigned item image upload URL.
41 #[derive(Debug, Deserialize)]
42 pub(super) struct ItemImagePresignRequest {
43 pub item_id: ItemId,
44 pub file_name: String,
45 pub content_type: String,
46 }
47
48 /// JSON input for confirming a completed item image upload.
49 #[derive(Debug, Deserialize)]
50 pub(super) struct ItemImageConfirmRequest {
51 pub item_id: ItemId,
52 pub s3_key: String,
53 }
54
55 /// Generate a presigned URL for uploading a project image
56 ///
57 /// POST /api/projects/image/presign
58 ///
59 /// Requires authentication. User must own the project.
60 #[tracing::instrument(skip_all, name = "storage::project_image_presign", fields(user_id = %user.id))]
61 pub(super) async fn project_image_presign(
62 State(db): State<PgPool>,
63 State(storage): State<AppStorage>,
64 AuthUser(user): AuthUser,
65 Json(req): Json<ProjectImagePresignRequest>,
66 ) -> Result<impl IntoResponse> {
67 user.check_not_suspended()?;
68 let s3 = storage.require_s3()?;
69
70 let file_type = FileType::Cover;
71 S3Client::validate_content_type(file_type, &req.content_type)?;
72 S3Client::validate_extension(file_type, &req.file_name)?;
73
74 // Verify user owns the project
75 let project = db::projects::get_project_by_id(&db, req.project_id)
76 .await?
77 .ok_or(AppError::NotFound)?;
78
79 if project.user_id != user.id {
80 return Err(AppError::Forbidden);
81 }
82
83 // Early quota check
84 db::creator_tiers::check_presign_allowed(&db, user.id, file_type).await?;
85
86 // Staging key (unserved); the scan worker promotes it to the content key and
87 // rebuilds the public `cover_image_url` on a Clean verdict (C1).
88 let s3_key = S3Client::generate_staging_key(&req.file_name);
89
90 // Track the pending upload so the reaper can clean it up if never confirmed
91 db::pending_uploads::record_pending_upload(&db, user.id, &s3_key, "main").await?;
92
93 let expires_in = 3600;
94 let upload_url = s3
95 .presign_upload(
96 &s3_key,
97 &req.content_type,
98 Some(expires_in),
99 Some(CACHE_CONTROL_IMMUTABLE),
100 None,
101 )
102 .await
103 .context("presign upload for project image")?;
104
105 Ok(Json(PresignUploadResponse {
106 upload_url,
107 s3_key: s3_key.into_string(),
108 expires_in,
109 cache_control: Some(CACHE_CONTROL_IMMUTABLE.to_string()),
110 max_file_bytes: None,
111 }))
112 }
113
114 /// Confirm a project image upload, scan, store URL
115 ///
116 /// POST /api/projects/image/confirm
117 ///
118 /// Requires authentication. User must own the project.
119 #[tracing::instrument(skip_all, name = "storage::project_image_confirm", fields(user_id = %user.id))]
120 pub(super) async fn project_image_confirm(
121 State(db): State<PgPool>,
122 State(storage): State<AppStorage>,
123 State(config): State<Config>,
124 State(scanning): State<Scanning>,
125 AuthUser(user): AuthUser,
126 Json(req): Json<ProjectImageConfirmRequest>,
127 ) -> Result<impl IntoResponse> {
128 user.check_not_suspended()?;
129 let s3 = storage.require_s3()?;
130
131 // Verify user owns the project
132 let project = db::projects::get_project_by_id(&db, req.project_id)
133 .await?
134 .ok_or(AppError::NotFound)?;
135
136 if project.user_id != user.id {
137 return Err(AppError::Forbidden);
138 }
139
140 // Authorize the staging key: a `staging/{uuid}` key has no project in its
141 // path, so ownership is proved via the `pending_uploads` row recorded at
142 // presign, not a prefix check. Gate before the size-reject path so an
143 // unowned (at most another user's in-flight) staging object is never enqueued
144 // for deletion.
145 if !db::pending_uploads::is_owned(&db, user.id, &req.s3_key, "main").await? {
146 return Err(AppError::BadRequest("Invalid upload key".to_string()));
147 }
148
149 // A single HEAD: `object_size` returns None when the object isn't there, so
150 // it doubles as the existence check (no separate object_exists round-trip).
151 let file_size_bytes = s3.object_size(&req.s3_key).await?.ok_or_else(|| {
152 AppError::BadRequest("Upload not found. Please try uploading again.".to_string())
153 })?;
154 if file_size_bytes as u64 > FileType::Cover.max_size() {
155 super::enqueue_s3_orphan(
156 &db,
157 &req.s3_key,
158 crate::storage::S3Bucket::Main,
159 "image_upload_rejected",
160 )
161 .await;
162 return Err(AppError::BadRequest(format!(
163 "File exceeds maximum size of {} MB",
164 FileType::Cover.max_size() / (1024 * 1024)
165 )));
166 }
167
168 // Idempotency: if the project already references this same s3_key, return success.
169 // The Run #6 audit caught a silent-data-loss bug here: without this check, a benign
170 // retry would queue `req.s3_key` (== current `cover_image_url`) for deletion.
171 if let Some(ref cur_url) = project.cover_image_url
172 && let Some(cur_key) = storage::extract_s3_key_from_url(
173 cur_url,
174 &config.cdn_base_url,
175 Some(s3.bucket()),
176 config.storage.as_ref().map(|c| c.endpoint.as_str()),
177 )
178 && cur_key == req.s3_key
179 {
180 // Still clear pending_uploads, orphan reaper would otherwise delete
181 // the live S3 object 24h later (Run #7 HIGH-1).
182 if let Err(e) =
183 db::pending_uploads::remove_pending_upload(&db, user.id, &req.s3_key, "main").await
184 {
185 tracing::warn!(error = ?e, key = %req.s3_key, "remove_pending_upload failed on idempotent re-confirm");
186 }
187 return Ok(Json(ProjectImageConfirmResponse {
188 success: true,
189 image_url: cur_url.clone(),
190 }));
191 }
192
193 // Enforce tier-based limits
194 let max_storage = match db::creator_tiers::check_upload_allowed(
195 &db,
196 user.id,
197 FileType::Cover,
198 file_size_bytes,
199 )
200 .await
201 {
202 Ok(max) => max,
203 Err(e) => {
204 super::enqueue_s3_orphan(
205 &db,
206 &req.s3_key,
207 crate::storage::S3Bucket::Main,
208 "image_upload_rejected",
209 )
210 .await;
211 return Err(e);
212 }
213 };
214
215 // Probe the old S3 object's size FIRST (async, before any tx). If the row
216 // references an old image we MUST determine its size or the storage counter
217 // drifts on every replacement; treating Err/Ok(None) as "no old image" would
218 // silently over-count. This probe stays outside the transaction so we never
219 // hold a DB connection across an S3 round-trip.
220 let mut replace_old_size: Option<i64> = None;
221 let old_key_to_delete: Option<String> = if let Some(ref old_url) = project.cover_image_url
222 && let Some(old_key) = storage::extract_s3_key_from_url(
223 old_url,
224 &config.cdn_base_url,
225 Some(s3.bucket()),
226 config.storage.as_ref().map(|c| c.endpoint.as_str()),
227 ) {
228 match s3.object_size(&old_key).await {
229 Ok(Some(old_size)) if old_size > 0 => {
230 replace_old_size = Some(old_size);
231 Some(old_key)
232 }
233 Ok(Some(_) | None) => {
234 // Old URL parsed but the object is gone (or zero-sized): treat as a
235 // fresh upload, nothing to refund. Still queue the old key for
236 // deletion in case of S3 eventual-consistency.
237 Some(old_key)
238 }
239 Err(e) => {
240 // S3 probe failed (transient). Refuse to write, letting a probe failure
241 // silently over-count storage on every replace is the bug this branch exists to prevent.
242 super::enqueue_s3_orphan(
243 &db,
244 &req.s3_key,
245 crate::storage::S3Bucket::Main,
246 "image_upload_rejected",
247 )
248 .await;
249 tracing::warn!(key = %old_key, error = ?e, "S3 probe failed during project image replace");
250 return Err(AppError::ServiceUnavailable(
251 "Could not verify previous image. Please try again.".to_string(),
252 ));
253 }
254 }
255 } else {
256 None
257 };
258
259 let image_url = storage::build_project_image_url(&config.cdn_base_url, &req.s3_key);
260
261 // Storage credit + project image URL UPDATE in ONE transaction. A rollback
262 // restores the counter, so the previous compensating `rollback_and_orphan`
263 // math (with a second S3 probe and swallowed `.ok()`s) is gone. `commit_upload`
264 // stays AFTER the commit. `Ok(false)` = ownership filter no-matched (project
265 // deleted/transferred mid-flight); the tx rolled back, nothing charged.
266 let committed: Result<bool> = async {
267 let mut tx = db.begin().await?;
268 db::creator_tiers::try_apply_storage_on(
269 &mut tx,
270 user.id,
271 replace_old_size,
272 file_size_bytes,
273 max_storage,
274 )
275 .await?;
276 let ok = db::projects::update_project_cover_cas(
277 &mut *tx,
278 req.project_id,
279 user.id,
280 project.cover_image_url.as_deref(),
281 &image_url,
282 &req.s3_key,
283 file_size_bytes,
284 )
285 .await?;
286 if !ok {
287 return Ok(false);
288 }
289 // Enqueue the OLD object for deletion in the SAME tx as the row UPDATE,
290 // closing the crash-between-commit-and-enqueue orphan window (ultra-fuzz
291 // Run #1 Storage LOW). is_s3_key_live guards the worker if a row still
292 // references it.
293 if let Some(old_key) = old_key_to_delete.as_deref() {
294 db::pending_s3_deletions::enqueue_deletions(
295 &mut *tx,
296 // A cover key is a private staging key until promote repoints it to
297 // the public content key, so the old cover may be in EITHER bucket.
298 // Enqueue both; the reaper no-ops the bucket the object isn't in
299 // (content keys are unique to one bucket) and `is_s3_key_live`
300 // still guards each bucket against a live reference.
301 &crate::storage::both_bucket_delete(old_key),
302 "project_image_replace",
303 )
304 .await?;
305 }
306 tx.commit().await?;
307 Ok(true)
308 }
309 .await;
310
311 match committed {
312 Err(e) => {
313 // tx rolled back, counter unchanged. Orphan-queue the new key for cleanup.
314 super::enqueue_s3_orphan(
315 &db,
316 &req.s3_key,
317 crate::storage::S3Bucket::Main,
318 "project_image_update_failed",
319 )
320 .await;
321 return Err(e);
322 }
323 Ok(false) => {
324 super::enqueue_s3_orphan(
325 &db,
326 &req.s3_key,
327 crate::storage::S3Bucket::Main,
328 "project_image_update_failed",
329 )
330 .await;
331 return Err(AppError::BadRequest(
332 "Project was modified concurrently. Please try uploading again.".to_string(),
333 ));
334 }
335 Ok(true) => {}
336 }
337
338 // Clear the pending upload record now that the upload is committed
339 db::pending_uploads::remove_pending_upload(&db, user.id, &req.s3_key, "main").await?;
340
341 // (The old S3 object was enqueued for deletion inside the commit tx above.)
342
343 // Scan enqueue AFTER the DB write commits (Phase 5 chronic fix, the same
344 // ordering rule that uploads/versions/media follow via `commit_upload`).
345 commit_upload(
346 &db,
347 scanning.scanner.as_ref(),
348 CommitTarget::ProjectImage(req.project_id),
349 &req.s3_key,
350 FileType::Cover,
351 user.id,
352 file_size_bytes,
353 )
354 .await?;
355
356 db::projects::bump_cache_generation(&db, req.project_id).await?;
357
358 tracing::info!(
359 "Project image confirmed: project={}, key={}, size={}",
360 req.project_id,
361 req.s3_key,
362 file_size_bytes
363 );
364
365 Ok(Json(ProjectImageConfirmResponse {
366 success: true,
367 image_url,
368 }))
369 }
370
371 /// Generate a presigned URL for uploading an item image (logo/cover)
372 ///
373 /// POST /api/items/image/presign
374 ///
375 /// Requires authentication. User must own the item.
376 #[tracing::instrument(skip_all, name = "storage::item_image_presign", fields(user_id = %user.id))]
377 pub(super) async fn item_image_presign(
378 State(db): State<PgPool>,
379 State(storage): State<AppStorage>,
380 AuthUser(user): AuthUser,
381 Json(req): Json<ItemImagePresignRequest>,
382 ) -> Result<impl IntoResponse> {
383 user.check_not_suspended()?;
384 let s3 = storage.require_s3()?;
385
386 let file_type = FileType::Cover;
387 S3Client::validate_content_type(file_type, &req.content_type)?;
388 S3Client::validate_extension(file_type, &req.file_name)?;
389
390 // Verify user owns the item
391 let owner = db::items::get_item_owner(&db, req.item_id)
392 .await?
393 .ok_or(AppError::NotFound)?;
394
395 if owner != user.id {
396 return Err(AppError::Forbidden);
397 }
398
399 // Early quota check
400 db::creator_tiers::check_presign_allowed(&db, user.id, file_type).await?;
401
402 // Staging key (unserved); the scan worker promotes it to the content key and
403 // rebuilds the public `cover_image_url` on a Clean verdict (C1).
404 let s3_key = S3Client::generate_staging_key(&req.file_name);
405
406 // Track the pending upload so the reaper can clean it up if never confirmed
407 db::pending_uploads::record_pending_upload(&db, user.id, &s3_key, "main").await?;
408
409 let expires_in = 3600;
410 let upload_url = s3
411 .presign_upload(
412 &s3_key,
413 &req.content_type,
414 Some(expires_in),
415 Some(CACHE_CONTROL_IMMUTABLE),
416 None,
417 )
418 .await
419 .context("presign upload for item image")?;
420
421 Ok(Json(PresignUploadResponse {
422 upload_url,
423 s3_key: s3_key.into_string(),
424 expires_in,
425 cache_control: Some(CACHE_CONTROL_IMMUTABLE.to_string()),
426 max_file_bytes: None,
427 }))
428 }
429
430 /// Confirm an item image upload, scan, store URL
431 ///
432 /// POST /api/items/image/confirm
433 ///
434 /// Requires authentication. User must own the item.
435 #[tracing::instrument(skip_all, name = "storage::item_image_confirm", fields(user_id = %user.id))]
436 pub(super) async fn item_image_confirm(
437 State(db): State<PgPool>,
438 State(storage): State<AppStorage>,
439 State(config): State<Config>,
440 State(scanning): State<Scanning>,
441 AuthUser(user): AuthUser,
442 Json(req): Json<ItemImageConfirmRequest>,
443 ) -> Result<impl IntoResponse> {
444 user.check_not_suspended()?;
445 let s3 = storage.require_s3()?;
446
447 // Verify user owns the item
448 let owner = db::items::get_item_owner(&db, req.item_id)
449 .await?
450 .ok_or(AppError::NotFound)?;
451
452 if owner != user.id {
453 return Err(AppError::Forbidden);
454 }
455
456 // Authorize the staging key: a `staging/{uuid}` key has no user/item in its
457 // path (and no `cover/` segment to pin), so ownership, and the fact that it
458 // was minted for THIS user's upload flow, is proved via the `pending_uploads`
459 // row recorded at presign, not a prefix check. Gate before the size-reject
460 // path so an unowned staging object is never enqueued for deletion.
461 if !db::pending_uploads::is_owned(&db, user.id, &req.s3_key, "main").await? {
462 return Err(AppError::BadRequest("Invalid upload key".to_string()));
463 }
464
465 // A single HEAD: `object_size` returns None when the object isn't there, so
466 // it doubles as the existence check (no separate object_exists round-trip).
467 let file_size_bytes = s3.object_size(&req.s3_key).await?.ok_or_else(|| {
468 AppError::BadRequest("Upload not found. Please try uploading again.".to_string())
469 })?;
470 if file_size_bytes as u64 > FileType::Cover.max_size() {
471 super::enqueue_s3_orphan(
472 &db,
473 &req.s3_key,
474 crate::storage::S3Bucket::Main,
475 "image_upload_rejected",
476 )
477 .await;
478 return Err(AppError::BadRequest(format!(
479 "File exceeds maximum size of {} MB",
480 FileType::Cover.max_size() / (1024 * 1024)
481 )));
482 }
483
484 // Idempotency: if cover_s3_key already matches, return success (no-op) BEFORE
485 // the tier/cap check, a creator at their storage cap must still be able to
486 // re-confirm a cover they already own (Run 9; matches project_image_confirm's
487 // ordering). Otherwise capture the existing cover for atomic replacement below.
488 let existing_item = db::items::get_item_by_id(&db, req.item_id).await?;
489 if let Some(ref item) = existing_item
490 && item.cover_s3_key.as_deref() == Some(&req.s3_key)
491 {
492 // Still clear pending_uploads, orphan reaper would otherwise delete
493 // the live S3 object 24h later (Run #7 HIGH-1).
494 if let Err(e) =
495 db::pending_uploads::remove_pending_upload(&db, user.id, &req.s3_key, "main").await
496 {
497 tracing::warn!(error = ?e, key = %req.s3_key, "remove_pending_upload failed on idempotent re-confirm");
498 }
499 return Ok(Json(super::images::ProjectImageConfirmResponse {
500 success: true,
501 image_url: item.cover_image_url.clone().unwrap_or_default(),
502 }));
503 }
504
505 // Enforce tier-based limits
506 let max_storage = match db::creator_tiers::check_upload_allowed(
507 &db,
508 user.id,
509 FileType::Cover,
510 file_size_bytes,
511 )
512 .await
513 {
514 Ok(max) => max,
515 Err(e) => {
516 super::enqueue_s3_orphan(
517 &db,
518 &req.s3_key,
519 crate::storage::S3Bucket::Main,
520 "image_upload_rejected",
521 )
522 .await;
523 return Err(e);
524 }
525 };
526
527 // The key the CAS must still see on the row for this confirm to win, the
528 // raw current cover key, independent of its recorded size (a zero/NULL-size
529 // old cover is still a key the column holds, and the CAS must match it or a
530 // legitimate first replace would be misread as a lost race).
531 let expected_old_key: Option<String> =
532 existing_item.as_ref().and_then(|i| i.cover_s3_key.clone());
533
534 // Old cover key+size come straight from the already-loaded item row (no S3
535 // probe needed). We refund/clean up only when the old object has a real
536 // size; split that decision once into the replace-size and the key to delete.
537 let (replace_old_size, old_key_to_delete): (Option<i64>, Option<String>) =
538 match existing_item.as_ref().and_then(|i| {
539 Some((
540 i.cover_s3_key.clone()?,
541 i.cover_file_size_bytes.unwrap_or(0),
542 ))
543 }) {
544 Some((key, size)) if size > 0 => (Some(size), Some(key)),
545 _ => (None, None),
546 };
547
548 let image_url = storage::build_project_image_url(&config.cdn_base_url, &req.s3_key);
549
550 // Storage credit + item cover UPDATE in ONE transaction (Run #7 HIGH-2 made
551 // these atomic via compensating actions; this makes them atomic via a real
552 // tx, a rollback restores the counter with no swallowed-`.ok()` math).
553 // `commit_upload` stays AFTER the commit. `Ok(false)` = ownership filter
554 // no-matched (item deleted/moved mid-flight); the tx rolled back, nothing charged.
555 let committed: Result<bool> = async {
556 let mut tx = db.begin().await?;
557 db::creator_tiers::try_apply_storage_on(
558 &mut tx,
559 user.id,
560 replace_old_size,
561 file_size_bytes,
562 max_storage,
563 )
564 .await?;
565 let ok = db::items::update_item_cover(
566 &mut *tx,
567 req.item_id,
568 user.id,
569 expected_old_key.as_deref(),
570 &image_url,
571 &req.s3_key,
572 file_size_bytes,
573 )
574 .await?;
575 if !ok {
576 return Ok(false);
577 }
578 // Enqueue the OLD cover for deletion in the SAME tx as the row UPDATE
579 // (ultra-fuzz Run #1 Storage LOW; matches the in-tx ordering the delete
580 // paths use). is_s3_key_live guards the worker against a live reference.
581 if let Some(old_key) = old_key_to_delete.as_deref() {
582 db::pending_s3_deletions::enqueue_deletions(
583 &mut *tx,
584 // Old cover may be a private staging key or a public content key;
585 // enqueue both (see `both_bucket_delete`).
586 &crate::storage::both_bucket_delete(old_key),
587 "item_image_replace",
588 )
589 .await?;
590 }
591 tx.commit().await?;
592 Ok(true)
593 }
594 .await;
595
596 match committed {
597 Err(e) => {
598 super::enqueue_s3_orphan(
599 &db,
600 &req.s3_key,
601 crate::storage::S3Bucket::Main,
602 "item_image_update_failed",
603 )
604 .await;
605 return Err(e);
606 }
607 Ok(false) => {
608 super::enqueue_s3_orphan(
609 &db,
610 &req.s3_key,
611 crate::storage::S3Bucket::Main,
612 "item_image_update_failed",
613 )
614 .await;
615 return Err(AppError::BadRequest(
616 "Item was modified concurrently. Please try uploading again.".to_string(),
617 ));
618 }
619 Ok(true) => {}
620 }
621
622 db::pending_uploads::remove_pending_upload(&db, user.id, &req.s3_key, "main").await?;
623
624 // (The old cover was enqueued for deletion inside the commit tx above.)
625
626 // Scan enqueue + scan_status flip AFTER the DB write, same ordering rule
627 // as uploads/versions/media. The Run #6 audit caught this same bug here.
628 commit_upload(
629 &db,
630 scanning.scanner.as_ref(),
631 CommitTarget::ItemImage(req.item_id),
632 &req.s3_key,
633 FileType::Cover,
634 user.id,
635 file_size_bytes,
636 )
637 .await?;
638
639 // Bump project cache
640 if let Some(item) = db::items::get_item_by_id(&db, req.item_id).await?
641 && let Err(e) = db::projects::bump_cache_generation(&db, item.project_id).await
642 {
643 tracing::warn!(project_id = %item.project_id, error = ?e, "failed to bump cache generation after image upload");
644 }
645
646 tracing::info!(
647 "Item image confirmed: item={}, key={}, size={}",
648 req.item_id,
649 req.s3_key,
650 file_size_bytes
651 );
652
653 Ok(Json(ProjectImageConfirmResponse {
654 success: true,
655 image_url,
656 }))
657 }
658