Skip to main content

max / makenotwork

Decompose AppState Phase 2: migrate storage to slices Migrate src/routes/storage/** off State<AppState>. Handlers extract PgPool + AppStorage (via require_s3/public_s3), adding Config (cdn_base_url/ storage) and Scanning where used (confirm/commit paths). commit_* helpers in mod.rs narrowed: - commit_upload/commit_rescan/enqueue_scan_for: (&PgPool, Option<&Arc<ScanPipeline>>, ...) - commit_promote_item/version: (&PgPool, &AppStorage, &Config, id) Their external callers (admin/uploads, api/internal/uploads, api/content_insertions) keep their own State<AppState> and project slice refs at the call: &state.db, &state.storage, state.scanner.as_ref(), &state.config. No behavior change; storage 58 + upload 38 + gallery 10 + image 10 + scan 25 tests pass. Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
Author: Max Johnson <me@maxj.phd> · 2026-07-13 23:46 UTC
Commit: 730c16478cbe7065ac9721af7e5aa18e60ccdc7f
Parent: 6297d7a
10 files changed, +281 insertions, -217 deletions
M server/src/routes/admin/uploads.rs +8 -6
@@ -263,7 +263,7 @@ pub(super) async fn admin_promote_item(
263 263 // Copy each held (staging) file to its content key and mark Clean in one
264 264 // step — a held file sits unserved at its staging key, so approving must run
265 265 // the same promote the scan worker's Clean path runs (C1).
266 - crate::routes::storage::commit_promote_item(&state, id).await?;
266 + crate::routes::storage::commit_promote_item(&state.db, &state.storage, &state.config, id).await?;
267 267 db::scan_admin_actions::log_item(
268 268 &state.db, id, admin.id, AdminAction::Promote,
269 269 Some("held_for_review"), Some("clean"),
@@ -299,7 +299,7 @@ pub(super) async fn admin_promote_version(
299 299 ) -> Result<Response> {
300 300 // See admin_promote_item: copy the held (staging) download to its content key
301 301 // and mark Clean together (C1).
302 - crate::routes::storage::commit_promote_version(&state, id).await?;
302 + crate::routes::storage::commit_promote_version(&state.db, &state.storage, &state.config, id).await?;
303 303 db::scan_admin_actions::log_version(
304 304 &state.db, id, admin.id, AdminAction::Promote,
305 305 Some("held_for_review"), Some("clean"),
@@ -361,7 +361,8 @@ async fn rescan_version_inner(
361 361 // to call scan_jobs::enqueue + update_*_scan_status directly, which made
362 362 // ordering bugs hard to fence at the type level.
363 363 crate::routes::storage::commit_rescan(
364 - state,
364 + &state.db,
365 + state.scanner.as_ref(),
365 366 crate::routes::storage::CommitTarget::Version(id),
366 367 &s3_key,
367 368 FileType::Download,
@@ -407,7 +408,8 @@ async fn rescan_item_inner(
407 408 };
408 409
409 410 crate::routes::storage::commit_rescan(
410 - state,
411 + &state.db,
412 + state.scanner.as_ref(),
411 413 crate::routes::storage::CommitTarget::Item(id),
412 414 &s3_key,
413 415 file_type,
@@ -509,7 +511,7 @@ pub(super) async fn admin_bulk_promote_held(
509 511 let admin_id = admin.id;
510 512 let vid = v.version_id;
511 513 set.spawn(async move {
512 - if let Err(e) = crate::routes::storage::commit_promote_version(&state, vid).await {
514 + if let Err(e) = crate::routes::storage::commit_promote_version(&state.db, &state.storage, &state.config, vid).await {
513 515 tracing::warn!(version_id = %vid, error = ?e, "bulk promote: version promote failed");
514 516 false
515 517 } else {
@@ -530,7 +532,7 @@ pub(super) async fn admin_bulk_promote_held(
530 532 let admin_id = admin.id;
531 533 let iid = i.item_id;
532 534 set.spawn(async move {
533 - if let Err(e) = crate::routes::storage::commit_promote_item(&state, iid).await {
535 + if let Err(e) = crate::routes::storage::commit_promote_item(&state.db, &state.storage, &state.config, iid).await {
534 536 tracing::warn!(item_id = %iid, error = ?e, "bulk promote: item promote failed");
535 537 false
536 538 } else {
M server/src/routes/api/content_insertions.rs +2 -1
@@ -219,7 +219,8 @@ pub(super) async fn confirm_insertion(
219 219 // worker flips it to 'clean'; on quarantine the row is purged and a WAM ticket
220 220 // filed. The creator still sees the pending clip in their management library.
221 221 commit_upload(
222 - &state,
222 + &state.db,
223 + state.scanner.as_ref(),
223 224 CommitTarget::ContentInsertion(insertion.id),
224 225 &req.s3_key,
225 226 FileType::Insertion,
M server/src/routes/api/internal/uploads.rs +2 -1
@@ -258,7 +258,8 @@ pub(super) async fn confirm_upload(
258 258 // Scan enqueue + scan_status flip AFTER the DB writes commit — chronic
259 259 // ordering invariant enforced via the shared commit_upload helper.
260 260 let _status = crate::routes::storage::commit_upload(
261 - &state,
261 + &state.db,
262 + state.scanner.as_ref(),
262 263 commit_target,
263 264 &req.s3_key,
264 265 file_type,
M server/src/routes/storage/downloads.rs +18 -16
@@ -6,13 +6,13 @@ use axum::{
6 6 Json,
7 7 };
8 8 use serde::Serialize;
9 + use sqlx::PgPool;
9 10
10 11 use crate::{
11 12 auth::MaybeUserVerified,
12 13 db::{self, ContentData, ItemId, VersionId},
13 14 error::{AppError, Result, ResultExt},
14 - pricing,
15 - AppState,
15 + pricing, AppStorage,
16 16 };
17 17
18 18 /// JSON response containing a presigned streaming/download URL.
@@ -64,23 +64,24 @@ async fn resolve_content_url(
64 64 /// - Paid items: Must be logged in and have purchased the item
65 65 #[tracing::instrument(skip_all, name = "storage::stream_url", fields(item_id))]
66 66 pub(super) async fn stream_url(
67 - State(state): State<AppState>,
67 + State(db): State<PgPool>,
68 + State(storage): State<AppStorage>,
68 69 MaybeUserVerified(maybe_user): MaybeUserVerified,
69 70 Path(item_id): Path<ItemId>,
70 71 ) -> Result<impl IntoResponse> {
71 72 tracing::Span::current().record("item_id", tracing::field::display(&item_id));
72 73
73 74 // Check if S3 is configured
74 - let s3 = state.require_s3()?;
75 + let s3 = storage.require_s3()?;
75 76
76 77 // Get the item
77 - let item = db::items::get_item_by_id(&state.db, item_id)
78 + let item = db::items::get_item_by_id(&db, item_id)
78 79 .await?
79 80 .ok_or(AppError::NotFound)?;
80 81
81 82 // Single-query access check: ownership, purchase, subscription, bundle
82 83 let user_id = maybe_user.as_ref().map(|u| u.id);
83 - let access = db::items::check_item_access(&state.db, item_id, user_id)
84 + let access = db::items::check_item_access(&db, item_id, user_id)
84 85 .await?
85 86 .ok_or(AppError::NotFound)?;
86 87 let is_creator = user_id.is_some_and(|uid| uid == access.owner_id);
@@ -142,11 +143,11 @@ pub(super) async fn stream_url(
142 143 ).await?;
143 144
144 145 // Increment total play count (includes replays)
145 - db::items::increment_play_count(&state.db, item_id).await?;
146 + db::items::increment_play_count(&db, item_id).await?;
146 147
147 148 // Track unique listeners for authenticated users
148 149 if let Some(ref user) = maybe_user {
149 - let _ = db::items::record_unique_play(&state.db, user.id, item_id).await;
150 + let _ = db::items::record_unique_play(&db, user.id, item_id).await;
150 151 }
151 152
152 153 Ok(Json(StreamUrlResponse {
@@ -162,15 +163,16 @@ pub(super) async fn stream_url(
162 163 /// Access control: free items are accessible to anyone, paid items require purchase.
163 164 #[tracing::instrument(skip_all, name = "storage::version_download", fields(version_id))]
164 165 pub(super) async fn version_download(
165 - State(state): State<AppState>,
166 + State(db): State<PgPool>,
167 + State(storage): State<AppStorage>,
166 168 MaybeUserVerified(maybe_user): MaybeUserVerified,
167 169 Path(version_id): Path<VersionId>,
168 170 ) -> Result<impl IntoResponse> {
169 171 tracing::Span::current().record("version_id", tracing::field::display(&version_id));
170 - let s3 = state.require_s3()?;
172 + let s3 = storage.require_s3()?;
171 173
172 174 // Fetch version
173 - let version = db::versions::get_version_by_id(&state.db, version_id)
175 + let version = db::versions::get_version_by_id(&db, version_id)
174 176 .await?
175 177 .ok_or(AppError::NotFound)?;
176 178
@@ -181,13 +183,13 @@ pub(super) async fn version_download(
181 183 .ok_or(AppError::NotFound)?;
182 184
183 185 // Fetch item for access control
184 - let item = db::items::get_item_by_id(&state.db, version.item_id)
186 + let item = db::items::get_item_by_id(&db, version.item_id)
185 187 .await?
186 188 .ok_or(AppError::NotFound)?;
187 189
188 190 // Single-query access check: ownership, purchase, subscription, bundle
189 191 let user_id = maybe_user.as_ref().map(|u| u.id);
190 - let access = db::items::check_item_access(&state.db, version.item_id, user_id)
192 + let access = db::items::check_item_access(&db, version.item_id, user_id)
191 193 .await?
192 194 .ok_or(AppError::NotFound)?;
193 195 let is_creator = user_id.is_some_and(|uid| uid == access.owner_id);
@@ -228,12 +230,12 @@ pub(super) async fn version_download(
228 230 ).await?;
229 231
230 232 // Increment per-version and item-level download counts
231 - db::versions::increment_download_count(&state.db, version_id).await?;
232 - db::items::increment_item_download_count(&state.db, version.item_id).await?;
233 + db::versions::increment_download_count(&db, version_id).await?;
234 + db::items::increment_item_download_count(&db, version.item_id).await?;
233 235
234 236 // Track per-user download for library "new version" indicators
235 237 if let Some(ref user) = maybe_user {
236 - let _ = db::versions::record_user_download(&state.db, user.id, version.item_id, version_id).await;
238 + let _ = db::versions::record_user_download(&db, user.id, version.item_id, version_id).await;
237 239 }
238 240
239 241 let license_url = if item.license_preset.is_some() {
M server/src/routes/storage/gallery.rs +48 -41
@@ -13,14 +13,16 @@ use axum::{
13 13 Json,
14 14 };
15 15 use serde::{Deserialize, Serialize};
16 + use sqlx::PgPool;
16 17 use uuid::Uuid;
17 18
18 19 use crate::{
19 20 auth::AuthUser,
21 + config::Config,
20 22 db::{self, ImageId, ItemId, ProjectId, UserId},
21 23 error::{AppError, Result, ResultExt},
22 24 storage::{self, FileType, S3Client, CACHE_CONTROL_IMMUTABLE},
23 - AppState,
25 + AppStorage, Scanning,
24 26 };
25 27
26 28 use super::{commit_upload, CommitTarget, PresignUploadResponse};
@@ -45,14 +47,14 @@ impl GalleryTarget {
45 47 /// Verify `user` owns the target entity. Returns the parsed kind. NotFound if
46 48 /// the entity does not exist, Forbidden if owned by someone else.
47 49 async fn require_owned(
48 - state: &AppState,
50 + db: &PgPool,
49 51 target: GalleryTarget,
50 52 target_id: Uuid,
51 53 user_id: UserId,
52 54 ) -> Result<()> {
53 55 let owner = match target {
54 - GalleryTarget::Item => db::items::get_item_owner(&state.db, ItemId::from(target_id)).await?,
55 - GalleryTarget::Project => db::projects::get_project_by_id(&state.db, ProjectId::from(target_id))
56 + GalleryTarget::Item => db::items::get_item_owner(db, ItemId::from(target_id)).await?,
57 + GalleryTarget::Project => db::projects::get_project_by_id(db, ProjectId::from(target_id))
56 58 .await?
57 59 .map(|p| p.user_id),
58 60 };
@@ -78,16 +80,16 @@ struct GalleryListItem {
78 80 /// Drives the wizard manager (the public page renders the same rows server-side).
79 81 #[tracing::instrument(skip_all, name = "storage::gallery_list", fields(user_id = %user.id, %target_type, %target_id))]
80 82 pub(super) async fn gallery_list(
81 - State(state): State<AppState>,
83 + State(db): State<PgPool>,
82 84 AuthUser(user): AuthUser,
83 85 Path((target_type, target_id)): Path<(String, Uuid)>,
84 86 ) -> Result<impl IntoResponse> {
85 87 let target = GalleryTarget::parse(&target_type)?;
86 - require_owned(&state, target, target_id, user.id).await?;
88 + require_owned(&db, target, target_id, user.id).await?;
87 89
88 90 let rows = match target {
89 - GalleryTarget::Item => db::gallery_images::list_for_item(&state.db, ItemId::from(target_id)).await?,
90 - GalleryTarget::Project => db::gallery_images::list_for_project(&state.db, ProjectId::from(target_id)).await?,
91 + GalleryTarget::Item => db::gallery_images::list_for_item(&db, ItemId::from(target_id)).await?,
92 + GalleryTarget::Project => db::gallery_images::list_for_project(&db, ProjectId::from(target_id)).await?,
91 93 };
92 94 let items: Vec<GalleryListItem> = rows
93 95 .into_iter()
@@ -115,24 +117,25 @@ pub struct GalleryPresignRequest {
115 117 /// POST /api/gallery/presign — presign a gallery image upload.
116 118 #[tracing::instrument(skip_all, name = "storage::gallery_presign", fields(user_id = %user.id))]
117 119 pub(super) async fn gallery_presign(
118 - State(state): State<AppState>,
120 + State(db): State<PgPool>,
121 + State(storage): State<AppStorage>,
119 122 AuthUser(user): AuthUser,
120 123 Json(req): Json<GalleryPresignRequest>,
121 124 ) -> Result<impl IntoResponse> {
122 125 user.check_not_suspended()?;
123 - let s3 = state.require_s3()?;
126 + let s3 = storage.require_s3()?;
124 127 let target = GalleryTarget::parse(&req.target_type)?;
125 128
126 129 let file_type = FileType::Cover;
127 130 S3Client::validate_content_type(file_type, &req.content_type)?;
128 131 S3Client::validate_extension(file_type, &req.file_name)?;
129 132
130 - require_owned(&state, target, req.target_id, user.id).await?;
133 + require_owned(&db, target, req.target_id, user.id).await?;
131 134
132 135 // Early per-entity cap check (authoritative re-check happens at confirm).
133 136 let count = match target {
134 - GalleryTarget::Item => db::gallery_images::count_for_item(&state.db, ItemId::from(req.target_id)).await?,
135 - GalleryTarget::Project => db::gallery_images::count_for_project(&state.db, ProjectId::from(req.target_id)).await?,
137 + GalleryTarget::Item => db::gallery_images::count_for_item(&db, ItemId::from(req.target_id)).await?,
138 + GalleryTarget::Project => db::gallery_images::count_for_project(&db, ProjectId::from(req.target_id)).await?,
136 139 };
137 140 if count >= db::gallery_images::MAX_GALLERY_IMAGES {
138 141 return Err(AppError::BadRequest(format!(
@@ -141,7 +144,7 @@ pub(super) async fn gallery_presign(
141 144 )));
142 145 }
143 146
144 - db::creator_tiers::check_presign_allowed(&state.db, user.id, file_type).await?;
147 + db::creator_tiers::check_presign_allowed(&db, user.id, file_type).await?;
145 148
146 149 // Validate the declared size (if any) before signing it into Content-Length.
147 150 super::validate_declared_upload_size(req.file_size_bytes, file_type, None)?;
@@ -151,7 +154,7 @@ pub(super) async fn gallery_presign(
151 154 // uuid also keeps multiple gallery uploads from colliding.
152 155 let s3_key = S3Client::generate_staging_key(&req.file_name);
153 156
154 - db::pending_uploads::record_pending_upload(&state.db, user.id, &s3_key, "main").await?;
157 + db::pending_uploads::record_pending_upload(&db, user.id, &s3_key, "main").await?;
155 158
156 159 let expires_in = 3600;
157 160 let upload_url = s3
@@ -200,15 +203,18 @@ enum GalleryOutcome {
200 203 /// POST /api/gallery/confirm — finalize a gallery image upload.
201 204 #[tracing::instrument(skip_all, name = "storage::gallery_confirm", fields(user_id = %user.id))]
202 205 pub(super) async fn gallery_confirm(
203 - State(state): State<AppState>,
206 + State(db): State<PgPool>,
207 + State(storage): State<AppStorage>,
208 + State(config): State<Config>,
209 + State(scanning): State<Scanning>,
204 210 AuthUser(user): AuthUser,
205 211 Json(req): Json<GalleryConfirmRequest>,
206 212 ) -> Result<impl IntoResponse> {
207 213 user.check_not_suspended()?;
208 - let s3 = state.require_s3()?;
214 + let s3 = storage.require_s3()?;
209 215 let target = GalleryTarget::parse(&req.target_type)?;
210 216
211 - require_owned(&state, target, req.target_id, user.id).await?;
217 + require_owned(&db, target, req.target_id, user.id).await?;
212 218
213 219 // Ownership of the staging key is proved inside the confirm transaction below
214 220 // (after the replay short-circuit) via `pending_uploads` — a `staging/{uuid}`
@@ -224,7 +230,7 @@ pub(super) async fn gallery_confirm(
224 230 AppError::BadRequest("Could not determine file size. Please try uploading again.".to_string())
225 231 })?;
226 232 if file_size_bytes as u64 > FileType::Cover.max_size() {
227 - super::enqueue_s3_orphan(&state.db, &req.s3_key, crate::storage::S3Bucket::Main, "gallery_upload_rejected").await;
233 + super::enqueue_s3_orphan(&db, &req.s3_key, crate::storage::S3Bucket::Main, "gallery_upload_rejected").await;
228 234 return Err(AppError::BadRequest(format!(
229 235 "File exceeds maximum size of {} MB",
230 236 FileType::Cover.max_size() / (1024 * 1024)
@@ -233,28 +239,28 @@ pub(super) async fn gallery_confirm(
233 239
234 240 // Authoritative cap re-check (presign's was best-effort/UX).
235 241 let count = match target {
236 - GalleryTarget::Item => db::gallery_images::count_for_item(&state.db, ItemId::from(req.target_id)).await?,
237 - GalleryTarget::Project => db::gallery_images::count_for_project(&state.db, ProjectId::from(req.target_id)).await?,
242 + GalleryTarget::Item => db::gallery_images::count_for_item(&db, ItemId::from(req.target_id)).await?,
243 + GalleryTarget::Project => db::gallery_images::count_for_project(&db, ProjectId::from(req.target_id)).await?,
238 244 };
239 245 if count >= db::gallery_images::MAX_GALLERY_IMAGES {
240 - super::enqueue_s3_orphan(&state.db, &req.s3_key, crate::storage::S3Bucket::Main, "gallery_upload_rejected").await;
246 + super::enqueue_s3_orphan(&db, &req.s3_key, crate::storage::S3Bucket::Main, "gallery_upload_rejected").await;
241 247 return Err(AppError::BadRequest(format!(
242 248 "Gallery is full (max {} images)",
243 249 db::gallery_images::MAX_GALLERY_IMAGES
244 250 )));
245 251 }
246 252
247 - let max_storage = match db::creator_tiers::check_upload_allowed(&state.db, user.id, FileType::Cover, file_size_bytes).await {
253 + let max_storage = match db::creator_tiers::check_upload_allowed(&db, user.id, FileType::Cover, file_size_bytes).await {
248 254 Ok(max) => max,
249 255 Err(e) => {
250 - super::enqueue_s3_orphan(&state.db, &req.s3_key, crate::storage::S3Bucket::Main, "gallery_upload_rejected").await;
256 + super::enqueue_s3_orphan(&db, &req.s3_key, crate::storage::S3Bucket::Main, "gallery_upload_rejected").await;
251 257 return Err(e);
252 258 }
253 259 };
254 260
255 261 let image_url = storage::build_project_image_url(
256 262 s3.as_ref(),
257 - state.config.cdn_base_url.as_deref(),
263 + config.cdn_base_url.as_deref(),
258 264 &req.s3_key,
259 265 )
260 266 .await?;
@@ -281,7 +287,7 @@ pub(super) async fn gallery_confirm(
281 287 // Returns the existing row on a replayed confirm (idempotent, no re-charge)
282 288 // or the freshly-inserted id otherwise.
283 289 let committed: Result<GalleryOutcome> = async {
284 - let mut tx = state.db.begin().await?;
290 + let mut tx = db.begin().await?;
285 291 // Serialize concurrent confirms for this gallery and re-count INSIDE the
286 292 // tx, so two inserts that both passed the best-effort pre-check above
287 293 // can't push it over MAX_GALLERY_IMAGES (Run #14 Storage LOW). A failure
@@ -314,7 +320,7 @@ pub(super) async fn gallery_confirm(
314 320 // consumed — so only new rows are gated. A `staging/{uuid}` key has no
315 321 // entity in its path, so this lookup (not a prefix check) is what stops a
316 322 // confirm from stealing another user's staging object.
317 - if !db::pending_uploads::is_owned(&state.db, user.id, &req.s3_key, "main").await? {
323 + if !db::pending_uploads::is_owned(&db, user.id, &req.s3_key, "main").await? {
318 324 return Err(AppError::BadRequest("Invalid upload key".to_string()));
319 325 }
320 326 let count_in_tx = match target {
@@ -361,16 +367,17 @@ pub(super) async fn gallery_confirm(
361 367 }
362 368 Ok(GalleryOutcome::Inserted(id)) => id,
363 369 Err(e) => {
364 - super::enqueue_s3_orphan(&state.db, &req.s3_key, crate::storage::S3Bucket::Main, "gallery_image_insert_failed").await;
370 + super::enqueue_s3_orphan(&db, &req.s3_key, crate::storage::S3Bucket::Main, "gallery_image_insert_failed").await;
365 371 return Err(e);
366 372 }
367 373 };
368 374
369 - db::pending_uploads::remove_pending_upload(&state.db, user.id, &req.s3_key, "main").await?;
375 + db::pending_uploads::remove_pending_upload(&db, user.id, &req.s3_key, "main").await?;
370 376
371 377 // Scan enqueue AFTER the DB write commits (the chronic-ordering rule).
372 378 commit_upload(
373 - &state,
379 + &db,
380 + scanning.scanner.as_ref(),
374 381 CommitTarget::GalleryImage(id.into()),
375 382 &req.s3_key,
376 383 FileType::Cover,
@@ -379,7 +386,7 @@ pub(super) async fn gallery_confirm(
379 386 )
380 387 .await?;
381 388
382 - bump_target_cache(&state, target, req.target_id).await;
389 + bump_target_cache(&db, target, req.target_id).await;
383 390
384 391 Ok(Json(GalleryConfirmResponse { success: true, id, image_url, alt }))
385 392 }
@@ -391,7 +398,7 @@ pub(super) async fn gallery_confirm(
391 398 /// DELETE /api/gallery/image/{target_type}/{image_id} — remove one gallery image.
392 399 #[tracing::instrument(skip_all, name = "storage::gallery_delete", fields(user_id = %user.id, %target_type, %image_id))]
393 400 pub(super) async fn gallery_delete(
394 - State(state): State<AppState>,
401 + State(db): State<PgPool>,
395 402 AuthUser(user): AuthUser,
396 403 Path((target_type, image_id)): Path<(String, ImageId)>,
397 404 ) -> Result<impl IntoResponse> {
@@ -404,7 +411,7 @@ pub(super) async fn gallery_delete(
404 411 // between commit and a post-commit enqueue can't orphan the object with no
405 412 // durable record (Run #18 Storage B6 — matches delete_version / media_delete).
406 413 let deleted: Option<db::gallery_images::GalleryImage> = {
407 - let mut tx = state.db.begin().await?;
414 + let mut tx = db.begin().await?;
408 415 let row = match target {
409 416 GalleryTarget::Item => db::gallery_images::delete_for_item(&mut *tx, image_id, user.id).await?,
410 417 GalleryTarget::Project => db::gallery_images::delete_for_project(&mut *tx, image_id, user.id).await?,
@@ -447,39 +454,39 @@ pub struct GalleryReorderRequest {
447 454 /// POST /api/gallery/reorder — set gallery display order.
448 455 #[tracing::instrument(skip_all, name = "storage::gallery_reorder", fields(user_id = %user.id))]
449 456 pub(super) async fn gallery_reorder(
450 - State(state): State<AppState>,
457 + State(db): State<PgPool>,
451 458 AuthUser(user): AuthUser,
452 459 Json(req): Json<GalleryReorderRequest>,
453 460 ) -> Result<impl IntoResponse> {
454 461 user.check_not_suspended()?;
455 462 let target = GalleryTarget::parse(&req.target_type)?;
456 - require_owned(&state, target, req.target_id, user.id).await?;
463 + require_owned(&db, target, req.target_id, user.id).await?;
457 464
458 465 match target {
459 466 GalleryTarget::Item => {
460 - db::gallery_images::reorder_item(&state.db, ItemId::from(req.target_id), &req.ordered_ids).await?
467 + db::gallery_images::reorder_item(&db, ItemId::from(req.target_id), &req.ordered_ids).await?
461 468 }
462 469 GalleryTarget::Project => {
463 - db::gallery_images::reorder_project(&state.db, ProjectId::from(req.target_id), &req.ordered_ids).await?
470 + db::gallery_images::reorder_project(&db, ProjectId::from(req.target_id), &req.ordered_ids).await?
464 471 }
465 472 }
466 473
467 - bump_target_cache(&state, target, req.target_id).await;
474 + bump_target_cache(&db, target, req.target_id).await;
468 475 Ok(Json(serde_json::json!({ "success": true })))
469 476 }
470 477
471 478 /// Bump the public-page cache generation for the affected project (best-effort).
472 - async fn bump_target_cache(state: &AppState, target: GalleryTarget, target_id: Uuid) {
479 + async fn bump_target_cache(db: &PgPool, target: GalleryTarget, target_id: Uuid) {
473 480 let project_id = match target {
474 481 GalleryTarget::Project => Some(ProjectId::from(target_id)),
475 - GalleryTarget::Item => db::items::get_item_by_id(&state.db, ItemId::from(target_id))
482 + GalleryTarget::Item => db::items::get_item_by_id(db, ItemId::from(target_id))
476 483 .await
477 484 .ok()
478 485 .flatten()
479 486 .map(|i| i.project_id),
480 487 };
481 488 if let Some(pid) = project_id
482 - && let Err(e) = db::projects::bump_cache_generation(&state.db, pid).await
489 + && let Err(e) = db::projects::bump_cache_generation(db, pid).await
483 490 {
484 491 tracing::warn!(project_id = %pid, error = ?e, "failed to bump cache generation after gallery change");
485 492 }
M server/src/routes/storage/images.rs +60 -48
@@ -6,13 +6,15 @@ use axum::{
6 6 Json,
7 7 };
8 8 use serde::{Deserialize, Serialize};
9 + use sqlx::PgPool;
9 10
10 11 use crate::{
11 12 auth::AuthUser,
13 + config::Config,
12 14 db::{self, ItemId, ProjectId},
13 15 error::{AppError, Result, ResultExt},
14 16 storage::{self, FileType, S3Client, CACHE_CONTROL_IMMUTABLE},
15 - AppState,
17 + AppStorage, Scanning,
16 18 };
17 19
18 20 use super::{commit_upload, CommitTarget, PresignUploadResponse};
@@ -61,19 +63,20 @@ pub struct ItemImageConfirmRequest {
61 63 /// Requires authentication. User must own the project.
62 64 #[tracing::instrument(skip_all, name = "storage::project_image_presign", fields(user_id = %user.id))]
63 65 pub(super) async fn project_image_presign(
64 - State(state): State<AppState>,
66 + State(db): State<PgPool>,
67 + State(storage): State<AppStorage>,
65 68 AuthUser(user): AuthUser,
66 69 Json(req): Json<ProjectImagePresignRequest>,
67 70 ) -> Result<impl IntoResponse> {
68 71 user.check_not_suspended()?;
69 - let s3 = state.require_s3()?;
72 + let s3 = storage.require_s3()?;
70 73
71 74 let file_type = FileType::Cover;
72 75 S3Client::validate_content_type(file_type, &req.content_type)?;
73 76 S3Client::validate_extension(file_type, &req.file_name)?;
74 77
75 78 // Verify user owns the project
76 - let project = db::projects::get_project_by_id(&state.db, req.project_id)
79 + let project = db::projects::get_project_by_id(&db, req.project_id)
77 80 .await?
78 81 .ok_or(AppError::NotFound)?;
79 82
@@ -82,14 +85,14 @@ pub(super) async fn project_image_presign(
82 85 }
83 86
84 87 // Early quota check
85 - db::creator_tiers::check_presign_allowed(&state.db, user.id, file_type).await?;
88 + db::creator_tiers::check_presign_allowed(&db, user.id, file_type).await?;
86 89
87 90 // Staging key (unserved); the scan worker promotes it to the content key and
88 91 // rebuilds the public `cover_image_url` on a Clean verdict (C1).
89 92 let s3_key = S3Client::generate_staging_key(&req.file_name);
90 93
91 94 // Track the pending upload so the reaper can clean it up if never confirmed
92 - db::pending_uploads::record_pending_upload(&state.db, user.id, &s3_key, "main").await?;
95 + db::pending_uploads::record_pending_upload(&db, user.id, &s3_key, "main").await?;
93 96
94 97 let expires_in = 3600;
95 98 let upload_url = s3.presign_upload(&s3_key, &req.content_type, Some(expires_in), Some(CACHE_CONTROL_IMMUTABLE), None)
@@ -112,15 +115,18 @@ pub(super) async fn project_image_presign(
112 115 /// Requires authentication. User must own the project.
113 116 #[tracing::instrument(skip_all, name = "storage::project_image_confirm", fields(user_id = %user.id))]
114 117 pub(super) async fn project_image_confirm(
115 - State(state): State<AppState>,
118 + State(db): State<PgPool>,
119 + State(storage): State<AppStorage>,
120 + State(config): State<Config>,
121 + State(scanning): State<Scanning>,
116 122 AuthUser(user): AuthUser,
117 123 Json(req): Json<ProjectImageConfirmRequest>,
118 124 ) -> Result<impl IntoResponse> {
119 125 user.check_not_suspended()?;
120 - let s3 = state.require_s3()?;
126 + let s3 = storage.require_s3()?;
121 127
122 128 // Verify user owns the project
123 - let project = db::projects::get_project_by_id(&state.db, req.project_id)
129 + let project = db::projects::get_project_by_id(&db, req.project_id)
124 130 .await?
125 131 .ok_or(AppError::NotFound)?;
126 132
@@ -133,7 +139,7 @@ pub(super) async fn project_image_confirm(
133 139 // presign — not a prefix check. Gate before the size-reject path so an
134 140 // unowned (at most another user's in-flight) staging object is never enqueued
135 141 // for deletion.
136 - if !db::pending_uploads::is_owned(&state.db, user.id, &req.s3_key, "main").await? {
142 + if !db::pending_uploads::is_owned(&db, user.id, &req.s3_key, "main").await? {
137 143 return Err(AppError::BadRequest("Invalid upload key".to_string()));
138 144 }
139 145
@@ -143,7 +149,7 @@ pub(super) async fn project_image_confirm(
143 149 AppError::BadRequest("Upload not found. Please try uploading again.".to_string())
144 150 })?;
145 151 if file_size_bytes as u64 > FileType::Cover.max_size() {
146 - super::enqueue_s3_orphan(&state.db, &req.s3_key, crate::storage::S3Bucket::Main, "image_upload_rejected").await;
152 + super::enqueue_s3_orphan(&db, &req.s3_key, crate::storage::S3Bucket::Main, "image_upload_rejected").await;
147 153 return Err(AppError::BadRequest(format!(
148 154 "File exceeds maximum size of {} MB",
149 155 FileType::Cover.max_size() / (1024 * 1024)
@@ -156,15 +162,15 @@ pub(super) async fn project_image_confirm(
156 162 if let Some(ref cur_url) = project.cover_image_url
157 163 && let Some(cur_key) = storage::extract_s3_key_from_url(
158 164 cur_url,
159 - state.config.cdn_base_url.as_deref(),
165 + config.cdn_base_url.as_deref(),
160 166 Some(s3.bucket()),
161 - state.config.storage.as_ref().map(|c| c.endpoint.as_str()),
167 + config.storage.as_ref().map(|c| c.endpoint.as_str()),
162 168 )
163 169 && cur_key == req.s3_key
164 170 {
165 171 // Still clear pending_uploads — orphan reaper would otherwise delete
166 172 // the live S3 object 24h later (Run #7 HIGH-1).
167 - if let Err(e) = db::pending_uploads::remove_pending_upload(&state.db, user.id, &req.s3_key, "main").await {
173 + if let Err(e) = db::pending_uploads::remove_pending_upload(&db, user.id, &req.s3_key, "main").await {
168 174 tracing::warn!(error = ?e, key = %req.s3_key, "remove_pending_upload failed on idempotent re-confirm");
169 175 }
170 176 return Ok(Json(ProjectImageConfirmResponse {
@@ -174,10 +180,10 @@ pub(super) async fn project_image_confirm(
174 180 }
175 181
176 182 // Enforce tier-based limits
177 - let max_storage = match db::creator_tiers::check_upload_allowed(&state.db, user.id, FileType::Cover, file_size_bytes).await {
183 + let max_storage = match db::creator_tiers::check_upload_allowed(&db, user.id, FileType::Cover, file_size_bytes).await {
178 184 Ok(max) => max,
179 185 Err(e) => {
180 - super::enqueue_s3_orphan(&state.db, &req.s3_key, crate::storage::S3Bucket::Main, "image_upload_rejected").await;
186 + super::enqueue_s3_orphan(&db, &req.s3_key, crate::storage::S3Bucket::Main, "image_upload_rejected").await;
181 187 return Err(e);
182 188 }
183 189 };
@@ -191,9 +197,9 @@ pub(super) async fn project_image_confirm(
191 197 let old_key_to_delete: Option<String> = if let Some(ref old_url) = project.cover_image_url
192 198 && let Some(old_key) = storage::extract_s3_key_from_url(
193 199 old_url,
194 - state.config.cdn_base_url.as_deref(),
200 + config.cdn_base_url.as_deref(),
195 201 Some(s3.bucket()),
196 - state.config.storage.as_ref().map(|c| c.endpoint.as_str()),
202 + config.storage.as_ref().map(|c| c.endpoint.as_str()),
197 203 )
198 204 {
199 205 match s3.object_size(&old_key).await {
@@ -210,7 +216,7 @@ pub(super) async fn project_image_confirm(
210 216 Err(e) => {
211 217 // S3 probe failed (transient). Refuse to write — letting a probe failure
212 218 // silently over-count storage on every replace is the bug this branch exists to prevent.
213 - super::enqueue_s3_orphan(&state.db, &req.s3_key, crate::storage::S3Bucket::Main, "image_upload_rejected").await;
219 + super::enqueue_s3_orphan(&db, &req.s3_key, crate::storage::S3Bucket::Main, "image_upload_rejected").await;
214 220 tracing::warn!(key = %old_key, error = ?e, "S3 probe failed during project image replace");
215 221 return Err(AppError::ServiceUnavailable(
216 222 "Could not verify previous image. Please try again.".to_string(),
@@ -224,7 +230,7 @@ pub(super) async fn project_image_confirm(
224 230 // Build permanent URL (async, before the tx).
225 231 let image_url = storage::build_project_image_url(
226 232 s3.as_ref(),
227 - state.config.cdn_base_url.as_deref(),
233 + config.cdn_base_url.as_deref(),
228 234 &req.s3_key,
229 235 ).await?;
230 236
@@ -234,7 +240,7 @@ pub(super) async fn project_image_confirm(
234 240 // stays AFTER the commit. `Ok(false)` = ownership filter no-matched (project
235 241 // deleted/transferred mid-flight); the tx rolled back, nothing charged.
236 242 let committed: Result<bool> = async {
237 - let mut tx = state.db.begin().await?;
243 + let mut tx = db.begin().await?;
238 244 db::creator_tiers::try_apply_storage_on(
239 245 &mut tx, user.id, replace_old_size, file_size_bytes, max_storage,
240 246 ).await?;
@@ -270,11 +276,11 @@ pub(super) async fn project_image_confirm(
270 276 match committed {
271 277 Err(e) => {
272 278 // tx rolled back — counter unchanged. Orphan-queue the new key for cleanup.
273 - super::enqueue_s3_orphan(&state.db, &req.s3_key, crate::storage::S3Bucket::Main, "project_image_update_failed").await;
279 + super::enqueue_s3_orphan(&db, &req.s3_key, crate::storage::S3Bucket::Main, "project_image_update_failed").await;
274 280 return Err(e);
275 281 }
276 282 Ok(false) => {
277 - super::enqueue_s3_orphan(&state.db, &req.s3_key, crate::storage::S3Bucket::Main, "project_image_update_failed").await;
283 + super::enqueue_s3_orphan(&db, &req.s3_key, crate::storage::S3Bucket::Main, "project_image_update_failed").await;
278 284 return Err(AppError::BadRequest(
279 285 "Project was modified concurrently. Please try uploading again.".to_string(),
280 286 ));
@@ -283,14 +289,15 @@ pub(super) async fn project_image_confirm(
283 289 }
284 290
285 291 // Clear the pending upload record now that the upload is committed
286 - db::pending_uploads::remove_pending_upload(&state.db, user.id, &req.s3_key, "main").await?;
292 + db::pending_uploads::remove_pending_upload(&db, user.id, &req.s3_key, "main").await?;
287 293
288 294 // (The old S3 object was enqueued for deletion inside the commit tx above.)
289 295
290 296 // Scan enqueue AFTER the DB write commits (Phase 5 chronic fix — the same
291 297 // ordering rule that uploads/versions/media follow via `commit_upload`).
292 298 commit_upload(
293 - &state,
299 + &db,
300 + scanning.scanner.as_ref(),
294 301 CommitTarget::ProjectImage(req.project_id),
295 302 &req.s3_key,
296 303 FileType::Cover,
@@ -299,7 +306,7 @@ pub(super) async fn project_image_confirm(
299 306 ).await?;
300 307
301 308 // Bump cache
302 - db::projects::bump_cache_generation(&state.db, req.project_id).await?;
309 + db::projects::bump_cache_generation(&db, req.project_id).await?;
303 310
304 311 tracing::info!(
305 312 "Project image confirmed: project={}, key={}, size={}",
@@ -321,19 +328,20 @@ pub(super) async fn project_image_confirm(
321 328 /// Requires authentication. User must own the item.
322 329 #[tracing::instrument(skip_all, name = "storage::item_image_presign", fields(user_id = %user.id))]
323 330 pub(super) async fn item_image_presign(
324 - State(state): State<AppState>,
331 + State(db): State<PgPool>,
332 + State(storage): State<AppStorage>,
325 333 AuthUser(user): AuthUser,
326 334 Json(req): Json<ItemImagePresignRequest>,
327 335 ) -> Result<impl IntoResponse> {
328 336 user.check_not_suspended()?;
329 - let s3 = state.require_s3()?;
337 + let s3 = storage.require_s3()?;
330 338
331 339 let file_type = FileType::Cover;
332 340 S3Client::validate_content_type(file_type, &req.content_type)?;
333 341 S3Client::validate_extension(file_type, &req.file_name)?;
334 342
335 343 // Verify user owns the item
336 - let owner = db::items::get_item_owner(&state.db, req.item_id)
344 + let owner = db::items::get_item_owner(&db, req.item_id)
337 345 .await?
338 346 .ok_or(AppError::NotFound)?;
339 347
@@ -342,14 +350,14 @@ pub(super) async fn item_image_presign(
342 350 }
343 351
344 352 // Early quota check
345 - db::creator_tiers::check_presign_allowed(&state.db, user.id, file_type).await?;
353 + db::creator_tiers::check_presign_allowed(&db, user.id, file_type).await?;
346 354
347 355 // Staging key (unserved); the scan worker promotes it to the content key and
348 356 // rebuilds the public `cover_image_url` on a Clean verdict (C1).
349 357 let s3_key = S3Client::generate_staging_key(&req.file_name);
350 358
351 359 // Track the pending upload so the reaper can clean it up if never confirmed
352 - db::pending_uploads::record_pending_upload(&state.db, user.id, &s3_key, "main").await?;
360 + db::pending_uploads::record_pending_upload(&db, user.id, &s3_key, "main").await?;
353 361
354 362 let expires_in = 3600;
355 363 let upload_url = s3.presign_upload(&s3_key, &req.content_type, Some(expires_in), Some(CACHE_CONTROL_IMMUTABLE), None)
@@ -372,15 +380,18 @@ pub(super) async fn item_image_presign(
372 380 /// Requires authentication. User must own the item.
373 381 #[tracing::instrument(skip_all, name = "storage::item_image_confirm", fields(user_id = %user.id))]
374 382 pub(super) async fn item_image_confirm(
375 - State(state): State<AppState>,
383 + State(db): State<PgPool>,
384 + State(storage): State<AppStorage>,
385 + State(config): State<Config>,
386 + State(scanning): State<Scanning>,
376 387 AuthUser(user): AuthUser,
377 388 Json(req): Json<ItemImageConfirmRequest>,
378 389 ) -> Result<impl IntoResponse> {
379 390 user.check_not_suspended()?;
380 - let s3 = state.require_s3()?;
391 + let s3 = storage.require_s3()?;
381 392
382 393 // Verify user owns the item
383 - let owner = db::items::get_item_owner(&state.db, req.item_id)
394 + let owner = db::items::get_item_owner(&db, req.item_id)
384 395 .await?
385 396 .ok_or(AppError::NotFound)?;
386 397
@@ -393,7 +404,7 @@ pub(super) async fn item_image_confirm(
393 404 // was minted for THIS user's upload flow — is proved via the `pending_uploads`
394 405 // row recorded at presign, not a prefix check. Gate before the size-reject
395 406 // path so an unowned staging object is never enqueued for deletion.
396 - if !db::pending_uploads::is_owned(&state.db, user.id, &req.s3_key, "main").await? {
407 + if !db::pending_uploads::is_owned(&db, user.id, &req.s3_key, "main").await? {
397 408 return Err(AppError::BadRequest(
398 409 "Invalid upload key".to_string(),
399 410 ));
@@ -405,7 +416,7 @@ pub(super) async fn item_image_confirm(
405 416 AppError::BadRequest("Upload not found. Please try uploading again.".to_string())
406 417 })?;
407 418 if file_size_bytes as u64 > FileType::Cover.max_size() {
408 - super::enqueue_s3_orphan(&state.db, &req.s3_key, crate::storage::S3Bucket::Main, "image_upload_rejected").await;
419 + super::enqueue_s3_orphan(&db, &req.s3_key, crate::storage::S3Bucket::Main, "image_upload_rejected").await;
409 420 return Err(AppError::BadRequest(format!(
410 421 "File exceeds maximum size of {} MB",
411 422 FileType::Cover.max_size() / (1024 * 1024)
@@ -416,13 +427,13 @@ pub(super) async fn item_image_confirm(
416 427 // the tier/cap check — a creator at their storage cap must still be able to
417 428 // re-confirm a cover they already own (Run 9; matches project_image_confirm's
418 429 // ordering). Otherwise capture the existing cover for atomic replacement below.
419 - let existing_item = db::items::get_item_by_id(&state.db, req.item_id).await?;
430 + let existing_item = db::items::get_item_by_id(&db, req.item_id).await?;
420 431 if let Some(ref item) = existing_item
421 432 && item.cover_s3_key.as_deref() == Some(&req.s3_key)
422 433 {
423 434 // Still clear pending_uploads — orphan reaper would otherwise delete
424 435 // the live S3 object 24h later (Run #7 HIGH-1).
425 - if let Err(e) = db::pending_uploads::remove_pending_upload(&state.db, user.id, &req.s3_key, "main").await {
436 + if let Err(e) = db::pending_uploads::remove_pending_upload(&db, user.id, &req.s3_key, "main").await {
426 437 tracing::warn!(error = ?e, key = %req.s3_key, "remove_pending_upload failed on idempotent re-confirm");
427 438 }
428 439 return Ok(Json(super::images::ProjectImageConfirmResponse {
@@ -432,10 +443,10 @@ pub(super) async fn item_image_confirm(
432 443 }
433 444
434 445 // Enforce tier-based limits
435 - let max_storage = match db::creator_tiers::check_upload_allowed(&state.db, user.id, FileType::Cover, file_size_bytes).await {
446 + let max_storage = match db::creator_tiers::check_upload_allowed(&db, user.id, FileType::Cover, file_size_bytes).await {
436 447 Ok(max) => max,
437 448 Err(e) => {
438 - super::enqueue_s3_orphan(&state.db, &req.s3_key, crate::storage::S3Bucket::Main, "image_upload_rejected").await;
449 + super::enqueue_s3_orphan(&db, &req.s3_key, crate::storage::S3Bucket::Main, "image_upload_rejected").await;
439 450 return Err(e);
440 451 }
441 452 };
@@ -458,7 +469,7 @@ pub(super) async fn item_image_confirm(
458 469
459 470 let image_url = storage::build_project_image_url(
460 471 s3.as_ref(),
461 - state.config.cdn_base_url.as_deref(),
472 + config.cdn_base_url.as_deref(),
462 473 &req.s3_key,
463 474 ).await?;
464 475
@@ -468,7 +479,7 @@ pub(super) async fn item_image_confirm(
468 479 // `commit_upload` stays AFTER the commit. `Ok(false)` = ownership filter
469 480 // no-matched (item deleted/moved mid-flight); the tx rolled back, nothing charged.
470 481 let committed: Result<bool> = async {
471 - let mut tx = state.db.begin().await?;
482 + let mut tx = db.begin().await?;
472 483 db::creator_tiers::try_apply_storage_on(
473 484 &mut tx, user.id, replace_old_size, file_size_bytes, max_storage,
474 485 ).await?;
@@ -499,11 +510,11 @@ pub(super) async fn item_image_confirm(
499 510
500 511 match committed {
501 512 Err(e) => {
502 - super::enqueue_s3_orphan(&state.db, &req.s3_key, crate::storage::S3Bucket::Main, "item_image_update_failed").await;
513 + super::enqueue_s3_orphan(&db, &req.s3_key, crate::storage::S3Bucket::Main, "item_image_update_failed").await;
503 514 return Err(e);
504 515 }
505 516 Ok(false) => {
506 - super::enqueue_s3_orphan(&state.db, &req.s3_key, crate::storage::S3Bucket::Main, "item_image_update_failed").await;
517 + super::enqueue_s3_orphan(&db, &req.s3_key, crate::storage::S3Bucket::Main, "item_image_update_failed").await;
507 518 return Err(AppError::BadRequest(
508 519 "Item was modified concurrently. Please try uploading again.".to_string(),
509 520 ));
@@ -511,14 +522,15 @@ pub(super) async fn item_image_confirm(
511 522 Ok(true) => {}
512 523 }
513 524
514 - db::pending_uploads::remove_pending_upload(&state.db, user.id, &req.s3_key, "main").await?;
525 + db::pending_uploads::remove_pending_upload(&db, user.id, &req.s3_key, "main").await?;
515 526
516 527 // (The old cover was enqueued for deletion inside the commit tx above.)
517 528
518 529 // Scan enqueue + scan_status flip AFTER the DB write — same ordering rule
519 530 // as uploads/versions/media. The Run #6 audit caught this same bug here.
520 531 commit_upload(
521 - &state,
532 + &db,
533 + scanning.scanner.as_ref(),
522 534 CommitTarget::ItemImage(req.item_id),
523 535 &req.s3_key,
524 536 FileType::Cover,
@@ -527,8 +539,8 @@ pub(super) async fn item_image_confirm(
527 539 ).await?;
528 540
529 541 // Bump project cache
530 - if let Some(item) = db::items::get_item_by_id(&state.db, req.item_id).await?
531 - && let Err(e) = db::projects::bump_cache_generation(&state.db, item.project_id).await
542 + if let Some(item) = db::items::get_item_by_id(&db, req.item_id).await?
543 + && let Err(e) = db::projects::bump_cache_generation(&db, item.project_id).await
532 544 {
533 545 tracing::warn!(project_id = %item.project_id, error = ?e, "failed to bump cache generation after image upload");
534 546 }
M server/src/routes/storage/media.rs +34 -26
@@ -11,13 +11,15 @@ use axum::{
11 11 Json,
12 12 };
13 13 use serde::{Deserialize, Serialize};
14 + use sqlx::PgPool;
14 15
15 16 use crate::{
16 17 auth::AuthUser,
18 + config::Config,
17 19 db::{self, MediaFileId},
18 20 error::{AppError, Result, ResultExt},
19 21 storage::{sanitize_filename, sanitize_folder, FileType, S3Client, CACHE_CONTROL_IMMUTABLE},
20 - AppState,
22 + AppStorage, Scanning,
21 23 };
22 24
23 25 use super::{commit_upload, CommitTarget, ConfirmUploadResponse, PresignUploadResponse};
@@ -134,12 +136,13 @@ fn file_to_response(f: &db::DbMediaFile, cdn_base: &str) -> MediaFileResponse {
134 136 /// POST /api/media/presign
135 137 #[tracing::instrument(skip_all, name = "media::presign", fields(user_id = %user.id))]
136 138 pub(super) async fn media_presign(
137 - State(state): State<AppState>,
139 + State(db): State<PgPool>,
140 + State(storage): State<AppStorage>,
138 141 AuthUser(user): AuthUser,
139 142 Json(req): Json<MediaPresignRequest>,
140 143 ) -> Result<impl IntoResponse> {
141 144 user.check_not_suspended()?;
142 - let s3 = state.require_s3()?;
145 + let s3 = storage.require_s3()?;
143 146
144 147 let (media_type, file_type) = classify_media(&req.content_type)?;
145 148 let _ = media_type; // used at confirm time
@@ -157,7 +160,7 @@ pub(super) async fn media_presign(
157 160 }
158 161
159 162 // Early quota check — images bypass tier, video requires BigFiles+
160 - db::creator_tiers::check_presign_allowed(&state.db, user.id, file_type).await?;
163 + db::creator_tiers::check_presign_allowed(&db, user.id, file_type).await?;
161 164
162 165 // Validate the declared size against the static per-type cap AND the user's
163 166 // tier per-file cap before signing it into Content-Length, so an oversize
@@ -165,7 +168,7 @@ pub(super) async fn media_presign(
165 168 // and only caught at confirm. `get_effective_max_file_bytes` returns None for
166 169 // images (they bypass the tier cap) and the tier limit for video.
167 170 let max_file_bytes =
168 - db::creator_tiers::get_effective_max_file_bytes(&state.db, user.id, file_type).await?;
171 + db::creator_tiers::get_effective_max_file_bytes(&db, user.id, file_type).await?;
169 172 super::validate_declared_upload_size(req.file_size_bytes, file_type, max_file_bytes)?;
170 173
171 174 // Filename uniqueness is enforced at confirm time by the
@@ -185,7 +188,7 @@ pub(super) async fn media_presign(
185 188 let s3_key = S3Client::generate_staging_key(&req.file_name);
186 189
187 190 // Track the pending upload so the reaper can clean it up if never confirmed
188 - db::pending_uploads::record_pending_upload(&state.db, user.id, &s3_key, "main").await?;
191 + db::pending_uploads::record_pending_upload(&db, user.id, &s3_key, "main").await?;
189 192
190 193 let expires_in = 3600;
191 194 let upload_url = s3
@@ -207,12 +210,14 @@ pub(super) async fn media_presign(
207 210 /// POST /api/media/confirm
208 211 #[tracing::instrument(skip_all, name = "media::confirm", fields(user_id = %user.id))]
209 212 pub(super) async fn media_confirm(
210 - State(state): State<AppState>,
213 + State(db): State<PgPool>,
214 + State(storage): State<AppStorage>,
215 + State(scanning): State<Scanning>,
211 216 AuthUser(user): AuthUser,
212 217 Json(req): Json<MediaConfirmRequest>,
213 218 ) -> Result<impl IntoResponse> {
214 219 user.check_not_suspended()?;
215 - let s3 = state.require_s3()?;
220 + let s3 = storage.require_s3()?;
216 221
217 222 let (media_type, file_type) = classify_media(&req.content_type)?;
218 223
@@ -225,7 +230,7 @@ pub(super) async fn media_confirm(
225 230 // not a prefix check. Gate before the sniff/size-reject paths so an unowned
226 231 // (at most another user's in-flight) staging object is never enqueued for
227 232 // deletion.
228 - if !db::pending_uploads::is_owned(&state.db, user.id, &req.s3_key, "main").await? {
233 + if !db::pending_uploads::is_owned(&db, user.id, &req.s3_key, "main").await? {
229 234 return Err(AppError::BadRequest(
230 235 "Invalid upload key".to_string(),
231 236 ));
@@ -243,7 +248,7 @@ pub(super) async fn media_confirm(
243 248 AppError::BadRequest("Could not determine file size. Please try uploading again.".to_string())
244 249 })?;
245 250 if file_size_bytes as u64 > file_type.max_size() {
246 - super::enqueue_s3_orphan(&state.db, &req.s3_key, crate::storage::S3Bucket::Main, "media_upload_rejected").await;
251 + super::enqueue_s3_orphan(&db, &req.s3_key, crate::storage::S3Bucket::Main, "media_upload_rejected").await;
247 252 return Err(AppError::BadRequest(format!(
248 253 "File exceeds maximum size of {} MB",
249 254 file_type.max_size() / (1024 * 1024)
@@ -279,7 +284,7 @@ pub(super) async fn media_confirm(
279 284 _ => false,
280 285 };
281 286 if mismatch {
282 - super::enqueue_s3_orphan(&state.db, &req.s3_key, crate::storage::S3Bucket::Main, "media_content_type_mismatch").await;
287 + super::enqueue_s3_orphan(&db, &req.s3_key, crate::storage::S3Bucket::Main, "media_content_type_mismatch").await;
283 288 let sniffed = match detected {
284 289 Some(infer::MatcherType::Image) => "image",
285 290 Some(infer::MatcherType::Video) => "video",
@@ -293,13 +298,13 @@ pub(super) async fn media_confirm(
293 298
294 299 // Tier enforcement
295 300 let max_storage = match db::creator_tiers::check_upload_allowed(
296 - &state.db, user.id, file_type, file_size_bytes,
301 + &db, user.id, file_type, file_size_bytes,
297 302 )
298 303 .await
299 304 {
300 305 Ok(max) => max,
301 306 Err(e) => {
302 - super::enqueue_s3_orphan(&state.db, &req.s3_key, crate::storage::S3Bucket::Main, "media_upload_rejected").await;
307 + super::enqueue_s3_orphan(&db, &req.s3_key, crate::storage::S3Bucket::Main, "media_upload_rejected").await;
303 308 return Err(e);
304 309 }
305 310 };
@@ -331,7 +336,7 @@ pub(super) async fn media_confirm(
331 336 // The unique index on (user_id, folder, filename) raises 23505 inside the
332 337 // tx; we catch the typed error after rollback and report a clean message.
333 338 let tx_result: Result<db::DbMediaFile> = async {
334 - let mut tx = state.db.begin().await?;
339 + let mut tx = db.begin().await?;
335 340 db::creator_tiers::try_increment_storage_on(&mut tx, user.id, file_size_bytes, max_storage).await?;
336 341 db::pending_uploads::remove_pending_upload(&mut *tx, user.id, &req.s3_key, "main").await?;
337 342 let row = db::media_files::create(
@@ -377,7 +382,7 @@ pub(super) async fn media_confirm(
377 382 }
378 383 // Any other failure: the tx rolled back and no row references this
379 384 // freshly-uploaded object, so it's a genuine orphan — clean it up.
380 - super::enqueue_s3_orphan(&state.db, &req.s3_key, crate::storage::S3Bucket::Main, "media_upload_rejected").await;
385 + super::enqueue_s3_orphan(&db, &req.s3_key, crate::storage::S3Bucket::Main, "media_upload_rejected").await;
381 386 return Err(e);
382 387 }
383 388 };
@@ -386,7 +391,8 @@ pub(super) async fn media_confirm(
386 391 // `commit_upload` helper. Always flips status (worker-or-now), so a no-scanner
387 392 // dev/test environment doesn't leave the row Pending forever.
388 393 let scan_status = commit_upload(
389 - &state,
394 + &db,
395 + scanning.scanner.as_ref(),
390 396 CommitTarget::Media(inserted.id),
391 397 &req.s3_key,
392 398 file_type,
@@ -412,7 +418,8 @@ pub(super) async fn media_confirm(
412 418 /// GET /api/media?folder={folder}
413 419 #[tracing::instrument(skip_all, name = "media::list", fields(user_id = %user.id))]
414 420 pub(super) async fn media_list(
415 - State(state): State<AppState>,
421 + State(db): State<PgPool>,
422 + State(config): State<Config>,
416 423 AuthUser(user): AuthUser,
417 424 Query(query): Query<MediaListQuery>,
418 425 ) -> Result<impl IntoResponse> {
@@ -420,7 +427,7 @@ pub(super) async fn media_list(
420 427 // without CDN config still render plausible URLs. In production a
421 428 // missing `cdn_base_url` is an operator-side misconfiguration; we log
422 429 // a WARN once per process so it surfaces without blocking the request.
423 - let cdn_base = if let Some(base) = state.config.cdn_base_url.as_deref() {
430 + let cdn_base = if let Some(base) = config.cdn_base_url.as_deref() {
424 431 base
425 432 } else {
426 433 static WARNED: std::sync::OnceLock<()> = std::sync::OnceLock::new();
@@ -433,13 +440,13 @@ pub(super) async fn media_list(
433 440 };
434 441
435 442 let files = db::media_files::list_by_user_folder(
436 - &state.db,
443 + &db,
437 444 user.id,
438 445 query.folder.as_deref(),
439 446 )
440 447 .await?;
441 448
442 - let folders = db::media_files::list_folders(&state.db, user.id).await?;
449 + let folders = db::media_files::list_folders(&db, user.id).await?;
443 450
444 451 let file_responses: Vec<MediaFileResponse> = files
445 452 .iter()
@@ -457,10 +464,10 @@ pub(super) async fn media_list(
457 464 /// GET /api/media/folders
458 465 #[tracing::instrument(skip_all, name = "media::folders", fields(user_id = %user.id))]
459 466 pub(super) async fn media_folders(
460 - State(state): State<AppState>,
467 + State(db): State<PgPool>,
461 468 AuthUser(user): AuthUser,
462 469 ) -> Result<impl IntoResponse> {
463 - let folders = db::media_files::list_folders(&state.db, user.id).await?;
470 + let folders = db::media_files::list_folders(&db, user.id).await?;
464 471 Ok(Json(MediaFoldersResponse { folders }))
465 472 }
466 473
@@ -469,16 +476,17 @@ pub(super) async fn media_folders(
469 476 /// DELETE /api/media/{id}
470 477 #[tracing::instrument(skip_all, name = "media::delete", fields(user_id = %user.id, media_id = %id))]
471 478 pub(super) async fn media_delete(
472 - State(state): State<AppState>,
479 + State(db): State<PgPool>,
480 + State(storage): State<AppStorage>,
473 481 AuthUser(user): AuthUser,
474 482 Path(id): Path<MediaFileId>,
475 483 ) -> Result<impl IntoResponse> {
476 484 user.check_not_suspended()?;
477 485 // Require S3 to be configured, but the actual delete goes through the
478 486 // durable queue (the sanctioned deletion path) rather than a direct call.
479 - state.require_s3()?;
487 + storage.require_s3()?;
480 488
481 - let file = db::media_files::get_by_id(&state.db, id)
489 + let file = db::media_files::get_by_id(&db, id)
482 490 .await?
483 491 .ok_or(AppError::NotFound)?;
484 492
@@ -503,7 +511,7 @@ pub(super) async fn media_delete(
503 511 // with no durable record (Run #18 Storage B6 — the same in-tx ordering
504 512 // delete_version adopted). The refund + enqueue use the DELETE's own returned
505 513 // row (`deleted`), not the pre-tx `get_by_id` read.
506 - let mut tx = state.db.begin().await?;
514 + let mut tx = db.begin().await?;
507 515 let deleted = db::media_files::delete(&mut *tx, id, user.id).await?;
508 516 if let Some(ref row) = deleted {
509 517 db::creator_tiers::decrement_storage_used(&mut *tx, user.id, row.file_size_bytes).await?;
M server/src/routes/storage/mod.rs +54 -33
@@ -7,19 +7,24 @@ pub(crate) mod media;
7 7 mod uploads;
8 8 mod versions;
9 9
10 + use std::sync::Arc;
11 +
10 12 use axum::routing::get;
11 13 use serde::Serialize;
14 + use sqlx::PgPool;
12 15 use tower_governor::GovernorLayer;
13 16 use uuid::Uuid;
14 17
15 18 use crate::{
19 + config::Config,
16 20 constants,
17 21 csrf::{delete_csrf, post_csrf, CsrfRouter},
18 22 db,
19 23 db::scan_jobs::ScanTargetKind,
20 24 error::{AppError, Result},
25 + scanning::ScanPipeline,
21 26 storage::FileType,
22 - AppState,
27 + AppState, AppStorage,
23 28 };
24 29
25 30 /// Validate a client-declared upload size before it is signed into the
@@ -261,7 +266,8 @@ impl CommitTarget {
261 266 fields(kind = ?target.kind(), target_id = %target.target_uuid(), %user_id, file_size_bytes, scan_status = tracing::field::Empty)
262 267 )]
263 268 pub(crate) async fn commit_upload(
264 - state: &AppState,
269 + db: &PgPool,
270 + scanner: Option<&Arc<ScanPipeline>>,
265 271 target: CommitTarget,
266 272 s3_key: &str,
267 273 file_type: FileType,
@@ -269,7 +275,8 @@ pub(crate) async fn commit_upload(
269 275 file_size_bytes: i64,
270 276 ) -> Result<db::FileScanStatus> {
271 277 let scan_status = enqueue_scan_for(
272 - state,
278 + db,
279 + scanner,
273 280 target.kind(),
274 281 target.target_uuid(),
275 282 s3_key,
@@ -280,13 +287,13 @@ pub(crate) async fn commit_upload(
280 287 .await?;
281 288 match target {
282 289 CommitTarget::Item(id) => {
283 - db::scanning::update_item_scan_status(&state.db, id, scan_status).await?;
290 + db::scanning::update_item_scan_status(db, id, scan_status).await?;
284 291 }
285 292 CommitTarget::Version(id) => {
286 - db::scanning::update_version_scan_status(&state.db, id, scan_status).await?;
293 + db::scanning::update_version_scan_status(db, id, scan_status).await?;
287 294 }
288 295 CommitTarget::Media(id) => {
289 - db::scanning::update_media_file_scan_status(&state.db, id, scan_status).await?;
296 + db::scanning::update_media_file_scan_status(db, id, scan_status).await?;
290 297 }
291 298 CommitTarget::ItemImage(_)
292 299 | CommitTarget::ProjectImage(_)
@@ -318,7 +325,8 @@ pub(crate) async fn commit_upload(
318 325 fields(kind = ?target.kind(), target_id = %target.target_uuid(), %user_id, file_size_bytes)
319 326 )]
320 327 pub(crate) async fn commit_rescan(
321 - state: &AppState,
328 + db: &PgPool,
329 + scanner: Option<&Arc<ScanPipeline>>,
322 330 target: CommitTarget,
323 331 s3_key: &str,
324 332 file_type: FileType,
@@ -326,7 +334,8 @@ pub(crate) async fn commit_rescan(
326 334 file_size_bytes: i64,
327 335 ) -> Result<db::FileScanStatus> {
328 336 enqueue_scan_for(
329 - state,
337 + db,
338 + scanner,
330 339 target.kind(),
331 340 target.target_uuid(),
332 341 s3_key,
@@ -338,13 +347,13 @@ pub(crate) async fn commit_rescan(
338 347 let pending = db::FileScanStatus::Pending;
339 348 match target {
340 349 CommitTarget::Item(id) => {
341 - db::scanning::update_item_scan_status(&state.db, id, pending).await?;
350 + db::scanning::update_item_scan_status(db, id, pending).await?;
342 351 }
343 352 CommitTarget::Version(id) => {
344 - db::scanning::update_version_scan_status(&state.db, id, pending).await?;
353 + db::scanning::update_version_scan_status(db, id, pending).await?;
345 354 }
346 355 CommitTarget::Media(id) => {
347 - db::scanning::update_media_file_scan_status(&state.db, id, pending).await?;
356 + db::scanning::update_media_file_scan_status(db, id, pending).await?;
348 357 }
349 358 CommitTarget::ItemImage(_)
350 359 | CommitTarget::ProjectImage(_)
@@ -368,11 +377,16 @@ pub(crate) async fn commit_rescan(
368 377 /// Idempotent: a file already at a `{owner}/c/…` content key is skipped, so a
369 378 /// bulk approve over a mixed set (some already promoted) is safe.
370 379 #[tracing::instrument(skip_all, name = "storage::commit_promote_item", fields(%item_id))]
371 - pub(crate) async fn commit_promote_item(state: &AppState, item_id: db::ItemId) -> Result<()> {
372 - let item = db::items::get_item_by_id(&state.db, item_id)
380 + pub(crate) async fn commit_promote_item(
381 + db: &PgPool,
382 + storage: &AppStorage,
383 + config: &Config,
384 + item_id: db::ItemId,
385 + ) -> Result<()> {
386 + let item = db::items::get_item_by_id(db, item_id)
373 387 .await?
374 388 .ok_or(AppError::NotFound)?;
375 - let owner = db::items::get_item_owner(&state.db, item_id)
389 + let owner = db::items::get_item_owner(db, item_id)
376 390 .await?
377 391 .ok_or(AppError::NotFound)?;
378 392 // Only the files still at a staging key need a copy; skip a file that is
@@ -392,9 +406,9 @@ pub(crate) async fn commit_promote_item(state: &AppState, item_id: db::ItemId) -
392 406 })
393 407 .collect();
394 408 if !staged.is_empty() {
395 - let s3 = state.require_s3()?;
409 + let s3 = storage.require_s3()?;
396 410 for (staging_key, file_type) in staged {
397 - let sha256 = db::scanning::latest_sha256_by_key(&state.db, &staging_key)
411 + let sha256 = db::scanning::latest_sha256_by_key(db, &staging_key)
398 412 .await?
399 413 .ok_or_else(|| {
400 414 AppError::Storage(format!(
@@ -402,12 +416,12 @@ pub(crate) async fn commit_promote_item(state: &AppState, item_id: db::ItemId) -
402 416 ))
403 417 })?;
404 418 crate::scanning::promote_staging_to_content(
405 - &state.db,
419 + db,
406 420 s3.as_ref(),
407 421 // Item audio is a gated Main-bucket kind; the public backend is
408 422 // unused here but passed for signature consistency.
409 - state.storage.public_s3.as_deref(),
410 - state.config.cdn_base_url.as_deref(),
423 + storage.public_s3.as_deref(),
424 + config.cdn_base_url.as_deref(),
411 425 ScanTargetKind::Item,
412 426 file_type,
413 427 item_id.into(),
@@ -422,7 +436,7 @@ pub(crate) async fn commit_promote_item(state: &AppState, item_id: db::ItemId) -
422 436 // `promote_gated` already set `scan_status = 'clean'` per promoted file; this
423 437 // also covers the no-staged-file case (all already content-keyed) so the
424 438 // admin action still lands the item Clean.
425 - db::scanning::update_item_scan_status(&state.db, item_id, db::FileScanStatus::Clean).await?;
439 + db::scanning::update_item_scan_status(db, item_id, db::FileScanStatus::Clean).await?;
426 440 Ok(())
427 441 }
428 442
@@ -430,11 +444,16 @@ pub(crate) async fn commit_promote_item(state: &AppState, item_id: db::ItemId) -
430 444 /// then mark the version Clean. See [`commit_promote_item`] for why the copy must
431 445 /// happen at approve time and why it is factored here.
432 446 #[tracing::instrument(skip_all, name = "storage::commit_promote_version", fields(%version_id))]
433 - pub(crate) async fn commit_promote_version(state: &AppState, version_id: db::VersionId) -> Result<()> {
434 - let version = db::versions::get_version_by_id(&state.db, version_id)
447 + pub(crate) async fn commit_promote_version(
448 + db: &PgPool,
449 + storage: &AppStorage,
450 + config: &Config,
451 + version_id: db::VersionId,
452 + ) -> Result<()> {
453 + let version = db::versions::get_version_by_id(db, version_id)
435 454 .await?
436 455 .ok_or(AppError::NotFound)?;
437 - let owner = db::items::get_item_owner(&state.db, version.item_id)
456 + let owner = db::items::get_item_owner(db, version.item_id)
438 457 .await?
439 458 .ok_or(AppError::NotFound)?;
440 459 // `require_s3` only when there is a staging file to copy — a held row with no
@@ -443,8 +462,8 @@ pub(crate) async fn commit_promote_version(state: &AppState, version_id: db::Ver
443 462 if let Some(staging_key) = version.s3_key.clone()
444 463 && staging_key.starts_with("staging/")
445 464 {
446 - let s3 = state.require_s3()?;
447 - let sha256 = db::scanning::latest_sha256_by_key(&state.db, &staging_key)
465 + let s3 = storage.require_s3()?;
466 + let sha256 = db::scanning::latest_sha256_by_key(db, &staging_key)
448 467 .await?
449 468 .ok_or_else(|| {
450 469 AppError::Storage(format!(
@@ -452,11 +471,11 @@ pub(crate) async fn commit_promote_version(state: &AppState, version_id: db::Ver
452 471 ))
453 472 })?;
454 473 crate::scanning::promote_staging_to_content(
455 - &state.db,
474 + db,
456 475 s3.as_ref(),
457 476 // Version downloads are a gated Main-bucket kind; public backend unused.
458 - state.storage.public_s3.as_deref(),
459 - state.config.cdn_base_url.as_deref(),
477 + storage.public_s3.as_deref(),
478 + config.cdn_base_url.as_deref(),
460 479 ScanTargetKind::Version,
461 480 FileType::Download,
462 481 version_id.into(),
@@ -467,7 +486,7 @@ pub(crate) async fn commit_promote_version(state: &AppState, version_id: db::Ver
467 486 )
468 487 .await?;
469 488 }
470 - db::scanning::update_version_scan_status(&state.db, version_id, db::FileScanStatus::Clean).await?;
489 + db::scanning::update_version_scan_status(db, version_id, db::FileScanStatus::Clean).await?;
471 490 Ok(())
472 491 }
473 492
@@ -479,13 +498,15 @@ pub(crate) async fn commit_promote_version(state: &AppState, version_id: db::Ver
479 498 /// future sibling handler. This function remains `pub(super)`-equivalent for
480 499 /// the `commit_upload` implementation and for the worker / admin tooling
481 500 /// that legitimately needs the lower-level op.
501 + #[allow(clippy::too_many_arguments)]
482 502 #[tracing::instrument(
483 503 skip_all,
484 504 name = "storage::enqueue_scan_for",
485 505 fields(kind = ?target_kind, %target_id, %user_id, file_size_bytes)
486 506 )]
487 507 async fn enqueue_scan_for(
488 - state: &AppState,
508 + db: &PgPool,
509 + scanner: Option<&Arc<ScanPipeline>>,
489 510 target_kind: ScanTargetKind,
490 511 target_id: Uuid,
491 512 s3_key: &str,
@@ -493,8 +514,8 @@ async fn enqueue_scan_for(
493 514 user_id: db::UserId,
494 515 file_size_bytes: i64,
495 516 ) -> Result<db::FileScanStatus> {
496 - if state.scanner.is_none() {
497 - let is_trusted = db::users::is_upload_trusted(&state.db, user_id).await?;
517 + if scanner.is_none() {
518 + let is_trusted = db::users::is_upload_trusted(db, user_id).await?;
498 519 let status = if is_trusted {
499 520 db::FileScanStatus::Clean
500 521 } else {
@@ -505,7 +526,7 @@ async fn enqueue_scan_for(
505 526 }
506 527
507 528 db::scan_jobs::enqueue(
508 - &state.db,
529 + db,
509 530 target_kind,
510 531 target_id,
511 532 s3_key,
M server/src/routes/storage/uploads.rs +28 -23
@@ -6,6 +6,7 @@ use axum::{
6 6 Json,
7 7 };
8 8 use serde::Deserialize;
9 + use sqlx::PgPool;
9 10 use std::str::FromStr;
10 11
11 12 use crate::{
@@ -13,7 +14,7 @@ use crate::{
13 14 db::{self, ItemId},
14 15 error::{AppError, Result, ResultExt},
15 16 storage::{FileType, S3Client, CACHE_CONTROL_IMMUTABLE},
16 - AppState,
17 + AppStorage, Scanning,
17 18 };
18 19
19 20 use super::{commit_upload, CommitTarget, ConfirmUploadResponse, PresignUploadResponse};
@@ -51,13 +52,14 @@ pub struct ConfirmUploadRequest {
51 52 /// Requires authentication. User must own the item.
52 53 #[tracing::instrument(skip_all, name = "storage::presign_upload", fields(user_id = %user.id))]
53 54 pub(super) async fn presign_upload(
54 - State(state): State<AppState>,
55 + State(db): State<PgPool>,
56 + State(storage): State<AppStorage>,
55 57 AuthUser(user): AuthUser,
56 58 Json(req): Json<PresignUploadRequest>,
57 59 ) -> Result<impl IntoResponse> {
58 60 user.check_not_suspended()?;
59 61 // Check if S3 is configured
60 - let s3 = state.require_s3()?;
62 + let s3 = storage.require_s3()?;
61 63
62 64 // Parse file type
63 65 let file_type = FileType::from_str(&req.file_type)
@@ -70,7 +72,7 @@ pub(super) async fn presign_upload(
70 72 S3Client::validate_extension(file_type, &req.file_name)?;
71 73
72 74 // Verify user owns the item
73 - let owner = db::items::get_item_owner(&state.db, req.item_id)
75 + let owner = db::items::get_item_owner(&db, req.item_id)
74 76 .await?
75 77 .ok_or(AppError::NotFound)?;
76 78
@@ -79,10 +81,10 @@ pub(super) async fn presign_upload(
79 81 }
80 82
81 83 // Early quota check (reject before generating presigned URL)
82 - db::creator_tiers::check_presign_allowed(&state.db, user.id, file_type).await?;
84 + db::creator_tiers::check_presign_allowed(&db, user.id, file_type).await?;
83 85
84 86 // 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?;
87 + let max_file_bytes = db::creator_tiers::get_effective_max_file_bytes(&db, user.id, file_type).await?;
86 88
87 89 // If the client declared the file size, validate it before signing — both
88 90 // against the static per-type cap and the tier-effective cap. The size is
@@ -98,7 +100,7 @@ pub(super) async fn presign_upload(
98 100 let s3_key = S3Client::generate_staging_key(&req.file_name);
99 101
100 102 // Track the pending upload so the reaper can clean it up if never confirmed
101 - db::pending_uploads::record_pending_upload(&state.db, user.id, &s3_key, "main").await?;
103 + db::pending_uploads::record_pending_upload(&db, user.id, &s3_key, "main").await?;
102 104
103 105 // Generate presigned upload URL with immutable cache headers
104 106 let expires_in = 3600; // 1 hour
@@ -122,20 +124,22 @@ pub(super) async fn presign_upload(
122 124 /// Requires authentication. User must own the item.
123 125 #[tracing::instrument(skip_all, name = "storage::confirm_upload", fields(user_id = %user.id))]
124 126 pub(super) async fn confirm_upload(
125 - State(state): State<AppState>,
127 + State(db): State<PgPool>,
128 + State(storage): State<AppStorage>,
129 + State(scanning): State<Scanning>,
126 130 AuthUser(user): AuthUser,
127 131 Json(req): Json<ConfirmUploadRequest>,
128 132 ) -> Result<impl IntoResponse> {
129 133 user.check_not_suspended()?;
130 134 // Check if S3 is configured
131 - let s3 = state.require_s3()?;
135 + let s3 = storage.require_s3()?;
132 136
133 137 // Parse file type
134 138 let file_type = FileType::from_str(&req.file_type)
135 139 .map_err(|_| AppError::BadRequest(format!("Invalid file type: {}", req.file_type)))?;
136 140
137 141 // Verify user owns the item
138 - let owner = db::items::get_item_owner(&state.db, req.item_id)
142 + let owner = db::items::get_item_owner(&db, req.item_id)
139 143 .await?
140 144 .ok_or(AppError::NotFound)?;
141 145
@@ -159,7 +163,7 @@ pub(super) async fn confirm_upload(
159 163 AppError::BadRequest("Could not determine file size. Please try uploading again.".to_string())
160 164 })?;
161 165 if file_size_bytes as u64 > file_type.max_size() {
162 - super::enqueue_s3_orphan(&state.db, &req.s3_key, crate::storage::S3Bucket::Main, "item_upload_rejected").await;
166 + super::enqueue_s3_orphan(&db, &req.s3_key, crate::storage::S3Bucket::Main, "item_upload_rejected").await;
163 167 let limit_mb = file_type.max_size() / (1024 * 1024);
164 168 let file_mb = file_size_bytes as u64 / (1024 * 1024);
165 169 return Err(AppError::FileTooLarge(format!(
@@ -169,10 +173,10 @@ pub(super) async fn confirm_upload(
169 173 }
170 174
171 175 // Enforce tier-based limits (per-file + storage cap)
172 - let max_storage = match db::creator_tiers::check_upload_allowed(&state.db, user.id, file_type, file_size_bytes).await {
176 + let max_storage = match db::creator_tiers::check_upload_allowed(&db, user.id, file_type, file_size_bytes).await {
173 177 Ok(max) => max,
174 178 Err(e) => {
175 - super::enqueue_s3_orphan(&state.db, &req.s3_key, crate::storage::S3Bucket::Main, "item_upload_rejected").await;
179 + super::enqueue_s3_orphan(&db, &req.s3_key, crate::storage::S3Bucket::Main, "item_upload_rejected").await;
176 180 return Err(e);
177 181 }
178 182 };
@@ -186,7 +190,7 @@ pub(super) async fn confirm_upload(
186 190 // which only /api/items/image/confirm writes — the old generic two-column
187 191 // path left it NULL and rendered an invisible cover (Run #13 SERIOUS).
188 192 if let crate::storage::GenericItemConfirm::UseRoute(route) = file_type.generic_item_confirm() {
189 - super::enqueue_s3_orphan(&state.db, &req.s3_key, crate::storage::S3Bucket::Main, "item_upload_rejected").await;
193 + super::enqueue_s3_orphan(&db, &req.s3_key, crate::storage::S3Bucket::Main, "item_upload_rejected").await;
190 194 return Err(AppError::BadRequest(format!(
191 195 "This file type isn't confirmed here — use {route}."
192 196 )));
@@ -197,7 +201,7 @@ pub(super) async fn confirm_upload(
197 201 let mut old_s3_key: Option<String> = None;
198 202 let mut replaced_old_size: i64 = 0;
199 203 let mut item_project_id: Option<db::ProjectId> = None;
200 - if let Some(item) = db::items::get_item_by_id(&state.db, req.item_id).await? {
204 + if let Some(item) = db::items::get_item_by_id(&db, req.item_id).await? {
201 205 item_project_id = Some(item.project_id);
202 206 let existing_key = match file_type {
203 207 FileType::Audio => item.audio_s3_key.as_deref(),
@@ -210,7 +214,7 @@ pub(super) async fn confirm_upload(
210 214 // Still clear the pending_uploads row — otherwise the orphan reaper
211 215 // will fire 24h later and delete the live S3 object out from under
212 216 // a perfectly happy DB row (Run #7 HIGH-1).
213 - if let Err(e) = db::pending_uploads::remove_pending_upload(&state.db, user.id, &req.s3_key, "main").await {
217 + if let Err(e) = db::pending_uploads::remove_pending_upload(&db, user.id, &req.s3_key, "main").await {
214 218 tracing::warn!(error = ?e, key = %req.s3_key, "remove_pending_upload failed on idempotent re-confirm");
215 219 }
216 220 return Ok(Json(ConfirmUploadResponse { success: true, pending_review: None }));
@@ -239,7 +243,7 @@ pub(super) async fn confirm_upload(
239 243 // would treat the not-yet-referenced staging object as dead and delete it —
240 244 // a cross-user griefing delete. Just reject; the real owner's confirm or the
241 245 // pending-upload reaper handles their key.
242 - if !db::pending_uploads::is_owned(&state.db, user.id, &req.s3_key, "main").await? {
246 + if !db::pending_uploads::is_owned(&db, user.id, &req.s3_key, "main").await? {
243 247 return Err(AppError::BadRequest("Invalid upload key".to_string()));
244 248 }
245 249
@@ -255,7 +259,7 @@ pub(super) async fn confirm_upload(
255 259 // signals the item vanished or lost the CAS race (tx rolled back, nothing
256 260 // charged).
257 261 let tx_result: Result<u64> = async {
258 - let mut tx = state.db.begin().await?;
262 + let mut tx = db.begin().await?;
259 263 db::creator_tiers::try_apply_storage_on(
260 264 &mut tx, user.id, is_replace.then_some(replaced_old_size), file_size_bytes, max_storage,
261 265 ).await?;
@@ -306,7 +310,7 @@ pub(super) async fn confirm_upload(
306 310 // filling in between), so a blind delete could destroy the live
307 311 // object the winner points at. Route through the orphan queue; its
308 312 // `is_s3_key_live` check skips any key a row still references.
309 - super::enqueue_s3_orphan(&state.db, &req.s3_key, crate::storage::S3Bucket::Main, "item_confirm_failed").await;
313 + super::enqueue_s3_orphan(&db, &req.s3_key, crate::storage::S3Bucket::Main, "item_confirm_failed").await;
310 314 return Err(e);
311 315 }
312 316 Ok(0) => {
@@ -316,7 +320,7 @@ pub(super) async fn confirm_upload(
316 320 // observed). Either way the tx rolled back, nothing was charged;
317 321 // route the now-unreferenced object through the orphan queue so the
318 322 // reaper still cleans it.
319 - super::enqueue_s3_orphan(&state.db, &req.s3_key, crate::storage::S3Bucket::Main, "item_upload_target_missing").await;
323 + super::enqueue_s3_orphan(&db, &req.s3_key, crate::storage::S3Bucket::Main, "item_upload_target_missing").await;
320 324 return Err(AppError::BadRequest(
321 325 "Item was modified concurrently. Please try uploading again.".to_string(),
322 326 ));
@@ -329,7 +333,8 @@ pub(super) async fn confirm_upload(
329 333 // this ordering. See `routes/storage/mod.rs::commit_upload` for the bug
330 334 // shapes this prevents.
331 335 let scan_status = commit_upload(
332 - &state,
336 + &db,
337 + scanning.scanner.as_ref(),
333 338 CommitTarget::Item(req.item_id),
334 339 &req.s3_key,
335 340 file_type,
@@ -341,12 +346,12 @@ pub(super) async fn confirm_upload(
341 346 // confirm tx above, so a crash here can't orphan it.)
342 347
343 348 // Clear the pending upload record now that the upload is confirmed
344 - db::pending_uploads::remove_pending_upload(&state.db, user.id, &req.s3_key, "main").await?;
349 + db::pending_uploads::remove_pending_upload(&db, user.id, &req.s3_key, "main").await?;
345 350
346 351 // Bump project cache generation so dashboard tabs reflect the new upload.
347 352 // Reuses the project_id read during idempotency above (no second fetch).
348 353 if let Some(project_id) = item_project_id
349 - && let Err(e) = db::projects::bump_cache_generation(&state.db, project_id).await
354 + && let Err(e) = db::projects::bump_cache_generation(&db, project_id).await
350 355 {
351 356 tracing::warn!(%project_id, error = ?e, "failed to bump cache generation after upload");
352 357 }
M server/src/routes/storage/versions.rs +27 -22
@@ -6,13 +6,14 @@ use axum::{
6 6 Json,
7 7 };
8 8 use serde::Deserialize;
9 + use sqlx::PgPool;
9 10
10 11 use crate::{
11 12 auth::AuthUser,
12 13 db::{self, VersionId},
13 14 error::{AppError, Result, ResultExt},
14 15 storage::{FileType, S3Client, CACHE_CONTROL_IMMUTABLE},
15 - AppState,
16 + AppStorage, Scanning,
16 17 };
17 18
18 19 use super::{commit_upload, CommitTarget, ConfirmUploadResponse, PresignUploadResponse};
@@ -42,13 +43,14 @@ pub struct VersionConfirmRequest {
42 43 /// Requires authentication. User must own the item (through version -> item -> project chain).
43 44 #[tracing::instrument(skip_all, name = "storage::version_presign_upload", fields(%version_id, user_id = %user.id))]
44 45 pub(super) async fn version_presign_upload(
45 - State(state): State<AppState>,
46 + State(db): State<PgPool>,
47 + State(storage): State<AppStorage>,
46 48 AuthUser(user): AuthUser,
47 49 Path(version_id): Path<VersionId>,
48 50 Json(req): Json<VersionPresignRequest>,
49 51 ) -> Result<impl IntoResponse> {
50 52 user.check_not_suspended()?;
51 - let s3 = state.require_s3()?;
53 + let s3 = storage.require_s3()?;
52 54
53 55 let file_type = FileType::Download;
54 56
@@ -57,11 +59,11 @@ pub(super) async fn version_presign_upload(
57 59 S3Client::validate_extension(file_type, &req.file_name)?;
58 60
59 61 // Fetch version and verify ownership through version -> item -> project chain
60 - let version = db::versions::get_version_by_id(&state.db, version_id)
62 + let version = db::versions::get_version_by_id(&db, version_id)
61 63 .await?
62 64 .ok_or(AppError::NotFound)?;
63 65
64 - let owner = db::items::get_item_owner(&state.db, version.item_id)
66 + let owner = db::items::get_item_owner(&db, version.item_id)
65 67 .await?
66 68 .ok_or(AppError::NotFound)?;
67 69
@@ -70,9 +72,9 @@ pub(super) async fn version_presign_upload(
70 72 }
71 73
72 74 // Early quota check
73 - db::creator_tiers::check_presign_allowed(&state.db, user.id, file_type).await?;
75 + db::creator_tiers::check_presign_allowed(&db, user.id, file_type).await?;
74 76
75 - let max_file_bytes = db::creator_tiers::get_effective_max_file_bytes(&state.db, user.id, file_type).await?;
77 + let max_file_bytes = db::creator_tiers::get_effective_max_file_bytes(&db, user.id, file_type).await?;
76 78
77 79 // Validate the declared size (if any) before signing it into Content-Length.
78 80 super::validate_declared_upload_size(req.file_size_bytes, file_type, max_file_bytes)?;
@@ -84,7 +86,7 @@ pub(super) async fn version_presign_upload(
84 86 let s3_key = S3Client::generate_staging_key(&req.file_name);
85 87
86 88 // Track the pending upload so the reaper can clean it up if never confirmed
87 - db::pending_uploads::record_pending_upload(&state.db, user.id, &s3_key, "main").await?;
89 + db::pending_uploads::record_pending_upload(&db, user.id, &s3_key, "main").await?;
88 90
89 91 let expires_in = 3600;
90 92 let upload_url = s3.presign_upload(&s3_key, &req.content_type, Some(expires_in), Some(CACHE_CONTROL_IMMUTABLE), req.file_size_bytes)
@@ -107,20 +109,22 @@ pub(super) async fn version_presign_upload(
107 109 /// Requires authentication. User must own the item.
108 110 #[tracing::instrument(skip_all, name = "storage::version_confirm_upload", fields(%version_id, user_id = %user.id))]
109 111 pub(super) async fn version_confirm_upload(
110 - State(state): State<AppState>,
112 + State(db): State<PgPool>,
113 + State(storage): State<AppStorage>,
114 + State(scanning): State<Scanning>,
111 115 AuthUser(user): AuthUser,
112 116 Path(version_id): Path<VersionId>,
113 117 Json(req): Json<VersionConfirmRequest>,
114 118 ) -> Result<impl IntoResponse> {
115 119 user.check_not_suspended()?;
116 - let s3 = state.require_s3()?;
120 + let s3 = storage.require_s3()?;
117 121
118 122 // Fetch version and verify ownership
119 - let version = db::versions::get_version_by_id(&state.db, version_id)
123 + let version = db::versions::get_version_by_id(&db, version_id)
120 124 .await?
121 125 .ok_or(AppError::NotFound)?;
122 126
123 - let owner = db::items::get_item_owner(&state.db, version.item_id)
127 + let owner = db::items::get_item_owner(&db, version.item_id)
124 128 .await?
125 129 .ok_or(AppError::NotFound)?;
126 130
@@ -135,7 +139,7 @@ pub(super) async fn version_confirm_upload(
135 139 if version.s3_key.as_deref() == Some(&req.s3_key) {
136 140 // Still clear pending_uploads — orphan reaper would otherwise delete the
137 141 // live S3 object 24h later (Run #7 HIGH-1).
138 - if let Err(e) = db::pending_uploads::remove_pending_upload(&state.db, user.id, &req.s3_key, "main").await {
142 + if let Err(e) = db::pending_uploads::remove_pending_upload(&db, user.id, &req.s3_key, "main").await {
139 143 tracing::warn!(error = ?e, key = %req.s3_key, "remove_pending_upload failed on idempotent re-confirm");
140 144 }
141 145 return Ok(Json(ConfirmUploadResponse { success: true, pending_review: None }));
@@ -146,7 +150,7 @@ pub(super) async fn version_confirm_upload(
146 150 // `pending_uploads` row recorded at presign — not a prefix check. Placed
147 151 // before the size/tier reject paths so an unowned (at most another user's
148 152 // in-flight) staging object is never enqueued for deletion.
149 - if !db::pending_uploads::is_owned(&state.db, user.id, &req.s3_key, "main").await? {
153 + if !db::pending_uploads::is_owned(&db, user.id, &req.s3_key, "main").await? {
150 154 return Err(AppError::BadRequest("Invalid upload key".to_string()));
151 155 }
152 156
@@ -157,7 +161,7 @@ pub(super) async fn version_confirm_upload(
157 161 AppError::BadRequest("Upload not found. Please try uploading again.".to_string())
158 162 })?;
159 163 if file_size_bytes as u64 > FileType::Download.max_size() {
160 - super::enqueue_s3_orphan(&state.db, &req.s3_key, crate::storage::S3Bucket::Main, "version_upload_rejected").await;
164 + super::enqueue_s3_orphan(&db, &req.s3_key, crate::storage::S3Bucket::Main, "version_upload_rejected").await;
161 165 let limit_mb = FileType::Download.max_size() / (1024 * 1024);
162 166 let file_mb = file_size_bytes as u64 / (1024 * 1024);
163 167 return Err(AppError::FileTooLarge(format!(
@@ -167,10 +171,10 @@ pub(super) async fn version_confirm_upload(
167 171 }
168 172
169 173 // Enforce tier-based limits (per-file + storage cap)
170 - let max_storage = match db::creator_tiers::check_upload_allowed(&state.db, user.id, FileType::Download, file_size_bytes).await {
174 + let max_storage = match db::creator_tiers::check_upload_allowed(&db, user.id, FileType::Download, file_size_bytes).await {
171 175 Ok(max) => max,
172 176 Err(e) => {
173 - super::enqueue_s3_orphan(&state.db, &req.s3_key, crate::storage::S3Bucket::Main, "version_upload_rejected").await;
177 + super::enqueue_s3_orphan(&db, &req.s3_key, crate::storage::S3Bucket::Main, "version_upload_rejected").await;
174 178 return Err(e);
175 179 }
176 180 };
@@ -189,7 +193,7 @@ pub(super) async fn version_confirm_upload(
189 193 // `commit_upload` stays AFTER the commit (the blessed scan-ordering path).
190 194 // `Ok(false)` = lost race (rolled back, nothing charged).
191 195 let committed: Result<bool> = async {
192 - let mut tx = state.db.begin().await?;
196 + let mut tx = db.begin().await?;
193 197 db::creator_tiers::try_apply_storage_on(
194 198 &mut tx, user.id, is_replace.then_some(old_size), file_size_bytes, max_storage,
195 199 ).await?;
@@ -234,7 +238,7 @@ pub(super) async fn version_confirm_upload(
234 238 // would then destroy the live object the winning confirm points at.
235 239 // Route through the orphan queue, whose `is_s3_key_live` check skips
236 240 // any key a row still references.
237 - super::enqueue_s3_orphan(&state.db, &req.s3_key, crate::storage::S3Bucket::Main, "version_confirm_failed").await;
241 + super::enqueue_s3_orphan(&db, &req.s3_key, crate::storage::S3Bucket::Main, "version_confirm_failed").await;
238 242 return Err(e);
239 243 }
240 244 Ok(false) => {
@@ -242,7 +246,7 @@ pub(super) async fn version_confirm_upload(
242 246 // s3_key out from under us. If it committed THIS key, a direct
243 247 // delete would 404 every fan download of the version the winner
244 248 // just published. The orphan queue's liveness check is the guard.
245 - super::enqueue_s3_orphan(&state.db, &req.s3_key, crate::storage::S3Bucket::Main, "version_confirm_lost_race").await;
249 + super::enqueue_s3_orphan(&db, &req.s3_key, crate::storage::S3Bucket::Main, "version_confirm_lost_race").await;
246 250 return Err(AppError::BadRequest(
247 251 "Version was modified concurrently. Please try uploading again.".to_string(),
248 252 ));
@@ -251,10 +255,11 @@ pub(super) async fn version_confirm_upload(
251 255 }
252 256
253 257 // Clear the pending upload record now that the upload is committed
254 - db::pending_uploads::remove_pending_upload(&state.db, user.id, &req.s3_key, "main").await?;
258 + db::pending_uploads::remove_pending_upload(&db, user.id, &req.s3_key, "main").await?;
255 259
256 260 let scan_status = commit_upload(
257 - &state,
261 + &db,
262 + scanning.scanner.as_ref(),
258 263 CommitTarget::Version(version_id),
259 264 &req.s3_key,
260 265 FileType::Download,