Add database level pagination and filtering for efficiency
This commit is contained in:
@@ -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>;
|
||||
}
|
||||
|
||||
// ─────────────────────────────────────────────────────
|
||||
|
||||
@@ -514,8 +514,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)
|
||||
@@ -535,7 +539,110 @@ impl SearchUseCase for SearchService {
|
||||
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| f.name.to_lowercase().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(),
|
||||
);
|
||||
|
||||
self.store_in_cache(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(
|
||||
self.file_repository.clone(),
|
||||
@@ -545,8 +652,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()
|
||||
|
||||
@@ -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)
|
||||
}
|
||||
}
|
||||
|
||||
// ---------------------------------------------------------------------------
|
||||
|
||||
@@ -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,248 @@ 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;
|
||||
let limit = criteria.limit;
|
||||
|
||||
// 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 {} OFFSET {}",
|
||||
order_column, order_dir, limit, offset
|
||||
))
|
||||
.bind(user_id)
|
||||
.bind(fid)
|
||||
.bind(&name_pattern)
|
||||
.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 {} OFFSET {}",
|
||||
order_column, order_dir, limit, offset
|
||||
))
|
||||
.bind(user_id)
|
||||
.bind(fid)
|
||||
.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 {} OFFSET {}",
|
||||
order_column, order_dir, limit, offset
|
||||
))
|
||||
.bind(user_id)
|
||||
.bind(&name_pattern)
|
||||
.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 {} OFFSET {}",
|
||||
order_column, order_dir, limit, offset
|
||||
))
|
||||
.bind(user_id)
|
||||
.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)
|
||||
}
|
||||
}
|
||||
|
||||
Reference in New Issue
Block a user