diff --git a/src/application/services/file_upload_service.rs b/src/application/services/file_upload_service.rs index fabcb5c1..824883ec 100644 --- a/src/application/services/file_upload_service.rs +++ b/src/application/services/file_upload_service.rs @@ -14,26 +14,7 @@ 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::{debug, info, warn}; - -/// Helper function to extract username from folder path string. -/// e.g. "My Folder - user1/subfolder/file.txt" → "user1" -fn extract_username_from_path(path: &str) -> Option { - if !path.contains("My Folder - ") { - return None; - } - let parts: Vec<&str> = path.split("My Folder - ").collect(); - if parts.len() <= 1 { - return None; - } - let remainder = parts[1].trim(); - let username = remainder.split('/').next().unwrap_or(remainder); - let username = username.trim(); - if username.is_empty() { - return None; - } - Some(username.to_string()) -} +use tracing::{info, warn}; /// Service for file upload operations. /// @@ -305,26 +286,35 @@ impl FileUploadService { // ── private helpers ────────────────────────────────────────── - /// Optionally update storage usage after a successful upload. + /// 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) { - if let Some(storage_service) = &self.storage_usage_service { - let file_path = file.path.clone(); - if let Some(username) = extract_username_from_path(&file_path) { - let service_clone = Arc::clone(storage_service); - tokio::spawn(async move { - match service_clone - .update_user_storage_usage_by_username(&username) - .await - { - Ok(usage) => debug!( - "Updated storage usage for user {} to {} bytes", - username, usage - ), - Err(e) => warn!("Failed to update storage usage for {}: {}", username, e), - } - }); + 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}"); } - } + }); } } diff --git a/src/application/services/storage_usage_service.rs b/src/application/services/storage_usage_service.rs index 20157a3c..a5bdb115 100644 --- a/src/application/services/storage_usage_service.rs +++ b/src/application/services/storage_usage_service.rs @@ -102,6 +102,32 @@ impl StorageUsageService { Ok(total_usage) } + /// Incrementally adjust one user's cached usage by `delta` bytes — O(1), + /// the per-upload counterpart to the O(N) full recompute above. An upload + /// adds `+size` (was a full `SUM(size)` over every file the user owns, i.e. + /// O(N) per upload and O(N²) for a bulk upload). Deletes/trash do not + /// decrement here (they never did); the periodic reconciliation sweep + /// ([`StorageUsagePort::update_all_users_storage_usage`]) remains the + /// correctness backstop for every mutation. Clamped at 0 so a late or + /// duplicate adjustment can never drive the counter negative. + pub async fn add_user_storage_usage_delta( + &self, + user_id: Uuid, + delta: i64, + ) -> Result<(), DomainError> { + sqlx::query( + "UPDATE auth.users + SET storage_used_bytes = GREATEST(0, storage_used_bytes + $2) + WHERE id = $1", + ) + .bind(user_id) + .bind(delta) + .execute(self.pool.as_ref()) + .await + .map_err(|e| DomainError::internal_error("StorageUsage", format!("usage delta: {e}")))?; + Ok(()) + } + /// Spawn a background task that periodically reconciles every user's cached /// `storage_used_bytes` against the actual sum of their files. ///