1use std::sync::Arc;
30
31use axum::extract::{DefaultBodyLimit, Extension, Multipart, Path};
32use axum::http::StatusCode;
33use axum::response::IntoResponse;
34use axum::routing::{delete, post};
35use axum::{Json, Router};
36use bytes::{Bytes, BytesMut};
37use serde::{Deserialize, Serialize};
38use uuid::Uuid;
39
40use crate::application::service::BucketService;
41use crate::application::service::FileMeta;
42use crate::application::service::FileService;
43use crate::application::service::MultipartUploadService;
44use crate::auth::{AuthExtractor, HasOwnerId};
45use crate::domain::entity::{Bucket, BucketStatus};
46use crate::error::{BucketError, BucketResult};
47use crate::storage::ObjectStorage;
48
49pub const DEFAULT_UPLOAD_BODY_LIMIT: usize = 256 * 1024 * 1024;
55pub const DEFAULT_CHUNK_BODY_LIMIT: usize = 16 * 1024 * 1024;
56
57pub struct UploadContext {
64 pub file_service: Arc<FileService>,
65 pub multipart_service: Arc<MultipartUploadService>,
66 pub bucket_service: Arc<BucketService>,
67 pub storage: Arc<dyn ObjectStorage>,
68}
69
70impl Clone for UploadContext {
71 fn clone(&self) -> Self {
72 Self {
73 file_service: self.file_service.clone(),
74 multipart_service: self.multipart_service.clone(),
75 bucket_service: self.bucket_service.clone(),
76 storage: self.storage.clone(),
77 }
78 }
79}
80
81#[derive(Debug, Clone, Copy)]
82pub struct UploadConfig {
83 pub single_shot_limit: usize,
84 pub chunk_limit: usize,
85}
86
87impl Default for UploadConfig {
88 fn default() -> Self {
89 Self {
90 single_shot_limit: DEFAULT_UPLOAD_BODY_LIMIT,
91 chunk_limit: DEFAULT_CHUNK_BODY_LIMIT,
92 }
93 }
94}
95
96pub fn upload_router<A>(ctx: UploadContext, config: UploadConfig) -> Router
102where
103 A: AuthExtractor<()> + HasOwnerId + Clone + 'static,
104 A::Rejection: IntoResponse,
105{
106 Router::new()
107 .route(
108 "/uploads",
109 post(single_shot_upload::<A>).layer(DefaultBodyLimit::max(config.single_shot_limit)),
110 )
111 .route("/uploads/sessions", post(initiate_session::<A>))
112 .route(
113 "/uploads/sessions/:id/parts/:part_number",
114 post(upload_part::<A>).layer(DefaultBodyLimit::max(config.chunk_limit)),
115 )
116 .route(
117 "/uploads/sessions/:id/complete",
118 post(complete_session::<A>),
119 )
120 .route("/uploads/sessions/:id", delete(abort_session::<A>))
121 .layer(Extension(Arc::new(ctx)))
122}
123
124#[derive(Debug, Serialize)]
127pub struct SingleShotResponse {
128 pub id: Uuid,
129 pub storage_key: String,
130 pub size_bytes: i64,
131 pub mime_type: String,
132}
133
134#[tracing::instrument(
135 name = "bucket.upload.single_shot",
136 skip_all,
137 fields(owner_id = %identity.owner_id())
138)]
139async fn single_shot_upload<A>(
140 Extension(ctx): Extension<Arc<UploadContext>>,
141 identity: A,
142 mut multipart: Multipart,
143) -> Result<(StatusCode, Json<SingleShotResponse>), BucketError>
144where
145 A: AuthExtractor<()> + HasOwnerId + 'static,
146 A::Rejection: IntoResponse,
147{
148 let owner_id = identity.owner_id();
149
150 let mut bucket_id: Option<Uuid> = None;
151 let mut path: Option<String> = None;
152 let mut owner_module: Option<String> = None;
153 let mut owner_entity: Option<String> = None;
154 let mut owner_entity_id: Option<Uuid> = None;
155 let mut storage_key: Option<String> = None;
156 let mut file_name: Option<String> = None;
157 let mut file_mime: Option<String> = None;
158 let mut file_bytes: Option<Bytes> = None;
159
160 while let Some(field) = multipart
161 .next_field()
162 .await
163 .map_err(|e| BucketError::Other(format!("multipart: {e}")))?
164 {
165 let name = field.name().unwrap_or("").to_string();
166 match name.as_str() {
167 "bucket_id" => bucket_id = Some(parse_uuid_field(field).await?),
168 "path" => path = Some(read_text_field(field).await?),
169 "owner_module" => owner_module = Some(read_text_field(field).await?),
170 "owner_entity" => owner_entity = Some(read_text_field(field).await?),
171 "owner_entity_id" => owner_entity_id = Some(parse_uuid_field(field).await?),
172 "storage_key" => storage_key = Some(read_text_field(field).await?),
173 "file" => {
174 file_name = field.file_name().map(str::to_string);
175 file_mime = field.content_type().map(str::to_string);
176 let bytes = field
177 .bytes()
178 .await
179 .map_err(|e| BucketError::Other(format!("read file field: {e}")))?;
180 file_bytes = Some(bytes);
181 }
182 _ => {
183 let _ = field.bytes().await;
184 }
185 }
186 }
187
188 let bucket_id = bucket_id.ok_or_else(|| BucketError::InvalidInput("missing bucket_id".into()))?;
189 let path = path.ok_or_else(|| BucketError::InvalidInput("missing path".into()))?;
190 let body = file_bytes.ok_or_else(|| BucketError::InvalidInput("missing file field".into()))?;
191 let original_name = file_name
192 .ok_or_else(|| BucketError::InvalidInput("file field has no filename".into()))?;
193 let mime_type = file_mime.unwrap_or_else(|| "application/octet-stream".to_string());
194
195 let bucket = load_bucket(&ctx.bucket_service, bucket_id).await?;
198 enforce_bucket_policy(&bucket, body.len() as i64, &mime_type)?;
199
200 ctx.multipart_service
202 .check_capacity(owner_id, body.len() as i64)
203 .await
204 .map_err(service_to_bucket_error)?;
205
206 let body_len = body.len() as i64;
207 let meta = FileMeta {
208 bucket_id,
209 owner_id,
210 original_name,
211 mime_type: mime_type.clone(),
212 path,
213 owner_module,
214 owner_entity,
215 owner_entity_id,
216 };
217
218 let file = match storage_key {
219 Some(key) => ctx.file_service.upload_with_key(&key, body, meta).await?,
220 None => ctx.file_service.upload(body, meta).await?,
221 };
222
223 if let Err(e) = ctx
226 .multipart_service
227 .record_completed_usage(owner_id, body_len)
228 .await
229 {
230 tracing::warn!(%owner_id, error = ?e, "record_completed_usage failed after single-shot upload");
231 }
232
233 Ok((
234 StatusCode::CREATED,
235 Json(SingleShotResponse {
236 id: file.id,
237 storage_key: file.storage_key,
238 size_bytes: file.size_bytes,
239 mime_type: file.mime_type,
240 }),
241 ))
242}
243
244#[derive(Debug, Deserialize)]
247pub struct InitiateRequest {
248 pub bucket_id: Uuid,
249 pub path: String,
250 pub filename: String,
251 pub mime_type: Option<String>,
252 pub file_size: i64,
253 pub chunk_size: Option<i32>,
254}
255
256#[derive(Debug, Serialize)]
257pub struct InitiateResponse {
258 pub session_id: Uuid,
259 pub chunk_size: i32,
260 pub total_chunks: i32,
261 pub expires_at: chrono::DateTime<chrono::Utc>,
262}
263
264#[tracing::instrument(
265 name = "bucket.upload.initiate",
266 skip_all,
267 fields(
268 owner_id = %identity.owner_id(),
269 bucket_id = %req.bucket_id,
270 file_size = req.file_size,
271 )
272)]
273async fn initiate_session<A>(
274 Extension(ctx): Extension<Arc<UploadContext>>,
275 identity: A,
276 Json(req): Json<InitiateRequest>,
277) -> Result<(StatusCode, Json<InitiateResponse>), BucketError>
278where
279 A: AuthExtractor<()> + HasOwnerId + 'static,
280 A::Rejection: IntoResponse,
281{
282 let owner_id = identity.owner_id();
283
284 let bucket = load_bucket(&ctx.bucket_service, req.bucket_id).await?;
287 enforce_bucket_policy(
288 &bucket,
289 req.file_size,
290 req.mime_type.as_deref().unwrap_or("application/octet-stream"),
291 )?;
292 ctx.multipart_service
293 .check_capacity(owner_id, req.file_size)
294 .await
295 .map_err(service_to_bucket_error)?;
296
297 let session = ctx
298 .multipart_service
299 .initiate(
300 req.bucket_id,
301 owner_id,
302 &req.path,
303 &req.filename,
304 req.mime_type.as_deref(),
305 req.file_size,
306 req.chunk_size,
307 )
308 .await
309 .map_err(service_to_bucket_error)?;
310
311 Ok((
312 StatusCode::CREATED,
313 Json(InitiateResponse {
314 session_id: session.id,
315 chunk_size: session.chunk_size,
316 total_chunks: session.total_chunks,
317 expires_at: session.expires_at,
318 }),
319 ))
320}
321
322#[derive(Debug, Serialize)]
323pub struct PartResponse {
324 pub session_id: Uuid,
325 pub part_number: i32,
326 pub uploaded_chunks: i32,
327 pub total_chunks: i32,
328}
329
330#[tracing::instrument(
331 name = "bucket.upload.part",
332 skip_all,
333 fields(
334 owner_id = %identity.owner_id(),
335 session_id = %session_id,
336 part_number,
337 )
338)]
339async fn upload_part<A>(
340 Extension(ctx): Extension<Arc<UploadContext>>,
341 identity: A,
342 Path((session_id, part_number)): Path<(Uuid, i32)>,
343 mut multipart: Multipart,
344) -> Result<Json<PartResponse>, BucketError>
345where
346 A: AuthExtractor<()> + HasOwnerId + 'static,
347 A::Rejection: IntoResponse,
348{
349 let owner_id = identity.owner_id();
350
351 let mut chunk: Option<Bytes> = None;
352 let mut content_type: Option<String> = None;
353 while let Some(field) = multipart
354 .next_field()
355 .await
356 .map_err(|e| BucketError::Other(format!("multipart: {e}")))?
357 {
358 if field.name() == Some("chunk") || field.name() == Some("file") {
359 content_type = field.content_type().map(str::to_string);
360 chunk = Some(
361 field
362 .bytes()
363 .await
364 .map_err(|e| BucketError::Other(format!("read chunk: {e}")))?,
365 );
366 break;
367 } else {
368 let _ = field.bytes().await;
369 }
370 }
371 let chunk = chunk.ok_or_else(|| BucketError::Other("missing chunk field".into()))?;
372
373 let part_key = part_storage_key(session_id, part_number);
374 ctx.storage
375 .put(
376 &part_key,
377 chunk,
378 content_type.as_deref().unwrap_or("application/octet-stream"),
379 )
380 .await?;
381
382 let session = ctx
383 .multipart_service
384 .record_part(session_id, owner_id, part_number)
385 .await
386 .map_err(service_to_bucket_error)?;
387
388 Ok(Json(PartResponse {
389 session_id: session.id,
390 part_number,
391 uploaded_chunks: session.uploaded_chunks,
392 total_chunks: session.total_chunks,
393 }))
394}
395
396#[derive(Debug, Deserialize, Default)]
397pub struct CompleteRequest {
398 pub storage_key: Option<String>,
401 pub owner_module: Option<String>,
402 pub owner_entity: Option<String>,
403 pub owner_entity_id: Option<Uuid>,
404}
405
406#[tracing::instrument(
407 name = "bucket.upload.complete",
408 skip_all,
409 fields(owner_id = %identity.owner_id(), session_id = %session_id)
410)]
411async fn complete_session<A>(
412 Extension(ctx): Extension<Arc<UploadContext>>,
413 identity: A,
414 Path(session_id): Path<Uuid>,
415 body: Option<Json<CompleteRequest>>,
416) -> Result<(StatusCode, Json<SingleShotResponse>), BucketError>
417where
418 A: AuthExtractor<()> + HasOwnerId + 'static,
419 A::Rejection: IntoResponse,
420{
421 let req = body.map(|Json(r)| r).unwrap_or_default();
422 let owner_id = identity.owner_id();
423
424 let session = ctx
425 .multipart_service
426 .complete(session_id, owner_id)
427 .await
428 .map_err(service_to_bucket_error)?;
429
430 let mut assembled = BytesMut::with_capacity(session.file_size.max(0) as usize);
434 for part_number in 1..=session.total_chunks {
435 let key = part_storage_key(session_id, part_number);
436 let part = ctx.storage.get(&key).await?;
437 assembled.extend_from_slice(&part);
438 }
439 let body = assembled.freeze();
440
441 let mime_type = session
442 .mime_type
443 .clone()
444 .unwrap_or_else(|| "application/octet-stream".to_string());
445 let meta = FileMeta {
446 bucket_id: session.bucket_id,
447 owner_id,
448 original_name: session.filename.clone(),
449 mime_type: mime_type.clone(),
450 path: session.path.clone(),
451 owner_module: req.owner_module,
452 owner_entity: req.owner_entity,
453 owner_entity_id: req.owner_entity_id,
454 };
455 let assembled_size = body.len() as i64;
456 let file = match req.storage_key {
457 Some(key) => ctx.file_service.upload_with_key(&key, body, meta).await?,
458 None => ctx.file_service.upload(body, meta).await?,
459 };
460
461 if let Err(e) = ctx
465 .multipart_service
466 .record_completed_usage(owner_id, assembled_size)
467 .await
468 {
469 tracing::warn!(%session_id, error = ?e, "record_completed_usage failed after assembly");
470 }
471 for part_number in 1..=session.total_chunks {
472 let key = part_storage_key(session_id, part_number);
473 if let Err(e) = ctx.storage.delete(&key).await {
474 tracing::warn!(%session_id, part_number, error = %e, "failed to delete staged part");
475 }
476 }
477 if let Err(e) = ctx.multipart_service.mark_completed(session_id).await {
478 tracing::warn!(%session_id, error = ?e, "mark_completed failed after assembly");
479 }
480
481 Ok((
482 StatusCode::CREATED,
483 Json(SingleShotResponse {
484 id: file.id,
485 storage_key: file.storage_key,
486 size_bytes: file.size_bytes,
487 mime_type: file.mime_type,
488 }),
489 ))
490}
491
492#[tracing::instrument(
493 name = "bucket.upload.abort",
494 skip_all,
495 fields(owner_id = %identity.owner_id(), session_id = %session_id)
496)]
497async fn abort_session<A>(
498 Extension(ctx): Extension<Arc<UploadContext>>,
499 identity: A,
500 Path(session_id): Path<Uuid>,
501) -> Result<StatusCode, BucketError>
502where
503 A: AuthExtractor<()> + HasOwnerId + 'static,
504 A::Rejection: IntoResponse,
505{
506 ctx.multipart_service
507 .abort(session_id, identity.owner_id())
508 .await
509 .map_err(service_to_bucket_error)?;
510 Ok(StatusCode::NO_CONTENT)
511}
512
513fn part_storage_key(session_id: Uuid, part_number: i32) -> String {
516 format!("sessions/{session_id}/parts/{part_number:08}")
517}
518
519async fn read_text_field(field: axum::extract::multipart::Field<'_>) -> BucketResult<String> {
520 field
521 .text()
522 .await
523 .map_err(|e| BucketError::Other(format!("read text field: {e}")))
524}
525
526async fn parse_uuid_field(field: axum::extract::multipart::Field<'_>) -> BucketResult<Uuid> {
527 let s = read_text_field(field).await?;
528 Uuid::parse_str(s.trim()).map_err(|e| BucketError::Other(format!("invalid uuid: {e}")))
529}
530
531fn service_to_bucket_error(e: crate::application::service::error::ServiceError) -> BucketError {
532 use crate::application::service::error::ServiceError;
533 match e {
534 ServiceError::NotFound => BucketError::NotFound,
535 ServiceError::Validation(m) => BucketError::InvalidInput(m),
540 ServiceError::AlreadyExists(m) => BucketError::Conflict(m),
541 other => BucketError::Other(other.to_string()),
542 }
543}
544
545async fn load_bucket(service: &BucketService, bucket_id: Uuid) -> BucketResult<Bucket> {
547 service
548 .find_by_id(&bucket_id.to_string())
549 .await
550 .map_err(|e| BucketError::Other(format!("bucket lookup: {e}")))?
551 .ok_or(BucketError::NotFound)
552}
553
554pub(crate) fn enforce_bucket_policy(
563 bucket: &Bucket,
564 size: i64,
565 mime_type: &str,
566) -> BucketResult<()> {
567 match bucket.status {
568 BucketStatus::Active => {}
569 BucketStatus::Readonly => {
570 return Err(BucketError::Conflict(format!(
571 "bucket {} is read-only",
572 bucket.id
573 )))
574 }
575 BucketStatus::Archived => {
576 return Err(BucketError::Conflict(format!(
577 "bucket {} is archived",
578 bucket.id
579 )))
580 }
581 BucketStatus::Deleted => {
582 return Err(BucketError::NotFound);
583 }
584 }
585 if bucket.is_deleted() {
586 return Err(BucketError::NotFound);
587 }
588 if let Some(max) = bucket.max_file_size {
589 if size > max {
590 return Err(BucketError::PayloadTooLarge(format!(
591 "{} bytes exceeds bucket limit of {} bytes",
592 size, max
593 )));
594 }
595 }
596 if !bucket.allowed_mime_types.is_empty()
597 && !bucket
598 .allowed_mime_types
599 .iter()
600 .any(|m| m.eq_ignore_ascii_case(mime_type))
601 {
602 return Err(BucketError::UnsupportedMediaType(format!(
603 "mime type `{}` is not allowed for bucket {}",
604 mime_type, bucket.id
605 )));
606 }
607 Ok(())
608}
609
610#[cfg(test)]
611mod tests {
612 use super::*;
613
614 #[test]
615 fn part_storage_key_is_stable_and_zero_padded() {
616 let sid = Uuid::parse_str("00000000-0000-0000-0000-000000000001").unwrap();
617 assert_eq!(
618 part_storage_key(sid, 1),
619 "sessions/00000000-0000-0000-0000-000000000001/parts/00000001"
620 );
621 assert_eq!(
622 part_storage_key(sid, 12345),
623 "sessions/00000000-0000-0000-0000-000000000001/parts/00012345"
624 );
625 }
626
627 #[test]
628 fn part_storage_keys_sort_lexicographically_by_part_number() {
629 let sid = Uuid::nil();
633 let mut keys: Vec<String> =
634 [1, 2, 10, 100, 9, 11].iter().map(|n| part_storage_key(sid, *n)).collect();
635 keys.sort();
636 let nums: Vec<i32> = keys
637 .iter()
638 .map(|k| {
639 k.rsplit('/').next().unwrap().parse::<i32>().unwrap()
640 })
641 .collect();
642 assert_eq!(nums, vec![1, 2, 9, 10, 11, 100]);
643 }
644
645 #[test]
646 fn upload_config_defaults_match_documented_limits() {
647 let c = UploadConfig::default();
648 assert_eq!(c.single_shot_limit, DEFAULT_UPLOAD_BODY_LIMIT);
649 assert_eq!(c.chunk_limit, DEFAULT_CHUNK_BODY_LIMIT);
650 assert_eq!(DEFAULT_UPLOAD_BODY_LIMIT, 256 * 1024 * 1024);
651 assert_eq!(DEFAULT_CHUNK_BODY_LIMIT, 16 * 1024 * 1024);
652 }
653
654 #[test]
655 fn service_error_not_found_maps_to_bucket_not_found() {
656 use crate::application::service::error::ServiceError;
657 let mapped = service_to_bucket_error(ServiceError::NotFound);
658 assert!(matches!(mapped, BucketError::NotFound));
659 }
660
661 #[test]
662 fn service_error_validation_maps_to_invalid_input() {
663 use crate::application::service::error::ServiceError;
664 let mapped = service_to_bucket_error(ServiceError::Validation("oops".into()));
665 assert!(matches!(mapped, BucketError::InvalidInput(m) if m == "oops"));
666 }
667
668 fn test_bucket() -> Bucket {
671 use crate::domain::entity::{BucketType, StorageBackend};
672 Bucket::new(
673 "test".into(),
674 "test".into(),
675 Uuid::new_v4(),
676 BucketType::default(),
677 BucketStatus::Active,
678 StorageBackend::Local,
679 "test".into(),
680 0,
681 0,
682 Vec::new(),
683 false,
684 false,
685 false,
686 )
687 }
688
689 #[test]
690 fn policy_allows_active_bucket_with_no_constraints() {
691 let b = test_bucket();
692 assert!(enforce_bucket_policy(&b, 100, "image/png").is_ok());
693 }
694
695 #[test]
696 fn policy_rejects_readonly_bucket_with_409() {
697 let mut b = test_bucket();
698 b.transition_to(crate::domain::state_machine::BucketState::Readonly).ok();
699 let err = enforce_bucket_policy(&b, 100, "image/png").unwrap_err();
700 assert!(matches!(err, BucketError::Conflict(_)));
701 }
702
703 #[test]
704 fn policy_rejects_archived_bucket_with_409() {
705 let mut b = test_bucket();
706 b.transition_to(crate::domain::state_machine::BucketState::Archived).ok();
707 let err = enforce_bucket_policy(&b, 100, "image/png").unwrap_err();
708 assert!(matches!(err, BucketError::Conflict(_)));
709 }
710
711 #[test]
712 fn policy_rejects_size_over_max_file_size_with_413() {
713 let mut b = test_bucket();
714 b.max_file_size = Some(1024);
715 let err = enforce_bucket_policy(&b, 2048, "image/png").unwrap_err();
716 assert!(matches!(err, BucketError::PayloadTooLarge(_)));
717 }
718
719 #[test]
720 fn policy_allows_size_at_max_boundary() {
721 let mut b = test_bucket();
722 b.max_file_size = Some(1024);
723 assert!(enforce_bucket_policy(&b, 1024, "image/png").is_ok());
724 }
725
726 #[test]
727 fn policy_rejects_mime_not_in_allowlist_with_415() {
728 let mut b = test_bucket();
729 b.allowed_mime_types = vec!["image/png".into(), "image/jpeg".into()];
730 let err = enforce_bucket_policy(&b, 100, "application/pdf").unwrap_err();
731 assert!(matches!(err, BucketError::UnsupportedMediaType(_)));
732 }
733
734 #[test]
735 fn policy_mime_match_is_case_insensitive() {
736 let mut b = test_bucket();
737 b.allowed_mime_types = vec!["image/PNG".into()];
738 assert!(enforce_bucket_policy(&b, 100, "IMAGE/png").is_ok());
739 }
740
741 #[test]
742 fn policy_empty_allowlist_means_any_mime_allowed() {
743 let b = test_bucket();
744 assert!(b.allowed_mime_types.is_empty());
745 assert!(enforce_bucket_policy(&b, 100, "anything/goes").is_ok());
746 }
747}