diff --git a/src/application/dtos/display_helpers.rs b/src/application/dtos/display_helpers.rs index 9f562bfb..384fd19e 100644 --- a/src/application/dtos/display_helpers.rs +++ b/src/application/dtos/display_helpers.rs @@ -1,12 +1,12 @@ -/// Shared display helpers for DTOs. -/// -/// These functions centralise the mime→icon / mime→category / size→human-string -/// logic so that every API response carries pre-computed display fields and the -/// frontend does **not** need to duplicate these mappings. -/// -/// The approach is: try MIME first (specific matches beat prefix matches), -/// then fall back to the file extension when the MIME is generic -/// (`application/octet-stream` or empty). +// Shared display helpers for DTOs. +// +// These functions centralise the mime→icon / mime→category / size→human-string +// logic so that every API response carries pre-computed display fields and the +// frontend does **not** need to duplicate these mappings. +// +// The approach is: try MIME first (specific matches beat prefix matches), +// then fall back to the file extension when the MIME is generic +// (`application/octet-stream` or empty). // ─── Private: extract lowercase extension from a filename ──────────── @@ -100,8 +100,6 @@ pub fn icon_class_for(name: &str, mime: &str) -> &'static str { || mime.contains("typescript") { return "fas fa-file-code"; - } else if mime.starts_with("text/markdown") { - return "fas fa-file-alt"; } else if mime.starts_with("text/") { return "fas fa-file-alt"; } diff --git a/src/application/ports/file_ports.rs b/src/application/ports/file_ports.rs index eba0027d..42d1202e 100644 --- a/src/application/ports/file_ports.rs +++ b/src/application/ports/file_ports.rs @@ -165,10 +165,7 @@ pub trait FileRetrievalUseCase: Send + Sync + 'static { /// Lists every file in the subtree rooted at `folder_id`. /// /// Default: falls back to `list_files(Some(folder_id))` (one level). - async fn list_files_in_subtree( - &self, - folder_id: &str, - ) -> Result, DomainError> { + async fn list_files_in_subtree(&self, folder_id: &str) -> Result, DomainError> { self.list_files(Some(folder_id)).await } @@ -183,7 +180,11 @@ pub trait FileRetrievalUseCase: Send + Sync + 'static { limit: i64, ) -> Result, DomainError> { let all = self.list_files(folder_id).await?; - Ok(all.into_iter().skip(offset as usize).take(limit as usize).collect()) + Ok(all + .into_iter() + .skip(offset as usize) + .take(limit as usize) + .collect()) } } diff --git a/src/application/ports/inbound.rs b/src/application/ports/inbound.rs index 027cd3d3..731b6d2f 100644 --- a/src/application/ports/inbound.rs +++ b/src/application/ports/inbound.rs @@ -76,10 +76,7 @@ pub trait FolderUseCase: Send + Sync + 'static { /// ordered by path. Uses ltree `<@` — single GiST-indexed query. /// /// Default: returns an empty vec (stubs / mocks). - async fn list_subtree_folders( - &self, - folder_id: &str, - ) -> Result, DomainError> { + async fn list_subtree_folders(&self, folder_id: &str) -> Result, DomainError> { let _ = folder_id; Ok(Vec::new()) } @@ -97,7 +94,11 @@ pub trait SearchUseCase: Send + Sync + 'static { /// /// `user_id` identifies the authenticated user so that SQL queries filter /// by owner and the result cache is isolated per tenant. - async fn search(&self, criteria: SearchCriteriaDto, user_id: &str) -> Result; + async fn search( + &self, + criteria: SearchCriteriaDto, + user_id: &str, + ) -> Result; /// Returns quick suggestions for autocomplete (lightweight, fast). async fn suggest( diff --git a/src/application/ports/storage_ports.rs b/src/application/ports/storage_ports.rs index 57b79212..0153107e 100644 --- a/src/application/ports/storage_ports.rs +++ b/src/application/ports/storage_ports.rs @@ -100,10 +100,7 @@ pub trait FileReadPort: Send + Sync + 'static { /// subtree is fetched in a single GiST-indexed query. /// /// Default: falls back to `list_files(Some(folder_id))` (one level). - async fn list_files_in_subtree( - &self, - folder_id: &str, - ) -> Result, DomainError> { + async fn list_files_in_subtree(&self, folder_id: &str) -> Result, DomainError> { self.list_files(Some(folder_id)).await } diff --git a/src/application/services/auth_application_service.rs b/src/application/services/auth_application_service.rs index d4214462..e95d1ee1 100644 --- a/src/application/services/auth_application_service.rs +++ b/src/application/services/auth_application_service.rs @@ -526,7 +526,10 @@ impl AuthApplicationService { } // Hash new password and update user - let new_hash = self.password_hasher.hash_password(&dto.new_password).await?; + let new_hash = self + .password_hasher + .hash_password(&dto.new_password) + .await?; user.update_password_hash(new_hash); // Save updated user @@ -642,10 +645,7 @@ impl AuthApplicationService { let admin_quota = self.capped_quota(&admin_role); // Hash the password (same as register / admin_create_user) - let password_hash = self - .password_hasher - .hash_password(&dto.password) - .await?; + let password_hash = self.password_hasher.hash_password(&dto.password).await?; // Create the new admin user let user = User::new( @@ -1156,10 +1156,8 @@ impl AuthApplicationService { let exchange_code = hex::encode(code_bytes); // Store auth response (auto-expires after 60 s via moka TTL) - self.pending_oidc_tokens.insert( - exchange_code.clone(), - PendingOidcToken { auth_response }, - ); + self.pending_oidc_tokens + .insert(exchange_code.clone(), PendingOidcToken { auth_response }); tracing::info!("OIDC login successful, one-time exchange code generated"); @@ -1169,13 +1167,16 @@ impl AuthApplicationService { /// Exchange a one-time code for the authentication tokens. /// The code is single-use and expires after 60 seconds (moka TTL). pub fn exchange_oidc_token(&self, one_time_code: &str) -> Result { - let pending = self.pending_oidc_tokens.remove(one_time_code).ok_or_else(|| { - DomainError::new( - ErrorKind::AccessDenied, - "OIDC", - "Invalid or expired exchange code. Please try logging in again.", - ) - })?; + let pending = self + .pending_oidc_tokens + .remove(one_time_code) + .ok_or_else(|| { + DomainError::new( + ErrorKind::AccessDenied, + "OIDC", + "Invalid or expired exchange code. Please try logging in again.", + ) + })?; Ok(pending.auth_response) } diff --git a/src/application/services/batch_operations.rs b/src/application/services/batch_operations.rs index ff894297..65e89bcb 100644 --- a/src/application/services/batch_operations.rs +++ b/src/application/services/batch_operations.rs @@ -713,7 +713,10 @@ impl BatchOperationService { match chunk { Ok(bytes) => { if let Err(e) = zip.write_all(&bytes) { - info!("Could not write zip chunk for {}: {}", file_dto.name, e); + info!( + "Could not write zip chunk for {}: {}", + file_dto.name, e + ); break; } } diff --git a/src/application/services/file_management_service.rs b/src/application/services/file_management_service.rs index a03eb8cf..397a3777 100644 --- a/src/application/services/file_management_service.rs +++ b/src/application/services/file_management_service.rs @@ -177,7 +177,10 @@ impl FileManagementUseCase for FileManagementService { .copy_folder_tree(source_folder_id, target_parent_id, dest_name) .await .map_err(|e| { - error!("Error copying folder tree (source: {}): {}", source_folder_id, e); + error!( + "Error copying folder tree (source: {}): {}", + source_folder_id, e + ); e })?; diff --git a/src/application/services/file_retrieval_service.rs b/src/application/services/file_retrieval_service.rs index f877dd21..a20025eb 100644 --- a/src/application/services/file_retrieval_service.rs +++ b/src/application/services/file_retrieval_service.rs @@ -88,7 +88,10 @@ impl FileRetrievalService { return None; } let format = OutputFormat::WebP; - match transcode.get_transcoded(id, content.clone(), mime, format).await { + match transcode + .get_transcoded(id, content.clone(), mime, format) + .await + { Ok((transcoded, webp_mime, true)) => { debug!( "🖼️ WebP transcode: {} -> {} bytes ({:.0}% smaller)", @@ -200,12 +203,7 @@ impl FileRetrievalService { let etag: Arc = format!("\"{}-{}\"", id, modified_at).into(); let ct: Arc = Arc::from(&*mime_type); cache - .put( - id.to_string(), - content_bytes.clone(), - etag, - ct, - ) + .put(id.to_string(), content_bytes.clone(), etag, ct) .await; } @@ -310,10 +308,7 @@ impl FileRetrievalUseCase for FileRetrievalService { self.file_read.get_file_range_stream(id, start, end).await } - async fn list_files_in_subtree( - &self, - folder_id: &str, - ) -> Result, DomainError> { + async fn list_files_in_subtree(&self, folder_id: &str) -> Result, DomainError> { let files = self.file_read.list_files_in_subtree(folder_id).await?; Ok(files.into_iter().map(FileDto::from).collect()) } @@ -324,7 +319,10 @@ impl FileRetrievalUseCase for FileRetrievalService { offset: i64, limit: i64, ) -> Result, DomainError> { - let files = self.file_read.list_files_batch(folder_id, offset, limit).await?; + let files = self + .file_read + .list_files_batch(folder_id, offset, limit) + .await?; Ok(files.into_iter().map(FileDto::from).collect()) } } diff --git a/src/application/services/folder_service.rs b/src/application/services/folder_service.rs index 7dc2e7f7..a695a795 100644 --- a/src/application/services/folder_service.rs +++ b/src/application/services/folder_service.rs @@ -182,10 +182,7 @@ impl FolderUseCase for FolderService { Ok(FolderDto::from(folder)) } - async fn list_subtree_folders( - &self, - folder_id: &str, - ) -> Result, DomainError> { + async fn list_subtree_folders(&self, folder_id: &str) -> Result, DomainError> { let folders = self.folder_storage.list_subtree_folders(folder_id).await?; Ok(folders.into_iter().map(FolderDto::from).collect()) } diff --git a/src/application/services/search_service.rs b/src/application/services/search_service.rs index ec0447a4..f25ede7c 100644 --- a/src/application/services/search_service.rs +++ b/src/application/services/search_service.rs @@ -347,19 +347,15 @@ impl SearchUseCase for SearchService { let start_idx = criteria.offset.min(total_count); let end_idx = (criteria.offset + criteria.limit).min(total_count); - let mut paginated_folders = Vec::new(); - let mut paginated_files = Vec::new(); + let folder_start = start_idx.min(folder_count); + let folder_end = end_idx.min(folder_count); + let paginated_folders = enriched_folders[folder_start..folder_end].to_vec(); - for i in start_idx..end_idx { - if i < folder_count { - paginated_folders.push(enriched_folders[i].clone()); - } else { - let file_idx = i - folder_count; - if file_idx < enriched_files.len() { - paginated_files.push(enriched_files[file_idx].clone()); - } - } - } + let file_start = start_idx.saturating_sub(folder_count); + let file_end = end_idx + .saturating_sub(folder_count) + .min(enriched_files.len()); + let paginated_files = enriched_files[file_start..file_end].to_vec(); let elapsed_ms = start.elapsed().as_millis() as u64; @@ -413,12 +409,10 @@ impl SearchUseCase for SearchService { // ── Sort folders (files already sorted by SQL ORDER BY) ── match criteria.sort_by.as_str() { "name" => { - enriched_folders - .sort_by(|a, b| a.name.to_lowercase().cmp(&b.name.to_lowercase())); + enriched_folders.sort_by(|a, b| a.name.to_lowercase().cmp(&b.name.to_lowercase())); } "name_desc" => { - enriched_folders - .sort_by(|a, b| b.name.to_lowercase().cmp(&a.name.to_lowercase())); + enriched_folders.sort_by(|a, b| b.name.to_lowercase().cmp(&a.name.to_lowercase())); } "date" => { enriched_folders.sort_by(|a, b| a.modified_at.cmp(&b.modified_at)); @@ -437,19 +431,15 @@ impl SearchUseCase for SearchService { let start_idx = criteria.offset.min(total_count); let end_idx = (criteria.offset + criteria.limit).min(total_count); - let mut paginated_folders = Vec::new(); - let mut paginated_files = Vec::new(); + let folder_start = start_idx.min(folder_count); + let folder_end = end_idx.min(folder_count); + let paginated_folders = enriched_folders[folder_start..folder_end].to_vec(); - for i in start_idx..end_idx { - if i < folder_count { - paginated_folders.push(enriched_folders[i].clone()); - } else { - let file_idx = i - folder_count; - if file_idx < enriched_files.len() { - paginated_files.push(enriched_files[file_idx].clone()); - } - } - } + let file_start = start_idx.saturating_sub(folder_count); + let file_end = end_idx + .saturating_sub(folder_count) + .min(enriched_files.len()); + let paginated_files = enriched_files[file_start..file_end].to_vec(); let elapsed_ms = start.elapsed().as_millis() as u64; @@ -496,7 +486,11 @@ impl SearchService { #[async_trait] impl SearchUseCase for SearchServiceStub { - async fn search(&self, _criteria: SearchCriteriaDto, _user_id: &str) -> Result { + async fn search( + &self, + _criteria: SearchCriteriaDto, + _user_id: &str, + ) -> Result { Ok(SearchResultsDto::empty()) } diff --git a/src/application/services/share_service.rs b/src/application/services/share_service.rs index 0e8315b8..7ecb27a4 100644 --- a/src/application/services/share_service.rs +++ b/src/application/services/share_service.rs @@ -132,8 +132,9 @@ impl ShareService { /// concurrently. This keeps RAM usage predictable (~19 MB × 2 = ~38 MB max) /// and avoids starving the Tokio blocking thread pool. async fn hash_password_async(&self, password: &str) -> Result { - let _permit = self.hash_semaphore.acquire().await - .map_err(|_| DomainError::internal_error("ShareService", "Hash semaphore closed".to_string()))?; + let _permit = self.hash_semaphore.acquire().await.map_err(|_| { + DomainError::internal_error("ShareService", "Hash semaphore closed".to_string()) + })?; self.password_hasher.hash_password(password).await } } diff --git a/src/common/config.rs b/src/common/config.rs index 525162e8..70d7c47a 100644 --- a/src/common/config.rs +++ b/src/common/config.rs @@ -209,9 +209,9 @@ impl Default for StorageConfig { fn default() -> Self { Self { root_dir: "storage".to_string(), - chunk_size: 1024 * 1024, // 1 MB - parallel_threshold: 100 * 1024 * 1024, // 100 MB - trash_retention_days: 30, // 30 days + chunk_size: 1024 * 1024, // 1 MB + parallel_threshold: 100 * 1024 * 1024, // 100 MB + trash_retention_days: 30, // 30 days max_upload_size: 10 * 1024 * 1024 * 1024, // 10 GB } } @@ -615,8 +615,7 @@ impl AppConfig { } // Storage limits - if let Ok(max_upload) = env::var("OXICLOUD_MAX_UPLOAD_SIZE") - .map(|v| v.parse::()) + if let Ok(max_upload) = env::var("OXICLOUD_MAX_UPLOAD_SIZE").map(|v| v.parse::()) && let Ok(val) = max_upload { config.storage.max_upload_size = val; diff --git a/src/common/di.rs b/src/common/di.rs index 25963643..a86442e4 100644 --- a/src/common/di.rs +++ b/src/common/di.rs @@ -36,10 +36,10 @@ use crate::application::services::{ use crate::common::config::AppConfig; use crate::common::errors::DomainError; use crate::domain::services::i18n_service::I18nService; +use crate::infrastructure::repositories::pg::SharePgRepository; use crate::infrastructure::repositories::pg::{ FileBlobReadRepository, FileBlobWriteRepository, FolderDbRepository, TrashDbRepository, }; -use crate::infrastructure::repositories::pg::SharePgRepository; use crate::infrastructure::services::file_content_cache::{ FileContentCache, FileContentCacheConfig, }; @@ -334,11 +334,13 @@ impl AppServiceFactory { // Build a password hasher for share password verification let password_hasher: Arc = - Arc::new(crate::infrastructure::services::password_hasher::Argon2PasswordHasher::new( - self.config.auth.hash_memory_cost, - self.config.auth.hash_time_cost, - self.config.auth.hash_parallelism, - )); + Arc::new( + crate::infrastructure::services::password_hasher::Argon2PasswordHasher::new( + self.config.auth.hash_memory_cost, + self.config.auth.hash_time_cost, + self.config.auth.hash_parallelism, + ), + ); let service = Arc::new(ShareService::new( Arc::new(self.config.clone()), @@ -470,7 +472,8 @@ impl AppServiceFactory { recent_service = Some(recent.clone()); apps.recent_service = Some(recent); - storage_usage_service = Some(self.create_storage_usage_service(&repos, &pool, &maintenance_pool)); + storage_usage_service = + Some(self.create_storage_usage_service(&repos, &pool, &maintenance_pool)); // Auth services if self.config.features.enable_auth { diff --git a/src/common/stubs.rs b/src/common/stubs.rs index 3099a1ca..38852377 100644 --- a/src/common/stubs.rs +++ b/src/common/stubs.rs @@ -621,7 +621,11 @@ pub struct StubSearchUseCase; #[async_trait] impl SearchUseCase for StubSearchUseCase { - async fn search(&self, _criteria: SearchCriteriaDto, _user_id: &str) -> Result { + async fn search( + &self, + _criteria: SearchCriteriaDto, + _user_id: &str, + ) -> Result { Ok(SearchResultsDto::empty()) } @@ -794,7 +798,14 @@ impl ContentCachePort for StubContentCachePort { None } - async fn put(&self, _file_id: String, _content: Bytes, _etag: Arc, _content_type: Arc) {} + async fn put( + &self, + _file_id: String, + _content: Bytes, + _etag: Arc, + _content_type: Arc, + ) { + } async fn invalidate(&self, _file_id: &str) {} diff --git a/src/domain/repositories/folder_repository.rs b/src/domain/repositories/folder_repository.rs index 5ac2c0d4..58bd2f7b 100644 --- a/src/domain/repositories/folder_repository.rs +++ b/src/domain/repositories/folder_repository.rs @@ -110,10 +110,7 @@ pub trait FolderRepository: Send + Sync + 'static { /// ordered by `path` so callers can iterate in directory order. /// /// Default: falls back to `list_folders` (one level only). - async fn list_subtree_folders( - &self, - folder_id: &str, - ) -> Result, DomainError> { + async fn list_subtree_folders(&self, folder_id: &str) -> Result, DomainError> { let _ = folder_id; Ok(Vec::new()) } @@ -154,17 +151,20 @@ pub trait FolderRepository: Send + Sync + 'static { recursive: bool, ) -> Result, DomainError> { // Recursive with folder_id → use optimised ltree scan - if recursive { - if let Some(fid) = parent_id { - return self.list_descendant_folders(fid, name_contains, user_id).await; - } + if recursive && let Some(fid) = parent_id { + return self + .list_descendant_folders(fid, name_contains, user_id) + .await; } // Fallback: load + filter in memory (stubs / mocks) let all = self.list_folders(parent_id).await?; match name_contains { Some(q) if !q.is_empty() => { let q = q.to_lowercase(); - Ok(all.into_iter().filter(|f| f.name().to_lowercase().contains(&q)).collect()) + Ok(all + .into_iter() + .filter(|f| f.name().to_lowercase().contains(&q)) + .collect()) } _ => Ok(all), } diff --git a/src/infrastructure/db.rs b/src/infrastructure/db.rs index cb0ad077..f153caae 100644 --- a/src/infrastructure/db.rs +++ b/src/infrastructure/db.rs @@ -110,24 +110,22 @@ async fn create_pool_with_retries( .connect(connection_string) .await { - Ok(pool) => { - match sqlx::query("SELECT 1").execute(&pool).await { - Ok(_) => { - tracing::info!("PostgreSQL {} pool established successfully", label); - return Ok(pool); - } - Err(e) => { - tracing::error!("Error verifying {} pool connection: {}", label, e); - if attempt >= MAX_ATTEMPTS { - return Err(anyhow::anyhow!( - "Error verifying PostgreSQL {} pool connection: {}", - label, - e - )); - } + Ok(pool) => match sqlx::query("SELECT 1").execute(&pool).await { + Ok(_) => { + tracing::info!("PostgreSQL {} pool established successfully", label); + return Ok(pool); + } + Err(e) => { + tracing::error!("Error verifying {} pool connection: {}", label, e); + if attempt >= MAX_ATTEMPTS { + return Err(anyhow::anyhow!( + "Error verifying PostgreSQL {} pool connection: {}", + label, + e + )); } } - } + }, Err(e) => { tracing::error!( "Error connecting to PostgreSQL {} pool (attempt {}/{}): {}", diff --git a/src/infrastructure/repositories/pg/favorites_pg_repository.rs b/src/infrastructure/repositories/pg/favorites_pg_repository.rs index 4b88bf7f..f844e596 100644 --- a/src/infrastructure/repositories/pg/favorites_pg_repository.rs +++ b/src/infrastructure/repositories/pg/favorites_pg_repository.rs @@ -222,7 +222,7 @@ impl FavoritesRepositoryPort for FavoritesPgRepository { let mut q = sqlx::query(&query); for (item_id, item_type) in chunk { - q = q.bind(&user_uuid).bind(item_id).bind(item_type); + q = q.bind(user_uuid).bind(item_id).bind(item_type); } let result = q.execute(&mut *tx).await.map_err(|e| { diff --git a/src/infrastructure/repositories/pg/file_blob_read_repository.rs b/src/infrastructure/repositories/pg/file_blob_read_repository.rs index a14bbd60..de61caeb 100644 --- a/src/infrastructure/repositories/pg/file_blob_read_repository.rs +++ b/src/infrastructure/repositories/pg/file_blob_read_repository.rs @@ -441,10 +441,7 @@ impl FileReadPort for FileBlobReadRepository { /// /// Single GiST-indexed query via ltree `<@`. /// Ordered by `(fo.path, fi.name)` so callers iterate in directory order. - async fn list_files_in_subtree( - &self, - folder_id: &str, - ) -> Result, DomainError> { + async fn list_files_in_subtree(&self, folder_id: &str) -> Result, DomainError> { let rows: Vec<( String, String, @@ -472,9 +469,7 @@ impl FileReadPort for FileBlobReadRepository { .bind(folder_id) .fetch_all(self.pool.as_ref()) .await - .map_err(|e| { - DomainError::internal_error("FileBlobRead", format!("subtree files: {e}")) - })?; + .map_err(|e| DomainError::internal_error("FileBlobRead", format!("subtree files: {e}")))?; rows.into_iter() .map(|(id, name, fid, fpath, size, mime, ca, ma, uid)| { @@ -520,11 +515,11 @@ impl FileReadPort for FileBlobReadRepository { conditions.push(format!("fi.folder_id = ${bind_idx}::uuid")); } - if let Some(name) = &criteria.name_contains { - if !name.is_empty() { - bind_idx += 1; - conditions.push(format!("LOWER(fi.name) LIKE ${bind_idx}")); - } + if let Some(name) = &criteria.name_contains + && !name.is_empty() + { + bind_idx += 1; + conditions.push(format!("LOWER(fi.name) LIKE ${bind_idx}")); } let where_clause = conditions.join(" AND "); @@ -546,19 +541,30 @@ impl FileReadPort for FileBlobReadRepository { ); // ── Bind parameters dynamically ────────────────────────────────── - let mut query = sqlx::query_as::<_, ( - String, String, Option, Option, - i64, String, i64, i64, Option, i64, - )>(&sql) - .bind(user_id); + let mut query = sqlx::query_as::< + _, + ( + String, + String, + Option, + Option, + i64, + String, + i64, + i64, + Option, + i64, + ), + >(&sql) + .bind(user_id); if let Some(fid) = folder_id { query = query.bind(fid); } - if let Some(name) = &criteria.name_contains { - if !name.is_empty() { - query = query.bind(format!("%{}%", name.to_lowercase())); - } + if let Some(name) = &criteria.name_contains + && !name.is_empty() + { + query = query.bind(format!("%{}%", name.to_lowercase())); } query = query.bind(limit).bind(offset); @@ -600,9 +606,7 @@ impl FileReadPort for FileBlobReadRepository { // When no root folder specified, delegate to existing paginated search let root_id = match root_folder_id { None => { - return self - .search_files_paginated(None, criteria, user_id) - .await; + return self.search_files_paginated(None, criteria, user_id).await; } Some(id) => id, }; @@ -631,19 +635,19 @@ impl FileReadPort for FileBlobReadRepository { "fo.lpath <@ (SELECT lpath FROM storage.folders WHERE id = $2::uuid)".to_string(), ); - if let Some(name) = &criteria.name_contains { - if !name.is_empty() { - bind_idx += 1; - conditions.push(format!("LOWER(fi.name) LIKE ${bind_idx}")); - } + if let Some(name) = &criteria.name_contains + && !name.is_empty() + { + bind_idx += 1; + conditions.push(format!("LOWER(fi.name) LIKE ${bind_idx}")); } - if let Some(types) = &criteria.file_types { - if !types.is_empty() { - bind_idx += 1; - conditions.push(format!( - "LOWER(SUBSTRING(fi.name FROM '\\.([^.]+)$')) = ANY(${bind_idx})" - )); - } + if let Some(types) = &criteria.file_types + && !types.is_empty() + { + bind_idx += 1; + conditions.push(format!( + "LOWER(SUBSTRING(fi.name FROM '\\.([^.]+)$')) = ANY(${bind_idx})" + )); } if criteria.created_after.is_some() { bind_idx += 1; @@ -698,24 +702,34 @@ impl FileReadPort for FileBlobReadRepository { ); // ── Bind parameters dynamically ── - let mut query = sqlx::query_as::<_, ( - String, String, Option, Option, - i64, String, i64, i64, Option, i64, - )>(&sql) - .bind(user_id) - .bind(root_id); + let mut query = sqlx::query_as::< + _, + ( + String, + String, + Option, + Option, + i64, + String, + i64, + i64, + Option, + i64, + ), + >(&sql) + .bind(user_id) + .bind(root_id); - if let Some(name) = &criteria.name_contains { - if !name.is_empty() { - query = query.bind(format!("%{}%", name.to_lowercase())); - } + if let Some(name) = &criteria.name_contains + && !name.is_empty() + { + query = query.bind(format!("%{}%", name.to_lowercase())); } - if let Some(types) = &criteria.file_types { - if !types.is_empty() { - let lower_types: Vec = - types.iter().map(|t| t.to_lowercase()).collect(); - query = query.bind(lower_types); - } + if let Some(types) = &criteria.file_types + && !types.is_empty() + { + let lower_types: Vec = types.iter().map(|t| t.to_lowercase()).collect(); + query = query.bind(lower_types); } if let Some(v) = criteria.created_after { query = query.bind(v as i64); @@ -739,12 +753,9 @@ impl FileReadPort for FileBlobReadRepository { query = query.bind(limit).bind(offset); // ── Execute single query ── - let rows = query - .fetch_all(self.pool.as_ref()) - .await - .map_err(|e| { - DomainError::internal_error("FileBlobRead", format!("subtree search: {e}")) - })?; + let rows = query.fetch_all(self.pool.as_ref()).await.map_err(|e| { + DomainError::internal_error("FileBlobRead", format!("subtree search: {e}")) + })?; let total_count = rows.first().map_or(0, |r| r.9) as usize; @@ -973,12 +984,9 @@ mod tests { .build(), }; - repo.hash_cache - .insert("a".to_string(), "ha".to_string()); - repo.hash_cache - .insert("b".to_string(), "hb".to_string()); - repo.hash_cache - .insert("c".to_string(), "hc".to_string()); + repo.hash_cache.insert("a".to_string(), "ha".to_string()); + repo.hash_cache.insert("b".to_string(), "hb".to_string()); + repo.hash_cache.insert("c".to_string(), "hc".to_string()); // Force moka to run pending eviction tasks repo.hash_cache.run_pending_tasks(); @@ -1016,7 +1024,8 @@ mod tests { } for h in handles { - h.join().expect("Thread must not panic — no poison possible with moka"); + h.join() + .expect("Thread must not panic — no poison possible with moka"); } } } diff --git a/src/infrastructure/repositories/pg/file_blob_write_repository.rs b/src/infrastructure/repositories/pg/file_blob_write_repository.rs index 0ca0535b..3f3de58b 100644 --- a/src/infrastructure/repositories/pg/file_blob_write_repository.rs +++ b/src/infrastructure/repositories/pg/file_blob_write_repository.rs @@ -166,14 +166,14 @@ impl FileBlobWriteRepository { }; // Decrement old blob ref (only if hash changed, best-effort) - if old_hash != new_hash { - if let Err(e) = self.dedup.remove_reference(&old_hash).await { - tracing::warn!( - "Failed to decrement old blob ref {}: {}", - &old_hash[..12], - e - ); - } + if old_hash != new_hash + && let Err(e) = self.dedup.remove_reference(&old_hash).await + { + tracing::warn!( + "Failed to decrement old blob ref {}: {}", + &old_hash[..12], + e + ); } Ok(()) @@ -701,10 +701,7 @@ impl FileWritePort for FileBlobWriteRepository { ); } } - DomainError::internal_error( - "FileBlobWrite", - format!("copy_folder_tree: {e}"), - ) + DomainError::internal_error("FileBlobWrite", format!("copy_folder_tree: {e}")) })?; tracing::info!( diff --git a/src/infrastructure/repositories/pg/folder_db_repository.rs b/src/infrastructure/repositories/pg/folder_db_repository.rs index 3f2d25d2..624c5c24 100644 --- a/src/infrastructure/repositories/pg/folder_db_repository.rs +++ b/src/infrastructure/repositories/pg/folder_db_repository.rs @@ -685,10 +685,7 @@ impl FolderRepository for FolderDbRepository { /// /// Single GiST-indexed query: `fo.lpath <@ (root's lpath)`. /// Ordered by `fo.path` so callers can iterate in directory order. - async fn list_subtree_folders( - &self, - folder_id: &str, - ) -> Result, DomainError> { + async fn list_subtree_folders(&self, folder_id: &str) -> Result, DomainError> { let sql = "SELECT fo.id::text, fo.name, fo.path, fo.parent_id::text, \ fo.user_id::text, \ EXTRACT(EPOCH FROM fo.created_at)::bigint, \ @@ -698,14 +695,21 @@ impl FolderRepository for FolderDbRepository { AND fo.lpath <@ (SELECT lpath FROM storage.folders WHERE id = $1::uuid) \ ORDER BY fo.path"; - let rows: Vec<(String, String, String, Option, Option, i64, i64)> = - sqlx::query_as(sql) - .bind(folder_id) - .fetch_all(self.pool()) - .await - .map_err(|e| { - DomainError::internal_error("FolderDb", format!("subtree folders: {e}")) - })?; + let rows: Vec<( + String, + String, + String, + Option, + Option, + i64, + i64, + )> = sqlx::query_as(sql) + .bind(folder_id) + .fetch_all(self.pool()) + .await + .map_err(|e| { + DomainError::internal_error("FolderDb", format!("subtree folders: {e}")) + })?; rows.into_iter() .map(|(id, name, path, pid, uid, ca, ma)| { @@ -728,10 +732,10 @@ impl FolderRepository for FolderDbRepository { recursive: bool, ) -> Result, DomainError> { // Recursive with folder scope → existing optimised ltree scan - if recursive { - if let Some(fid) = parent_id { - return self.list_descendant_folders(fid, name_contains, user_id).await; - } + if recursive && let Some(fid) = parent_id { + return self + .list_descendant_folders(fid, name_contains, user_id) + .await; } // Build optional name filter @@ -761,22 +765,27 @@ impl FolderRepository for FolderDbRepository { ORDER BY fo.name" ); - let rows: Vec<(String, String, String, Option, Option, i64, i64)> = - if let Some(ref pattern) = name_pattern { - sqlx::query_as(&sql) - .bind(user_id) - .bind(pattern) - .fetch_all(self.pool()) - .await - } else { - sqlx::query_as(&sql) - .bind(user_id) - .fetch_all(self.pool()) - .await - } - .map_err(|e| { - DomainError::internal_error("FolderDb", format!("search_folders: {e}")) - })?; + let rows: Vec<( + String, + String, + String, + Option, + Option, + i64, + i64, + )> = if let Some(ref pattern) = name_pattern { + sqlx::query_as(&sql) + .bind(user_id) + .bind(pattern) + .fetch_all(self.pool()) + .await + } else { + sqlx::query_as(&sql) + .bind(user_id) + .fetch_all(self.pool()) + .await + } + .map_err(|e| DomainError::internal_error("FolderDb", format!("search_folders: {e}")))?; return rows .into_iter() @@ -820,37 +829,42 @@ impl FolderRepository for FolderDbRepository { ) }; - let rows: Vec<(String, String, String, Option, Option, i64, i64)> = - if let Some(pid) = parent_id { - if let Some(ref pattern) = name_pattern { - sqlx::query_as(&sql) - .bind(pid) - .bind(user_id) - .bind(pattern) - .fetch_all(self.pool()) - .await - } else { - sqlx::query_as(&sql) - .bind(pid) - .bind(user_id) - .fetch_all(self.pool()) - .await - } - } else if let Some(ref pattern) = name_pattern { + let rows: Vec<( + String, + String, + String, + Option, + Option, + i64, + i64, + )> = if let Some(pid) = parent_id { + if let Some(ref pattern) = name_pattern { sqlx::query_as(&sql) + .bind(pid) .bind(user_id) .bind(pattern) .fetch_all(self.pool()) .await } else { sqlx::query_as(&sql) + .bind(pid) .bind(user_id) .fetch_all(self.pool()) .await } - .map_err(|e| { - DomainError::internal_error("FolderDb", format!("search_folders: {e}")) - })?; + } else if let Some(ref pattern) = name_pattern { + sqlx::query_as(&sql) + .bind(user_id) + .bind(pattern) + .fetch_all(self.pool()) + .await + } else { + sqlx::query_as(&sql) + .bind(user_id) + .fetch_all(self.pool()) + .await + } + .map_err(|e| DomainError::internal_error("FolderDb", format!("search_folders: {e}")))?; rows.into_iter() .map(|(id, name, path, pid, uid, ca, ma)| { @@ -870,9 +884,10 @@ impl FolderRepository for FolderDbRepository { user_id: &str, ) -> Result, DomainError> { let (where_extra, name_pattern) = match name_contains { - Some(name) if !name.is_empty() => { - (" AND LOWER(fo.name) LIKE $3", Some(format!("%{}%", name.to_lowercase()))) - } + Some(name) if !name.is_empty() => ( + " AND LOWER(fo.name) LIKE $3", + Some(format!("%{}%", name.to_lowercase())), + ), _ => ("", None), }; @@ -890,24 +905,29 @@ impl FolderRepository for FolderDbRepository { ORDER BY fo.name" ); - let rows: Vec<(String, String, String, Option, Option, i64, i64)> = - if let Some(ref pattern) = name_pattern { - sqlx::query_as(&sql) - .bind(user_id) - .bind(folder_id) - .bind(pattern) - .fetch_all(self.pool()) - .await - } else { - sqlx::query_as(&sql) - .bind(user_id) - .bind(folder_id) - .fetch_all(self.pool()) - .await - } - .map_err(|e| { - DomainError::internal_error("FolderDb", format!("descendant search: {e}")) - })?; + let rows: Vec<( + String, + String, + String, + Option, + Option, + i64, + i64, + )> = if let Some(ref pattern) = name_pattern { + sqlx::query_as(&sql) + .bind(user_id) + .bind(folder_id) + .bind(pattern) + .fetch_all(self.pool()) + .await + } else { + sqlx::query_as(&sql) + .bind(user_id) + .bind(folder_id) + .fetch_all(self.pool()) + .await + } + .map_err(|e| DomainError::internal_error("FolderDb", format!("descendant search: {e}")))?; rows.into_iter() .map(|(id, name, path, pid, uid, ca, ma)| { diff --git a/src/infrastructure/repositories/pg/share_pg_repository.rs b/src/infrastructure/repositories/pg/share_pg_repository.rs index d5790def..069fea77 100644 --- a/src/infrastructure/repositories/pg/share_pg_repository.rs +++ b/src/infrastructure/repositories/pg/share_pg_repository.rs @@ -1,308 +1,307 @@ -use async_trait::async_trait; -use sqlx::{PgPool, Row}; -use std::sync::Arc; - -use crate::{ - application::ports::share_ports::ShareStoragePort, - common::errors::DomainError, - domain::entities::share::{Share, ShareItemType, SharePermissions}, -}; - -/// PostgreSQL implementation of [`ShareStoragePort`]. -/// -/// Replaces the legacy file-based `ShareFsRepository` that read/wrote the -/// entire `shares.json` on every operation. Each method now issues a single -/// indexed SQL statement — O(1) lookups, ACID transactions, and no data-race -/// risk. -pub struct SharePgRepository { - db_pool: Arc, -} - -impl SharePgRepository { - pub fn new(db_pool: Arc) -> Self { - Self { db_pool } - } - - /// Maps a [`sqlx::postgres::PgRow`] to the domain [`Share`] entity. - fn row_to_entity(row: &sqlx::postgres::PgRow) -> Result { - let id: String = row.try_get("id").map_err(|e| { - DomainError::internal_error("Share", format!("Failed to read id: {e}")) - })?; - let item_id: String = row.try_get("item_id").map_err(|e| { - DomainError::internal_error("Share", format!("Failed to read item_id: {e}")) - })?; - let item_name: Option = row.try_get("item_name").unwrap_or(None); - let item_type_str: String = row.try_get("item_type").map_err(|e| { - DomainError::internal_error("Share", format!("Failed to read item_type: {e}")) - })?; - let token: String = row.try_get("token").map_err(|e| { - DomainError::internal_error("Share", format!("Failed to read token: {e}")) - })?; - let password_hash: Option = row.try_get("password_hash").unwrap_or(None); - let expires_at: Option = row.try_get("expires_at").unwrap_or(None); - let permissions_read: bool = row.try_get("permissions_read").unwrap_or(true); - let permissions_write: bool = row.try_get("permissions_write").unwrap_or(false); - let permissions_reshare: bool = row.try_get("permissions_reshare").unwrap_or(false); - let created_at: i64 = row.try_get("created_at").map_err(|e| { - DomainError::internal_error("Share", format!("Failed to read created_at: {e}")) - })?; - let created_by: String = row.try_get("created_by").map_err(|e| { - DomainError::internal_error("Share", format!("Failed to read created_by: {e}")) - })?; - let access_count: i64 = row.try_get("access_count").unwrap_or(0); - - let item_type = - ShareItemType::try_from(item_type_str.as_str()).unwrap_or(ShareItemType::File); - let permissions = SharePermissions::new(permissions_read, permissions_write, permissions_reshare); - - Ok(Share::from_raw( - id, - item_id, - item_name, - item_type, - token, - password_hash, - expires_at.map(|v| v as u64), - permissions, - created_at as u64, - created_by, - access_count as u64, - )) - } -} - -#[async_trait] -impl ShareStoragePort for SharePgRepository { - async fn save_share(&self, share: &Share) -> Result { - let row = sqlx::query( - r#" - INSERT INTO storage.shares - (id, item_id, item_name, item_type, token, password_hash, - expires_at, permissions_read, permissions_write, permissions_reshare, - created_at, created_by, access_count) - VALUES - ($1::UUID, $2, $3, $4, $5, $6, - $7, $8, $9, $10, - $11, $12, $13) - ON CONFLICT (id) DO UPDATE SET - item_name = EXCLUDED.item_name, - password_hash = EXCLUDED.password_hash, - expires_at = EXCLUDED.expires_at, - permissions_read = EXCLUDED.permissions_read, - permissions_write = EXCLUDED.permissions_write, - permissions_reshare = EXCLUDED.permissions_reshare, - access_count = EXCLUDED.access_count - RETURNING - id::TEXT, item_id, item_name, item_type, token, password_hash, - expires_at, permissions_read, permissions_write, permissions_reshare, - created_at, created_by, access_count - "#, - ) - .bind(share.id()) - .bind(share.item_id()) - .bind(share.item_name()) - .bind(share.item_type().to_string()) - .bind(share.token()) - .bind(share.password_hash()) - .bind(share.expires_at().map(|v| v as i64)) - .bind(share.permissions().read()) - .bind(share.permissions().write()) - .bind(share.permissions().reshare()) - .bind(share.created_at() as i64) - .bind(share.created_by()) - .bind(share.access_count() as i64) - .fetch_one(&*self.db_pool) - .await - .map_err(|e| { - tracing::error!("Database error saving share: {}", e); - DomainError::internal_error("Share", format!("Failed to save share: {e}")) - })?; - - Self::row_to_entity(&row) - } - - async fn find_share_by_id(&self, id: &str) -> Result { - let row = sqlx::query( - r#" - SELECT id::TEXT, item_id, item_name, item_type, token, password_hash, - expires_at, permissions_read, permissions_write, permissions_reshare, - created_at, created_by, access_count - FROM storage.shares - WHERE id = $1::UUID - "#, - ) - .bind(id) - .fetch_optional(&*self.db_pool) - .await - .map_err(|e| { - tracing::error!("Database error finding share by id: {}", e); - DomainError::internal_error("Share", format!("Failed to find share: {e}")) - })?; - - match row { - Some(r) => Self::row_to_entity(&r), - None => Err(DomainError::not_found( - "Share", - format!("Share with ID {id} not found"), - )), - } - } - - async fn find_share_by_token(&self, token: &str) -> Result { - let row = sqlx::query( - r#" - SELECT id::TEXT, item_id, item_name, item_type, token, password_hash, - expires_at, permissions_read, permissions_write, permissions_reshare, - created_at, created_by, access_count - FROM storage.shares - WHERE token = $1 - "#, - ) - .bind(token) - .fetch_optional(&*self.db_pool) - .await - .map_err(|e| { - tracing::error!("Database error finding share by token: {}", e); - DomainError::internal_error("Share", format!("Failed to find share by token: {e}")) - })?; - - match row { - Some(r) => Self::row_to_entity(&r), - None => Err(DomainError::not_found( - "Share", - format!("Share with token {token} not found"), - )), - } - } - - async fn find_shares_by_item( - &self, - item_id: &str, - item_type: &ShareItemType, - ) -> Result, DomainError> { - let rows = sqlx::query( - r#" - SELECT id::TEXT, item_id, item_name, item_type, token, password_hash, - expires_at, permissions_read, permissions_write, permissions_reshare, - created_at, created_by, access_count - FROM storage.shares - WHERE item_id = $1 AND item_type = $2 - ORDER BY created_at DESC - "#, - ) - .bind(item_id) - .bind(item_type.to_string()) - .fetch_all(&*self.db_pool) - .await - .map_err(|e| { - tracing::error!("Database error finding shares by item: {}", e); - DomainError::internal_error("Share", format!("Failed to find shares by item: {e}")) - })?; - - rows.iter().map(Self::row_to_entity).collect() - } - - async fn update_share(&self, share: &Share) -> Result { - let row = sqlx::query( - r#" - UPDATE storage.shares SET - item_name = $2, - password_hash = $3, - expires_at = $4, - permissions_read = $5, - permissions_write = $6, - permissions_reshare = $7, - access_count = $8 - WHERE id = $1::UUID - RETURNING - id::TEXT, item_id, item_name, item_type, token, password_hash, - expires_at, permissions_read, permissions_write, permissions_reshare, - created_at, created_by, access_count - "#, - ) - .bind(share.id()) - .bind(share.item_name()) - .bind(share.password_hash()) - .bind(share.expires_at().map(|v| v as i64)) - .bind(share.permissions().read()) - .bind(share.permissions().write()) - .bind(share.permissions().reshare()) - .bind(share.access_count() as i64) - .fetch_optional(&*self.db_pool) - .await - .map_err(|e| { - tracing::error!("Database error updating share: {}", e); - DomainError::internal_error("Share", format!("Failed to update share: {e}")) - })?; - - match row { - Some(r) => Self::row_to_entity(&r), - None => Err(DomainError::not_found( - "Share", - format!("Share with ID {} not found for update", share.id()), - )), - } - } - - async fn delete_share(&self, id: &str) -> Result<(), DomainError> { - let result = sqlx::query( - "DELETE FROM storage.shares WHERE id = $1::UUID", - ) - .bind(id) - .execute(&*self.db_pool) - .await - .map_err(|e| { - tracing::error!("Database error deleting share: {}", e); - DomainError::internal_error("Share", format!("Failed to delete share: {e}")) - })?; - - if result.rows_affected() == 0 { - return Err(DomainError::not_found( - "Share", - format!("Share with ID {id} not found for deletion"), - )); - } - - Ok(()) - } - - async fn find_shares_by_user( - &self, - user_id: &str, - offset: usize, - limit: usize, - ) -> Result<(Vec, usize), DomainError> { - // Single query with window function — count + rows in one roundtrip - let rows = sqlx::query( - r#" - SELECT id::TEXT, item_id, item_name, item_type, token, password_hash, - expires_at, permissions_read, permissions_write, permissions_reshare, - created_at, created_by, access_count, - COUNT(*) OVER() AS total_count - FROM storage.shares - WHERE created_by = $1 - ORDER BY created_at DESC - LIMIT $2 OFFSET $3 - "#, - ) - .bind(user_id) - .bind(limit as i64) - .bind(offset as i64) - .fetch_all(&*self.db_pool) - .await - .map_err(|e| { - tracing::error!("Database error finding shares by user: {}", e); - DomainError::internal_error("Share", format!("Failed to find shares by user: {e}")) - })?; - - let total: usize = rows - .first() - .and_then(|r| r.try_get::("total_count").ok()) - .unwrap_or(0) as usize; - - let shares: Result, DomainError> = - rows.iter().map(Self::row_to_entity).collect(); - - Ok((shares?, total)) - } -} +use async_trait::async_trait; +use sqlx::{PgPool, Row}; +use std::sync::Arc; + +use crate::{ + application::ports::share_ports::ShareStoragePort, + common::errors::DomainError, + domain::entities::share::{Share, ShareItemType, SharePermissions}, +}; + +/// PostgreSQL implementation of [`ShareStoragePort`]. +/// +/// Replaces the legacy file-based `ShareFsRepository` that read/wrote the +/// entire `shares.json` on every operation. Each method now issues a single +/// indexed SQL statement — O(1) lookups, ACID transactions, and no data-race +/// risk. +pub struct SharePgRepository { + db_pool: Arc, +} + +impl SharePgRepository { + pub fn new(db_pool: Arc) -> Self { + Self { db_pool } + } + + /// Maps a [`sqlx::postgres::PgRow`] to the domain [`Share`] entity. + fn row_to_entity(row: &sqlx::postgres::PgRow) -> Result { + let id: String = row + .try_get("id") + .map_err(|e| DomainError::internal_error("Share", format!("Failed to read id: {e}")))?; + let item_id: String = row.try_get("item_id").map_err(|e| { + DomainError::internal_error("Share", format!("Failed to read item_id: {e}")) + })?; + let item_name: Option = row.try_get("item_name").unwrap_or(None); + let item_type_str: String = row.try_get("item_type").map_err(|e| { + DomainError::internal_error("Share", format!("Failed to read item_type: {e}")) + })?; + let token: String = row.try_get("token").map_err(|e| { + DomainError::internal_error("Share", format!("Failed to read token: {e}")) + })?; + let password_hash: Option = row.try_get("password_hash").unwrap_or(None); + let expires_at: Option = row.try_get("expires_at").unwrap_or(None); + let permissions_read: bool = row.try_get("permissions_read").unwrap_or(true); + let permissions_write: bool = row.try_get("permissions_write").unwrap_or(false); + let permissions_reshare: bool = row.try_get("permissions_reshare").unwrap_or(false); + let created_at: i64 = row.try_get("created_at").map_err(|e| { + DomainError::internal_error("Share", format!("Failed to read created_at: {e}")) + })?; + let created_by: String = row.try_get("created_by").map_err(|e| { + DomainError::internal_error("Share", format!("Failed to read created_by: {e}")) + })?; + let access_count: i64 = row.try_get("access_count").unwrap_or(0); + + let item_type = + ShareItemType::try_from(item_type_str.as_str()).unwrap_or(ShareItemType::File); + let permissions = + SharePermissions::new(permissions_read, permissions_write, permissions_reshare); + + Ok(Share::from_raw( + id, + item_id, + item_name, + item_type, + token, + password_hash, + expires_at.map(|v| v as u64), + permissions, + created_at as u64, + created_by, + access_count as u64, + )) + } +} + +#[async_trait] +impl ShareStoragePort for SharePgRepository { + async fn save_share(&self, share: &Share) -> Result { + let row = sqlx::query( + r#" + INSERT INTO storage.shares + (id, item_id, item_name, item_type, token, password_hash, + expires_at, permissions_read, permissions_write, permissions_reshare, + created_at, created_by, access_count) + VALUES + ($1::UUID, $2, $3, $4, $5, $6, + $7, $8, $9, $10, + $11, $12, $13) + ON CONFLICT (id) DO UPDATE SET + item_name = EXCLUDED.item_name, + password_hash = EXCLUDED.password_hash, + expires_at = EXCLUDED.expires_at, + permissions_read = EXCLUDED.permissions_read, + permissions_write = EXCLUDED.permissions_write, + permissions_reshare = EXCLUDED.permissions_reshare, + access_count = EXCLUDED.access_count + RETURNING + id::TEXT, item_id, item_name, item_type, token, password_hash, + expires_at, permissions_read, permissions_write, permissions_reshare, + created_at, created_by, access_count + "#, + ) + .bind(share.id()) + .bind(share.item_id()) + .bind(share.item_name()) + .bind(share.item_type().to_string()) + .bind(share.token()) + .bind(share.password_hash()) + .bind(share.expires_at().map(|v| v as i64)) + .bind(share.permissions().read()) + .bind(share.permissions().write()) + .bind(share.permissions().reshare()) + .bind(share.created_at() as i64) + .bind(share.created_by()) + .bind(share.access_count() as i64) + .fetch_one(&*self.db_pool) + .await + .map_err(|e| { + tracing::error!("Database error saving share: {}", e); + DomainError::internal_error("Share", format!("Failed to save share: {e}")) + })?; + + Self::row_to_entity(&row) + } + + async fn find_share_by_id(&self, id: &str) -> Result { + let row = sqlx::query( + r#" + SELECT id::TEXT, item_id, item_name, item_type, token, password_hash, + expires_at, permissions_read, permissions_write, permissions_reshare, + created_at, created_by, access_count + FROM storage.shares + WHERE id = $1::UUID + "#, + ) + .bind(id) + .fetch_optional(&*self.db_pool) + .await + .map_err(|e| { + tracing::error!("Database error finding share by id: {}", e); + DomainError::internal_error("Share", format!("Failed to find share: {e}")) + })?; + + match row { + Some(r) => Self::row_to_entity(&r), + None => Err(DomainError::not_found( + "Share", + format!("Share with ID {id} not found"), + )), + } + } + + async fn find_share_by_token(&self, token: &str) -> Result { + let row = sqlx::query( + r#" + SELECT id::TEXT, item_id, item_name, item_type, token, password_hash, + expires_at, permissions_read, permissions_write, permissions_reshare, + created_at, created_by, access_count + FROM storage.shares + WHERE token = $1 + "#, + ) + .bind(token) + .fetch_optional(&*self.db_pool) + .await + .map_err(|e| { + tracing::error!("Database error finding share by token: {}", e); + DomainError::internal_error("Share", format!("Failed to find share by token: {e}")) + })?; + + match row { + Some(r) => Self::row_to_entity(&r), + None => Err(DomainError::not_found( + "Share", + format!("Share with token {token} not found"), + )), + } + } + + async fn find_shares_by_item( + &self, + item_id: &str, + item_type: &ShareItemType, + ) -> Result, DomainError> { + let rows = sqlx::query( + r#" + SELECT id::TEXT, item_id, item_name, item_type, token, password_hash, + expires_at, permissions_read, permissions_write, permissions_reshare, + created_at, created_by, access_count + FROM storage.shares + WHERE item_id = $1 AND item_type = $2 + ORDER BY created_at DESC + "#, + ) + .bind(item_id) + .bind(item_type.to_string()) + .fetch_all(&*self.db_pool) + .await + .map_err(|e| { + tracing::error!("Database error finding shares by item: {}", e); + DomainError::internal_error("Share", format!("Failed to find shares by item: {e}")) + })?; + + rows.iter().map(Self::row_to_entity).collect() + } + + async fn update_share(&self, share: &Share) -> Result { + let row = sqlx::query( + r#" + UPDATE storage.shares SET + item_name = $2, + password_hash = $3, + expires_at = $4, + permissions_read = $5, + permissions_write = $6, + permissions_reshare = $7, + access_count = $8 + WHERE id = $1::UUID + RETURNING + id::TEXT, item_id, item_name, item_type, token, password_hash, + expires_at, permissions_read, permissions_write, permissions_reshare, + created_at, created_by, access_count + "#, + ) + .bind(share.id()) + .bind(share.item_name()) + .bind(share.password_hash()) + .bind(share.expires_at().map(|v| v as i64)) + .bind(share.permissions().read()) + .bind(share.permissions().write()) + .bind(share.permissions().reshare()) + .bind(share.access_count() as i64) + .fetch_optional(&*self.db_pool) + .await + .map_err(|e| { + tracing::error!("Database error updating share: {}", e); + DomainError::internal_error("Share", format!("Failed to update share: {e}")) + })?; + + match row { + Some(r) => Self::row_to_entity(&r), + None => Err(DomainError::not_found( + "Share", + format!("Share with ID {} not found for update", share.id()), + )), + } + } + + async fn delete_share(&self, id: &str) -> Result<(), DomainError> { + let result = sqlx::query("DELETE FROM storage.shares WHERE id = $1::UUID") + .bind(id) + .execute(&*self.db_pool) + .await + .map_err(|e| { + tracing::error!("Database error deleting share: {}", e); + DomainError::internal_error("Share", format!("Failed to delete share: {e}")) + })?; + + if result.rows_affected() == 0 { + return Err(DomainError::not_found( + "Share", + format!("Share with ID {id} not found for deletion"), + )); + } + + Ok(()) + } + + async fn find_shares_by_user( + &self, + user_id: &str, + offset: usize, + limit: usize, + ) -> Result<(Vec, usize), DomainError> { + // Single query with window function — count + rows in one roundtrip + let rows = sqlx::query( + r#" + SELECT id::TEXT, item_id, item_name, item_type, token, password_hash, + expires_at, permissions_read, permissions_write, permissions_reshare, + created_at, created_by, access_count, + COUNT(*) OVER() AS total_count + FROM storage.shares + WHERE created_by = $1 + ORDER BY created_at DESC + LIMIT $2 OFFSET $3 + "#, + ) + .bind(user_id) + .bind(limit as i64) + .bind(offset as i64) + .fetch_all(&*self.db_pool) + .await + .map_err(|e| { + tracing::error!("Database error finding shares by user: {}", e); + DomainError::internal_error("Share", format!("Failed to find shares by user: {e}")) + })?; + + let total: usize = rows + .first() + .and_then(|r| r.try_get::("total_count").ok()) + .unwrap_or(0) as usize; + + let shares: Result, DomainError> = + rows.iter().map(Self::row_to_entity).collect(); + + Ok((shares?, total)) + } +} diff --git a/src/infrastructure/services/chunked_upload_service.rs b/src/infrastructure/services/chunked_upload_service.rs index 18ec96fb..cc8c5e57 100644 --- a/src/infrastructure/services/chunked_upload_service.rs +++ b/src/infrastructure/services/chunked_upload_service.rs @@ -130,7 +130,7 @@ impl UploadSession { /// Build the completed-chunks bitmask (1 bit per chunk). fn build_progress_bitmask(&self) -> Vec { - let len = (self.chunks.len() + 7) / 8; + let len = self.chunks.len().div_ceil(8); let mut bitmask = vec![0u8; len]; for chunk in &self.chunks { if chunk.status == ChunkStatus::Complete { @@ -157,8 +157,8 @@ impl UploadSession { /// Persist the full session metadata once (on create). async fn persist_metadata(&self) -> Result<(), String> { let path = self.temp_dir.join(SESSION_META_FILE); - let json = serde_json::to_vec(self) - .map_err(|e| format!("Failed to serialise session: {e}"))?; + let json = + serde_json::to_vec(self).map_err(|e| format!("Failed to serialise session: {e}"))?; // Atomic write: write to .tmp then rename let tmp = self.temp_dir.join("session.json.tmp"); fs::write(&tmp, &json) @@ -208,9 +208,7 @@ impl ChunkedUploadService { }; if recovered_count > 0 { - tracing::info!( - "♻️ Recovered {recovered_count} chunked-upload session(s) from disk" - ); + tracing::info!("♻️ Recovered {recovered_count} chunked-upload session(s) from disk"); } // Start cleanup task @@ -357,14 +355,13 @@ impl ChunkedUploadService { let dir_name = path.file_name().and_then(|n| n.to_str()).unwrap_or(""); let sessions = sessions.read().await; - if !sessions.contains_key(dir_name) { - if let Ok(metadata) = fs::metadata(&path).await - && let Ok(modified) = metadata.modified() - && modified.elapsed().unwrap_or_default() > SESSION_EXPIRATION - { - let _ = fs::remove_dir_all(&path).await; - tracing::info!("🧹 Cleaned orphaned upload dir: {:?}", path); - } + if !sessions.contains_key(dir_name) + && let Ok(metadata) = fs::metadata(&path).await + && let Ok(modified) = metadata.modified() + && modified.elapsed().unwrap_or_default() > SESSION_EXPIRATION + { + let _ = fs::remove_dir_all(&path).await; + tracing::info!("🧹 Cleaned orphaned upload dir: {:?}", path); } } } @@ -504,11 +501,10 @@ impl ChunkedUploadService { // worker free for other connections. if let Some(ref expected_checksum) = checksum { let data_clone = data.clone(); // Bytes::clone is O(1) — just an Arc increment - let actual_checksum = tokio::task::spawn_blocking(move || { - format!("{:x}", md5::compute(&data_clone)) - }) - .await - .map_err(|e| format!("MD5 checksum task failed: {e}"))?; + let actual_checksum = + tokio::task::spawn_blocking(move || format!("{:x}", md5::compute(&data_clone))) + .await + .map_err(|e| format!("MD5 checksum task failed: {e}"))?; if actual_checksum != *expected_checksum { return Err(format!( @@ -576,10 +572,7 @@ impl ChunkedUploadService { } /// Get upload status - async fn get_status_inner( - &self, - upload_id: &str, - ) -> Result { + async fn get_status_inner(&self, upload_id: &str) -> Result { let sessions = self.sessions.read().await; let session = sessions .get(upload_id) @@ -645,12 +638,17 @@ impl ChunkedUploadService { let chunks_meta: Vec<(usize, PathBuf)> = session .chunks .iter() - .map(|c| (c.index, session.temp_dir.join(format!("chunk_{:06}", c.index)))) + .map(|c| { + ( + c.index, + session.temp_dir.join(format!("chunk_{:06}", c.index)), + ) + }) .collect(); let total_size = session.total_size; let hash = tokio::task::spawn_blocking(move || -> Result { - use std::io::{Read, Write, BufWriter as StdBufWriter}; + use std::io::{BufWriter as StdBufWriter, Read, Write}; let raw_output = std::fs::OpenOptions::new() .create(true) @@ -721,10 +719,10 @@ impl ChunkedUploadService { /// Finalize upload: remove session from RAM and clean up temp directory async fn finalize_upload_inner(&self, upload_id: &str) -> Result<(), String> { let mut sessions = self.sessions.write().await; - if let Some(session) = sessions.remove(upload_id) { - if let Err(e) = fs::remove_dir_all(&session.temp_dir).await { - tracing::warn!("Failed to cleanup upload {}: {}", upload_id, e); - } + if let Some(session) = sessions.remove(upload_id) + && let Err(e) = fs::remove_dir_all(&session.temp_dir).await + { + tracing::warn!("Failed to cleanup upload {}: {}", upload_id, e); } Ok(()) } @@ -919,8 +917,7 @@ mod tests { }; let json = serde_json::to_vec(&session).expect("serialise"); - let restored: UploadSession = - serde_json::from_slice(&json).expect("deserialise"); + let restored: UploadSession = serde_json::from_slice(&json).expect("deserialise"); assert_eq!(restored.id, session.id); assert_eq!(restored.filename, session.filename); @@ -995,7 +992,9 @@ mod tests { let recovered = ChunkedUploadService::recover_sessions(&base).await; assert_eq!(recovered.len(), 1); - let session = recovered.get(&upload_id).expect("session must be recovered"); + let session = recovered + .get(&upload_id) + .expect("session must be recovered"); assert_eq!(session.filename, "bigfile.bin"); assert_eq!(session.folder_id, Some("folder-x".into())); assert_eq!(session.chunks[0].status, ChunkStatus::Complete); @@ -1050,10 +1049,8 @@ mod tests { assert!(status.pending_chunks.is_empty()); // 4. Complete (assemble) - let (path, filename, _folder, _ct, size, hash) = service - .complete_upload_inner(&id) - .await - .expect("complete"); + let (path, filename, _folder, _ct, size, hash) = + service.complete_upload_inner(&id).await.expect("complete"); assert_eq!(filename, "test.txt"); assert_eq!(size, 1024); assert!(!hash.is_empty()); @@ -1077,7 +1074,13 @@ mod tests { let service = ChunkedUploadService::new(base.clone()).await; let resp = service - .create_session_inner("x.bin".into(), None, "application/octet-stream".into(), 512, Some(512)) + .create_session_inner( + "x.bin".into(), + None, + "application/octet-stream".into(), + 512, + Some(512), + ) .await .expect("create"); @@ -1155,12 +1158,18 @@ mod tests { chunk_size: 512, chunks: vec![ ChunkInfo { - index: 0, offset: 0, size: 512, - status: ChunkStatus::Pending, checksum: None, + index: 0, + offset: 0, + size: 512, + status: ChunkStatus::Pending, + checksum: None, }, ChunkInfo { - index: 1, offset: 512, size: 512, - status: ChunkStatus::Pending, checksum: None, + index: 1, + offset: 512, + size: 512, + status: ChunkStatus::Pending, + checksum: None, }, ], created_at: Utc::now(), @@ -1171,14 +1180,20 @@ mod tests { // Write metadata let json = serde_json::to_vec(&session).unwrap(); - fs::write(session_dir.join(SESSION_META_FILE), &json).await.unwrap(); + fs::write(session_dir.join(SESSION_META_FILE), &json) + .await + .unwrap(); // Write progress marking both chunks complete let bitmask = vec![0b00000011u8]; // bits 0 and 1 - fs::write(session_dir.join(PROGRESS_FILE), &bitmask).await.unwrap(); + fs::write(session_dir.join(PROGRESS_FILE), &bitmask) + .await + .unwrap(); // But only create chunk_000000 on disk — chunk_000001 is "missing" - fs::write(session_dir.join("chunk_000000"), &[0u8; 512]).await.unwrap(); + fs::write(session_dir.join("chunk_000000"), &[0u8; 512]) + .await + .unwrap(); let recovered = ChunkedUploadService::recover_sessions(&base).await; let s = recovered.get("partial-session").expect("must be recovered"); diff --git a/src/infrastructure/services/dedup_service.rs b/src/infrastructure/services/dedup_service.rs index f26300b7..5f2bf34a 100644 --- a/src/infrastructure/services/dedup_service.rs +++ b/src/infrastructure/services/dedup_service.rs @@ -712,8 +712,7 @@ impl DedupService { let blob_root = blob_root.clone(); async move { let prefix = &hash[0..2]; - let blob_path = - blob_root.join(prefix).join(format!("{}.blob", hash)); + let blob_path = blob_root.join(prefix).join(format!("{}.blob", hash)); let mut issues = Vec::new(); diff --git a/src/infrastructure/services/file_content_cache.rs b/src/infrastructure/services/file_content_cache.rs index 65d514b3..b60e8506 100644 --- a/src/infrastructure/services/file_content_cache.rs +++ b/src/infrastructure/services/file_content_cache.rs @@ -124,7 +124,13 @@ impl FileContentCache { /// Put file content into cache /// /// Moka handles eviction automatically based on weight (content size). - pub async fn put(&self, file_id: String, content: Bytes, etag: Arc, content_type: Arc) { + pub async fn put( + &self, + file_id: String, + content: Bytes, + etag: Arc, + content_type: Arc, + ) { let size = content.len(); // Don't cache if too large @@ -258,24 +264,14 @@ mod tests { // A file within the limit should be cached let small = Bytes::from(vec![0u8; 50]); cache - .put( - "small".to_string(), - small, - "e1".into(), - "app/bin".into(), - ) + .put("small".to_string(), small, "e1".into(), "app/bin".into()) .await; assert!(cache.get("small").await.is_some()); // A file exceeding max_file_size is rejected by our own logic let big = Bytes::from(vec![1u8; 51]); cache - .put( - "big".to_string(), - big, - "e2".into(), - "app/bin".into(), - ) + .put("big".to_string(), big, "e2".into(), "app/bin".into()) .await; assert!( cache.get("big").await.is_none(), @@ -296,12 +292,7 @@ mod tests { let content = Bytes::from("test"); cache - .put( - "file1".to_string(), - content, - "e".into(), - "t".into(), - ) + .put("file1".to_string(), content, "e".into(), "t".into()) .await; assert!(cache.get("file1").await.is_some()); diff --git a/src/infrastructure/services/image_transcode_service.rs b/src/infrastructure/services/image_transcode_service.rs index 2fca238c..1e5f73ee 100644 --- a/src/infrastructure/services/image_transcode_service.rs +++ b/src/infrastructure/services/image_transcode_service.rs @@ -258,11 +258,7 @@ impl ImageTranscodeService { original_size, transcoded_size ); - return Ok(( - original_content, - original_mime.to_string(), - false, - )); + return Ok((original_content, original_mime.to_string(), false)); } let saved = original_size - transcoded_size; diff --git a/src/infrastructure/services/password_hasher.rs b/src/infrastructure/services/password_hasher.rs index 78bb5a5e..4ab1a338 100644 --- a/src/infrastructure/services/password_hasher.rs +++ b/src/infrastructure/services/password_hasher.rs @@ -39,13 +39,12 @@ impl Argon2PasswordHasher { /// /// Panics at startup if the parameters are invalid (caught immediately). pub fn new(memory_cost: u32, time_cost: u32, parallelism: u32) -> Self { - let params = Params::new(memory_cost, time_cost, parallelism, None) - .unwrap_or_else(|e| { - panic!( - "Invalid Argon2 parameters (m={}, t={}, p={}): {}", - memory_cost, time_cost, parallelism, e - ) - }); + let params = Params::new(memory_cost, time_cost, parallelism, None).unwrap_or_else(|e| { + panic!( + "Invalid Argon2 parameters (m={}, t={}, p={}): {}", + memory_cost, time_cost, parallelism, e + ) + }); tracing::info!( "Argon2PasswordHasher initialized: m_cost={} KiB, t_cost={}, p_cost={}", diff --git a/src/infrastructure/services/thumbnail_service.rs b/src/infrastructure/services/thumbnail_service.rs index 04dd25d8..206115e5 100644 --- a/src/infrastructure/services/thumbnail_service.rs +++ b/src/infrastructure/services/thumbnail_service.rs @@ -192,11 +192,7 @@ impl ThumbnailService { } // 2. Generate thumbnail (CPU-bound, runs in spawn_blocking) - tracing::info!( - "🎨 Generating thumbnail: {} {:?}", - file_id_owned, - size - ); + tracing::info!("🎨 Generating thumbnail: {} {:?}", file_id_owned, size); match self.generate_thumbnail(&original_owned, size).await { Ok(bytes) => { // Save to disk (best-effort — don't fail the request) @@ -245,7 +241,10 @@ impl ThumbnailService { let max_dim = size.max_dimension(); // Acquire semaphore permit — bounds peak RAM from concurrent decodes - let _permit = self.decode_semaphore.acquire().await + let _permit = self + .decode_semaphore + .acquire() + .await .map_err(|_| ThumbnailError::TaskError("Decode semaphore closed".into()))?; // Run image processing in blocking thread pool @@ -278,9 +277,9 @@ impl ThumbnailService { // Adaptive filter: faster filters for smaller sizes where // quality difference vs Lanczos3 is imperceptible let filter = match size { - ThumbnailSize::Icon => FilterType::Triangle, // 150px — max speed + ThumbnailSize::Icon => FilterType::Triangle, // 150px — max speed ThumbnailSize::Preview => FilterType::CatmullRom, // 400px — good balance - ThumbnailSize::Large => FilterType::CatmullRom, // 800px — sufficient quality + ThumbnailSize::Large => FilterType::CatmullRom, // 800px — sufficient quality }; let thumbnail = img.resize(new_width, new_height, filter); @@ -312,7 +311,10 @@ impl ThumbnailService { let _permit = match self.decode_semaphore.acquire().await { Ok(p) => p, Err(_) => { - tracing::warn!("Decode semaphore closed, skipping thumbnails for {}", file_id); + tracing::warn!( + "Decode semaphore closed, skipping thumbnails for {}", + file_id + ); return; } }; @@ -333,8 +335,8 @@ impl ThumbnailService { ))); } - let img = image::open(&path) - .map_err(|e| ThumbnailError::ImageError(e.to_string()))?; + let img = + image::open(&path).map_err(|e| ThumbnailError::ImageError(e.to_string()))?; let (orig_w, orig_h) = (img.width(), img.height()); @@ -352,18 +354,15 @@ impl ThumbnailService { }; let filter = match size { - ThumbnailSize::Icon => FilterType::Triangle, + ThumbnailSize::Icon => FilterType::Triangle, ThumbnailSize::Preview => FilterType::CatmullRom, - ThumbnailSize::Large => FilterType::CatmullRom, + ThumbnailSize::Large => FilterType::CatmullRom, }; let thumb = img.resize(new_w, new_h, filter); let mut buf = Vec::new(); thumb - .write_to( - &mut std::io::Cursor::new(&mut buf), - ImageFormat::WebP, - ) + .write_to(&mut std::io::Cursor::new(&mut buf), ImageFormat::WebP) .map_err(|e| ThumbnailError::ImageError(e.to_string()))?; Ok((size, Bytes::from(buf))) @@ -376,15 +375,11 @@ impl ThumbnailService { let thumbnails = match results { Ok(Ok(t)) => t, Ok(Err(e)) => { - tracing::warn!( - "Thumbnail generation failed for {}: {}", file_id, e - ); + tracing::warn!("Thumbnail generation failed for {}: {}", file_id, e); return; } Err(e) => { - tracing::warn!( - "Thumbnail task panicked for {}: {}", file_id, e - ); + tracing::warn!("Thumbnail task panicked for {}: {}", file_id, e); return; } }; diff --git a/src/infrastructure/services/zip_service.rs b/src/infrastructure/services/zip_service.rs index 7787f395..e74a1e57 100644 --- a/src/infrastructure/services/zip_service.rs +++ b/src/infrastructure/services/zip_service.rs @@ -8,8 +8,8 @@ use crate::{ use async_trait::async_trait; use async_zip::base::write::ZipFileWriter; use async_zip::{Compression, ZipEntryBuilder}; -use futures::io::AsyncWriteExt as FuturesWriteExt; use futures::StreamExt; +use futures::io::AsyncWriteExt as FuturesWriteExt; use std::collections::HashMap; use std::sync::Arc; use tempfile::NamedTempFile; @@ -99,17 +99,13 @@ impl ZipService { .folder_service .list_subtree_folders(folder_id) .await - .map_err(|e| { - ZipError::FolderContentsError(format!("subtree folders: {}", e)) - })?; + .map_err(|e| ZipError::FolderContentsError(format!("subtree folders: {}", e)))?; let all_files = self .file_service .list_files_in_subtree(folder_id) .await - .map_err(|e| { - ZipError::FolderContentsError(format!("subtree files: {}", e)) - })?; + .map_err(|e| ZipError::FolderContentsError(format!("subtree files: {}", e)))?; info!( "ZIP subtree: {} folders, {} files", @@ -157,8 +153,7 @@ impl ZipService { let zip_dir = format!("{}/", folder_zip_path(&folder.path)); // Directory entry (Stored, zero-length body) - let dir_entry = - ZipEntryBuilder::new(zip_dir.clone().into(), Compression::Stored); + let dir_entry = ZipEntryBuilder::new(zip_dir.clone().into(), Compression::Stored); match zip.write_entry_whole(dir_entry, &[]).await { Ok(()) => debug!("Folder added to ZIP: {}", zip_dir), Err(e) => { diff --git a/src/interfaces/api/handlers/file_handler.rs b/src/interfaces/api/handlers/file_handler.rs index 26e05d21..a3f5b243 100644 --- a/src/interfaces/api/handlers/file_handler.rs +++ b/src/interfaces/api/handlers/file_handler.rs @@ -476,13 +476,8 @@ impl FileHandler { Ok((_file, content)) => match content { OptimizedFileContent::Bytes { data, mime_type, .. - } => Self::build_cached_response( - data, - &mime_type, - &disposition, - &etag, - ) - .into_response(), + } => Self::build_cached_response(data, &mime_type, &disposition, &etag) + .into_response(), OptimizedFileContent::Mmap(mmap_data) => Response::builder() .status(StatusCode::OK) .header(header::CONTENT_TYPE, &file_dto.mime_type) @@ -566,10 +561,8 @@ impl FileHandler { tracing::info!("Found {} files", files.len()); let mut resp = (StatusCode::OK, Json(files)).into_response(); - resp.headers_mut().insert( - header::ETAG, - header::HeaderValue::from_str(&etag).unwrap(), - ); + resp.headers_mut() + .insert(header::ETAG, header::HeaderValue::from_str(&etag).unwrap()); resp } Err(err) => { @@ -601,7 +594,11 @@ impl FileHandler { }; // Generate thumbnails for supported images in background - if state.core.thumbnail_service.is_supported_image(&file.mime_type) { + if state + .core + .thumbnail_service + .is_supported_image(&file.mime_type) + { let file_id = file.id.clone(); let file_path_rel = file.path.clone(); let thumbnail_service = state.core.thumbnail_service.clone(); diff --git a/src/interfaces/api/handlers/folder_handler.rs b/src/interfaces/api/handlers/folder_handler.rs index e224ab23..716cf6d3 100644 --- a/src/interfaces/api/handlers/folder_handler.rs +++ b/src/interfaces/api/handlers/folder_handler.rs @@ -89,17 +89,16 @@ impl FolderHandler { match service.get_folder(&id).await { Ok(folder) => { // Access check: folder must belong to the requesting user - if let Some(ref owner) = folder.owner_id { - if owner != &auth_user.id { - tracing::warn!( - "get_folder: user '{}' attempted to access folder '{}' owned by '{}'", - auth_user.id, - id, - owner - ); - return (StatusCode::NOT_FOUND, "Folder not found".to_string()) - .into_response(); - } + if let Some(ref owner) = folder.owner_id + && owner != &auth_user.id + { + tracing::warn!( + "get_folder: user '{}' attempted to access folder '{}' owned by '{}'", + auth_user.id, + id, + owner + ); + return (StatusCode::NOT_FOUND, "Folder not found".to_string()).into_response(); } (StatusCode::OK, Json(folder)).into_response() } @@ -198,8 +197,13 @@ impl FolderHandler { /// Compute a lightweight ETag from the maximum `modified_at` timestamp /// and item count. No body buffering required. - fn compute_listing_etag(folders: &[crate::application::dtos::folder_dto::FolderDto], files: &[crate::application::dtos::file_dto::FileDto]) -> String { - let max_mod = folders.iter().map(|f| f.modified_at) + fn compute_listing_etag( + folders: &[crate::application::dtos::folder_dto::FolderDto], + files: &[crate::application::dtos::file_dto::FileDto], + ) -> String { + let max_mod = folders + .iter() + .map(|f| f.modified_at) .chain(files.iter().map(|f| f.modified_at)) .max() .unwrap_or(0); @@ -249,10 +253,8 @@ impl FolderHandler { let listing = FolderListingDto { folders, files }; let mut resp = (StatusCode::OK, Json(listing)).into_response(); - resp.headers_mut().insert( - header::ETAG, - header::HeaderValue::from_str(&etag).unwrap(), - ); + resp.headers_mut() + .insert(header::ETAG, header::HeaderValue::from_str(&etag).unwrap()); resp } (Err(err), _) | (_, Err(err)) => { @@ -441,10 +443,7 @@ impl FolderHandler { } }; - tracing::info!( - "ZIP file created successfully, size: {} bytes", - file_size - ); + tracing::info!("ZIP file created successfully, size: {} bytes", file_size); // Split the NamedTempFile into the already-open std File // and the TempPath (auto-deletes on drop). This reuses diff --git a/src/interfaces/api/handlers/webdav_handler.rs b/src/interfaces/api/handlers/webdav_handler.rs index 4e1f207e..57486815 100644 --- a/src/interfaces/api/handlers/webdav_handler.rs +++ b/src/interfaces/api/handlers/webdav_handler.rs @@ -284,13 +284,8 @@ async fn handle_propfind( let mut xml_writer = Writer::new(&mut buf); WebDavAdapter::write_multistatus_start(&mut xml_writer) .map_err(|e| AppError::internal_error(format!("XML write error: {}", e)))?; - WebDavAdapter::write_file_entry( - &mut xml_writer, - &file, - &propfind_request, - &base_href, - ) - .map_err(|e| AppError::internal_error(format!("XML write error: {}", e)))?; + WebDavAdapter::write_file_entry(&mut xml_writer, &file, &propfind_request, &base_href) + .map_err(|e| AppError::internal_error(format!("XML write error: {}", e)))?; WebDavAdapter::write_multistatus_end(&mut xml_writer) .map_err(|e| AppError::internal_error(format!("XML write error: {}", e)))?; } @@ -329,9 +324,9 @@ async fn build_streaming_propfind_response( { let mut w = Writer::new(&mut buf); WebDavAdapter::write_multistatus_start(&mut w) - .map_err(|e| std::io::Error::new(std::io::ErrorKind::Other, e.to_string()))?; + .map_err(|e| std::io::Error::other(e.to_string()))?; WebDavAdapter::write_folder_entry(&mut w, &folder, &propfind_request, &base_href) - .map_err(|e| std::io::Error::new(std::io::ErrorKind::Other, e.to_string()))?; + .map_err(|e| std::io::Error::other(e.to_string()))?; } yield Bytes::from(buf); @@ -353,7 +348,7 @@ async fn build_streaming_propfind_response( let result = folder_service .list_folders_paginated(fid_ref, &pag) .await - .map_err(|e| std::io::Error::new(std::io::ErrorKind::Other, e.to_string()))?; + .map_err(|e| std::io::Error::other(e.to_string()))?; if result.items.is_empty() { break; @@ -365,7 +360,7 @@ async fn build_streaming_propfind_response( for subfolder in &result.items { let href = format!("{}{}/", base_href, subfolder.name); WebDavAdapter::write_folder_entry(&mut w, subfolder, &propfind_request, &href) - .map_err(|e| std::io::Error::new(std::io::ErrorKind::Other, e.to_string()))?; + .map_err(|e| std::io::Error::other(e.to_string()))?; } } let has_more = result.pagination.has_next; @@ -383,7 +378,7 @@ async fn build_streaming_propfind_response( let batch: Vec = file_retrieval_service .list_files_batch(fid_ref, offset, PROPFIND_BATCH_SIZE) .await - .map_err(|e| std::io::Error::new(std::io::ErrorKind::Other, e.to_string()))?; + .map_err(|e| std::io::Error::other(e.to_string()))?; if batch.is_empty() { break; @@ -396,7 +391,7 @@ async fn build_streaming_propfind_response( for file in &batch { let href = format!("{}{}", base_href, file.name); WebDavAdapter::write_file_entry(&mut w, file, &propfind_request, &href) - .map_err(|e| std::io::Error::new(std::io::ErrorKind::Other, e.to_string()))?; + .map_err(|e| std::io::Error::other(e.to_string()))?; } } yield Bytes::from(chunk); @@ -413,13 +408,14 @@ async fn build_streaming_propfind_response( { let mut w = Writer::new(&mut buf); WebDavAdapter::write_multistatus_end(&mut w) - .map_err(|e| std::io::Error::new(std::io::ErrorKind::Other, e.to_string()))?; + .map_err(|e| std::io::Error::other(e.to_string()))?; } yield Bytes::from(buf); }; use futures::TryStreamExt; - let stream = stream.map_err(|e: std::io::Error| -> Box { Box::new(e) }); + let stream = stream + .map_err(|e: std::io::Error| -> Box { Box::new(e) }); Ok(Response::builder() .status(StatusCode::MULTI_STATUS) @@ -453,7 +449,9 @@ async fn handle_proppatch( // Read request body (XML — bounded to 1 MB) let body_bytes = body::to_bytes(req.into_body(), MAX_XML_BODY) .await - .map_err(|e| AppError::payload_too_large(format!("PROPPATCH body too large or unreadable: {}", e)))?; + .map_err(|e| { + AppError::payload_too_large(format!("PROPPATCH body too large or unreadable: {}", e)) + })?; let (props_to_set, props_to_remove) = WebDavAdapter::parse_proppatch(body_bytes.reader()) .map_err(|e| AppError::bad_request(format!("Failed to parse PROPPATCH request: {}", e)))?; @@ -661,12 +659,13 @@ async fn handle_put( ))); } hasher.update(chunk); - file.write_all(chunk) - .await - .map_err(|e| AppError::internal_error(format!("Failed to write to temp file: {}", e)))?; + file.write_all(chunk).await.map_err(|e| { + AppError::internal_error(format!("Failed to write to temp file: {}", e)) + })?; } } - file.flush().await + file.flush() + .await .map_err(|e| AppError::internal_error(format!("Failed to flush temp file: {}", e)))?; drop(file); @@ -1109,10 +1108,7 @@ async fn handle_copy( .create_folder(create_dto) .await .map_err(|e| { - AppError::internal_error(format!( - "Failed to create destination folder: {}", - e - )) + AppError::internal_error(format!("Failed to create destination folder: {}", e)) })?; } } else { diff --git a/src/interfaces/api/handlers/wopi_handler.rs b/src/interfaces/api/handlers/wopi_handler.rs index d4028058..74957c2c 100644 --- a/src/interfaces/api/handlers/wopi_handler.rs +++ b/src/interfaces/api/handlers/wopi_handler.rs @@ -13,7 +13,7 @@ use axum::{ Router, body::Body, extract::{Path, Query, State}, - http::{HeaderMap, StatusCode, Request}, + http::{HeaderMap, Request, StatusCode}, response::{Html, IntoResponse, Response}, routing::{get, post}, }; diff --git a/src/interfaces/errors.rs b/src/interfaces/errors.rs index 71747b6d..47b9053a 100644 --- a/src/interfaces/errors.rs +++ b/src/interfaces/errors.rs @@ -98,11 +98,7 @@ impl AppError { /// Create a 413 Payload Too Large error. pub fn payload_too_large(message: impl Into) -> Self { - Self::new( - StatusCode::PAYLOAD_TOO_LARGE, - message, - "PayloadTooLarge", - ) + Self::new(StatusCode::PAYLOAD_TOO_LARGE, message, "PayloadTooLarge") } } diff --git a/src/main.rs b/src/main.rs index 9d8bc813..7d363037 100644 --- a/src/main.rs +++ b/src/main.rs @@ -4,8 +4,8 @@ use std::sync::Arc; use axum::Router; use axum::extract::DefaultBodyLimit; -use tower_http::trace::TraceLayer; use tower_http::limit::RequestBodyLimitLayer; +use tower_http::trace::TraceLayer; use tracing_subscriber::{layer::SubscriberExt, util::SubscriberInitExt}; /// OxiCloud - Cloud Storage Platform