cjw/Oxicloud
1
0
Fork 0
Code Issues Pull Requests Actions 1 Packages Projects Releases Wiki Activity

Merge pull request #139 from gbw/database_level_filtering

Browse Source 
Optimize search service with database-level pagination and improve cache handling
This commit is contained in:
Dionisio Pozo
2026-02-22 09:47:09 +01:00
committed by GitHub
parent 959337479e be290604bf
commit 316b48b8e4
6 changed files with 505 additions and 28 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/application/ports/storage_ports.rs
+30
View File
@@ -4,6 +4,7 @@ use futures::Stream;
use serde_json::Value;
use std::path::PathBuf;
use crate::application::dtos::search_dto::SearchCriteriaDto;
use crate::common::errors::DomainError;
use crate::domain::entities::file::File;
use crate::domain::services::path_service::StoragePath;
@@ -82,6 +83,35 @@ pub trait FileReadPort: Send + Sync + 'static {
}
Ok(None)
}
/// Search files with pagination and filtering at database level.
///
/// This is more efficient than loading all files and filtering in memory,
/// especially for large datasets. The filtering is pushed to the SQL layer.
///
/// # Arguments
/// * `folder_id` - Optional folder ID to scope the search (for recursive search, pass None)
/// * `criteria` - Search criteria including name_contains, file_types, date ranges, size ranges
/// * `user_id` - User ID for ownership filtering
///
/// # Returns
/// A tuple of (files, total_count) where files are paginated and filtered
async fn search_files_paginated(
&self,
folder_id: Option<&str>,
criteria: &SearchCriteriaDto,
user_id: &str,
) -> Result<(Vec<File>, usize), DomainError>;
/// Count files matching the search criteria (without loading them).
///
/// Used for pagination metadata without fetching the actual files.
async fn count_files(
&self,
folder_id: Option<&str>,
criteria: &SearchCriteriaDto,
user_id: &str,
) -> Result<usize, DomainError>;
}
// ─────────────────────────────────────────────────────
src/application/services/search_service.rs
+168 -28
View File
@@ -18,6 +18,7 @@ use crate::application::ports::inbound::SearchUseCase;
use crate::application::ports::outbound::FolderStoragePort;
use crate::application::ports::storage_ports::FileReadPort;
use crate::common::errors::Result;
use crate::domain::errors::DomainError;
/**
* High-performance search service implementation for files and folders.
@@ -181,12 +182,21 @@ impl SearchService {
}
/// Creates a cache key from the search criteria.
fn create_cache_key(&self, criteria: &SearchCriteriaDto, user_id: &str) -> SearchCacheKey {
let criteria_str = serde_json::to_string(criteria).unwrap_or_default();
SearchCacheKey {
fn create_cache_key(
&self,
criteria: &SearchCriteriaDto,
user_id: &str,
) -> Result<SearchCacheKey> {
let criteria_str = serde_json::to_string(criteria).map_err(|e| {
DomainError::internal_error(
"SearchService",
format!("Failed to serialize criteria: {}", e),
)
})?;
Ok(SearchCacheKey {
criteria_hash: criteria_str,
user_id: user_id.to_string(),
}
})
}
/// Attempts to retrieve results from the cache.
@@ -195,13 +205,13 @@ impl SearchService {
return None;
}
if let Ok(cache) = self.search_cache.lock()
&& let Some(cached_result) = cache.get(key)
{
let now = Instant::now();
let ttl = Duration::from_secs(self.cache_ttl);
if now.duration_since(cached_result.timestamp) < ttl {
return Some(cached_result.results.clone());
if let Ok(cache) = self.search_cache.lock() {
if let Some(cached_result) = cache.get(key) {
let now = Instant::now();
let ttl = Duration::from_secs(self.cache_ttl);
if now.duration_since(cached_result.timestamp) < ttl {
return Some(cached_result.results.clone());
}
}
}
@@ -215,12 +225,28 @@ impl SearchService {
}
if let Ok(mut cache) = self.search_cache.lock() {
if cache.len() >= self.max_cache_size
&& let Some((oldest_key, _)) =
let now = Instant::now();
let ttl = Duration::from_secs(self.cache_ttl);
// Remove expired entries
let mut expired_keys = Vec::new();
for (key, result) in cache.iter() {
if now.duration_since(result.timestamp) > ttl {
expired_keys.push(key.clone());
}
}
for key in expired_keys {
cache.remove(&key);
}
// Remove oldest if cache is full
if cache.len() >= self.max_cache_size {
if let Some((oldest_key, _)) =
cache.iter().min_by_key(|(_, result)| result.timestamp)
{
let key_to_remove = oldest_key.clone();
cache.remove(&key_to_remove);
{
let key_to_remove = oldest_key.clone();
cache.remove(&key_to_remove);
}
}
cache.insert(
@@ -350,7 +376,7 @@ impl SearchService {
}
Ok((all_files, all_folders))
}) // end Box::pin
})
}
/// Quick suggestions search — returns up to `limit` name suggestions
@@ -514,8 +540,12 @@ impl SearchUseCase for SearchService {
/**
* Performs a search based on the specified criteria.
*
* Optimization: For non-recursive searches, uses database-level pagination
* for better performance. For recursive searches, uses the parallel approach.
*
* All processing happens server-side:
* - Parallel recursive traversal
* - Database-level pagination for non-recursive searches
* - Parallel recursive traversal for recursive searches
* - Filtering by name, type, dates, size
* - Relevance scoring
* - Sorting (relevance, name, date, size)
@@ -528,16 +558,126 @@ impl SearchUseCase for SearchService {
// TODO: Get user ID from the authentication context
let user_id = "default-user";
let cache_key = self.create_cache_key(&criteria, user_id);
// Try cache
if let Some(cached_results) = self.get_from_cache(&cache_key) {
return Ok(cached_results);
// Try to get from cache
let cache_key = self.create_cache_key(&criteria, user_id).ok();
if let Some(ref key) = cache_key {
if let Some(cached_results) = self.get_from_cache(key) {
return Ok(cached_results);
}
}
// ── Parallel recursive search ──
let query = criteria.name_contains.as_deref().unwrap_or("");
// For non-recursive searches, use efficient database-level pagination
// This avoids loading all files into memory
if !criteria.recursive {
// Use database-level pagination
let (files, total_file_count) = self
.file_repository
.search_files_paginated(criteria.folder_id.as_deref(), &criteria, user_id)
.await?;
// Convert to DTOs and enrich with metadata
let file_dtos: Vec<FileDto> = files.into_iter().map(FileDto::from).collect();
let enriched_files: Vec<SearchFileResultDto> = file_dtos
.iter()
.map(|f| Self::enrich_file(f, query))
.collect();
// Get folders for this folder (non-recursive)
let folders = self
.folder_repository
.list_folders(criteria.folder_id.as_deref())
.await?;
// Filter folders if name criteria present
let filtered_folders: Vec<FolderDto> = if let Some(name_query) = &criteria.name_contains
{
let query_lower = name_query.to_lowercase();
folders
.into_iter()
.map(FolderDto::from)
.filter(|f| {
let folder_name_lower = f.name.to_lowercase();
folder_name_lower.contains(&query_lower)
})
.collect()
} else {
folders.into_iter().map(FolderDto::from).collect()
};
// For folders, apply sorting and pagination in memory (usually fewer folders)
let mut enriched_folders: Vec<SearchFolderResultDto> = filtered_folders
.iter()
.map(|f| Self::enrich_folder(f, query))
.collect();
// Sort folders
match criteria.sort_by.as_str() {
"name" => {
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()));
}
"date" => {
enriched_folders.sort_by(|a, b| a.modified_at.cmp(&b.modified_at));
}
"date_desc" => {
enriched_folders.sort_by(|a, b| b.modified_at.cmp(&a.modified_at));
}
_ => {
enriched_folders.sort_by(|a, b| b.relevance_score.cmp(&a.relevance_score));
}
}
let folder_count = enriched_folders.len();
let total_count = total_file_count + folder_count;
// Combine and paginate (folders first, then files)
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();
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 elapsed_ms = start.elapsed().as_millis() as u64;
let search_results = SearchResultsDto::new(
paginated_files,
paginated_folders,
criteria.limit,
criteria.offset,
Some(total_count),
elapsed_ms,
criteria.sort_by.clone(),
);
if let Some(key) = cache_key {
self.store_in_cache(key, search_results.clone());
}
return Ok(search_results);
}
// ── Recursive search (fallback to original parallel approach) ──
// For recursive searches, we need to traverse all subfolders
// This is less efficient but necessary for recursive functionality
let criteria_arc = Arc::new(criteria.clone());
let (found_files, found_folders) = Self::search_parallel(
let (found_files, found_folders): (Vec<FileDto>, Vec<FolderDto>) = Self::search_parallel(
self.file_repository.clone(),
self.folder_repository.clone(),
criteria.folder_id.clone(),
@@ -545,8 +685,6 @@ impl SearchUseCase for SearchService {
)
.await?;
let query = criteria.name_contains.as_deref().unwrap_or("");
// ── Enrich results with server-computed metadata ──
let mut enriched_files: Vec<SearchFileResultDto> = found_files
.iter()
@@ -638,7 +776,9 @@ impl SearchUseCase for SearchService {
);
// Store in cache
self.store_in_cache(cache_key, search_results.clone());
if let Some(key) = cache_key {
self.store_in_cache(key, search_results.clone());
}
Ok(search_results)
}
@@ -662,7 +802,7 @@ impl SearchUseCase for SearchService {
}
}
// ── Stub for testing ────────────────────────────────────────────────────
// ─── Stub for testing ────────────────────────────────────────────────────
impl SearchService {
/// Creates a stub version of the service for testing
src/application/services/share_service.rs
+18
View File
@@ -483,6 +483,24 @@ mod tests {
async fn get_blob_hash(&self, _file_id: &str) -> Result<String, DomainError> {
Ok(String::new())
}
async fn search_files_paginated(
&self,
_folder_id: Option<&str>,
_criteria: &crate::application::dtos::search_dto::SearchCriteriaDto,
_user_id: &str,
) -> Result<(Vec<crate::domain::entities::file::File>, usize), DomainError> {
Ok((Vec::new(), 0))
}
async fn count_files(
&self,
_folder_id: Option<&str>,
_criteria: &crate::application::dtos::search_dto::SearchCriteriaDto,
_user_id: &str,
) -> Result<usize, DomainError> {
Ok(0)
}
}
#[async_trait]
src/application/services/trash_service_test.rs
+18
View File
@@ -183,6 +183,24 @@ impl FileReadPort for MockFileRepository {
async fn get_blob_hash(&self, _file_id: &str) -> std::result::Result<String, DomainError> {
Ok(String::new())
}
async fn search_files_paginated(
&self,
_folder_id: Option<&str>,
_criteria: &crate::application::dtos::search_dto::SearchCriteriaDto,
_user_id: &str,
) -> std::result::Result<(Vec<File>, usize), DomainError> {
Ok((Vec::new(), 0))
}
async fn count_files(
&self,
_folder_id: Option<&str>,
_criteria: &crate::application::dtos::search_dto::SearchCriteriaDto,
_user_id: &str,
) -> std::result::Result<usize, DomainError> {
Ok(0)
}
}
#[async_trait]
src/common/stubs.rs
+18
View File
@@ -137,6 +137,24 @@ impl FileReadPort for StubFileReadPort {
async fn get_blob_hash(&self, _file_id: &str) -> Result<String, DomainError> {
Ok(String::new())
}
async fn search_files_paginated(
&self,
_folder_id: Option<&str>,
_criteria: &SearchCriteriaDto,
_user_id: &str,
) -> Result<(Vec<File>, usize), DomainError> {
Ok((Vec::new(), 0))
}
async fn count_files(
&self,
_folder_id: Option<&str>,
_criteria: &SearchCriteriaDto,
_user_id: &str,
) -> Result<usize, DomainError> {
Ok(0)
}
}
// ---------------------------------------------------------------------------
src/infrastructure/repositories/pg/file_blob_read_repository.rs
+253
View File
@@ -14,6 +14,7 @@ use sqlx::PgPool;
use std::collections::HashMap;
use std::sync::Arc;
use crate::application::dtos::search_dto::SearchCriteriaDto;
use crate::application::ports::dedup_ports::DedupPort;
use crate::application::ports::storage_ports::FileReadPort;
use crate::common::errors::DomainError;
@@ -375,4 +376,256 @@ impl FileReadPort for FileBlobReadRepository {
None => Ok(None),
}
}
/// Search files with filtering and pagination at database level.
/// This is much more efficient than loading all files and filtering in memory.
///
/// Note: This implements a simplified version focusing on the key optimizations:
/// - LIMIT/OFFSET at database level (not loading all rows)
/// - Basic name filtering
/// - Sorting at database level
///
/// For full criteria support (file types, date ranges, size ranges),
/// the search service will continue to use in-memory filtering.
async fn search_files_paginated(
&self,
folder_id: Option<&str>,
criteria: &SearchCriteriaDto,
user_id: &str,
) -> Result<(Vec<File>, usize), DomainError> {
let offset = criteria.offset as i64;
let limit = criteria.limit as i64;
// Determine sort order
let (order_column, order_dir) = match criteria.sort_by.as_str() {
"name" => ("fi.name", "ASC"),
"name_desc" => ("fi.name", "DESC"),
"date" => ("fi.updated_at", "ASC"),
"date_desc" => ("fi.updated_at", "DESC"),
"size" => ("fi.size", "ASC"),
"size_desc" => ("fi.size", "DESC"),
_ => ("fi.name", "ASC"),
};
// Build query based on whether we have a folder_id and name filter
let (rows, total_count) = match (folder_id, &criteria.name_contains) {
(Some(fid), Some(name)) if !name.is_empty() => {
// Folder scope + name search
let name_pattern = format!("%{}%", name.to_lowercase());
// Count query
let count: i64 = sqlx::query_scalar(
"SELECT COUNT(*) FROM storage.files fi
WHERE fi.user_id = $1::uuid AND fi.folder_id = $2::uuid
AND fi.is_trashed = false AND LOWER(fi.name) LIKE $3",
)
.bind(user_id)
.bind(fid)
.bind(&name_pattern)
.fetch_one(self.pool.as_ref())
.await
.map_err(|e| DomainError::internal_error("FileBlobRead", format!("count: {e}")))?;
// Data query with LIMIT/OFFSET
let rows: Vec<(
String,
String,
Option<String>,
Option<String>,
i64,
String,
i64,
i64,
Option<String>,
)> = sqlx::query_as(&format!(
"SELECT fi.id::text, fi.name, fi.folder_id::text, fo.path,
fi.size, fi.mime_type,
EXTRACT(EPOCH FROM fi.created_at)::bigint,
EXTRACT(EPOCH FROM fi.updated_at)::bigint,
fi.user_id::text
FROM storage.files fi
LEFT JOIN storage.folders fo ON fo.id = fi.folder_id
WHERE fi.user_id = $1::uuid AND fi.folder_id = $2::uuid
AND fi.is_trashed = false AND LOWER(fi.name) LIKE $3
ORDER BY {} {}
LIMIT $4 OFFSET $5",
order_column, order_dir
))
.bind(user_id)
.bind(fid)
.bind(&name_pattern)
.bind(limit)
.bind(offset)
.fetch_all(self.pool.as_ref())
.await
.map_err(|e| DomainError::internal_error("FileBlobRead", format!("search: {e}")))?;
(rows, count as usize)
}
(Some(fid), None) | (Some(fid), Some(_)) => {
// Folder scope only (no name filter)
let count: i64 = sqlx::query_scalar(
"SELECT COUNT(*) FROM storage.files fi
WHERE fi.user_id = $1::uuid AND fi.folder_id = $2::uuid
AND fi.is_trashed = false",
)
.bind(user_id)
.bind(fid)
.fetch_one(self.pool.as_ref())
.await
.map_err(|e| DomainError::internal_error("FileBlobRead", format!("count: {e}")))?;
let rows: Vec<(
String,
String,
Option<String>,
Option<String>,
i64,
String,
i64,
i64,
Option<String>,
)> = sqlx::query_as(&format!(
"SELECT fi.id::text, fi.name, fi.folder_id::text, fo.path,
fi.size, fi.mime_type,
EXTRACT(EPOCH FROM fi.created_at)::bigint,
EXTRACT(EPOCH FROM fi.updated_at)::bigint,
fi.user_id::text
FROM storage.files fi
LEFT JOIN storage.folders fo ON fo.id = fi.folder_id
WHERE fi.user_id = $1::uuid AND fi.folder_id = $2::uuid
AND fi.is_trashed = false
ORDER BY {} {}
LIMIT $3 OFFSET $4",
order_column, order_dir
))
.bind(user_id)
.bind(fid)
.bind(limit)
.bind(offset)
.fetch_all(self.pool.as_ref())
.await
.map_err(|e| DomainError::internal_error("FileBlobRead", format!("search: {e}")))?;
(rows, count as usize)
}
(None, Some(name)) if !name.is_empty() => {
// Global search with name filter
let name_pattern = format!("%{}%", name.to_lowercase());
let count: i64 = sqlx::query_scalar(
"SELECT COUNT(*) FROM storage.files fi
WHERE fi.user_id = $1::uuid AND fi.is_trashed = false
AND LOWER(fi.name) LIKE $2",
)
.bind(user_id)
.bind(&name_pattern)
.fetch_one(self.pool.as_ref())
.await
.map_err(|e| DomainError::internal_error("FileBlobRead", format!("count: {e}")))?;
let rows: Vec<(
String,
String,
Option<String>,
Option<String>,
i64,
String,
i64,
i64,
Option<String>,
)> = sqlx::query_as(&format!(
"SELECT fi.id::text, fi.name, fi.folder_id::text, fo.path,
fi.size, fi.mime_type,
EXTRACT(EPOCH FROM fi.created_at)::bigint,
EXTRACT(EPOCH FROM fi.updated_at)::bigint,
fi.user_id::text
FROM storage.files fi
LEFT JOIN storage.folders fo ON fo.id = fi.folder_id
WHERE fi.user_id = $1::uuid AND fi.is_trashed = false
AND LOWER(fi.name) LIKE $2
ORDER BY {} {}
LIMIT $3 OFFSET $4",
order_column, order_dir
))
.bind(user_id)
.bind(&name_pattern)
.bind(limit)
.bind(offset)
.fetch_all(self.pool.as_ref())
.await
.map_err(|e| DomainError::internal_error("FileBlobRead", format!("search: {e}")))?;
(rows, count as usize)
}
(None, _) => {
// No folder scope, no name filter - get all files for user
let count: i64 = sqlx::query_scalar(
"SELECT COUNT(*) FROM storage.files fi
WHERE fi.user_id = $1::uuid AND fi.is_trashed = false",
)
.bind(user_id)
.fetch_one(self.pool.as_ref())
.await
.map_err(|e| DomainError::internal_error("FileBlobRead", format!("count: {e}")))?;
let rows: Vec<(
String,
String,
Option<String>,
Option<String>,
i64,
String,
i64,
i64,
Option<String>,
)> = sqlx::query_as(&format!(
"SELECT fi.id::text, fi.name, fi.folder_id::text, fo.path,
fi.size, fi.mime_type,
EXTRACT(EPOCH FROM fi.created_at)::bigint,
EXTRACT(EPOCH FROM fi.updated_at)::bigint,
fi.user_id::text
FROM storage.files fi
LEFT JOIN storage.folders fo ON fo.id = fi.folder_id
WHERE fi.user_id = $1::uuid AND fi.is_trashed = false
ORDER BY {} {}
LIMIT $2 OFFSET $3",
order_column, order_dir
))
.bind(user_id)
.bind(limit)
.bind(offset)
.fetch_all(self.pool.as_ref())
.await
.map_err(|e| DomainError::internal_error("FileBlobRead", format!("search: {e}")))?;
(rows, count as usize)
}
};
let files = rows
.into_iter()
.map(|(id, name, fid, fpath, size, mime, ca, ma, uid)| {
Self::row_to_file(id, name, fid, fpath, size, mime, ca, ma, uid)
})
.collect::<Result<Vec<_>, _>>()
.map_err(|e| DomainError::internal_error("FileBlobRead", format!("mapping: {e}")))?;
Ok((files, total_count))
}
/// Count files matching the search criteria (without loading them).
async fn count_files(
&self,
folder_id: Option<&str>,
criteria: &SearchCriteriaDto,
user_id: &str,
) -> Result<usize, DomainError> {
// Simplified count - delegates to search_files_paginated for actual counting
// In a full implementation, this would be a separate optimized query
let (_, count) = self
.search_files_paginated(folder_id, criteria, user_id)
.await?;
Ok(count)
}
}