use axum::{ Json, body::Body, extract::{Multipart, Path, Query, State}, http::{HeaderMap, Response, StatusCode, header}, response::IntoResponse, }; use bytes::Bytes; use http_range_header::parse_range_header; use serde::Deserialize; use std::collections::HashMap; use utoipa::ToSchema; use crate::application::ports::external_mount_ports::MountStat; use crate::application::ports::file_ports::{ FileManagementUseCase, FileRetrievalUseCase, FileUploadUseCase, RangeContent, }; use crate::application::ports::storage_ports::{FileReadPort, StorageUsagePort}; use crate::application::ports::thumbnail_ports::ThumbnailPort; use crate::application::ports::{file_ports::OptimizedFileContent, folder_ports::FolderUseCase}; use crate::application::services::external_mount_router::ResolvedId; use crate::application::services::mount_registry::MountConfig; use crate::common::di::AppState; use crate::domain::errors::DomainError; use crate::domain::services::external_mount_id::{NodeId, virtual_file_etag}; use crate::interfaces::errors::AppError; use crate::interfaces::middleware::auth::AuthUser; use crate::interfaces::range_requests::not_modified_response; use crate::interfaces::upload_ingest; use crate::{application::dtos::file_dto::FileDto, domain::services::authorization::Permission}; use std::sync::Arc; /** * Type aliases for dependency injection state. */ /// Global application state for dependency injection type GlobalState = Arc; /** * API handler for file-related operations. * * Acts as a thin HTTP adapter in the hexagonal architecture: it parses requests, * delegates business logic to application services, and maps results to HTTP * responses. No infrastructure or strategy logic lives here. */ pub struct FileHandler; impl FileHandler { // ── Why no #[utoipa::path] here? ───────────────────────────────────────────── // utoipa 5.4.0's proc macro generates helper structs / impls inside its expansion. // Rust allows struct definitions at module scope but forbids them inside impl blocks, // so `#[utoipa::path]` fails on every method in this impl block regardless of HTTP // verb or annotation content. All route handlers are free functions below. // TODO: collapse after utoipa upgrade. // ═══════════════════════════════════════════════════════════════════════ // UPLOAD // ═══════════════════════════════════════════════════════════════════════ /// Streaming file upload — bounded RAM regardless of file size. /// /// The multipart body is streamed straight into the CDC chunk store: /// chunking, hashing and dedup checks happen while the bytes arrive. /// No spool file, no re-read — chunks the store already has are never /// written to disk at all. pub async fn upload_file( State(state): State, auth_user: AuthUser, multipart: Multipart, ) -> impl IntoResponse { match Self::upload_file_inner(&state, &auth_user, multipart).await { Ok((mut file, _blob_hash)) => { crate::interfaces::api::handlers::caller_flags::enrich_file_flags( &state, &mut file, auth_user.id, ) .await; Self::created_json_response(&file).into_response() } Err(response) => response.into_response(), } } /// Instant upload: create a file from a blob the caller already owns. /// /// Zero content bytes travel — the client proved possession of the /// content by hash (it computed BLAKE3 locally and confirmed via /// `GET /api/dedup/check/{hash}`), so the server only bumps the blob's /// reference count and registers the metadata row. /// /// All authorization (folder Create permission, hash ownership with /// anti-enumeration, quota) lives in the application service. pub(super) async fn create_file_by_hash_impl( State(state): State, auth_user: AuthUser, Json(request): Json, ) -> impl IntoResponse { // Hash shape check — same contract as /api/dedup/check/{hash}. if request.hash.len() != 64 || !request.hash.chars().all(|c| c.is_ascii_hexdigit()) { return AppError::bad_request( "Invalid hash format. Expected BLAKE3 (64 hex characters)", ) .into_response(); } // Basename only — same path-traversal guard as the multipart upload. let filename = request .name .rsplit('/') .next() .unwrap_or(&request.name) .rsplit('\\') .next() .unwrap_or(&request.name) .to_string(); if filename.is_empty() { return AppError::bad_request("File name must not be empty").into_response(); } match state .applications .file_upload_service .create_file_from_owned_blob_with_perms( auth_user.id, filename, request.folder_id, &request.hash, ) .await { Ok(mut file) => { crate::interfaces::api::handlers::caller_flags::enrich_file_flags( &state, &mut file, auth_user.id, ) .await; Self::created_json_response(&file).into_response() } Err(err) => { // Anti-enumeration shape: every "caller cannot reach this // hash" outcome collapses into the same 404 with an // `upload_path` hint, regardless of whether the hash exists // globally, is owned by another tenant, or got GC'd in a // race against trash-empty. Hides the cross-tenant content // existence oracle and tells the client where to fall back. // // Three NotFound("Blob", _) paths in the service map here: // 1. user_owns_blob_reference returned false // 2. get_blob_metadata returned None (blob row vanished) // 3. add_reference lost the race with GC (rows_affected==0) use crate::common::errors::ErrorKind; if err.kind == ErrorKind::NotFound && err.entity_type == "Blob" { return Response::builder() .status(StatusCode::NOT_FOUND) .header(header::CONTENT_TYPE, "application/json") .body(Body::from( r#"{"error":"blob_not_owned_by_caller","upload_path":"/api/files/upload"}"#, )) .unwrap() .into_response(); } Self::domain_error_response(err).into_response() } } } /// Core upload logic shared by [`Self::upload_file`] and /// [`Self::upload_file_with_thumbnails`]. /// /// Returns `(FileDto, blob_hash)` on success. The blob hash is the /// BLAKE3 digest computed during the streaming ingest and is /// propagated without an extra database round-trip so that callers /// (e.g. thumbnail generation) can resolve the physical blob path /// immediately. async fn upload_file_inner( state: &GlobalState, auth_user: &AuthUser, mut multipart: Multipart, ) -> Result<(crate::application::dtos::file_dto::FileDto, String), Response> { let upload_service = &state.applications.file_upload_service; let mut folder_id: Option = None; tracing::debug!("📤 Processing streaming file upload (hash-on-write)"); // caveat: if folder_id field is given after check can fails while let Some(field) = multipart.next_field().await.unwrap_or(None) { let name = field.name().unwrap_or("").to_string(); if name == "folder_id" { let v = field.text().await.unwrap_or_default(); if !v.is_empty() { folder_id = Some(v); } continue; } if name == "file" { let raw_filename = field.file_name().unwrap_or("unnamed").to_string(); // Browsers send the full relative path (e.g. "Screenshots/file.png") // as the filename for folder uploads via webkitRelativePath. // Strip path components to get the basename only. // This also prevents path-traversal attacks. let filename = raw_filename .rsplit('/') .next() .unwrap_or(&raw_filename) .rsplit('\\') .next() .unwrap_or(&raw_filename) .to_string(); let content_type = field .content_type() .unwrap_or("application/octet-stream") .to_string(); // ── Fail-fast pre-check: verify the caller can Create inside // the target folder BEFORE spooling the multipart body to disk. // The upload service re-checks at write time — this is a // UX/resource optimization, not the security boundary. if let Some(ref fid) = folder_id && let Err(err) = state .applications .folder_service_concrete .require_permission(auth_user.id, Permission::Create, fid) .await { tracing::warn!( "⛔ UPLOAD REJECTED: user='{}' folder='{}' err='{}'", auth_user.username, fid, err ); return Err(Self::domain_error_response(err)); } // ── Early quota check (before spooling to disk) ────── if let Some(storage_svc) = state.storage_usage_service.as_ref() { let estimated_size = field .headers() .get(header::CONTENT_LENGTH) .and_then(|v| v.to_str().ok()) .and_then(|s| s.parse::().ok()) .unwrap_or(0); if let Err(err) = storage_svc .check_storage_quota(auth_user.id, estimated_size) .await { tracing::warn!( "⛔ UPLOAD REJECTED (early quota): user={}, file={}, est_size={}", auth_user.username, filename, estimated_size ); return Err(Self::quota_error_response(err)); } } // ── External mount destination? Stream to the provider ── // Detected BEFORE the CAS ingest so the bytes never touch // BLAKE3/dedup. Authorization happens inside the service. if let Some(ref fid) = folder_id { let (mount_cfg, parent_node) = match state.mount_router.classify(fid) { ResolvedId::MountRoot { cfg } => (Some(cfg), NodeId::default()), ResolvedId::MountChild { cfg, node_id } => (Some(cfg), node_id), ResolvedId::Regular => (None, NodeId::default()), }; if let Some(cfg) = mount_cfg { use futures::StreamExt; let body: crate::application::ports::external_mount_ports::MountByteStream< '_, > = Box::pin( upload_ingest::multipart_field_stream(field) .map(|r| r.map_err(|e| std::io::Error::other(e.to_string()))), ); return match state .applications .external_upload_service .write_file(&cfg, &parent_node, &filename, body, auth_user.id) .await { Ok(file) => Ok((file, String::new())), Err(err) => Err(Self::domain_error_response(err)), }; } } // ── Stream the field into the CDC chunk store ──────── // Chunking (FastCDC) + hashing (BLAKE3) + dedup checks + // MIME sniffing all happen while the bytes arrive; chunks // the store already has never touch the disk. Size is // capped globally by DefaultBodyLimit. let dedup = &state.core.dedup_service; let source = upload_ingest::multipart_field_stream(field); let ingested = match upload_ingest::ingest_stream_to_cas( source, dedup, &filename, &content_type, usize::MAX, None, ) .await { Ok(ingested) => ingested, Err(e) => { tracing::error!("❌ UPLOAD INGEST FAILED: {} - {}", filename, e.message); return Err(e.into_response()); } }; // ── Quota enforcement (exact size now known) ───────── if let Some(storage_svc) = state.storage_usage_service.as_ref() && let Err(err) = storage_svc .check_storage_quota(auth_user.id, ingested.size) .await { upload_ingest::discard_ingested(dedup, &ingested).await; tracing::warn!( "⛔ UPLOAD REJECTED (user quota): user={}, file={}, size={}", auth_user.username, filename, ingested.size ); return Err(Self::quota_error_response(err)); } // ── Per-drive quota enforcement (D4) ───────────────── // Sibling to the per-user check above: same read-only // SELECT shape, same discard-then-507 outcome. Skipped // when there's no folder_id (root-level upload — no // drive to charge; folder service refuses these // independently). Unlimited-quota drives (`NULL`) // short-circuit inside the service. if let Some(storage_svc) = state.storage_usage_service.as_ref() && let Some(fid_str) = folder_id.as_deref() && let Ok(fid) = uuid::Uuid::parse_str(fid_str) && let Err(err) = storage_svc .check_drive_quota_by_folder(fid, ingested.size) .await { upload_ingest::discard_ingested(dedup, &ingested).await; tracing::warn!( "⛔ UPLOAD REJECTED (drive quota): user={}, folder={}, file={}, size={}", auth_user.username, fid, filename, ingested.size ); return Err(Self::quota_error_response(err)); } // ── Register the file row against the ingested blob ── let hash = ingested.hash.clone(); let size = ingested.size; match upload_service .upload_file_streaming( filename.clone(), folder_id, ingested.content_type.clone(), ingested.stored(), auth_user.id, ) .await { Ok(file) => { tracing::info!( "✅ STREAMING UPLOAD: {} ({} bytes, ID: {})", filename, size, file.id ); return Ok((file, hash)); } Err(err) => { tracing::error!("❌ UPLOAD FAILED: {} - {}", filename, err); return Err(Self::domain_error_response(err)); } } } } Err(( StatusCode::BAD_REQUEST, Json(serde_json::json!({ "error": "No file provided" })), ) .into_response()) } // ═══════════════════════════════════════════════════════════════════════ // THUMBNAILS // ═══════════════════════════════════════════════════════════════════════ /// Get a thumbnail for a file (image or video). /// /// **Cache-first**: if the thumbnail already exists in the moka in-memory /// cache or on disk, serve it immediately — **zero DB queries**. The /// ownership check was already performed when the thumbnail was first /// generated (at upload) or uploaded (PUT by the owner). UUIDv4 file IDs /// have 122 bits of entropy, making enumeration infeasible. /// /// **ETag / 304**: responses carry an immutable ETag. If the browser /// sends `If-None-Match` matching the ETag, we return 304 Not Modified /// without touching cache or DB — pure header round-trip. /// /// The DB path is only taken on a **cache miss for images** where the /// thumbnail hasn't been generated yet (first access after upload if /// background generation hasn't finished). pub(super) async fn get_thumbnail_impl( State(state): State, auth_user: AuthUser, headers: &HeaderMap, Path((id, size)): Path<(String, String)>, ) -> impl IntoResponse + use<> { use crate::application::ports::thumbnail_ports::{ThumbnailFormat, ThumbnailSize}; // check first that user can access this resource if let Err(err) = state .applications .file_management_service .require_permission(auth_user.id, Permission::Read, &id) .await { return AppError::from(err).into_response(); } let thumbnail_service = &state.core.thumbnail_service; let thumb_size = match size.as_str() { "icon" => ThumbnailSize::Icon, "preview" => ThumbnailSize::Preview, "large" => ThumbnailSize::Large, _ => { return ( StatusCode::BAD_REQUEST, Json(serde_json::json!({ "error": "Invalid thumbnail size. Use: icon, preview, or large" })), ) .into_response(); } }; // Content negotiation: WebP for clients that advertise it (~97%), JPEG // otherwise. `Vary: Accept` keeps shared/browser caches from handing a // WebP body to a JPEG-only client (or vice-versa). let format = ThumbnailFormat::from_accept(headers.get(header::ACCEPT).and_then(|v| v.to_str().ok())); // ── ETag short-circuit (Solution C) ────────────────────────── // Thumbnails are immutable — the ETag never changes for a given // (file_id, size, format) triple. If the browser already has it, return // 304 with zero I/O or DB work. Format is in the ETag so a client that // switched codecs doesn't get a stale 304. let etag = { let (s, f) = (thumb_size.as_str(), format.as_str()); let mut e = String::with_capacity(9 + id.len() + s.len() + f.len()); e.push_str("\"thumb-"); e.push_str(&id); e.push('-'); e.push_str(s); e.push('-'); e.push_str(f); e.push('"'); e }; if let Some(if_none_match) = headers.get(header::IF_NONE_MATCH) && let Ok(val) = if_none_match.to_str() && (val == etag || val == "*") { return Response::builder() .status(StatusCode::NOT_MODIFIED) .header(header::ETAG, &etag) .header(header::VARY, header::ACCEPT.as_str()) .header(header::CACHE_CONTROL, "public, max-age=31536000, immutable") .body(Body::empty()) .unwrap() .into_response(); } // ── Cache-first path (Solution A) ──────────────────────────── // Try moka (RAM) → disk before touching the database. // If the thumbnail exists it was authorized at creation time. if let Some(data) = thumbnail_service .get_cached_thumbnail(&id, None, thumb_size.into(), format) .await { return Response::builder() .status(StatusCode::OK) .header( header::CONTENT_TYPE, crate::common::mime_detect::thumbnail_content_type(&data), ) .header(header::CONTENT_LENGTH, data.len()) .header(header::CACHE_CONTROL, "public, max-age=31536000, immutable") .header(header::ETAG, &etag) .header(header::VARY, header::ACCEPT.as_str()) .body(Body::from(data)) .unwrap() .into_response(); } // ── Cache miss — need DB for ownership + blob resolution ───── let file_retrieval_service = &state.applications.file_retrieval_service; let file = match file_retrieval_service .get_file_or_trashed_with_perms(&id, auth_user.id) .await { Ok(f) => f, Err(err) => { return AppError::from(err).into_response(); } }; // Images and videos both store blob-hash thumbnails (videos via an // eagerly-extracted frame); anything else has nothing to thumbnail → 204. let is_image = thumbnail_service.is_supported_image(&file.mime_type); let is_video = file.mime_type.starts_with("video/"); if !is_image && !is_video { return Response::builder() .status(StatusCode::NO_CONTENT) .header(header::CACHE_CONTROL, "no-store") .body(Body::empty()) .unwrap() .into_response(); } // Resolve the blob hash (content-addressable storage). let blob_hash = match state .repositories .file_read_repository .get_blob_hash(&id) .await { Ok(hash) => hash, Err(_) => { return AppError::internal_error("File blob not found").into_response(); } }; if let Some(data) = thumbnail_service .get_cached_thumbnail(&id, Some(&blob_hash), thumb_size.into(), format) .await { return Response::builder() .status(StatusCode::OK) .header( header::CONTENT_TYPE, crate::common::mime_detect::thumbnail_content_type(&data), ) .header(header::CONTENT_LENGTH, data.len()) .header(header::CACHE_CONTROL, "public, max-age=31536000, immutable") .header(header::ETAG, &etag) .header(header::VARY, header::ACCEPT.as_str()) .body(Body::from(data)) .unwrap() .into_response(); } // Videos: thumbnails are produced eagerly server-side (ffmpeg) on upload, // and persisted WebP-only. Serve that WebP regardless of the negotiated // format — a JPEG/`*/*`/no-Accept client still gets it, correctly labelled // via byte-sniffing — otherwise non-WebP clients would 204 forever despite // a valid thumbnail on disk. We never image-decode a video, so a genuine // miss (generation in flight or unavailable) returns 204. if is_video { if let Some(data) = thumbnail_service .get_cached_thumbnail( &id, Some(&blob_hash), thumb_size.into(), ThumbnailFormat::Webp, ) .await { return Response::builder() .status(StatusCode::OK) .header( header::CONTENT_TYPE, crate::common::mime_detect::thumbnail_content_type(&data), ) .header(header::CONTENT_LENGTH, data.len()) .header(header::CACHE_CONTROL, "public, max-age=31536000, immutable") .header(header::ETAG, &etag) .header(header::VARY, header::ACCEPT.as_str()) .body(Body::from(data)) .unwrap() .into_response(); } return Response::builder() .status(StatusCode::NO_CONTENT) .header(header::CACHE_CONTROL, "no-store") .body(Body::empty()) .unwrap() .into_response(); } match thumbnail_service .get_thumbnail_from_blob( &id, &blob_hash, thumb_size.into(), format, state.core.dedup_service.clone(), ) .await { Ok(data) => Response::builder() .status(StatusCode::OK) .header( header::CONTENT_TYPE, crate::common::mime_detect::thumbnail_content_type(&data), ) .header(header::CONTENT_LENGTH, data.len()) .header(header::CACHE_CONTROL, "public, max-age=31536000, immutable") .header(header::ETAG, &etag) .header(header::VARY, header::ACCEPT.as_str()) .body(Body::from(data)) .unwrap() .into_response(), Err(err) => AppError::internal_error(format!("Thumbnail generation failed: {}", err)) .into_response(), } } // ═══════════════════════════════════════════════════════════════════════ // UPLOAD THUMBNAIL (client-generated, e.g. video frames) // ═══════════════════════════════════════════════════════════════════════ /// Accept a client-generated thumbnail (e.g. video frame extracted via /// `