Skip to main content

max / makenotwork

11.3 KB · 296 lines History Blame Raw
1 //! Storage API routes for S3 file uploads and streaming
2
3 mod downloads;
4 mod images;
5 pub(crate) mod media;
6 mod uploads;
7 mod versions;
8
9 use axum::routing::get;
10 use serde::Serialize;
11 use tower_governor::GovernorLayer;
12 use uuid::Uuid;
13
14 use crate::{
15 constants,
16 csrf::{delete_csrf, post_csrf, CsrfRouter},
17 db,
18 db::scan_jobs::ScanTargetKind,
19 error::Result,
20 storage::FileType,
21 AppState,
22 };
23
24 /// Enqueue an orphaned S3 key for the pending-deletion worker.
25 ///
26 /// Use this when an upload has already crossed the durability boundary —
27 /// storage credit committed, DB row inserted, or an old object queued for
28 /// replacement — and a downstream step then failed. The queue worker retries
29 /// until the delete succeeds (or 404s), so a transient S3 failure doesn't
30 /// leak the object permanently.
31 ///
32 /// Pre-credit / pre-DB rejection paths (size cap, type-mismatch, tier check
33 /// fail) may still use `s3.delete_object(...).await.ok()` directly: nothing
34 /// in the DB references the key yet, so a swallowed error is at worst a 24h
35 /// orphan that the pending_uploads reaper will collect on its own schedule.
36 pub(crate) async fn enqueue_s3_orphan(pool: &sqlx::PgPool, s3_key: &str, source: &'static str) {
37 if let Err(e) = db::pending_s3_deletions::enqueue_deletions(
38 pool,
39 &[(s3_key.to_string(), "main".to_string())],
40 source,
41 )
42 .await
43 {
44 tracing::warn!(error = ?e, key = %s3_key, source = %source, "failed to enqueue orphan S3 key");
45 }
46 }
47
48 /// Register S3 upload and streaming routes.
49 ///
50 /// Upload routes (presign + confirm) are rate limited per IP (see `constants::UPLOAD_RATE_LIMIT_*`).
51 /// Stream/download endpoints are unlimited (presigned URLs already expire in 1 hour).
52 pub fn storage_routes() -> CsrfRouter<AppState> {
53 let upload_rate_limit = crate::helpers::rate_limiter_ms(constants::UPLOAD_RATE_LIMIT_MS, constants::UPLOAD_RATE_LIMIT_BURST);
54
55 let upload_routes = CsrfRouter::new()
56 .route("/api/upload/presign", post_csrf(uploads::presign_upload))
57 .route("/api/upload/confirm", post_csrf(uploads::confirm_upload))
58 .route("/api/versions/{version_id}/upload/presign", post_csrf(versions::version_presign_upload))
59 .route("/api/versions/{version_id}/upload/confirm", post_csrf(versions::version_confirm_upload))
60 .route("/api/projects/image/presign", post_csrf(images::project_image_presign))
61 .route("/api/projects/image/confirm", post_csrf(images::project_image_confirm))
62 .route("/api/items/image/presign", post_csrf(images::item_image_presign))
63 .route("/api/items/image/confirm", post_csrf(images::item_image_confirm))
64 .route("/api/media/presign", post_csrf(media::media_presign))
65 .route("/api/media/confirm", post_csrf(media::media_confirm))
66 .route_get("/api/media", get(media::media_list))
67 .route_get("/api/media/folders", get(media::media_folders))
68 .route("/api/media/{id}", delete_csrf(media::media_delete))
69 .route_layer(GovernorLayer {
70 config: upload_rate_limit,
71 });
72
73 let stream_rate_limit = crate::helpers::rate_limiter_ms(constants::STREAM_RATE_LIMIT_MS, constants::STREAM_RATE_LIMIT_BURST);
74
75 let stream_routes = CsrfRouter::new()
76 .route_get("/api/stream/{item_id}", get(downloads::stream_url))
77 .route_get("/api/versions/{version_id}/download", get(downloads::version_download))
78 .route_layer(GovernorLayer {
79 config: stream_rate_limit,
80 });
81
82 upload_routes.merge(stream_routes)
83 }
84
85 // =============================================================================
86 // Shared Request/Response Types
87 // =============================================================================
88
89 /// JSON response containing the presigned upload URL and S3 key.
90 #[derive(Debug, Serialize)]
91 pub struct PresignUploadResponse {
92 pub upload_url: String,
93 pub s3_key: String,
94 pub expires_in: u64,
95 /// Cache-Control header the client must send with the S3 PUT (part of the presigned signature).
96 #[serde(skip_serializing_if = "Option::is_none")]
97 pub cache_control: Option<String>,
98 /// Maximum file size in bytes for this upload (for client-side pre-validation).
99 #[serde(skip_serializing_if = "Option::is_none")]
100 pub max_file_bytes: Option<u64>,
101 }
102
103 /// JSON response confirming a successful upload.
104 #[derive(Debug, Serialize)]
105 pub struct ConfirmUploadResponse {
106 pub success: bool,
107 /// When true, the file was uploaded but is pending manual review before
108 /// it becomes available to fans. The creator should see a "pending review"
109 /// indicator instead of assuming the file is live.
110 #[serde(skip_serializing_if = "Option::is_none")]
111 pub pending_review: Option<bool>,
112 }
113
114 // =============================================================================
115 // Helpers
116 // =============================================================================
117
118 /// Discriminates which entity an upload commit applies to, and carries the
119 /// per-target ID and the corresponding scan_status update.
120 ///
121 /// Construct one of these in your handler AFTER the entity's DB write has
122 /// committed, then pass it to [`commit_upload`]. Order matters — see
123 /// [`commit_upload`] docs.
124 pub(crate) enum CommitTarget {
125 /// An item (Audio/Cover/Video s3_key column on `items`).
126 Item(db::ItemId),
127 /// A version (`versions` table).
128 Version(db::VersionId),
129 /// A media library file (`media_files` table).
130 Media(db::MediaFileId),
131 /// A project cover image (URL stored on `projects.cover_image_url`;
132 /// project images do not carry a per-row scan_status column — the worker
133 /// only logs and creates a WAM ticket on quarantine).
134 ProjectImage(db::ProjectId),
135 /// An item image (`items.cover_s3_key`); shares the item's scan_status.
136 ItemImage(db::ItemId),
137 /// A content insertion clip; no per-row scan_status column.
138 ContentInsertion(db::ContentInsertionId),
139 }
140
141 impl CommitTarget {
142 fn kind(&self) -> ScanTargetKind {
143 match self {
144 CommitTarget::Item(_) => ScanTargetKind::Item,
145 CommitTarget::ItemImage(_) => ScanTargetKind::ItemImage,
146 CommitTarget::Version(_) => ScanTargetKind::Version,
147 CommitTarget::Media(_) => ScanTargetKind::Media,
148 CommitTarget::ProjectImage(_) => ScanTargetKind::ProjectImage,
149 CommitTarget::ContentInsertion(_) => ScanTargetKind::ContentInsertion,
150 }
151 }
152
153 fn target_uuid(&self) -> Uuid {
154 match self {
155 CommitTarget::Item(id) | CommitTarget::ItemImage(id) => (*id).into(),
156 CommitTarget::Version(id) => (*id).into(),
157 CommitTarget::Media(id) => (*id).into(),
158 CommitTarget::ProjectImage(id) => (*id).into(),
159 CommitTarget::ContentInsertion(id) => (*id).into(),
160 }
161 }
162 }
163
164 /// Enqueue a scan job and write the resulting status onto the target entity.
165 ///
166 /// **Call this AFTER the DB write that commits the upload has succeeded.**
167 /// Calling it earlier produces three known bug shapes — chronic across four
168 /// audit runs — which is why the lower-level pieces (`enqueue_scan_for`,
169 /// `update_*_scan_status`) are gated behind this single entry point:
170 ///
171 /// 1. A handler that early-returns (idempotent re-confirm, route mismatch,
172 /// quota rejection) leaks a `scan_jobs` row and flips a Clean status back
173 /// to Pending, blocking every fan's download until a rescan.
174 /// 2. A failed DB write leaves a dangling scan_jobs row pointing at an S3
175 /// key that's about to be deleted.
176 /// 3. The worker can race the still-uncommitted entity row.
177 ///
178 /// Use [`CommitTarget`] to bind the target id + per-target status updater.
179 /// The function returns the `FileScanStatus` that was written (callers use
180 /// this to populate the `pending_review` field of `ConfirmUploadResponse`).
181 pub(crate) async fn commit_upload(
182 state: &AppState,
183 target: CommitTarget,
184 s3_key: &str,
185 file_type: FileType,
186 user_id: db::UserId,
187 file_size_bytes: i64,
188 ) -> Result<db::FileScanStatus> {
189 let scan_status = enqueue_scan_for(
190 state,
191 target.kind(),
192 target.target_uuid(),
193 s3_key,
194 file_type,
195 user_id,
196 file_size_bytes,
197 )
198 .await?;
199 match target {
200 CommitTarget::Item(id) | CommitTarget::ItemImage(id) => {
201 db::scanning::update_item_scan_status(&state.db, id, scan_status).await?;
202 }
203 CommitTarget::Version(id) => {
204 db::scanning::update_version_scan_status(&state.db, id, scan_status).await?;
205 }
206 CommitTarget::Media(id) => {
207 db::scanning::update_media_file_scan_status(&state.db, id, scan_status).await?;
208 }
209 CommitTarget::ProjectImage(_) | CommitTarget::ContentInsertion(_) => {
210 // No per-row scan_status column; worker logs + creates WAM ticket.
211 }
212 }
213 Ok(scan_status)
214 }
215
216 /// Admin-rescan entry point. The entity already exists; we just need to
217 /// re-run the scan pipeline against its existing `s3_key`. Enqueues the
218 /// scan job then flips the per-row `scan_status` to Pending in the same
219 /// order `commit_upload` uses for first-scan, so admin handlers can't
220 /// invert it (the chronic disease the seal was built to prevent).
221 pub(crate) async fn commit_rescan(
222 state: &AppState,
223 target: CommitTarget,
224 s3_key: &str,
225 file_type: FileType,
226 user_id: db::UserId,
227 file_size_bytes: i64,
228 ) -> Result<db::FileScanStatus> {
229 enqueue_scan_for(
230 state,
231 target.kind(),
232 target.target_uuid(),
233 s3_key,
234 file_type,
235 user_id,
236 file_size_bytes,
237 )
238 .await?;
239 let pending = db::FileScanStatus::Pending;
240 match target {
241 CommitTarget::Item(id) | CommitTarget::ItemImage(id) => {
242 db::scanning::update_item_scan_status(&state.db, id, pending).await?;
243 }
244 CommitTarget::Version(id) => {
245 db::scanning::update_version_scan_status(&state.db, id, pending).await?;
246 }
247 CommitTarget::Media(id) => {
248 db::scanning::update_media_file_scan_status(&state.db, id, pending).await?;
249 }
250 CommitTarget::ProjectImage(_) | CommitTarget::ContentInsertion(_) => {
251 // No per-row scan_status column.
252 }
253 }
254 Ok(pending)
255 }
256
257 /// Enqueue an async scan job for an uploaded file and return the initial
258 /// `scan_status` to write onto the target entity.
259 ///
260 /// **Storage handlers should not call this directly** — use [`commit_upload`]
261 /// so the ordering invariant (scan-after-DB-commit) cannot be inverted by a
262 /// future sibling handler. This function remains `pub(super)`-equivalent for
263 /// the `commit_upload` implementation and for the worker / admin tooling
264 /// that legitimately needs the lower-level op.
265 async fn enqueue_scan_for(
266 state: &AppState,
267 target_kind: ScanTargetKind,
268 target_id: Uuid,
269 s3_key: &str,
270 file_type: FileType,
271 user_id: db::UserId,
272 file_size_bytes: i64,
273 ) -> Result<db::FileScanStatus> {
274 if state.scanner.is_none() {
275 let is_trusted = db::users::is_upload_trusted(&state.db, user_id).await?;
276 return Ok(if is_trusted {
277 db::FileScanStatus::Clean
278 } else {
279 db::FileScanStatus::HeldForReview
280 });
281 }
282
283 db::scan_jobs::enqueue(
284 &state.db,
285 target_kind,
286 target_id,
287 s3_key,
288 file_type,
289 user_id,
290 file_size_bytes,
291 )
292 .await?;
293
294 Ok(db::FileScanStatus::Pending)
295 }
296