Skip to main content

max / makenotwork

12.2 KB · 343 lines History Blame Raw
1 //! Presigned upload and confirm handlers for version files.
2
3 use axum::{
4 Json,
5 extract::{Path, State},
6 response::IntoResponse,
7 };
8 use serde::Deserialize;
9 use sqlx::PgPool;
10
11 use crate::{
12 AppStorage, Scanning,
13 auth::AuthUser,
14 db::{self, VersionId},
15 error::{AppError, Result, ResultExt},
16 storage::{CACHE_CONTROL_IMMUTABLE, FileType, S3Client},
17 };
18
19 use super::{CommitTarget, ConfirmUploadResponse, PresignUploadResponse, commit_upload};
20
21 /// JSON input for requesting a presigned version upload URL.
22 #[derive(Debug, Deserialize)]
23 pub(super) struct VersionPresignRequest {
24 pub file_name: String,
25 pub content_type: String,
26 /// Optional declared size; when present it is signed into the presigned
27 /// URL's `Content-Length` so S3 rejects oversized PUTs at the protocol
28 /// level (version downloads are up to 500 MB).
29 #[serde(default)]
30 pub file_size_bytes: Option<i64>,
31 }
32
33 /// JSON input for confirming a completed version upload.
34 #[derive(Debug, Deserialize)]
35 pub(super) struct VersionConfirmRequest {
36 pub s3_key: String,
37 }
38
39 /// Generate a presigned URL for uploading a version file to S3
40 ///
41 /// POST /api/versions/{version_id}/upload/presign
42 ///
43 /// Requires authentication. User must own the item (through version -> item -> project chain).
44 #[tracing::instrument(skip_all, name = "storage::version_presign_upload", fields(%version_id, user_id = %user.id))]
45 pub(super) async fn version_presign_upload(
46 State(db): State<PgPool>,
47 State(storage): State<AppStorage>,
48 AuthUser(user): AuthUser,
49 Path(version_id): Path<VersionId>,
50 Json(req): Json<VersionPresignRequest>,
51 ) -> Result<impl IntoResponse> {
52 user.check_not_suspended()?;
53 let s3 = storage.require_s3()?;
54
55 let file_type = FileType::Download;
56
57 // Validate content type and extension
58 S3Client::validate_content_type(file_type, &req.content_type)?;
59 S3Client::validate_extension(file_type, &req.file_name)?;
60
61 // Fetch version and verify ownership through version -> item -> project chain
62 let version = db::versions::get_version_by_id(&db, version_id)
63 .await?
64 .ok_or(AppError::NotFound)?;
65
66 let owner = db::items::get_item_owner(&db, version.item_id)
67 .await?
68 .ok_or(AppError::NotFound)?;
69
70 if owner != user.id {
71 return Err(AppError::Forbidden);
72 }
73
74 // Early quota check
75 db::creator_tiers::check_presign_allowed(&db, user.id, file_type).await?;
76
77 let max_file_bytes =
78 db::creator_tiers::get_effective_max_file_bytes(&db, user.id, file_type).await?;
79
80 // Validate the declared size (if any) before signing it into Content-Length.
81 super::validate_declared_upload_size(req.file_size_bytes, file_type, max_file_bytes)?;
82
83 // Staging key (unserved); the scan worker promotes it to the content key on
84 // a Clean verdict. The random staging uuid also guarantees two versions that
85 // share a filename never collide onto one object (the property the old
86 // version-id-woven key gave us, ultra-fuzz Run #1 Storage HIGH).
87 let s3_key = S3Client::generate_staging_key(&req.file_name);
88
89 // Track the pending upload so the reaper can clean it up if never confirmed
90 db::pending_uploads::record_pending_upload(&db, user.id, &s3_key, "main").await?;
91
92 let expires_in = 3600;
93 let upload_url = s3
94 .presign_upload(
95 &s3_key,
96 &req.content_type,
97 Some(expires_in),
98 Some(CACHE_CONTROL_IMMUTABLE),
99 req.file_size_bytes,
100 )
101 .await
102 .context("presign upload for version file")?;
103
104 Ok(Json(PresignUploadResponse {
105 upload_url,
106 s3_key: s3_key.into_string(),
107 expires_in,
108 cache_control: Some(CACHE_CONTROL_IMMUTABLE.to_string()),
109 max_file_bytes,
110 }))
111 }
112
113 /// Confirm that a version file upload has completed and update the database
114 ///
115 /// POST /api/versions/{version_id}/upload/confirm
116 ///
117 /// Requires authentication. User must own the item.
118 #[tracing::instrument(skip_all, name = "storage::version_confirm_upload", fields(%version_id, user_id = %user.id))]
119 pub(super) async fn version_confirm_upload(
120 State(db): State<PgPool>,
121 State(storage): State<AppStorage>,
122 State(scanning): State<Scanning>,
123 AuthUser(user): AuthUser,
124 Path(version_id): Path<VersionId>,
125 Json(req): Json<VersionConfirmRequest>,
126 ) -> Result<impl IntoResponse> {
127 user.check_not_suspended()?;
128 let s3 = storage.require_s3()?;
129
130 // Fetch version and verify ownership
131 let version = db::versions::get_version_by_id(&db, version_id)
132 .await?
133 .ok_or(AppError::NotFound)?;
134
135 let owner = db::items::get_item_owner(&db, version.item_id)
136 .await?
137 .ok_or(AppError::NotFound)?;
138
139 if owner != user.id {
140 return Err(AppError::Forbidden);
141 }
142
143 // Idempotent re-confirm: the version already references this exact key.
144 // Handled BEFORE the ownership gate and the scan enqueue, re-confirming an
145 // already-Clean version must not knock it back to Pending, and the pending
146 // row was consumed by the first confirm so the gate below would reject it.
147 if version.s3_key.as_deref() == Some(&req.s3_key) {
148 // Still clear pending_uploads, orphan reaper would otherwise delete the
149 // live S3 object 24h later (Run #7 HIGH-1).
150 if let Err(e) =
151 db::pending_uploads::remove_pending_upload(&db, user.id, &req.s3_key, "main").await
152 {
153 tracing::warn!(error = ?e, key = %req.s3_key, "remove_pending_upload failed on idempotent re-confirm");
154 }
155 return Ok(Json(ConfirmUploadResponse {
156 success: true,
157 pending_review: None,
158 }));
159 }
160
161 // Authorize the staging key for a fresh confirm. A `staging/{uuid}` key has
162 // no user/item/version in its path, so ownership is proved via the
163 // `pending_uploads` row recorded at presign, not a prefix check. Placed
164 // before the size/tier reject paths so an unowned (at most another user's
165 // in-flight) staging object is never enqueued for deletion.
166 if !db::pending_uploads::is_owned(&db, user.id, &req.s3_key, "main").await? {
167 return Err(AppError::BadRequest("Invalid upload key".to_string()));
168 }
169
170 // A single HEAD: `object_size` returns None when the object isn't there, so
171 // it doubles as the existence check (no separate object_exists round-trip).
172 // Versions are always downloads, so enforce that size limit.
173 let file_size_bytes = s3.object_size(&req.s3_key).await?.ok_or_else(|| {
174 AppError::BadRequest("Upload not found. Please try uploading again.".to_string())
175 })?;
176 if file_size_bytes as u64 > FileType::Download.max_size() {
177 super::enqueue_s3_orphan(
178 &db,
179 &req.s3_key,
180 crate::storage::S3Bucket::Main,
181 "version_upload_rejected",
182 )
183 .await;
184 let limit_mb = FileType::Download.max_size() / (1024 * 1024);
185 let file_mb = file_size_bytes as u64 / (1024 * 1024);
186 return Err(AppError::FileTooLarge(format!(
187 "File is {file_mb} MB but the maximum for download files is {limit_mb} MB."
188 )));
189 }
190
191 // Enforce tier-based limits (per-file + storage cap)
192 let max_storage = match db::creator_tiers::check_upload_allowed(
193 &db,
194 user.id,
195 FileType::Download,
196 file_size_bytes,
197 )
198 .await
199 {
200 Ok(max) => max,
201 Err(e) => {
202 super::enqueue_s3_orphan(
203 &db,
204 &req.s3_key,
205 crate::storage::S3Bucket::Main,
206 "version_upload_rejected",
207 )
208 .await;
209 return Err(e);
210 }
211 };
212
213 let old_s3_key = version.s3_key.clone();
214 let old_size = version.file_size_bytes.unwrap_or(0);
215 let is_replace = old_s3_key.is_some() && old_size > 0;
216
217 // Extract file name from the s3_key (last path segment)
218 let file_name = req
219 .s3_key
220 .rsplit('/')
221 .next()
222 .map(std::string::ToString::to_string);
223
224 // Storage credit + version UPDATE in ONE transaction. The expected-old guard
225 // (`s3_key IS NOT DISTINCT FROM`) returns no row if another confirm raced
226 // ahead; we leave the tx uncommitted so the rollback undoes the storage
227 // change with no compensating math (the previous swallowed-`.ok()` path).
228 // `commit_upload` stays AFTER the commit (the blessed scan-ordering path).
229 // `Ok(false)` = lost race (rolled back, nothing charged).
230 let committed: Result<bool> = async {
231 let mut tx = db.begin().await?;
232 db::creator_tiers::try_apply_storage_on(
233 &mut tx,
234 user.id,
235 is_replace.then_some(old_size),
236 file_size_bytes,
237 max_storage,
238 )
239 .await?;
240 let updated = db::versions::update_version_file(
241 &mut *tx,
242 version_id,
243 old_s3_key.as_deref(),
244 &req.s3_key,
245 Some(file_size_bytes),
246 file_name.as_deref(),
247 )
248 .await?;
249 if updated.is_none() {
250 // Lost race, drop tx to roll back the storage change.
251 return Ok(false);
252 }
253 // Enqueue the OLD key for deletion in the SAME tx as the row UPDATE, so
254 // a crash between commit and a post-commit enqueue can't orphan it with
255 // no durable record (ultra-fuzz Run #1 Storage LOW; mirrors the in-tx
256 // ordering delete_version uses). After commit the row points at the new
257 // key, so the old key is non-live; the worker's is_s3_key_live check is
258 // the backstop if anything still references it.
259 if let Some(old_key) = old_s3_key.as_deref() {
260 db::pending_s3_deletions::enqueue_deletions(
261 &mut *tx,
262 &[(old_key.to_string(), "main".to_string())],
263 "version_replace",
264 )
265 .await?;
266 }
267 tx.commit().await?;
268 Ok(true)
269 }
270 .await;
271
272 match committed {
273 Err(e) => {
274 // The tx rolled back, so nothing this request wrote references the
275 // key, but a concurrent double-confirm of this same key could have
276 // committed it onto the row before our `try_apply_storage_on`
277 // errored (e.g. the storage cap filled in between). A blind delete
278 // would then destroy the live object the winning confirm points at.
279 // Route through the orphan queue, whose `is_s3_key_live` check skips
280 // any key a row still references.
281 super::enqueue_s3_orphan(
282 &db,
283 &req.s3_key,
284 crate::storage::S3Bucket::Main,
285 "version_confirm_failed",
286 )
287 .await;
288 return Err(e);
289 }
290 Ok(false) => {
291 // Lost the CAS race: a concurrent confirm swapped the version's
292 // s3_key out from under us. If it committed THIS key, a direct
293 // delete would 404 every fan download of the version the winner
294 // just published. The orphan queue's liveness check is the guard.
295 super::enqueue_s3_orphan(
296 &db,
297 &req.s3_key,
298 crate::storage::S3Bucket::Main,
299 "version_confirm_lost_race",
300 )
301 .await;
302 return Err(AppError::BadRequest(
303 "Version was modified concurrently. Please try uploading again.".to_string(),
304 ));
305 }
306 Ok(true) => {}
307 }
308
309 // Clear the pending upload record now that the upload is committed
310 db::pending_uploads::remove_pending_upload(&db, user.id, &req.s3_key, "main").await?;
311
312 let scan_status = commit_upload(
313 &db,
314 scanning.scanner.as_ref(),
315 CommitTarget::Version(version_id),
316 &req.s3_key,
317 FileType::Download,
318 user.id,
319 file_size_bytes,
320 )
321 .await?;
322
323 // (The old S3 key was enqueued for deletion inside the commit tx above.)
324
325 tracing::info!(
326 key = %req.s3_key,
327 size = file_size_bytes,
328 is_replace,
329 ?scan_status,
330 "version upload confirmed"
331 );
332
333 let pending_review = if scan_status == db::FileScanStatus::HeldForReview {
334 Some(true)
335 } else {
336 None
337 };
338 Ok(Json(ConfirmUploadResponse {
339 success: true,
340 pending_review,
341 }))
342 }
343