Skip to main content

max / makenotwork

15.4 KB · 384 lines History Blame Raw
1 //! Presigned upload and confirm handlers for item content.
2
3 use axum::{
4 extract::State,
5 response::IntoResponse,
6 Json,
7 };
8 use serde::Deserialize;
9 use std::str::FromStr;
10
11 use crate::{
12 auth::AuthUser,
13 db::{self, ItemId},
14 error::{AppError, Result, ResultExt},
15 storage::{FileType, S3Client, CACHE_CONTROL_IMMUTABLE},
16 AppState,
17 };
18
19 use super::{commit_upload, CommitTarget, ConfirmUploadResponse, PresignUploadResponse};
20
21 /// JSON input for requesting a presigned S3 upload URL.
22 ///
23 /// `file_size_bytes` is optional for compatibility with older clients that
24 /// don't know the size ahead of time, but when supplied it is signed into
25 /// the presigned URL's `Content-Length` and S3 will reject any PUT whose
26 /// actual body length differs — protocol-level enforcement of the per-file
27 /// cap, no bandwidth wasted on oversized uploads that the confirm step
28 /// would have rejected anyway.
29 #[derive(Debug, Deserialize)]
30 pub struct PresignUploadRequest {
31 pub item_id: ItemId,
32 pub file_type: String,
33 pub file_name: String,
34 pub content_type: String,
35 #[serde(default)]
36 pub file_size_bytes: Option<i64>,
37 }
38
39 /// JSON input for confirming a completed S3 upload.
40 #[derive(Debug, Deserialize)]
41 pub struct ConfirmUploadRequest {
42 pub item_id: ItemId,
43 pub file_type: String,
44 pub s3_key: String,
45 }
46
47 /// Generate a presigned URL for uploading a file to S3
48 ///
49 /// POST /api/upload/presign
50 ///
51 /// Requires authentication. User must own the item.
52 #[tracing::instrument(skip_all, name = "storage::presign_upload")]
53 pub(super) async fn presign_upload(
54 State(state): State<AppState>,
55 AuthUser(user): AuthUser,
56 Json(req): Json<PresignUploadRequest>,
57 ) -> Result<impl IntoResponse> {
58 user.check_not_suspended()?;
59 // Check if S3 is configured
60 let s3 = state.require_s3()?;
61
62 // Parse file type
63 let file_type = FileType::from_str(&req.file_type)
64 .map_err(|_| AppError::BadRequest(format!("Invalid file type: {}", req.file_type)))?;
65
66 // Validate content type
67 S3Client::validate_content_type(file_type, &req.content_type)?;
68
69 // Validate file extension
70 S3Client::validate_extension(file_type, &req.file_name)?;
71
72 // Verify user owns the item
73 let owner = db::items::get_item_owner(&state.db, req.item_id)
74 .await?
75 .ok_or(AppError::NotFound)?;
76
77 if owner != user.id {
78 return Err(AppError::Forbidden);
79 }
80
81 // Early quota check (reject before generating presigned URL)
82 db::creator_tiers::check_presign_allowed(&state.db, user.id, file_type).await?;
83
84 // Get the effective max file size for client-side pre-validation
85 let max_file_bytes = db::creator_tiers::get_effective_max_file_bytes(&state.db, user.id, file_type).await?;
86
87 // If the client declared the file size, validate it before signing —
88 // both against the static per-type cap and the tier-effective cap. We
89 // bind it as Content-Length so S3 rejects oversized PUTs at the protocol
90 // level. Clients omitting `file_size_bytes` fall back to the old behavior
91 // (no protocol-level enforcement; confirm step still validates).
92 if let Some(size) = req.file_size_bytes {
93 if size <= 0 {
94 return Err(AppError::BadRequest("file_size_bytes must be positive".to_string()));
95 }
96 if size as u64 > file_type.max_size() {
97 let limit_mb = file_type.max_size() / (1024 * 1024);
98 let file_mb = size as u64 / (1024 * 1024);
99 return Err(AppError::FileTooLarge(format!(
100 "File is {} MB but the maximum for {} files is {} MB.",
101 file_mb, file_type.as_str(), limit_mb
102 )));
103 }
104 if let Some(tier_cap) = max_file_bytes
105 && (size as u64) > tier_cap
106 {
107 let limit_mb = tier_cap / (1024 * 1024);
108 return Err(AppError::FileTooLarge(format!(
109 "File exceeds your tier's per-file limit of {} MB.",
110 limit_mb
111 )));
112 }
113 }
114
115 // Generate S3 key
116 let s3_key = S3Client::generate_key(user.id, req.item_id, file_type, &req.file_name);
117
118 // Track the pending upload so the reaper can clean it up if never confirmed
119 db::pending_uploads::record_pending_upload(&state.db, user.id, &s3_key, "main").await?;
120
121 // Generate presigned upload URL with immutable cache headers
122 let expires_in = 3600; // 1 hour
123 let upload_url = s3.presign_upload(&s3_key, &req.content_type, Some(expires_in), Some(CACHE_CONTROL_IMMUTABLE), req.file_size_bytes)
124 .await
125 .context("presign upload for item content")?;
126
127 Ok(Json(PresignUploadResponse {
128 upload_url,
129 s3_key,
130 expires_in,
131 cache_control: Some(CACHE_CONTROL_IMMUTABLE.to_string()),
132 max_file_bytes,
133 }))
134 }
135
136 /// Confirm that an upload has completed and update the database
137 ///
138 /// POST /api/upload/confirm
139 ///
140 /// Requires authentication. User must own the item.
141 #[tracing::instrument(skip_all, name = "storage::confirm_upload")]
142 pub(super) async fn confirm_upload(
143 State(state): State<AppState>,
144 AuthUser(user): AuthUser,
145 Json(req): Json<ConfirmUploadRequest>,
146 ) -> Result<impl IntoResponse> {
147 user.check_not_suspended()?;
148 // Check if S3 is configured
149 let s3 = state.require_s3()?;
150
151 // Parse file type
152 let file_type = FileType::from_str(&req.file_type)
153 .map_err(|_| AppError::BadRequest(format!("Invalid file type: {}", req.file_type)))?;
154
155 // Verify user owns the item
156 let owner = db::items::get_item_owner(&state.db, req.item_id)
157 .await?
158 .ok_or(AppError::NotFound)?;
159
160 if owner != user.id {
161 return Err(AppError::Forbidden);
162 }
163
164 // Validate S3 key belongs to this user + item (prevent cross-user file reference)
165 let expected_prefix = format!("{}/{}/", user.id, req.item_id);
166 if !req.s3_key.starts_with(&expected_prefix) {
167 return Err(AppError::BadRequest(
168 "Invalid upload key".to_string(),
169 ));
170 }
171
172 // Verify the object exists in S3
173 if !s3.object_exists(&req.s3_key).await? {
174 return Err(AppError::BadRequest(
175 "Upload not found. Please try uploading again.".to_string(),
176 ));
177 }
178
179 // Enforce file size limit (static per-type limit)
180 let file_size_bytes = s3.object_size(&req.s3_key).await?.ok_or_else(|| {
181 AppError::BadRequest("Could not determine file size. Please try uploading again.".to_string())
182 })?;
183 if file_size_bytes as u64 > file_type.max_size() {
184 s3.delete_object(&req.s3_key).await.ok();
185 let limit_mb = file_type.max_size() / (1024 * 1024);
186 let file_mb = file_size_bytes as u64 / (1024 * 1024);
187 return Err(AppError::FileTooLarge(format!(
188 "File is {} MB but the maximum for {} files is {} MB.",
189 file_mb, file_type.as_str(), limit_mb
190 )));
191 }
192
193 // Enforce tier-based limits (per-file + storage cap)
194 let max_storage = match db::creator_tiers::check_upload_allowed(&state.db, user.id, file_type, file_size_bytes).await {
195 Ok(max) => max,
196 Err(e) => {
197 s3.delete_object(&req.s3_key).await.ok();
198 return Err(e);
199 }
200 };
201
202 // Reject file types that have their own dedicated confirm routes BEFORE
203 // any scan enqueue or scan_status flip — otherwise a misrouted-but-valid
204 // item_id flips that item's scan_status to Pending, blocks every fan's
205 // download, and leaks a scan_jobs row for an S3 key we're about to delete.
206 match file_type {
207 FileType::Audio | FileType::Cover | FileType::Video => {}
208 FileType::Download => {
209 s3.delete_object(&req.s3_key).await.ok();
210 return Err(AppError::BadRequest(
211 "Use /api/versions/{version_id}/upload/* routes for download files".to_string(),
212 ));
213 }
214 FileType::Insertion => {
215 s3.delete_object(&req.s3_key).await.ok();
216 return Err(AppError::BadRequest(
217 "Use /api/users/me/insertions/* routes for insertion clips".to_string(),
218 ));
219 }
220 FileType::MediaImage | FileType::MediaVideo => {
221 s3.delete_object(&req.s3_key).await.ok();
222 return Err(AppError::BadRequest(
223 "Use /api/media/* routes for media library uploads".to_string(),
224 ));
225 }
226 }
227
228 // Idempotency: if the matching s3_key field already equals this key, return success (no-op).
229 // If replacing a different file, use atomic replace_storage to avoid drift.
230 let mut old_s3_key: Option<String> = None;
231 let mut replaced_old_size: i64 = 0;
232 if let Some(item) = db::items::get_item_by_id(&state.db, req.item_id).await? {
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) = db::pending_uploads::remove_pending_upload(&state.db, user.id, &req.s3_key).await {
245 tracing::warn!(error = ?e, key = %req.s3_key, "remove_pending_upload failed on idempotent re-confirm");
246 }
247 return Ok(Json(ConfirmUploadResponse { success: true, pending_review: None }));
248 }
249 if let Some(old_key) = existing_key {
250 let old_size = match file_type {
251 FileType::Audio => item.audio_file_size_bytes.unwrap_or(0),
252 FileType::Cover => item.cover_file_size_bytes.unwrap_or(0),
253 FileType::Video => item.video_file_size_bytes.unwrap_or(0),
254 _ => 0,
255 };
256 // Atomically decrement old + increment new in a single UPDATE
257 if let Err(e) = db::creator_tiers::try_replace_storage(
258 &state.db, user.id, old_size, file_size_bytes, max_storage,
259 ).await {
260 s3.delete_object(&req.s3_key).await.ok();
261 return Err(e);
262 }
263 old_s3_key = Some(old_key.to_string());
264 replaced_old_size = old_size;
265 } else {
266 // No existing file — plain increment
267 if let Err(e) = db::creator_tiers::try_increment_storage(&state.db, user.id, file_size_bytes, max_storage).await {
268 s3.delete_object(&req.s3_key).await.ok();
269 return Err(e);
270 }
271 }
272 } else {
273 // Item not found yet (shouldn't happen but handle gracefully)
274 if let Err(e) = db::creator_tiers::try_increment_storage(&state.db, user.id, file_size_bytes, max_storage).await {
275 s3.delete_object(&req.s3_key).await.ok();
276 return Err(e);
277 }
278 }
279
280 // Update the item's S3 key and file size in a transaction so both succeed or fail together.
281 // On error, roll back and decrement the storage counter we just incremented.
282 let (s3_col, size_col) = match file_type {
283 FileType::Audio => ("audio_s3_key", "audio_file_size_bytes"),
284 FileType::Cover => ("cover_s3_key", "cover_file_size_bytes"),
285 FileType::Video => ("video_s3_key", "video_file_size_bytes"),
286 // Already rejected above — defensive error instead of panic
287 FileType::Download | FileType::Insertion | FileType::MediaImage | FileType::MediaVideo => {
288 return Err(AppError::BadRequest(format!("File type {:?} not supported for item uploads", file_type)));
289 }
290 };
291 let update_sql = format!(
292 "UPDATE items SET {} = $2, {} = $3, updated_at = NOW() WHERE id = $1 AND project_id IN (SELECT id FROM projects WHERE user_id = $4)",
293 s3_col, size_col,
294 );
295 let update_result = sqlx::query(&update_sql)
296 .bind(req.item_id)
297 .bind(&req.s3_key)
298 .bind(file_size_bytes)
299 .bind(user.id)
300 .execute(&state.db)
301 .await;
302
303 // Roll back the storage change. On a replace, we already decremented old_size
304 // and incremented file_size_bytes — undo both by swapping back. Pass i64::MAX
305 // as cap since we're restoring a state that was previously within cap.
306 let rollback_storage = || async {
307 if old_s3_key.is_some() {
308 db::creator_tiers::try_replace_storage(
309 &state.db, user.id, file_size_bytes, replaced_old_size, i64::MAX,
310 ).await.ok();
311 } else {
312 db::creator_tiers::decrement_storage_used(&state.db, user.id, file_size_bytes).await.ok();
313 }
314 };
315
316 match update_result {
317 Err(e) => {
318 rollback_storage().await;
319 s3.delete_object(&req.s3_key).await.ok();
320 return Err(e.into());
321 }
322 Ok(res) if res.rows_affected() == 0 => {
323 // Item was deleted (or transferred out from under the ownership
324 // filter) between our earlier `get_item_owner` check and this
325 // UPDATE. Without this guard, storage credit stayed incremented,
326 // `pending_uploads` got cleared a few lines down, and `commit_upload`
327 // enqueued a scan job against a ghost target — the S3 object then
328 // leaked permanently with the counter over-charged. Route the new
329 // object through the orphan queue so the reaper still cleans it.
330 rollback_storage().await;
331 super::enqueue_s3_orphan(&state.db, &req.s3_key, "item_upload_target_missing").await;
332 return Err(AppError::BadRequest(
333 "Item was modified concurrently. Please try uploading again.".to_string(),
334 ));
335 }
336 Ok(_) => {}
337 }
338
339 // Scan enqueue + scan_status flip happens AFTER the DB UPDATE commits via
340 // the shared `commit_upload` helper, which is the only blessed path for
341 // this ordering. See `routes/storage/mod.rs::commit_upload` for the bug
342 // shapes this prevents.
343 let scan_status = commit_upload(
344 &state,
345 CommitTarget::Item(req.item_id),
346 &req.s3_key,
347 file_type,
348 user.id,
349 file_size_bytes,
350 ).await?;
351
352 // Delete the old S3 object now that the replacement is committed. Route
353 // through the orphan queue so a transient S3 failure here doesn't leak the
354 // object forever — the worker retries until the delete succeeds (or 404s).
355 if let Some(old_key) = &old_s3_key {
356 super::enqueue_s3_orphan(&state.db, old_key, "item_upload_replace").await;
357 }
358
359 // Clear the pending upload record now that the upload is confirmed
360 db::pending_uploads::remove_pending_upload(&state.db, user.id, &req.s3_key).await?;
361
362 // Bump project cache generation so dashboard tabs reflect the new upload
363 if let Some(item) = db::items::get_item_by_id(&state.db, req.item_id).await?
364 && let Err(e) = db::projects::bump_cache_generation(&state.db, item.project_id).await
365 {
366 tracing::warn!(project_id = %item.project_id, error = ?e, "failed to bump cache generation after upload");
367 }
368
369 tracing::info!(
370 "Upload confirmed: item={}, type={:?}, key={}, size={}",
371 req.item_id,
372 file_type,
373 req.s3_key,
374 file_size_bytes
375 );
376
377 let pending_review = if scan_status == db::FileScanStatus::HeldForReview {
378 Some(true)
379 } else {
380 None
381 };
382 Ok(Json(ConfirmUploadResponse { success: true, pending_review }))
383 }
384