Skip to main content

max / makenotwork

16.7 KB · 427 lines History Blame Raw
1 //! Presigned upload and confirm handlers for item content.
2
3 use axum::{Json, extract::State, response::IntoResponse};
4 use serde::Deserialize;
5 use sqlx::PgPool;
6 use std::str::FromStr;
7
8 use crate::{
9 AppStorage, Scanning,
10 auth::AuthUser,
11 db::{self, ItemId},
12 error::{AppError, Result, ResultExt},
13 storage::{CACHE_CONTROL_IMMUTABLE, FileType, S3Client},
14 };
15
16 use super::{CommitTarget, ConfirmUploadResponse, PresignUploadResponse, commit_upload};
17
18 /// JSON input for requesting a presigned S3 upload URL.
19 ///
20 /// `file_size_bytes` is optional for compatibility with older clients that
21 /// don't know the size ahead of time, but when supplied it is signed into
22 /// the presigned URL's `Content-Length` and S3 will reject any PUT whose
23 /// actual body length differs, protocol-level enforcement of the per-file
24 /// cap, no bandwidth wasted on oversized uploads that the confirm step
25 /// would have rejected anyway.
26 #[derive(Debug, Deserialize)]
27 pub(super) struct PresignUploadRequest {
28 pub item_id: ItemId,
29 pub file_type: String,
30 pub file_name: String,
31 pub content_type: String,
32 #[serde(default)]
33 pub file_size_bytes: Option<i64>,
34 }
35
36 /// JSON input for confirming a completed S3 upload.
37 #[derive(Debug, Deserialize)]
38 pub(super) struct ConfirmUploadRequest {
39 pub item_id: ItemId,
40 pub file_type: String,
41 pub s3_key: String,
42 }
43
44 /// Generate a presigned URL for uploading a file to S3
45 ///
46 /// POST /api/upload/presign
47 ///
48 /// Requires authentication. User must own the item.
49 #[tracing::instrument(skip_all, name = "storage::presign_upload", fields(user_id = %user.id))]
50 pub(super) async fn presign_upload(
51 State(db): State<PgPool>,
52 State(storage): State<AppStorage>,
53 AuthUser(user): AuthUser,
54 Json(req): Json<PresignUploadRequest>,
55 ) -> Result<impl IntoResponse> {
56 user.check_not_suspended()?;
57 // Check if S3 is configured
58 let s3 = storage.require_s3()?;
59
60 // Parse file type
61 let file_type = FileType::from_str(&req.file_type)
62 .map_err(|_| AppError::BadRequest(format!("Invalid file type: {}", req.file_type)))?;
63
64 S3Client::validate_content_type(file_type, &req.content_type)?;
65
66 S3Client::validate_extension(file_type, &req.file_name)?;
67
68 // Verify user owns the item
69 let owner = db::items::get_item_owner(&db, req.item_id)
70 .await?
71 .ok_or(AppError::NotFound)?;
72
73 if owner != user.id {
74 return Err(AppError::Forbidden);
75 }
76
77 // Early quota check (reject before generating presigned URL)
78 db::creator_tiers::check_presign_allowed(&db, user.id, file_type).await?;
79
80 // Get the effective max file size for client-side pre-validation
81 let max_file_bytes =
82 db::creator_tiers::get_effective_max_file_bytes(&db, user.id, file_type).await?;
83
84 // If the client declared the file size, validate it before signing, both
85 // against the static per-type cap and the tier-effective cap. The size is
86 // bound as Content-Length below so S3 rejects oversized PUTs at the protocol
87 // level. Clients omitting `file_size_bytes` fall back to the old behavior
88 // (no protocol-level enforcement; confirm step still validates).
89 super::validate_declared_upload_size(req.file_size_bytes, file_type, max_file_bytes)?;
90
91 // Presign to an unserved staging key, never the served key. After a Clean
92 // scan the worker copies the object to its content-addressed served key and
93 // deletes this staging object, so a re-PUT to this presigned URL after the
94 // scan can't change the bytes a buyer is served (C1).
95 let s3_key = S3Client::generate_staging_key(&req.file_name);
96
97 // Track the pending upload so the reaper can clean it up if never confirmed
98 db::pending_uploads::record_pending_upload(&db, user.id, &s3_key, "main").await?;
99
100 // Generate presigned upload URL with immutable cache headers
101 let expires_in = 3600; // 1 hour
102 let upload_url = s3
103 .presign_upload(
104 &s3_key,
105 &req.content_type,
106 Some(expires_in),
107 Some(CACHE_CONTROL_IMMUTABLE),
108 req.file_size_bytes,
109 )
110 .await
111 .context("presign upload for item content")?;
112
113 Ok(Json(PresignUploadResponse {
114 upload_url,
115 s3_key: s3_key.into_string(),
116 expires_in,
117 cache_control: Some(CACHE_CONTROL_IMMUTABLE.to_string()),
118 max_file_bytes,
119 }))
120 }
121
122 /// Confirm that an upload has completed and update the database
123 ///
124 /// POST /api/upload/confirm
125 ///
126 /// Requires authentication. User must own the item.
127 #[tracing::instrument(skip_all, name = "storage::confirm_upload", fields(user_id = %user.id))]
128 pub(super) async fn confirm_upload(
129 State(db): State<PgPool>,
130 State(storage): State<AppStorage>,
131 State(scanning): State<Scanning>,
132 AuthUser(user): AuthUser,
133 Json(req): Json<ConfirmUploadRequest>,
134 ) -> Result<impl IntoResponse> {
135 user.check_not_suspended()?;
136 // Check if S3 is configured
137 let s3 = storage.require_s3()?;
138
139 // Parse file type
140 let file_type = FileType::from_str(&req.file_type)
141 .map_err(|_| AppError::BadRequest(format!("Invalid file type: {}", req.file_type)))?;
142
143 // Verify user owns the item
144 let owner = db::items::get_item_owner(&db, req.item_id)
145 .await?
146 .ok_or(AppError::NotFound)?;
147
148 if owner != user.id {
149 return Err(AppError::Forbidden);
150 }
151
152 // Ownership of the staging key is proved below (after the idempotent
153 // re-confirm short-circuit) via `pending_uploads`, since a `staging/{uuid}`
154 // key carries no user/item in its path for a prefix check to bind.
155
156 // Verify the object exists in S3
157 if !s3.object_exists(&req.s3_key).await? {
158 return Err(AppError::BadRequest(
159 "Upload not found. Please try uploading again.".to_string(),
160 ));
161 }
162
163 // Enforce file size limit (static per-type limit)
164 let file_size_bytes = s3.object_size(&req.s3_key).await?.ok_or_else(|| {
165 AppError::BadRequest(
166 "Could not determine file size. Please try uploading again.".to_string(),
167 )
168 })?;
169 if file_size_bytes as u64 > file_type.max_size() {
170 super::enqueue_s3_orphan(
171 &db,
172 &req.s3_key,
173 crate::storage::S3Bucket::Main,
174 "item_upload_rejected",
175 )
176 .await;
177 let limit_mb = file_type.max_size() / (1024 * 1024);
178 let file_mb = file_size_bytes as u64 / (1024 * 1024);
179 return Err(AppError::FileTooLarge(format!(
180 "File is {} MB but the maximum for {} files is {} MB.",
181 file_mb,
182 file_type.as_str(),
183 limit_mb
184 )));
185 }
186
187 // Enforce tier-based limits (per-file + storage cap)
188 let max_storage =
189 match db::creator_tiers::check_upload_allowed(&db, user.id, file_type, file_size_bytes)
190 .await
191 {
192 Ok(max) => max,
193 Err(e) => {
194 super::enqueue_s3_orphan(
195 &db,
196 &req.s3_key,
197 crate::storage::S3Bucket::Main,
198 "item_upload_rejected",
199 )
200 .await;
201 return Err(e);
202 }
203 };
204
205 // Resolve, from the single exhaustive declaration on `FileType`, how this
206 // type is confirmed on an `items` row, and reject types that belong to a
207 // dedicated route BEFORE any scan enqueue or scan_status flip. (A misrouted
208 // but valid item_id would otherwise flip scan_status to Pending, block every
209 // fan's download, and leak a scan_jobs row for an S3 key we're about to
210 // delete.) Cover is rejected here now: it also needs `cover_image_url`,
211 // which only /api/items/image/confirm writes, the old generic two-column
212 // path left it NULL and rendered an invisible cover (Run #13 SERIOUS).
213 if let crate::storage::GenericItemConfirm::UseRoute(route) = file_type.generic_item_confirm() {
214 super::enqueue_s3_orphan(
215 &db,
216 &req.s3_key,
217 crate::storage::S3Bucket::Main,
218 "item_upload_rejected",
219 )
220 .await;
221 return Err(AppError::BadRequest(format!(
222 "This file type isn't confirmed here. Use {route}."
223 )));
224 }
225
226 // Idempotency + replace detection. We read the item ONCE here and reuse its
227 // project_id for the cache bump below (previously a second get_item_by_id).
228 let mut old_s3_key: Option<String> = None;
229 let mut replaced_old_size: i64 = 0;
230 let mut item_project_id: Option<db::ProjectId> = None;
231 if let Some(item) = db::items::get_item_by_id(&db, req.item_id).await? {
232 item_project_id = Some(item.project_id);
233 let existing_key = match file_type {
234 FileType::Audio => item.audio_s3_key.as_deref(),
235 FileType::Cover => item.cover_s3_key.as_deref(),
236 FileType::Video => item.video_s3_key.as_deref(),
237 _ => None,
238 };
239 if existing_key == Some(&req.s3_key) {
240 // Idempotent re-confirm: the entity already references this s3_key.
241 // Still clear the pending_uploads row, otherwise the orphan reaper
242 // will fire 24h later and delete the live S3 object out from under
243 // a perfectly happy DB row (Run #7 HIGH-1).
244 if let Err(e) =
245 db::pending_uploads::remove_pending_upload(&db, user.id, &req.s3_key, "main").await
246 {
247 tracing::warn!(error = ?e, key = %req.s3_key, "remove_pending_upload failed on idempotent re-confirm");
248 }
249 return Ok(Json(ConfirmUploadResponse {
250 success: true,
251 pending_review: None,
252 }));
253 }
254 if let Some(old_key) = existing_key {
255 old_s3_key = Some(old_key.to_string());
256 replaced_old_size = match file_type {
257 FileType::Audio => item.audio_file_size_bytes.unwrap_or(0),
258 FileType::Cover => item.cover_file_size_bytes.unwrap_or(0),
259 FileType::Video => item.video_file_size_bytes.unwrap_or(0),
260 _ => 0,
261 };
262 }
263 }
264 let is_replace = old_s3_key.is_some();
265
266 // Authorize the staging key for a *fresh* confirm: the caller must have
267 // presigned it (recorded against them in `pending_uploads`). The idempotent
268 // re-confirm above already returned, it consumed the pending row on the
269 // first confirm and is authorized by the item ownership check + the entity
270 // already referencing the key, so this gate only guards new writes. A
271 // `staging/{uuid}` key has no owner in its path, so this lookup (not a prefix
272 // check) is what prevents confirming another user's staging object.
273 // Do NOT enqueue this key for orphan deletion on failure: an unowned staging
274 // key is (at most) another user's in-flight upload, and the deletion queue
275 // would treat the not-yet-referenced staging object as dead and delete it,
276 // a cross-user griefing delete. Just reject; the real owner's confirm or the
277 // pending-upload reaper handles their key.
278 if !db::pending_uploads::is_owned(&db, user.id, &req.s3_key, "main").await? {
279 return Err(AppError::BadRequest("Invalid upload key".to_string()));
280 }
281
282 // Storage credit + item UPDATE in ONE transaction. A rollback restores both,
283 // so a mid-write failure can't leave the storage counter inflated against a
284 // row that never got the key (the previous compensating-action path with
285 // swallowed `.ok()` errors). The write goes through the sealed
286 // `update_item_file_cas`, whose `IS NOT DISTINCT FROM old_key` guard makes a
287 // concurrent double-confirm lose the race (`LostRace`) instead of
288 // double-crediting storage or clobbering the live object. `commit_upload`
289 // (scan enqueue + scan_status flip) stays AFTER the commit, that ordering
290 // is the blessed path; see `routes/storage/mod.rs::commit_upload`. `Ok(0)`
291 // signals the item vanished or lost the CAS race (tx rolled back, nothing
292 // charged).
293 let tx_result: Result<u64> = async {
294 let mut tx = db.begin().await?;
295 db::creator_tiers::try_apply_storage_on(
296 &mut tx,
297 user.id,
298 is_replace.then_some(replaced_old_size),
299 file_size_bytes,
300 max_storage,
301 )
302 .await?;
303 match db::items::update_item_file_cas(
304 &mut *tx,
305 req.item_id,
306 user.id,
307 file_type,
308 old_s3_key.as_deref(),
309 &req.s3_key,
310 file_size_bytes,
311 )
312 .await?
313 {
314 db::items::FileConfirmOutcome::LostRace => {
315 // Leave the tx uncommitted, drop rolls it back, undoing the
316 // storage change with no manual compensation.
317 Ok(0)
318 }
319 db::items::FileConfirmOutcome::Committed => {
320 // Enqueue the OLD object for deletion in the SAME tx as the row
321 // swap, closing the crash-between-commit-and-enqueue orphan
322 // window (ultra-fuzz Run #1 Storage LOW; mirrors the project-
323 // image replace path in images.rs). After commit the row points
324 // at the new key, so the old key is non-live; the worker's
325 // is_s3_key_live check is the backstop if a row still references
326 // it.
327 if let Some(old_key) = old_s3_key.as_deref() {
328 db::pending_s3_deletions::enqueue_deletions(
329 &mut *tx,
330 &[(old_key.to_string(), "main".to_string())],
331 "item_upload_replace",
332 )
333 .await?;
334 }
335 tx.commit().await?;
336 Ok(1)
337 }
338 }
339 }
340 .await;
341
342 match tx_result {
343 Err(e) => {
344 // tx rolled back, storage counter unchanged. A concurrent
345 // double-confirm of this same key could have committed it onto the
346 // item row before our `try_apply_storage_on` errored (storage cap
347 // filling in between), so a blind delete could destroy the live
348 // object the winner points at. Route through the orphan queue; its
349 // `is_s3_key_live` check skips any key a row still references.
350 super::enqueue_s3_orphan(
351 &db,
352 &req.s3_key,
353 crate::storage::S3Bucket::Main,
354 "item_confirm_failed",
355 )
356 .await;
357 return Err(e);
358 }
359 Ok(0) => {
360 // CAS matched zero rows: the item was deleted/transferred out from
361 // under the ownership filter, OR a concurrent confirm won the race
362 // and swapped the target column (so it no longer holds the key we
363 // observed). Either way the tx rolled back, nothing was charged;
364 // route the now-unreferenced object through the orphan queue so the
365 // reaper still cleans it.
366 super::enqueue_s3_orphan(
367 &db,
368 &req.s3_key,
369 crate::storage::S3Bucket::Main,
370 "item_upload_target_missing",
371 )
372 .await;
373 return Err(AppError::BadRequest(
374 "Item was modified concurrently. Please try uploading again.".to_string(),
375 ));
376 }
377 Ok(_) => {}
378 }
379
380 // Scan enqueue + scan_status flip happens AFTER the DB UPDATE commits via
381 // the shared `commit_upload` helper, which is the only blessed path for
382 // this ordering. See `routes/storage/mod.rs::commit_upload` for the bug
383 // shapes this prevents.
384 let scan_status = commit_upload(
385 &db,
386 scanning.scanner.as_ref(),
387 CommitTarget::Item(req.item_id),
388 &req.s3_key,
389 file_type,
390 user.id,
391 file_size_bytes,
392 )
393 .await?;
394
395 // (The old S3 object on a replace was enqueued for deletion inside the
396 // confirm tx above, so a crash here can't orphan it.)
397
398 // Clear the pending upload record now that the upload is confirmed
399 db::pending_uploads::remove_pending_upload(&db, user.id, &req.s3_key, "main").await?;
400
401 // Bump project cache generation so dashboard tabs reflect the new upload.
402 // Reuses the project_id read during idempotency above (no second fetch).
403 if let Some(project_id) = item_project_id
404 && let Err(e) = db::projects::bump_cache_generation(&db, project_id).await
405 {
406 tracing::warn!(%project_id, error = ?e, "failed to bump cache generation after upload");
407 }
408
409 tracing::info!(
410 "Upload confirmed: item={}, type={:?}, key={}, size={}",
411 req.item_id,
412 file_type,
413 req.s3_key,
414 file_size_bytes
415 );
416
417 let pending_review = if scan_status == db::FileScanStatus::HeldForReview {
418 Some(true)
419 } else {
420 None
421 };
422 Ok(Json(ConfirmUploadResponse {
423 success: true,
424 pending_review,
425 }))
426 }
427