fix: resolve all clippy warnings and enforce cargo fmt

- display_helpers: convert module doc-comments to regular comments,
  merge identical text/markdown + text/ branches
- search_service: replace needless range loops with slice-based pagination
- folder_repository, folder_db_repository: collapse nested if statements
- favorites_pg_repository: remove unnecessary borrow on generic arg
- file_blob_read_repository: collapse 6 nested if-let blocks
- file_blob_write_repository: collapse nested if for dedup ref decrement
- chunked_upload_service: use div_ceil(), collapse 2 nested if blocks
- folder_handler: collapse nested if-let for owner check
- webdav_handler: replace 7x io::Error::new(ErrorKind::Other, ..) with
  io::Error::other(..)
- cargo fmt applied to all files

Passes: cargo clippy --all-targets --all-features -- -D warnings
This commit is contained in:
Dionisio
2026-02-25 10:28:34 +01:00
parent 093400ce72
commit 97cf6402e2
34 changed files with 769 additions and 761 deletions
+9 -11
View File
@@ -1,12 +1,12 @@
/// Shared display helpers for DTOs. // Shared display helpers for DTOs.
/// //
/// These functions centralise the mime→icon / mime→category / size→human-string // These functions centralise the mime→icon / mime→category / size→human-string
/// logic so that every API response carries pre-computed display fields and the // logic so that every API response carries pre-computed display fields and the
/// frontend does **not** need to duplicate these mappings. // frontend does **not** need to duplicate these mappings.
/// //
/// The approach is: try MIME first (specific matches beat prefix matches), // The approach is: try MIME first (specific matches beat prefix matches),
/// then fall back to the file extension when the MIME is generic // then fall back to the file extension when the MIME is generic
/// (`application/octet-stream` or empty). // (`application/octet-stream` or empty).
// ─── Private: extract lowercase extension from a filename ──────────── // ─── 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") || mime.contains("typescript")
{ {
return "fas fa-file-code"; return "fas fa-file-code";
} else if mime.starts_with("text/markdown") {
return "fas fa-file-alt";
} else if mime.starts_with("text/") { } else if mime.starts_with("text/") {
return "fas fa-file-alt"; return "fas fa-file-alt";
} }
+6 -5
View File
@@ -165,10 +165,7 @@ pub trait FileRetrievalUseCase: Send + Sync + 'static {
/// Lists every file in the subtree rooted at `folder_id`. /// Lists every file in the subtree rooted at `folder_id`.
/// ///
/// Default: falls back to `list_files(Some(folder_id))` (one level). /// Default: falls back to `list_files(Some(folder_id))` (one level).
async fn list_files_in_subtree( async fn list_files_in_subtree(&self, folder_id: &str) -> Result<Vec<FileDto>, DomainError> {
&self,
folder_id: &str,
) -> Result<Vec<FileDto>, DomainError> {
self.list_files(Some(folder_id)).await self.list_files(Some(folder_id)).await
} }
@@ -183,7 +180,11 @@ pub trait FileRetrievalUseCase: Send + Sync + 'static {
limit: i64, limit: i64,
) -> Result<Vec<FileDto>, DomainError> { ) -> Result<Vec<FileDto>, DomainError> {
let all = self.list_files(folder_id).await?; 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())
} }
} }
+6 -5
View File
@@ -76,10 +76,7 @@ pub trait FolderUseCase: Send + Sync + 'static {
/// ordered by path. Uses ltree `<@` — single GiST-indexed query. /// ordered by path. Uses ltree `<@` — single GiST-indexed query.
/// ///
/// Default: returns an empty vec (stubs / mocks). /// Default: returns an empty vec (stubs / mocks).
async fn list_subtree_folders( async fn list_subtree_folders(&self, folder_id: &str) -> Result<Vec<FolderDto>, DomainError> {
&self,
folder_id: &str,
) -> Result<Vec<FolderDto>, DomainError> {
let _ = folder_id; let _ = folder_id;
Ok(Vec::new()) Ok(Vec::new())
} }
@@ -97,7 +94,11 @@ pub trait SearchUseCase: Send + Sync + 'static {
/// ///
/// `user_id` identifies the authenticated user so that SQL queries filter /// `user_id` identifies the authenticated user so that SQL queries filter
/// by owner and the result cache is isolated per tenant. /// by owner and the result cache is isolated per tenant.
async fn search(&self, criteria: SearchCriteriaDto, user_id: &str) -> Result<SearchResultsDto, DomainError>; async fn search(
&self,
criteria: SearchCriteriaDto,
user_id: &str,
) -> Result<SearchResultsDto, DomainError>;
/// Returns quick suggestions for autocomplete (lightweight, fast). /// Returns quick suggestions for autocomplete (lightweight, fast).
async fn suggest( async fn suggest(
+1 -4
View File
@@ -100,10 +100,7 @@ pub trait FileReadPort: Send + Sync + 'static {
/// subtree is fetched in a single GiST-indexed query. /// subtree is fetched in a single GiST-indexed query.
/// ///
/// Default: falls back to `list_files(Some(folder_id))` (one level). /// Default: falls back to `list_files(Some(folder_id))` (one level).
async fn list_files_in_subtree( async fn list_files_in_subtree(&self, folder_id: &str) -> Result<Vec<File>, DomainError> {
&self,
folder_id: &str,
) -> Result<Vec<File>, DomainError> {
self.list_files(Some(folder_id)).await self.list_files(Some(folder_id)).await
} }
@@ -526,7 +526,10 @@ impl AuthApplicationService {
} }
// Hash new password and update user // 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); user.update_password_hash(new_hash);
// Save updated user // Save updated user
@@ -642,10 +645,7 @@ impl AuthApplicationService {
let admin_quota = self.capped_quota(&admin_role); let admin_quota = self.capped_quota(&admin_role);
// Hash the password (same as register / admin_create_user) // Hash the password (same as register / admin_create_user)
let password_hash = self let password_hash = self.password_hasher.hash_password(&dto.password).await?;
.password_hasher
.hash_password(&dto.password)
.await?;
// Create the new admin user // Create the new admin user
let user = User::new( let user = User::new(
@@ -1156,10 +1156,8 @@ impl AuthApplicationService {
let exchange_code = hex::encode(code_bytes); let exchange_code = hex::encode(code_bytes);
// Store auth response (auto-expires after 60 s via moka TTL) // Store auth response (auto-expires after 60 s via moka TTL)
self.pending_oidc_tokens.insert( self.pending_oidc_tokens
exchange_code.clone(), .insert(exchange_code.clone(), PendingOidcToken { auth_response });
PendingOidcToken { auth_response },
);
tracing::info!("OIDC login successful, one-time exchange code generated"); tracing::info!("OIDC login successful, one-time exchange code generated");
@@ -1169,7 +1167,10 @@ impl AuthApplicationService {
/// Exchange a one-time code for the authentication tokens. /// Exchange a one-time code for the authentication tokens.
/// The code is single-use and expires after 60 seconds (moka TTL). /// The code is single-use and expires after 60 seconds (moka TTL).
pub fn exchange_oidc_token(&self, one_time_code: &str) -> Result<AuthResponseDto, DomainError> { pub fn exchange_oidc_token(&self, one_time_code: &str) -> Result<AuthResponseDto, DomainError> {
let pending = self.pending_oidc_tokens.remove(one_time_code).ok_or_else(|| { let pending = self
.pending_oidc_tokens
.remove(one_time_code)
.ok_or_else(|| {
DomainError::new( DomainError::new(
ErrorKind::AccessDenied, ErrorKind::AccessDenied,
"OIDC", "OIDC",
+4 -1
View File
@@ -713,7 +713,10 @@ impl BatchOperationService {
match chunk { match chunk {
Ok(bytes) => { Ok(bytes) => {
if let Err(e) = zip.write_all(&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; break;
} }
} }
@@ -177,7 +177,10 @@ impl FileManagementUseCase for FileManagementService {
.copy_folder_tree(source_folder_id, target_parent_id, dest_name) .copy_folder_tree(source_folder_id, target_parent_id, dest_name)
.await .await
.map_err(|e| { .map_err(|e| {
error!("Error copying folder tree (source: {}): {}", source_folder_id, e); error!(
"Error copying folder tree (source: {}): {}",
source_folder_id, e
);
e e
})?; })?;
@@ -88,7 +88,10 @@ impl FileRetrievalService {
return None; return None;
} }
let format = OutputFormat::WebP; 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)) => { Ok((transcoded, webp_mime, true)) => {
debug!( debug!(
"🖼️ WebP transcode: {} -> {} bytes ({:.0}% smaller)", "🖼️ WebP transcode: {} -> {} bytes ({:.0}% smaller)",
@@ -200,12 +203,7 @@ impl FileRetrievalService {
let etag: Arc<str> = format!("\"{}-{}\"", id, modified_at).into(); let etag: Arc<str> = format!("\"{}-{}\"", id, modified_at).into();
let ct: Arc<str> = Arc::from(&*mime_type); let ct: Arc<str> = Arc::from(&*mime_type);
cache cache
.put( .put(id.to_string(), content_bytes.clone(), etag, ct)
id.to_string(),
content_bytes.clone(),
etag,
ct,
)
.await; .await;
} }
@@ -310,10 +308,7 @@ impl FileRetrievalUseCase for FileRetrievalService {
self.file_read.get_file_range_stream(id, start, end).await self.file_read.get_file_range_stream(id, start, end).await
} }
async fn list_files_in_subtree( async fn list_files_in_subtree(&self, folder_id: &str) -> Result<Vec<FileDto>, DomainError> {
&self,
folder_id: &str,
) -> Result<Vec<FileDto>, DomainError> {
let files = self.file_read.list_files_in_subtree(folder_id).await?; let files = self.file_read.list_files_in_subtree(folder_id).await?;
Ok(files.into_iter().map(FileDto::from).collect()) Ok(files.into_iter().map(FileDto::from).collect())
} }
@@ -324,7 +319,10 @@ impl FileRetrievalUseCase for FileRetrievalService {
offset: i64, offset: i64,
limit: i64, limit: i64,
) -> Result<Vec<FileDto>, DomainError> { ) -> Result<Vec<FileDto>, 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()) Ok(files.into_iter().map(FileDto::from).collect())
} }
} }
+1 -4
View File
@@ -182,10 +182,7 @@ impl FolderUseCase for FolderService {
Ok(FolderDto::from(folder)) Ok(FolderDto::from(folder))
} }
async fn list_subtree_folders( async fn list_subtree_folders(&self, folder_id: &str) -> Result<Vec<FolderDto>, DomainError> {
&self,
folder_id: &str,
) -> Result<Vec<FolderDto>, DomainError> {
let folders = self.folder_storage.list_subtree_folders(folder_id).await?; let folders = self.folder_storage.list_subtree_folders(folder_id).await?;
Ok(folders.into_iter().map(FolderDto::from).collect()) Ok(folders.into_iter().map(FolderDto::from).collect())
} }
+23 -29
View File
@@ -347,19 +347,15 @@ impl SearchUseCase for SearchService {
let start_idx = criteria.offset.min(total_count); let start_idx = criteria.offset.min(total_count);
let end_idx = (criteria.offset + criteria.limit).min(total_count); let end_idx = (criteria.offset + criteria.limit).min(total_count);
let mut paginated_folders = Vec::new(); let folder_start = start_idx.min(folder_count);
let mut paginated_files = Vec::new(); 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 { let file_start = start_idx.saturating_sub(folder_count);
if i < folder_count { let file_end = end_idx
paginated_folders.push(enriched_folders[i].clone()); .saturating_sub(folder_count)
} else { .min(enriched_files.len());
let file_idx = i - folder_count; let paginated_files = enriched_files[file_start..file_end].to_vec();
if file_idx < enriched_files.len() {
paginated_files.push(enriched_files[file_idx].clone());
}
}
}
let elapsed_ms = start.elapsed().as_millis() as u64; 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) ── // ── Sort folders (files already sorted by SQL ORDER BY) ──
match criteria.sort_by.as_str() { match criteria.sort_by.as_str() {
"name" => { "name" => {
enriched_folders enriched_folders.sort_by(|a, b| a.name.to_lowercase().cmp(&b.name.to_lowercase()));
.sort_by(|a, b| a.name.to_lowercase().cmp(&b.name.to_lowercase()));
} }
"name_desc" => { "name_desc" => {
enriched_folders enriched_folders.sort_by(|a, b| b.name.to_lowercase().cmp(&a.name.to_lowercase()));
.sort_by(|a, b| b.name.to_lowercase().cmp(&a.name.to_lowercase()));
} }
"date" => { "date" => {
enriched_folders.sort_by(|a, b| a.modified_at.cmp(&b.modified_at)); 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 start_idx = criteria.offset.min(total_count);
let end_idx = (criteria.offset + criteria.limit).min(total_count); let end_idx = (criteria.offset + criteria.limit).min(total_count);
let mut paginated_folders = Vec::new(); let folder_start = start_idx.min(folder_count);
let mut paginated_files = Vec::new(); 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 { let file_start = start_idx.saturating_sub(folder_count);
if i < folder_count { let file_end = end_idx
paginated_folders.push(enriched_folders[i].clone()); .saturating_sub(folder_count)
} else { .min(enriched_files.len());
let file_idx = i - folder_count; let paginated_files = enriched_files[file_start..file_end].to_vec();
if file_idx < enriched_files.len() {
paginated_files.push(enriched_files[file_idx].clone());
}
}
}
let elapsed_ms = start.elapsed().as_millis() as u64; let elapsed_ms = start.elapsed().as_millis() as u64;
@@ -496,7 +486,11 @@ impl SearchService {
#[async_trait] #[async_trait]
impl SearchUseCase for SearchServiceStub { impl SearchUseCase for SearchServiceStub {
async fn search(&self, _criteria: SearchCriteriaDto, _user_id: &str) -> Result<SearchResultsDto> { async fn search(
&self,
_criteria: SearchCriteriaDto,
_user_id: &str,
) -> Result<SearchResultsDto> {
Ok(SearchResultsDto::empty()) Ok(SearchResultsDto::empty())
} }
+3 -2
View File
@@ -132,8 +132,9 @@ impl ShareService {
/// concurrently. This keeps RAM usage predictable (~19 MB × 2 = ~38 MB max) /// concurrently. This keeps RAM usage predictable (~19 MB × 2 = ~38 MB max)
/// and avoids starving the Tokio blocking thread pool. /// and avoids starving the Tokio blocking thread pool.
async fn hash_password_async(&self, password: &str) -> Result<String, DomainError> { async fn hash_password_async(&self, password: &str) -> Result<String, DomainError> {
let _permit = self.hash_semaphore.acquire().await let _permit = self.hash_semaphore.acquire().await.map_err(|_| {
.map_err(|_| DomainError::internal_error("ShareService", "Hash semaphore closed".to_string()))?; DomainError::internal_error("ShareService", "Hash semaphore closed".to_string())
})?;
self.password_hasher.hash_password(password).await self.password_hasher.hash_password(password).await
} }
} }
+1 -2
View File
@@ -615,8 +615,7 @@ impl AppConfig {
} }
// Storage limits // Storage limits
if let Ok(max_upload) = env::var("OXICLOUD_MAX_UPLOAD_SIZE") if let Ok(max_upload) = env::var("OXICLOUD_MAX_UPLOAD_SIZE").map(|v| v.parse::<usize>())
.map(|v| v.parse::<usize>())
&& let Ok(val) = max_upload && let Ok(val) = max_upload
{ {
config.storage.max_upload_size = val; config.storage.max_upload_size = val;
+7 -4
View File
@@ -36,10 +36,10 @@ use crate::application::services::{
use crate::common::config::AppConfig; use crate::common::config::AppConfig;
use crate::common::errors::DomainError; use crate::common::errors::DomainError;
use crate::domain::services::i18n_service::I18nService; use crate::domain::services::i18n_service::I18nService;
use crate::infrastructure::repositories::pg::SharePgRepository;
use crate::infrastructure::repositories::pg::{ use crate::infrastructure::repositories::pg::{
FileBlobReadRepository, FileBlobWriteRepository, FolderDbRepository, TrashDbRepository, FileBlobReadRepository, FileBlobWriteRepository, FolderDbRepository, TrashDbRepository,
}; };
use crate::infrastructure::repositories::pg::SharePgRepository;
use crate::infrastructure::services::file_content_cache::{ use crate::infrastructure::services::file_content_cache::{
FileContentCache, FileContentCacheConfig, FileContentCache, FileContentCacheConfig,
}; };
@@ -334,11 +334,13 @@ impl AppServiceFactory {
// Build a password hasher for share password verification // Build a password hasher for share password verification
let password_hasher: Arc<dyn crate::application::ports::auth_ports::PasswordHasherPort> = let password_hasher: Arc<dyn crate::application::ports::auth_ports::PasswordHasherPort> =
Arc::new(crate::infrastructure::services::password_hasher::Argon2PasswordHasher::new( Arc::new(
crate::infrastructure::services::password_hasher::Argon2PasswordHasher::new(
self.config.auth.hash_memory_cost, self.config.auth.hash_memory_cost,
self.config.auth.hash_time_cost, self.config.auth.hash_time_cost,
self.config.auth.hash_parallelism, self.config.auth.hash_parallelism,
)); ),
);
let service = Arc::new(ShareService::new( let service = Arc::new(ShareService::new(
Arc::new(self.config.clone()), Arc::new(self.config.clone()),
@@ -470,7 +472,8 @@ impl AppServiceFactory {
recent_service = Some(recent.clone()); recent_service = Some(recent.clone());
apps.recent_service = Some(recent); 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 // Auth services
if self.config.features.enable_auth { if self.config.features.enable_auth {
+13 -2
View File
@@ -621,7 +621,11 @@ pub struct StubSearchUseCase;
#[async_trait] #[async_trait]
impl SearchUseCase for StubSearchUseCase { impl SearchUseCase for StubSearchUseCase {
async fn search(&self, _criteria: SearchCriteriaDto, _user_id: &str) -> Result<SearchResultsDto, DomainError> { async fn search(
&self,
_criteria: SearchCriteriaDto,
_user_id: &str,
) -> Result<SearchResultsDto, DomainError> {
Ok(SearchResultsDto::empty()) Ok(SearchResultsDto::empty())
} }
@@ -794,7 +798,14 @@ impl ContentCachePort for StubContentCachePort {
None None
} }
async fn put(&self, _file_id: String, _content: Bytes, _etag: Arc<str>, _content_type: Arc<str>) {} async fn put(
&self,
_file_id: String,
_content: Bytes,
_etag: Arc<str>,
_content_type: Arc<str>,
) {
}
async fn invalidate(&self, _file_id: &str) {} async fn invalidate(&self, _file_id: &str) {}
+9 -9
View File
@@ -110,10 +110,7 @@ pub trait FolderRepository: Send + Sync + 'static {
/// ordered by `path` so callers can iterate in directory order. /// ordered by `path` so callers can iterate in directory order.
/// ///
/// Default: falls back to `list_folders` (one level only). /// Default: falls back to `list_folders` (one level only).
async fn list_subtree_folders( async fn list_subtree_folders(&self, folder_id: &str) -> Result<Vec<Folder>, DomainError> {
&self,
folder_id: &str,
) -> Result<Vec<Folder>, DomainError> {
let _ = folder_id; let _ = folder_id;
Ok(Vec::new()) Ok(Vec::new())
} }
@@ -154,17 +151,20 @@ pub trait FolderRepository: Send + Sync + 'static {
recursive: bool, recursive: bool,
) -> Result<Vec<Folder>, DomainError> { ) -> Result<Vec<Folder>, DomainError> {
// Recursive with folder_id → use optimised ltree scan // Recursive with folder_id → use optimised ltree scan
if recursive { if recursive && let Some(fid) = parent_id {
if let Some(fid) = parent_id { return self
return self.list_descendant_folders(fid, name_contains, user_id).await; .list_descendant_folders(fid, name_contains, user_id)
} .await;
} }
// Fallback: load + filter in memory (stubs / mocks) // Fallback: load + filter in memory (stubs / mocks)
let all = self.list_folders(parent_id).await?; let all = self.list_folders(parent_id).await?;
match name_contains { match name_contains {
Some(q) if !q.is_empty() => { Some(q) if !q.is_empty() => {
let q = q.to_lowercase(); 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), _ => Ok(all),
} }
+2 -4
View File
@@ -110,8 +110,7 @@ async fn create_pool_with_retries(
.connect(connection_string) .connect(connection_string)
.await .await
{ {
Ok(pool) => { Ok(pool) => match sqlx::query("SELECT 1").execute(&pool).await {
match sqlx::query("SELECT 1").execute(&pool).await {
Ok(_) => { Ok(_) => {
tracing::info!("PostgreSQL {} pool established successfully", label); tracing::info!("PostgreSQL {} pool established successfully", label);
return Ok(pool); return Ok(pool);
@@ -126,8 +125,7 @@ async fn create_pool_with_retries(
)); ));
} }
} }
} },
}
Err(e) => { Err(e) => {
tracing::error!( tracing::error!(
"Error connecting to PostgreSQL {} pool (attempt {}/{}): {}", "Error connecting to PostgreSQL {} pool (attempt {}/{}): {}",
@@ -222,7 +222,7 @@ impl FavoritesRepositoryPort for FavoritesPgRepository {
let mut q = sqlx::query(&query); let mut q = sqlx::query(&query);
for (item_id, item_type) in chunk { 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| { let result = q.execute(&mut *tx).await.map_err(|e| {
@@ -441,10 +441,7 @@ impl FileReadPort for FileBlobReadRepository {
/// ///
/// Single GiST-indexed query via ltree `<@`. /// Single GiST-indexed query via ltree `<@`.
/// Ordered by `(fo.path, fi.name)` so callers iterate in directory order. /// Ordered by `(fo.path, fi.name)` so callers iterate in directory order.
async fn list_files_in_subtree( async fn list_files_in_subtree(&self, folder_id: &str) -> Result<Vec<File>, DomainError> {
&self,
folder_id: &str,
) -> Result<Vec<File>, DomainError> {
let rows: Vec<( let rows: Vec<(
String, String,
String, String,
@@ -472,9 +469,7 @@ impl FileReadPort for FileBlobReadRepository {
.bind(folder_id) .bind(folder_id)
.fetch_all(self.pool.as_ref()) .fetch_all(self.pool.as_ref())
.await .await
.map_err(|e| { .map_err(|e| DomainError::internal_error("FileBlobRead", format!("subtree files: {e}")))?;
DomainError::internal_error("FileBlobRead", format!("subtree files: {e}"))
})?;
rows.into_iter() rows.into_iter()
.map(|(id, name, fid, fpath, size, mime, ca, ma, uid)| { .map(|(id, name, fid, fpath, size, mime, ca, ma, uid)| {
@@ -520,12 +515,12 @@ impl FileReadPort for FileBlobReadRepository {
conditions.push(format!("fi.folder_id = ${bind_idx}::uuid")); conditions.push(format!("fi.folder_id = ${bind_idx}::uuid"));
} }
if let Some(name) = &criteria.name_contains { if let Some(name) = &criteria.name_contains
if !name.is_empty() { && !name.is_empty()
{
bind_idx += 1; bind_idx += 1;
conditions.push(format!("LOWER(fi.name) LIKE ${bind_idx}")); conditions.push(format!("LOWER(fi.name) LIKE ${bind_idx}"));
} }
}
let where_clause = conditions.join(" AND "); let where_clause = conditions.join(" AND ");
let limit_bind = bind_idx + 1; let limit_bind = bind_idx + 1;
@@ -546,20 +541,31 @@ impl FileReadPort for FileBlobReadRepository {
); );
// ── Bind parameters dynamically ────────────────────────────────── // ── Bind parameters dynamically ──────────────────────────────────
let mut query = sqlx::query_as::<_, ( let mut query = sqlx::query_as::<
String, String, Option<String>, Option<String>, _,
i64, String, i64, i64, Option<String>, i64, (
)>(&sql) String,
String,
Option<String>,
Option<String>,
i64,
String,
i64,
i64,
Option<String>,
i64,
),
>(&sql)
.bind(user_id); .bind(user_id);
if let Some(fid) = folder_id { if let Some(fid) = folder_id {
query = query.bind(fid); query = query.bind(fid);
} }
if let Some(name) = &criteria.name_contains { if let Some(name) = &criteria.name_contains
if !name.is_empty() { && !name.is_empty()
{
query = query.bind(format!("%{}%", name.to_lowercase())); query = query.bind(format!("%{}%", name.to_lowercase()));
} }
}
query = query.bind(limit).bind(offset); query = query.bind(limit).bind(offset);
// ── Execute single query ───────────────────────────────────────── // ── Execute single query ─────────────────────────────────────────
@@ -600,9 +606,7 @@ impl FileReadPort for FileBlobReadRepository {
// When no root folder specified, delegate to existing paginated search // When no root folder specified, delegate to existing paginated search
let root_id = match root_folder_id { let root_id = match root_folder_id {
None => { None => {
return self return self.search_files_paginated(None, criteria, user_id).await;
.search_files_paginated(None, criteria, user_id)
.await;
} }
Some(id) => id, Some(id) => id,
}; };
@@ -631,20 +635,20 @@ impl FileReadPort for FileBlobReadRepository {
"fo.lpath <@ (SELECT lpath FROM storage.folders WHERE id = $2::uuid)".to_string(), "fo.lpath <@ (SELECT lpath FROM storage.folders WHERE id = $2::uuid)".to_string(),
); );
if let Some(name) = &criteria.name_contains { if let Some(name) = &criteria.name_contains
if !name.is_empty() { && !name.is_empty()
{
bind_idx += 1; bind_idx += 1;
conditions.push(format!("LOWER(fi.name) LIKE ${bind_idx}")); conditions.push(format!("LOWER(fi.name) LIKE ${bind_idx}"));
} }
} if let Some(types) = &criteria.file_types
if let Some(types) = &criteria.file_types { && !types.is_empty()
if !types.is_empty() { {
bind_idx += 1; bind_idx += 1;
conditions.push(format!( conditions.push(format!(
"LOWER(SUBSTRING(fi.name FROM '\\.([^.]+)$')) = ANY(${bind_idx})" "LOWER(SUBSTRING(fi.name FROM '\\.([^.]+)$')) = ANY(${bind_idx})"
)); ));
} }
}
if criteria.created_after.is_some() { if criteria.created_after.is_some() {
bind_idx += 1; bind_idx += 1;
conditions.push(format!( conditions.push(format!(
@@ -698,25 +702,35 @@ impl FileReadPort for FileBlobReadRepository {
); );
// ── Bind parameters dynamically ── // ── Bind parameters dynamically ──
let mut query = sqlx::query_as::<_, ( let mut query = sqlx::query_as::<
String, String, Option<String>, Option<String>, _,
i64, String, i64, i64, Option<String>, i64, (
)>(&sql) String,
String,
Option<String>,
Option<String>,
i64,
String,
i64,
i64,
Option<String>,
i64,
),
>(&sql)
.bind(user_id) .bind(user_id)
.bind(root_id); .bind(root_id);
if let Some(name) = &criteria.name_contains { if let Some(name) = &criteria.name_contains
if !name.is_empty() { && !name.is_empty()
{
query = query.bind(format!("%{}%", name.to_lowercase())); query = query.bind(format!("%{}%", name.to_lowercase()));
} }
} if let Some(types) = &criteria.file_types
if let Some(types) = &criteria.file_types { && !types.is_empty()
if !types.is_empty() { {
let lower_types: Vec<String> = let lower_types: Vec<String> = types.iter().map(|t| t.to_lowercase()).collect();
types.iter().map(|t| t.to_lowercase()).collect();
query = query.bind(lower_types); query = query.bind(lower_types);
} }
}
if let Some(v) = criteria.created_after { if let Some(v) = criteria.created_after {
query = query.bind(v as i64); query = query.bind(v as i64);
} }
@@ -739,10 +753,7 @@ impl FileReadPort for FileBlobReadRepository {
query = query.bind(limit).bind(offset); query = query.bind(limit).bind(offset);
// ── Execute single query ── // ── Execute single query ──
let rows = query let rows = query.fetch_all(self.pool.as_ref()).await.map_err(|e| {
.fetch_all(self.pool.as_ref())
.await
.map_err(|e| {
DomainError::internal_error("FileBlobRead", format!("subtree search: {e}")) DomainError::internal_error("FileBlobRead", format!("subtree search: {e}"))
})?; })?;
@@ -973,12 +984,9 @@ mod tests {
.build(), .build(),
}; };
repo.hash_cache repo.hash_cache.insert("a".to_string(), "ha".to_string());
.insert("a".to_string(), "ha".to_string()); repo.hash_cache.insert("b".to_string(), "hb".to_string());
repo.hash_cache repo.hash_cache.insert("c".to_string(), "hc".to_string());
.insert("b".to_string(), "hb".to_string());
repo.hash_cache
.insert("c".to_string(), "hc".to_string());
// Force moka to run pending eviction tasks // Force moka to run pending eviction tasks
repo.hash_cache.run_pending_tasks(); repo.hash_cache.run_pending_tasks();
@@ -1016,7 +1024,8 @@ mod tests {
} }
for h in handles { 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");
} }
} }
} }
@@ -166,15 +166,15 @@ impl FileBlobWriteRepository {
}; };
// Decrement old blob ref (only if hash changed, best-effort) // Decrement old blob ref (only if hash changed, best-effort)
if old_hash != new_hash { if old_hash != new_hash
if let Err(e) = self.dedup.remove_reference(&old_hash).await { && let Err(e) = self.dedup.remove_reference(&old_hash).await
{
tracing::warn!( tracing::warn!(
"Failed to decrement old blob ref {}: {}", "Failed to decrement old blob ref {}: {}",
&old_hash[..12], &old_hash[..12],
e e
); );
} }
}
Ok(()) Ok(())
} }
@@ -701,10 +701,7 @@ impl FileWritePort for FileBlobWriteRepository {
); );
} }
} }
DomainError::internal_error( DomainError::internal_error("FileBlobWrite", format!("copy_folder_tree: {e}"))
"FileBlobWrite",
format!("copy_folder_tree: {e}"),
)
})?; })?;
tracing::info!( tracing::info!(
@@ -685,10 +685,7 @@ impl FolderRepository for FolderDbRepository {
/// ///
/// Single GiST-indexed query: `fo.lpath <@ (root's lpath)`. /// Single GiST-indexed query: `fo.lpath <@ (root's lpath)`.
/// Ordered by `fo.path` so callers can iterate in directory order. /// Ordered by `fo.path` so callers can iterate in directory order.
async fn list_subtree_folders( async fn list_subtree_folders(&self, folder_id: &str) -> Result<Vec<Folder>, DomainError> {
&self,
folder_id: &str,
) -> Result<Vec<Folder>, DomainError> {
let sql = "SELECT fo.id::text, fo.name, fo.path, fo.parent_id::text, \ let sql = "SELECT fo.id::text, fo.name, fo.path, fo.parent_id::text, \
fo.user_id::text, \ fo.user_id::text, \
EXTRACT(EPOCH FROM fo.created_at)::bigint, \ EXTRACT(EPOCH FROM fo.created_at)::bigint, \
@@ -698,8 +695,15 @@ impl FolderRepository for FolderDbRepository {
AND fo.lpath <@ (SELECT lpath FROM storage.folders WHERE id = $1::uuid) \ AND fo.lpath <@ (SELECT lpath FROM storage.folders WHERE id = $1::uuid) \
ORDER BY fo.path"; ORDER BY fo.path";
let rows: Vec<(String, String, String, Option<String>, Option<String>, i64, i64)> = let rows: Vec<(
sqlx::query_as(sql) String,
String,
String,
Option<String>,
Option<String>,
i64,
i64,
)> = sqlx::query_as(sql)
.bind(folder_id) .bind(folder_id)
.fetch_all(self.pool()) .fetch_all(self.pool())
.await .await
@@ -728,10 +732,10 @@ impl FolderRepository for FolderDbRepository {
recursive: bool, recursive: bool,
) -> Result<Vec<Folder>, DomainError> { ) -> Result<Vec<Folder>, DomainError> {
// Recursive with folder scope → existing optimised ltree scan // Recursive with folder scope → existing optimised ltree scan
if recursive { if recursive && let Some(fid) = parent_id {
if let Some(fid) = parent_id { return self
return self.list_descendant_folders(fid, name_contains, user_id).await; .list_descendant_folders(fid, name_contains, user_id)
} .await;
} }
// Build optional name filter // Build optional name filter
@@ -761,8 +765,15 @@ impl FolderRepository for FolderDbRepository {
ORDER BY fo.name" ORDER BY fo.name"
); );
let rows: Vec<(String, String, String, Option<String>, Option<String>, i64, i64)> = let rows: Vec<(
if let Some(ref pattern) = name_pattern { String,
String,
String,
Option<String>,
Option<String>,
i64,
i64,
)> = if let Some(ref pattern) = name_pattern {
sqlx::query_as(&sql) sqlx::query_as(&sql)
.bind(user_id) .bind(user_id)
.bind(pattern) .bind(pattern)
@@ -774,9 +785,7 @@ impl FolderRepository for FolderDbRepository {
.fetch_all(self.pool()) .fetch_all(self.pool())
.await .await
} }
.map_err(|e| { .map_err(|e| DomainError::internal_error("FolderDb", format!("search_folders: {e}")))?;
DomainError::internal_error("FolderDb", format!("search_folders: {e}"))
})?;
return rows return rows
.into_iter() .into_iter()
@@ -820,8 +829,15 @@ impl FolderRepository for FolderDbRepository {
) )
}; };
let rows: Vec<(String, String, String, Option<String>, Option<String>, i64, i64)> = let rows: Vec<(
if let Some(pid) = parent_id { String,
String,
String,
Option<String>,
Option<String>,
i64,
i64,
)> = if let Some(pid) = parent_id {
if let Some(ref pattern) = name_pattern { if let Some(ref pattern) = name_pattern {
sqlx::query_as(&sql) sqlx::query_as(&sql)
.bind(pid) .bind(pid)
@@ -848,9 +864,7 @@ impl FolderRepository for FolderDbRepository {
.fetch_all(self.pool()) .fetch_all(self.pool())
.await .await
} }
.map_err(|e| { .map_err(|e| DomainError::internal_error("FolderDb", format!("search_folders: {e}")))?;
DomainError::internal_error("FolderDb", format!("search_folders: {e}"))
})?;
rows.into_iter() rows.into_iter()
.map(|(id, name, path, pid, uid, ca, ma)| { .map(|(id, name, path, pid, uid, ca, ma)| {
@@ -870,9 +884,10 @@ impl FolderRepository for FolderDbRepository {
user_id: &str, user_id: &str,
) -> Result<Vec<Folder>, DomainError> { ) -> Result<Vec<Folder>, DomainError> {
let (where_extra, name_pattern) = match name_contains { let (where_extra, name_pattern) = match name_contains {
Some(name) if !name.is_empty() => { Some(name) if !name.is_empty() => (
(" AND LOWER(fo.name) LIKE $3", Some(format!("%{}%", name.to_lowercase()))) " AND LOWER(fo.name) LIKE $3",
} Some(format!("%{}%", name.to_lowercase())),
),
_ => ("", None), _ => ("", None),
}; };
@@ -890,8 +905,15 @@ impl FolderRepository for FolderDbRepository {
ORDER BY fo.name" ORDER BY fo.name"
); );
let rows: Vec<(String, String, String, Option<String>, Option<String>, i64, i64)> = let rows: Vec<(
if let Some(ref pattern) = name_pattern { String,
String,
String,
Option<String>,
Option<String>,
i64,
i64,
)> = if let Some(ref pattern) = name_pattern {
sqlx::query_as(&sql) sqlx::query_as(&sql)
.bind(user_id) .bind(user_id)
.bind(folder_id) .bind(folder_id)
@@ -905,9 +927,7 @@ impl FolderRepository for FolderDbRepository {
.fetch_all(self.pool()) .fetch_all(self.pool())
.await .await
} }
.map_err(|e| { .map_err(|e| DomainError::internal_error("FolderDb", format!("descendant search: {e}")))?;
DomainError::internal_error("FolderDb", format!("descendant search: {e}"))
})?;
rows.into_iter() rows.into_iter()
.map(|(id, name, path, pid, uid, ca, ma)| { .map(|(id, name, path, pid, uid, ca, ma)| {
@@ -25,9 +25,9 @@ impl SharePgRepository {
/// Maps a [`sqlx::postgres::PgRow`] to the domain [`Share`] entity. /// Maps a [`sqlx::postgres::PgRow`] to the domain [`Share`] entity.
fn row_to_entity(row: &sqlx::postgres::PgRow) -> Result<Share, DomainError> { fn row_to_entity(row: &sqlx::postgres::PgRow) -> Result<Share, DomainError> {
let id: String = row.try_get("id").map_err(|e| { let id: String = row
DomainError::internal_error("Share", format!("Failed to read id: {e}")) .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| { let item_id: String = row.try_get("item_id").map_err(|e| {
DomainError::internal_error("Share", format!("Failed to read item_id: {e}")) DomainError::internal_error("Share", format!("Failed to read item_id: {e}"))
})?; })?;
@@ -53,7 +53,8 @@ impl SharePgRepository {
let item_type = let item_type =
ShareItemType::try_from(item_type_str.as_str()).unwrap_or(ShareItemType::File); ShareItemType::try_from(item_type_str.as_str()).unwrap_or(ShareItemType::File);
let permissions = SharePermissions::new(permissions_read, permissions_write, permissions_reshare); let permissions =
SharePermissions::new(permissions_read, permissions_write, permissions_reshare);
Ok(Share::from_raw( Ok(Share::from_raw(
id, id,
@@ -245,9 +246,7 @@ impl ShareStoragePort for SharePgRepository {
} }
async fn delete_share(&self, id: &str) -> Result<(), DomainError> { async fn delete_share(&self, id: &str) -> Result<(), DomainError> {
let result = sqlx::query( let result = sqlx::query("DELETE FROM storage.shares WHERE id = $1::UUID")
"DELETE FROM storage.shares WHERE id = $1::UUID",
)
.bind(id) .bind(id)
.execute(&*self.db_pool) .execute(&*self.db_pool)
.await .await
@@ -130,7 +130,7 @@ impl UploadSession {
/// Build the completed-chunks bitmask (1 bit per chunk). /// Build the completed-chunks bitmask (1 bit per chunk).
fn build_progress_bitmask(&self) -> Vec<u8> { fn build_progress_bitmask(&self) -> Vec<u8> {
let len = (self.chunks.len() + 7) / 8; let len = self.chunks.len().div_ceil(8);
let mut bitmask = vec![0u8; len]; let mut bitmask = vec![0u8; len];
for chunk in &self.chunks { for chunk in &self.chunks {
if chunk.status == ChunkStatus::Complete { if chunk.status == ChunkStatus::Complete {
@@ -157,8 +157,8 @@ impl UploadSession {
/// Persist the full session metadata once (on create). /// Persist the full session metadata once (on create).
async fn persist_metadata(&self) -> Result<(), String> { async fn persist_metadata(&self) -> Result<(), String> {
let path = self.temp_dir.join(SESSION_META_FILE); let path = self.temp_dir.join(SESSION_META_FILE);
let json = serde_json::to_vec(self) let json =
.map_err(|e| format!("Failed to serialise session: {e}"))?; serde_json::to_vec(self).map_err(|e| format!("Failed to serialise session: {e}"))?;
// Atomic write: write to .tmp then rename // Atomic write: write to .tmp then rename
let tmp = self.temp_dir.join("session.json.tmp"); let tmp = self.temp_dir.join("session.json.tmp");
fs::write(&tmp, &json) fs::write(&tmp, &json)
@@ -208,9 +208,7 @@ impl ChunkedUploadService {
}; };
if recovered_count > 0 { if recovered_count > 0 {
tracing::info!( tracing::info!("♻️ Recovered {recovered_count} chunked-upload session(s) from disk");
"♻️ Recovered {recovered_count} chunked-upload session(s) from disk"
);
} }
// Start cleanup task // Start cleanup task
@@ -357,8 +355,8 @@ impl ChunkedUploadService {
let dir_name = path.file_name().and_then(|n| n.to_str()).unwrap_or(""); let dir_name = path.file_name().and_then(|n| n.to_str()).unwrap_or("");
let sessions = sessions.read().await; let sessions = sessions.read().await;
if !sessions.contains_key(dir_name) { if !sessions.contains_key(dir_name)
if let Ok(metadata) = fs::metadata(&path).await && let Ok(metadata) = fs::metadata(&path).await
&& let Ok(modified) = metadata.modified() && let Ok(modified) = metadata.modified()
&& modified.elapsed().unwrap_or_default() > SESSION_EXPIRATION && modified.elapsed().unwrap_or_default() > SESSION_EXPIRATION
{ {
@@ -370,7 +368,6 @@ impl ChunkedUploadService {
} }
} }
} }
}
// ── Core operations ────────────────────────────────────────────────── // ── Core operations ──────────────────────────────────────────────────
@@ -504,9 +501,8 @@ impl ChunkedUploadService {
// worker free for other connections. // worker free for other connections.
if let Some(ref expected_checksum) = checksum { if let Some(ref expected_checksum) = checksum {
let data_clone = data.clone(); // Bytes::clone is O(1) — just an Arc increment let data_clone = data.clone(); // Bytes::clone is O(1) — just an Arc increment
let actual_checksum = tokio::task::spawn_blocking(move || { let actual_checksum =
format!("{:x}", md5::compute(&data_clone)) tokio::task::spawn_blocking(move || format!("{:x}", md5::compute(&data_clone)))
})
.await .await
.map_err(|e| format!("MD5 checksum task failed: {e}"))?; .map_err(|e| format!("MD5 checksum task failed: {e}"))?;
@@ -576,10 +572,7 @@ impl ChunkedUploadService {
} }
/// Get upload status /// Get upload status
async fn get_status_inner( async fn get_status_inner(&self, upload_id: &str) -> Result<UploadStatusResponseDto, String> {
&self,
upload_id: &str,
) -> Result<UploadStatusResponseDto, String> {
let sessions = self.sessions.read().await; let sessions = self.sessions.read().await;
let session = sessions let session = sessions
.get(upload_id) .get(upload_id)
@@ -645,12 +638,17 @@ impl ChunkedUploadService {
let chunks_meta: Vec<(usize, PathBuf)> = session let chunks_meta: Vec<(usize, PathBuf)> = session
.chunks .chunks
.iter() .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(); .collect();
let total_size = session.total_size; let total_size = session.total_size;
let hash = tokio::task::spawn_blocking(move || -> Result<String, String> { let hash = tokio::task::spawn_blocking(move || -> Result<String, String> {
use std::io::{Read, Write, BufWriter as StdBufWriter}; use std::io::{BufWriter as StdBufWriter, Read, Write};
let raw_output = std::fs::OpenOptions::new() let raw_output = std::fs::OpenOptions::new()
.create(true) .create(true)
@@ -721,11 +719,11 @@ impl ChunkedUploadService {
/// Finalize upload: remove session from RAM and clean up temp directory /// Finalize upload: remove session from RAM and clean up temp directory
async fn finalize_upload_inner(&self, upload_id: &str) -> Result<(), String> { async fn finalize_upload_inner(&self, upload_id: &str) -> Result<(), String> {
let mut sessions = self.sessions.write().await; let mut sessions = self.sessions.write().await;
if let Some(session) = sessions.remove(upload_id) { if let Some(session) = sessions.remove(upload_id)
if let Err(e) = fs::remove_dir_all(&session.temp_dir).await { && let Err(e) = fs::remove_dir_all(&session.temp_dir).await
{
tracing::warn!("Failed to cleanup upload {}: {}", upload_id, e); tracing::warn!("Failed to cleanup upload {}: {}", upload_id, e);
} }
}
Ok(()) Ok(())
} }
@@ -919,8 +917,7 @@ mod tests {
}; };
let json = serde_json::to_vec(&session).expect("serialise"); let json = serde_json::to_vec(&session).expect("serialise");
let restored: UploadSession = let restored: UploadSession = serde_json::from_slice(&json).expect("deserialise");
serde_json::from_slice(&json).expect("deserialise");
assert_eq!(restored.id, session.id); assert_eq!(restored.id, session.id);
assert_eq!(restored.filename, session.filename); assert_eq!(restored.filename, session.filename);
@@ -995,7 +992,9 @@ mod tests {
let recovered = ChunkedUploadService::recover_sessions(&base).await; let recovered = ChunkedUploadService::recover_sessions(&base).await;
assert_eq!(recovered.len(), 1); 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.filename, "bigfile.bin");
assert_eq!(session.folder_id, Some("folder-x".into())); assert_eq!(session.folder_id, Some("folder-x".into()));
assert_eq!(session.chunks[0].status, ChunkStatus::Complete); assert_eq!(session.chunks[0].status, ChunkStatus::Complete);
@@ -1050,10 +1049,8 @@ mod tests {
assert!(status.pending_chunks.is_empty()); assert!(status.pending_chunks.is_empty());
// 4. Complete (assemble) // 4. Complete (assemble)
let (path, filename, _folder, _ct, size, hash) = service let (path, filename, _folder, _ct, size, hash) =
.complete_upload_inner(&id) service.complete_upload_inner(&id).await.expect("complete");
.await
.expect("complete");
assert_eq!(filename, "test.txt"); assert_eq!(filename, "test.txt");
assert_eq!(size, 1024); assert_eq!(size, 1024);
assert!(!hash.is_empty()); assert!(!hash.is_empty());
@@ -1077,7 +1074,13 @@ mod tests {
let service = ChunkedUploadService::new(base.clone()).await; let service = ChunkedUploadService::new(base.clone()).await;
let resp = service 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 .await
.expect("create"); .expect("create");
@@ -1155,12 +1158,18 @@ mod tests {
chunk_size: 512, chunk_size: 512,
chunks: vec![ chunks: vec![
ChunkInfo { ChunkInfo {
index: 0, offset: 0, size: 512, index: 0,
status: ChunkStatus::Pending, checksum: None, offset: 0,
size: 512,
status: ChunkStatus::Pending,
checksum: None,
}, },
ChunkInfo { ChunkInfo {
index: 1, offset: 512, size: 512, index: 1,
status: ChunkStatus::Pending, checksum: None, offset: 512,
size: 512,
status: ChunkStatus::Pending,
checksum: None,
}, },
], ],
created_at: Utc::now(), created_at: Utc::now(),
@@ -1171,14 +1180,20 @@ mod tests {
// Write metadata // Write metadata
let json = serde_json::to_vec(&session).unwrap(); 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 // Write progress marking both chunks complete
let bitmask = vec![0b00000011u8]; // bits 0 and 1 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" // 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 recovered = ChunkedUploadService::recover_sessions(&base).await;
let s = recovered.get("partial-session").expect("must be recovered"); let s = recovered.get("partial-session").expect("must be recovered");
+1 -2
View File
@@ -712,8 +712,7 @@ impl DedupService {
let blob_root = blob_root.clone(); let blob_root = blob_root.clone();
async move { async move {
let prefix = &hash[0..2]; let prefix = &hash[0..2];
let blob_path = let blob_path = blob_root.join(prefix).join(format!("{}.blob", hash));
blob_root.join(prefix).join(format!("{}.blob", hash));
let mut issues = Vec::new(); let mut issues = Vec::new();
@@ -124,7 +124,13 @@ impl FileContentCache {
/// Put file content into cache /// Put file content into cache
/// ///
/// Moka handles eviction automatically based on weight (content size). /// Moka handles eviction automatically based on weight (content size).
pub async fn put(&self, file_id: String, content: Bytes, etag: Arc<str>, content_type: Arc<str>) { pub async fn put(
&self,
file_id: String,
content: Bytes,
etag: Arc<str>,
content_type: Arc<str>,
) {
let size = content.len(); let size = content.len();
// Don't cache if too large // Don't cache if too large
@@ -258,24 +264,14 @@ mod tests {
// A file within the limit should be cached // A file within the limit should be cached
let small = Bytes::from(vec![0u8; 50]); let small = Bytes::from(vec![0u8; 50]);
cache cache
.put( .put("small".to_string(), small, "e1".into(), "app/bin".into())
"small".to_string(),
small,
"e1".into(),
"app/bin".into(),
)
.await; .await;
assert!(cache.get("small").await.is_some()); assert!(cache.get("small").await.is_some());
// A file exceeding max_file_size is rejected by our own logic // A file exceeding max_file_size is rejected by our own logic
let big = Bytes::from(vec![1u8; 51]); let big = Bytes::from(vec![1u8; 51]);
cache cache
.put( .put("big".to_string(), big, "e2".into(), "app/bin".into())
"big".to_string(),
big,
"e2".into(),
"app/bin".into(),
)
.await; .await;
assert!( assert!(
cache.get("big").await.is_none(), cache.get("big").await.is_none(),
@@ -296,12 +292,7 @@ mod tests {
let content = Bytes::from("test"); let content = Bytes::from("test");
cache cache
.put( .put("file1".to_string(), content, "e".into(), "t".into())
"file1".to_string(),
content,
"e".into(),
"t".into(),
)
.await; .await;
assert!(cache.get("file1").await.is_some()); assert!(cache.get("file1").await.is_some());
@@ -258,11 +258,7 @@ impl ImageTranscodeService {
original_size, original_size,
transcoded_size transcoded_size
); );
return Ok(( return Ok((original_content, original_mime.to_string(), false));
original_content,
original_mime.to_string(),
false,
));
} }
let saved = original_size - transcoded_size; let saved = original_size - transcoded_size;
@@ -39,8 +39,7 @@ impl Argon2PasswordHasher {
/// ///
/// Panics at startup if the parameters are invalid (caught immediately). /// Panics at startup if the parameters are invalid (caught immediately).
pub fn new(memory_cost: u32, time_cost: u32, parallelism: u32) -> Self { pub fn new(memory_cost: u32, time_cost: u32, parallelism: u32) -> Self {
let params = Params::new(memory_cost, time_cost, parallelism, None) let params = Params::new(memory_cost, time_cost, parallelism, None).unwrap_or_else(|e| {
.unwrap_or_else(|e| {
panic!( panic!(
"Invalid Argon2 parameters (m={}, t={}, p={}): {}", "Invalid Argon2 parameters (m={}, t={}, p={}): {}",
memory_cost, time_cost, parallelism, e memory_cost, time_cost, parallelism, e
@@ -192,11 +192,7 @@ impl ThumbnailService {
} }
// 2. Generate thumbnail (CPU-bound, runs in spawn_blocking) // 2. Generate thumbnail (CPU-bound, runs in spawn_blocking)
tracing::info!( tracing::info!("🎨 Generating thumbnail: {} {:?}", file_id_owned, size);
"🎨 Generating thumbnail: {} {:?}",
file_id_owned,
size
);
match self.generate_thumbnail(&original_owned, size).await { match self.generate_thumbnail(&original_owned, size).await {
Ok(bytes) => { Ok(bytes) => {
// Save to disk (best-effort — don't fail the request) // Save to disk (best-effort — don't fail the request)
@@ -245,7 +241,10 @@ impl ThumbnailService {
let max_dim = size.max_dimension(); let max_dim = size.max_dimension();
// Acquire semaphore permit — bounds peak RAM from concurrent decodes // 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()))?; .map_err(|_| ThumbnailError::TaskError("Decode semaphore closed".into()))?;
// Run image processing in blocking thread pool // Run image processing in blocking thread pool
@@ -312,7 +311,10 @@ impl ThumbnailService {
let _permit = match self.decode_semaphore.acquire().await { let _permit = match self.decode_semaphore.acquire().await {
Ok(p) => p, Ok(p) => p,
Err(_) => { Err(_) => {
tracing::warn!("Decode semaphore closed, skipping thumbnails for {}", file_id); tracing::warn!(
"Decode semaphore closed, skipping thumbnails for {}",
file_id
);
return; return;
} }
}; };
@@ -333,8 +335,8 @@ impl ThumbnailService {
))); )));
} }
let img = image::open(&path) let img =
.map_err(|e| ThumbnailError::ImageError(e.to_string()))?; image::open(&path).map_err(|e| ThumbnailError::ImageError(e.to_string()))?;
let (orig_w, orig_h) = (img.width(), img.height()); let (orig_w, orig_h) = (img.width(), img.height());
@@ -360,10 +362,7 @@ impl ThumbnailService {
let mut buf = Vec::new(); let mut buf = Vec::new();
thumb thumb
.write_to( .write_to(&mut std::io::Cursor::new(&mut buf), ImageFormat::WebP)
&mut std::io::Cursor::new(&mut buf),
ImageFormat::WebP,
)
.map_err(|e| ThumbnailError::ImageError(e.to_string()))?; .map_err(|e| ThumbnailError::ImageError(e.to_string()))?;
Ok((size, Bytes::from(buf))) Ok((size, Bytes::from(buf)))
@@ -376,15 +375,11 @@ impl ThumbnailService {
let thumbnails = match results { let thumbnails = match results {
Ok(Ok(t)) => t, Ok(Ok(t)) => t,
Ok(Err(e)) => { Ok(Err(e)) => {
tracing::warn!( tracing::warn!("Thumbnail generation failed for {}: {}", file_id, e);
"Thumbnail generation failed for {}: {}", file_id, e
);
return; return;
} }
Err(e) => { Err(e) => {
tracing::warn!( tracing::warn!("Thumbnail task panicked for {}: {}", file_id, e);
"Thumbnail task panicked for {}: {}", file_id, e
);
return; return;
} }
}; };
+4 -9
View File
@@ -8,8 +8,8 @@ use crate::{
use async_trait::async_trait; use async_trait::async_trait;
use async_zip::base::write::ZipFileWriter; use async_zip::base::write::ZipFileWriter;
use async_zip::{Compression, ZipEntryBuilder}; use async_zip::{Compression, ZipEntryBuilder};
use futures::io::AsyncWriteExt as FuturesWriteExt;
use futures::StreamExt; use futures::StreamExt;
use futures::io::AsyncWriteExt as FuturesWriteExt;
use std::collections::HashMap; use std::collections::HashMap;
use std::sync::Arc; use std::sync::Arc;
use tempfile::NamedTempFile; use tempfile::NamedTempFile;
@@ -99,17 +99,13 @@ impl ZipService {
.folder_service .folder_service
.list_subtree_folders(folder_id) .list_subtree_folders(folder_id)
.await .await
.map_err(|e| { .map_err(|e| ZipError::FolderContentsError(format!("subtree folders: {}", e)))?;
ZipError::FolderContentsError(format!("subtree folders: {}", e))
})?;
let all_files = self let all_files = self
.file_service .file_service
.list_files_in_subtree(folder_id) .list_files_in_subtree(folder_id)
.await .await
.map_err(|e| { .map_err(|e| ZipError::FolderContentsError(format!("subtree files: {}", e)))?;
ZipError::FolderContentsError(format!("subtree files: {}", e))
})?;
info!( info!(
"ZIP subtree: {} folders, {} files", "ZIP subtree: {} folders, {} files",
@@ -157,8 +153,7 @@ impl ZipService {
let zip_dir = format!("{}/", folder_zip_path(&folder.path)); let zip_dir = format!("{}/", folder_zip_path(&folder.path));
// Directory entry (Stored, zero-length body) // Directory entry (Stored, zero-length body)
let dir_entry = let dir_entry = ZipEntryBuilder::new(zip_dir.clone().into(), Compression::Stored);
ZipEntryBuilder::new(zip_dir.clone().into(), Compression::Stored);
match zip.write_entry_whole(dir_entry, &[]).await { match zip.write_entry_whole(dir_entry, &[]).await {
Ok(()) => debug!("Folder added to ZIP: {}", zip_dir), Ok(()) => debug!("Folder added to ZIP: {}", zip_dir),
Err(e) => { Err(e) => {
+8 -11
View File
@@ -476,12 +476,7 @@ impl FileHandler {
Ok((_file, content)) => match content { Ok((_file, content)) => match content {
OptimizedFileContent::Bytes { OptimizedFileContent::Bytes {
data, mime_type, .. data, mime_type, ..
} => Self::build_cached_response( } => Self::build_cached_response(data, &mime_type, &disposition, &etag)
data,
&mime_type,
&disposition,
&etag,
)
.into_response(), .into_response(),
OptimizedFileContent::Mmap(mmap_data) => Response::builder() OptimizedFileContent::Mmap(mmap_data) => Response::builder()
.status(StatusCode::OK) .status(StatusCode::OK)
@@ -566,10 +561,8 @@ impl FileHandler {
tracing::info!("Found {} files", files.len()); tracing::info!("Found {} files", files.len());
let mut resp = (StatusCode::OK, Json(files)).into_response(); let mut resp = (StatusCode::OK, Json(files)).into_response();
resp.headers_mut().insert( resp.headers_mut()
header::ETAG, .insert(header::ETAG, header::HeaderValue::from_str(&etag).unwrap());
header::HeaderValue::from_str(&etag).unwrap(),
);
resp resp
} }
Err(err) => { Err(err) => {
@@ -601,7 +594,11 @@ impl FileHandler {
}; };
// Generate thumbnails for supported images in background // 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_id = file.id.clone();
let file_path_rel = file.path.clone(); let file_path_rel = file.path.clone();
let thumbnail_service = state.core.thumbnail_service.clone(); let thumbnail_service = state.core.thumbnail_service.clone();
+14 -15
View File
@@ -89,17 +89,16 @@ impl FolderHandler {
match service.get_folder(&id).await { match service.get_folder(&id).await {
Ok(folder) => { Ok(folder) => {
// Access check: folder must belong to the requesting user // Access check: folder must belong to the requesting user
if let Some(ref owner) = folder.owner_id { if let Some(ref owner) = folder.owner_id
if owner != &auth_user.id { && owner != &auth_user.id
{
tracing::warn!( tracing::warn!(
"get_folder: user '{}' attempted to access folder '{}' owned by '{}'", "get_folder: user '{}' attempted to access folder '{}' owned by '{}'",
auth_user.id, auth_user.id,
id, id,
owner owner
); );
return (StatusCode::NOT_FOUND, "Folder not found".to_string()) return (StatusCode::NOT_FOUND, "Folder not found".to_string()).into_response();
.into_response();
}
} }
(StatusCode::OK, Json(folder)).into_response() (StatusCode::OK, Json(folder)).into_response()
} }
@@ -198,8 +197,13 @@ impl FolderHandler {
/// Compute a lightweight ETag from the maximum `modified_at` timestamp /// Compute a lightweight ETag from the maximum `modified_at` timestamp
/// and item count. No body buffering required. /// 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 { fn compute_listing_etag(
let max_mod = folders.iter().map(|f| f.modified_at) 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)) .chain(files.iter().map(|f| f.modified_at))
.max() .max()
.unwrap_or(0); .unwrap_or(0);
@@ -249,10 +253,8 @@ impl FolderHandler {
let listing = FolderListingDto { folders, files }; let listing = FolderListingDto { folders, files };
let mut resp = (StatusCode::OK, Json(listing)).into_response(); let mut resp = (StatusCode::OK, Json(listing)).into_response();
resp.headers_mut().insert( resp.headers_mut()
header::ETAG, .insert(header::ETAG, header::HeaderValue::from_str(&etag).unwrap());
header::HeaderValue::from_str(&etag).unwrap(),
);
resp resp
} }
(Err(err), _) | (_, Err(err)) => { (Err(err), _) | (_, Err(err)) => {
@@ -441,10 +443,7 @@ impl FolderHandler {
} }
}; };
tracing::info!( tracing::info!("ZIP file created successfully, size: {} bytes", file_size);
"ZIP file created successfully, size: {} bytes",
file_size
);
// Split the NamedTempFile into the already-open std File // Split the NamedTempFile into the already-open std File
// and the TempPath (auto-deletes on drop). This reuses // and the TempPath (auto-deletes on drop). This reuses
+20 -24
View File
@@ -284,12 +284,7 @@ async fn handle_propfind(
let mut xml_writer = Writer::new(&mut buf); let mut xml_writer = Writer::new(&mut buf);
WebDavAdapter::write_multistatus_start(&mut xml_writer) WebDavAdapter::write_multistatus_start(&mut xml_writer)
.map_err(|e| AppError::internal_error(format!("XML write error: {}", e)))?; .map_err(|e| AppError::internal_error(format!("XML write error: {}", e)))?;
WebDavAdapter::write_file_entry( WebDavAdapter::write_file_entry(&mut xml_writer, &file, &propfind_request, &base_href)
&mut xml_writer,
&file,
&propfind_request,
&base_href,
)
.map_err(|e| AppError::internal_error(format!("XML write error: {}", e)))?; .map_err(|e| AppError::internal_error(format!("XML write error: {}", e)))?;
WebDavAdapter::write_multistatus_end(&mut xml_writer) WebDavAdapter::write_multistatus_end(&mut xml_writer)
.map_err(|e| AppError::internal_error(format!("XML write error: {}", e)))?; .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); let mut w = Writer::new(&mut buf);
WebDavAdapter::write_multistatus_start(&mut w) 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) 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); yield Bytes::from(buf);
@@ -353,7 +348,7 @@ async fn build_streaming_propfind_response(
let result = folder_service let result = folder_service
.list_folders_paginated(fid_ref, &pag) .list_folders_paginated(fid_ref, &pag)
.await .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() { if result.items.is_empty() {
break; break;
@@ -365,7 +360,7 @@ async fn build_streaming_propfind_response(
for subfolder in &result.items { for subfolder in &result.items {
let href = format!("{}{}/", base_href, subfolder.name); let href = format!("{}{}/", base_href, subfolder.name);
WebDavAdapter::write_folder_entry(&mut w, subfolder, &propfind_request, &href) 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; let has_more = result.pagination.has_next;
@@ -383,7 +378,7 @@ async fn build_streaming_propfind_response(
let batch: Vec<FileDto> = file_retrieval_service let batch: Vec<FileDto> = file_retrieval_service
.list_files_batch(fid_ref, offset, PROPFIND_BATCH_SIZE) .list_files_batch(fid_ref, offset, PROPFIND_BATCH_SIZE)
.await .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() { if batch.is_empty() {
break; break;
@@ -396,7 +391,7 @@ async fn build_streaming_propfind_response(
for file in &batch { for file in &batch {
let href = format!("{}{}", base_href, file.name); let href = format!("{}{}", base_href, file.name);
WebDavAdapter::write_file_entry(&mut w, file, &propfind_request, &href) 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); yield Bytes::from(chunk);
@@ -413,13 +408,14 @@ async fn build_streaming_propfind_response(
{ {
let mut w = Writer::new(&mut buf); let mut w = Writer::new(&mut buf);
WebDavAdapter::write_multistatus_end(&mut w) 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); yield Bytes::from(buf);
}; };
use futures::TryStreamExt; use futures::TryStreamExt;
let stream = stream.map_err(|e: std::io::Error| -> Box<dyn std::error::Error + Send + Sync> { Box::new(e) }); let stream = stream
.map_err(|e: std::io::Error| -> Box<dyn std::error::Error + Send + Sync> { Box::new(e) });
Ok(Response::builder() Ok(Response::builder()
.status(StatusCode::MULTI_STATUS) .status(StatusCode::MULTI_STATUS)
@@ -453,7 +449,9 @@ async fn handle_proppatch(
// Read request body (XML — bounded to 1 MB) // Read request body (XML — bounded to 1 MB)
let body_bytes = body::to_bytes(req.into_body(), MAX_XML_BODY) let body_bytes = body::to_bytes(req.into_body(), MAX_XML_BODY)
.await .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()) 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)))?; .map_err(|e| AppError::bad_request(format!("Failed to parse PROPPATCH request: {}", e)))?;
@@ -661,12 +659,13 @@ async fn handle_put(
))); )));
} }
hasher.update(chunk); hasher.update(chunk);
file.write_all(chunk) file.write_all(chunk).await.map_err(|e| {
AppError::internal_error(format!("Failed to write to temp file: {}", e))
})?;
}
}
file.flush()
.await .await
.map_err(|e| AppError::internal_error(format!("Failed to write to temp file: {}", e)))?;
}
}
file.flush().await
.map_err(|e| AppError::internal_error(format!("Failed to flush temp file: {}", e)))?; .map_err(|e| AppError::internal_error(format!("Failed to flush temp file: {}", e)))?;
drop(file); drop(file);
@@ -1109,10 +1108,7 @@ async fn handle_copy(
.create_folder(create_dto) .create_folder(create_dto)
.await .await
.map_err(|e| { .map_err(|e| {
AppError::internal_error(format!( AppError::internal_error(format!("Failed to create destination folder: {}", e))
"Failed to create destination folder: {}",
e
))
})?; })?;
} }
} else { } else {
+1 -1
View File
@@ -13,7 +13,7 @@ use axum::{
Router, Router,
body::Body, body::Body,
extract::{Path, Query, State}, extract::{Path, Query, State},
http::{HeaderMap, StatusCode, Request}, http::{HeaderMap, Request, StatusCode},
response::{Html, IntoResponse, Response}, response::{Html, IntoResponse, Response},
routing::{get, post}, routing::{get, post},
}; };
+1 -5
View File
@@ -98,11 +98,7 @@ impl AppError {
/// Create a 413 Payload Too Large error. /// Create a 413 Payload Too Large error.
pub fn payload_too_large(message: impl Into<String>) -> Self { pub fn payload_too_large(message: impl Into<String>) -> Self {
Self::new( Self::new(StatusCode::PAYLOAD_TOO_LARGE, message, "PayloadTooLarge")
StatusCode::PAYLOAD_TOO_LARGE,
message,
"PayloadTooLarge",
)
} }
} }
+1 -1
View File
@@ -4,8 +4,8 @@ use std::sync::Arc;
use axum::Router; use axum::Router;
use axum::extract::DefaultBodyLimit; use axum::extract::DefaultBodyLimit;
use tower_http::trace::TraceLayer;
use tower_http::limit::RequestBodyLimitLayer; use tower_http::limit::RequestBodyLimitLayer;
use tower_http::trace::TraceLayer;
use tracing_subscriber::{layer::SubscriberExt, util::SubscriberInitExt}; use tracing_subscriber::{layer::SubscriberExt, util::SubscriberInitExt};
/// OxiCloud - Cloud Storage Platform /// OxiCloud - Cloud Storage Platform