use std::sync::Arc; use uuid::Uuid; use crate::application::dtos::file_dto::FileDto; use crate::application::ports::authorization_ports::AuthorizationEngine; use crate::application::ports::file_lifecycle::FileLifecycleHook; use crate::application::ports::file_ports::{FileUploadUseCase, StoredBlob}; use crate::application::ports::storage_ports::{FileReadPort, FileWritePort, StorageUsagePort}; use crate::application::services::storage_usage_service::StorageUsageService; use crate::common::errors::DomainError; use crate::domain::services::authorization::{Permission, Resource, Subject}; use crate::infrastructure::repositories::pg::FileBlobReadRepository; use crate::infrastructure::repositories::pg::FileBlobWriteRepository; use crate::infrastructure::services::dedup_service::DedupService; use crate::infrastructure::services::file_content_cache::FileContentCache; use crate::infrastructure::services::pg_acl_engine::PgAclEngine; use tracing::{info, warn}; /// Service for file upload operations. /// /// Content never passes through this service: the interface layer streams /// the request body straight into the CDC chunk store (no spool file, no /// full-body buffering) and hands over a [`StoredBlob`] reference. This /// service registers the metadata row, keeps caches coherent and fires /// lifecycle hooks. Blob-reference ownership is consumed by the write /// port, which releases it on failure — callers never compensate. pub struct FileUploadService { /// Write port — registers file rows against ingested blobs. file_write: Arc, /// Read port — needed for WebDAV/WOPI update-by-path. file_read: Option>, /// Optional storage usage tracking storage_usage_service: Option>, /// Content cache — invalidated on file update so stale content is never served. content_cache: Option>, /// Single lifecycle dispatcher — fires on_file_created / on_file_updated. file_lifecycle_hook: Option>, /// Dependencies of the instant-upload path /// (`create_file_from_owned_blob_with_perms`); `None` in minimal test /// wiring. instant_upload: Option, } /// Everything the instant-upload path needs beyond the upload service's own /// ports: permission checks, the dedup index, and quota enforcement. struct InstantUploadDeps { authz: Arc, dedup: Arc, quota: Arc, } impl FileUploadService { /// Constructor with write port only (minimal). pub fn new(file_repository: Arc) -> Self { Self { file_write: file_repository, file_read: None, storage_usage_service: None, content_cache: None, file_lifecycle_hook: None, instant_upload: None, } } /// Constructor for blob-storage model: write + read ports. pub fn new_with_read( file_write: Arc, file_read: Arc, ) -> Self { Self { file_write, file_read: Some(file_read), storage_usage_service: None, content_cache: None, file_lifecycle_hook: None, instant_upload: None, } } /// Wires the authorization engine, dedup index and quota service that /// power the instant-upload path. pub fn with_instant_upload( mut self, authz: Arc, dedup: Arc, quota: Arc, ) -> Self { self.instant_upload = Some(InstantUploadDeps { authz, dedup, quota, }); self } /// Configures the content cache for invalidation on file updates. pub fn with_content_cache(mut self, cache: Arc) -> Self { self.content_cache = Some(cache); self } /// Registers the lifecycle hook dispatcher (thumbnails, audio metadata, …). pub fn with_file_lifecycle_hook(mut self, hook: Arc) -> Self { self.file_lifecycle_hook = Some(hook); self } /// Configures the storage usage service pub fn with_storage_usage_service( mut self, storage_usage_service: Arc, ) -> Self { self.storage_usage_service = Some(storage_usage_service); self } // ── Instant upload (zero content bytes) ────────────────────── /// Register a new file row pointing at a blob the caller **already /// owns** — the instant-upload path: the client proved it has the /// content by hash, so no bytes travel and no chunk is written. Pure /// metadata: one ref_count bump + one row INSERT. /// /// Security model (mirrors `GET /api/dedup/check/{hash}`): /// - The caller must have `Create` permission on the target folder. /// - The hash is only claimable when the caller owns at least one /// non-trashed file referencing it — never a global content oracle. /// A non-owned hash returns `NotFound` (anti-enumeration: same shape /// as "no such blob") and emits an `instant_upload.rejected` audit /// event with the real reason. /// - Quota is enforced on the logical size, exactly like a byte upload. pub async fn create_file_from_owned_blob_with_perms( &self, caller_id: Uuid, name: String, folder_id: String, hash: &str, ) -> Result { let Some(InstantUploadDeps { authz, dedup, quota, }) = &self.instant_upload else { return Err(DomainError::internal_error( "FileUpload", "instant upload is not wired (authz/dedup/quota missing)", )); }; // ── AuthZ: Create on the target folder ─────────────────── let folder_uuid = Uuid::parse_str(&folder_id) .map_err(|_| DomainError::not_found("Folder", folder_id.clone()))?; authz .require( Subject::User(caller_id), Permission::Create, Resource::Folder(folder_uuid), ) .await?; // ── Ownership: only blobs the caller can already read ──── if !dedup .user_owns_blob_reference(hash, &caller_id.to_string()) .await { tracing::info!( target: "audit", event = "instant_upload.rejected", reason = "hash_not_owned", caller_id = %caller_id, blob_hash = %hash, "👮🏻‍♂️ Instant upload rejected: caller owns no file referencing the claimed hash", ); return Err(DomainError::not_found("Blob", hash)); } let Some(metadata) = dedup.get_blob_metadata(hash).await else { // Lost a race with the last-reference delete — same shape as // "never existed". return Err(DomainError::not_found("Blob", hash)); }; // ── Quota on the logical size, before taking any reference ── quota.check_storage_quota(caller_id, metadata.size).await?; // The manifest knows the original content type; fall back to the // new name's extension when the stored one is generic. let claimed = metadata.content_type.as_deref().unwrap_or(""); let content_type = match crate::common::mime_detect::refine_content_type(&[], &name, claimed) { ct if ct.is_empty() => "application/octet-stream".to_string(), ct => ct, }; // Take the reference the row registration will consume (it releases // it again on any failure). A concurrent GC between the ownership // check and this bump surfaces as NotFound — the client falls back // to a normal byte upload. dedup.add_reference(hash).await?; let dto = self .upload_file_streaming( name, Some(folder_id), content_type, StoredBlob { hash: hash.to_string(), size: metadata.size, is_new_blob: false, }, caller_id, ) .await?; info!( "⚡ INSTANT UPLOAD: {} ({} bytes, 0 transferred, ID: {})", dto.name, metadata.size, dto.id ); Ok(dto) } /// Swap an existing file's content to an already-ingested blob — the /// update mode of the delta-upload commit. The caller needs `Write` /// permission on the file; the blob reference is consumed (released on /// failure by the write port, like every other registration path). pub async fn update_file_content_by_id_with_perms( &self, caller_id: Uuid, file_id: &str, blob: StoredBlob, ) -> Result { let Some(InstantUploadDeps { authz, .. }) = &self.instant_upload else { return Err(DomainError::internal_error( "FileUpload", "instant upload is not wired (authz/dedup/quota missing)", )); }; let Some(file_read) = &self.file_read else { return Err(DomainError::internal_error( "FileUpload", "read port is not wired", )); }; let file_uuid = Uuid::parse_str(file_id) .map_err(|_| DomainError::not_found("File", file_id.to_string()))?; authz .require( Subject::User(caller_id), Permission::Update, Resource::File(file_uuid), ) .await?; let file = file_read.get_file(file_id).await?; let (new_hash, updated_at) = self .file_write .update_file_content_with_blob(file_id, &blob.hash, blob.size, None, caller_id) .await?; // The file maps to a different blob now — stale cached content must // never be served for the rest of its TTI window. if let Some(cc) = &self.content_cache { cc.invalidate(file_id).await; } let parts = file.into_parts(); let updated = crate::domain::entities::file::File::with_timestamps_and_blob_hash( parts.id, parts.name, parts.storage_path, blob.size, parts.mime_type, parts.folder_id, parts.created_at, updated_at as u64, parts.owner_id, new_hash, ) .map_err(|e| DomainError::internal_error("FileUpload", format!("rebuild entity: {e}")))?; let dto = FileDto::from(updated); if let Some(hook) = &self.file_lifecycle_hook { hook.on_file_updated(file_id, &dto.content_hash, &dto.mime_type); } Ok(dto) } // ── private helpers ────────────────────────────────────────── /// Bump the owner's cached storage usage after a successful upload. /// /// Incremental (`+size`, O(1)) and fire-and-forget on a background task, so /// it adds neither latency nor a `SUM(size)` over the user's whole library /// to the upload path (the previous full recompute was O(N) per upload, /// O(N²) for a bulk upload). Keyed by the file's `owner_id`; drift — e.g. /// deletes, which don't decrement — is reconciled by the periodic sweep. A /// DTO without a resolvable owner is simply left to that sweep. fn maybe_update_storage_usage(&self, file: &FileDto) { let Some(storage_service) = &self.storage_usage_service else { return; }; let Some(owner) = file .owner_id .as_deref() .and_then(|s| Uuid::parse_str(s).ok()) else { return; }; let delta = file.size as i64; let service_clone = Arc::clone(storage_service); tokio::spawn(async move { if let Err(e) = service_clone .add_user_storage_usage_delta(owner, delta) .await { warn!("Failed to bump storage usage for {owner}: {e}"); } }); } } impl FileUploadUseCase for FileUploadService { /// Register a new file row pointing at an already-ingested blob. async fn upload_file_streaming( &self, name: String, folder_id: Option, content_type: String, blob: StoredBlob, caller_id: Uuid, ) -> Result { let file = self .file_write .save_file_with_blob( name.clone(), folder_id, content_type, &blob.hash, blob.size, caller_id, ) .await?; let dto = FileDto::from(file); info!( "📡 STREAMING UPLOAD: {} ({} bytes, ID: {})", name, blob.size, dto.id ); self.maybe_update_storage_usage(&dto); if let Some(hook) = &self.file_lifecycle_hook { hook.on_file_created(&dto.id, &dto.content_hash, &dto.mime_type, blob.is_new_blob); } Ok(dto) } /// Swap the content of the file at `path` to an already-ingested blob, /// creating the file when it doesn't exist (WebDAV/NextCloud/WOPI PUT). async fn update_file_streaming( &self, path: &str, drive_id: Uuid, blob: StoredBlob, content_type: &str, modified_at: Option, caller_id: Uuid, ) -> Result { // Try to find the existing file first if let Some(file_read) = &self.file_read && let Some(file) = file_read.find_file_by_path(path, drive_id).await? { let file_id = file.id().to_string(); let (new_hash, updated_at) = self .file_write .update_file_content_with_blob( &file_id, &blob.hash, blob.size, modified_at, caller_id, ) .await?; // Invalidate content cache — file content has changed. if let Some(cc) = &self.content_cache { cc.invalidate(&file_id).await; } // Rebuild the fresh DTO from the entity already in hand plus the // values the UPDATE just returned — a re-read would only fetch // what we already know, at one extra round-trip per overwrite // (WebDAV sync clients overwrite constantly). let parts = file.into_parts(); let updated = crate::domain::entities::file::File::with_timestamps_and_blob_hash( parts.id, parts.name, parts.storage_path, blob.size, parts.mime_type, parts.folder_id, parts.created_at, updated_at as u64, parts.owner_id, new_hash, ) .map_err(|e| { DomainError::internal_error("FileUpload", format!("rebuild entity: {e}")) })?; let dto = FileDto::from(updated); if let Some(hook) = &self.file_lifecycle_hook { hook.on_file_updated(&file_id, &dto.content_hash, content_type); } return Ok(dto); } // File doesn't exist — create it via streaming upload let path_normalized = path.trim_start_matches('/').trim_end_matches('/'); let (_, filename) = if let Some(idx) = path_normalized.rfind('/') { (&path_normalized[..idx], &path_normalized[idx + 1..]) } else { ("", path_normalized) }; // get_parent_folder_id expects the full file path — it strips the // last segment (filename) internally to find the parent folder. // `drive_id` scopes the parent lookup to the same drive as the // incoming write (post-D0 `storage.folders.path` repeats across // drives). let parent_id = if path_normalized.contains('/') { if let Some(file_read) = &self.file_read { file_read .get_parent_folder_id(path_normalized, drive_id) .await .ok() } else { None } } else { None }; let is_new_blob = blob.is_new_blob; let created = self .file_write .save_file_with_blob( filename.to_string(), parent_id, content_type.to_string(), &blob.hash, blob.size, caller_id, ) .await?; let dto = FileDto::from(created); if let Some(hook) = &self.file_lifecycle_hook { hook.on_file_created(&dto.id, &dto.content_hash, content_type, is_new_blob); } Ok(dto) } }