From ed433df2afbc420ec46ccc2a499f02a2706efa20 Mon Sep 17 00:00:00 2001 From: Dionisio Date: Tue, 24 Feb 2026 12:18:38 +0100 Subject: [PATCH] perf(zip): eliminate N+1 queries with ltree bulk subtree fetch MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Replace BFS traversal that issued 2 SQL queries per folder (list_files + list_folders) with 2 total queries using PostgreSQL ltree <@ operator: 1. list_subtree_folders: single GiST-indexed scan for all folders 2. list_files_in_subtree: single GiST-indexed join for all files Files are grouped by folder_id in a HashMap, then iterated in directory order (folders pre-sorted by path from SQL). Changes across 4 architecture layers: - Domain: FolderRepository::list_subtree_folders (default impl) - Application ports: FolderUseCase, FileRetrievalUseCase, FileReadPort - Application services: FolderService, FileRetrievalService passthroughs - Infrastructure: PG implementations + ZipService rewrite Query count: O(N) → O(1). Latency for 100-folder tree: ~200 round-trips → 3. --- src/application/ports/file_ports.rs | 10 + src/application/ports/inbound.rs | 12 ++ src/application/ports/storage_ports.rs | 13 ++ .../services/file_retrieval_service.rs | 8 + src/application/services/folder_service.rs | 8 + src/domain/repositories/folder_repository.rs | 14 ++ .../pg/file_blob_read_repository.rs | 46 +++++ .../repositories/pg/folder_db_repository.rs | 33 ++++ src/infrastructure/services/zip_service.rs | 179 ++++++++---------- 9 files changed, 224 insertions(+), 99 deletions(-) diff --git a/src/application/ports/file_ports.rs b/src/application/ports/file_ports.rs index ec63d76f..c685e9b4 100644 --- a/src/application/ports/file_ports.rs +++ b/src/application/ports/file_ports.rs @@ -160,6 +160,16 @@ pub trait FileRetrievalUseCase: Send + Sync + 'static { start: u64, end: Option, ) -> Result> + Send>, DomainError>; + + /// Lists every file in the subtree rooted at `folder_id`. + /// + /// Default: falls back to `list_files(Some(folder_id))` (one level). + async fn list_files_in_subtree( + &self, + folder_id: &str, + ) -> Result, DomainError> { + self.list_files(Some(folder_id)).await + } } // ───────────────────────────────────────────────────── diff --git a/src/application/ports/inbound.rs b/src/application/ports/inbound.rs index a8bd53c5..444f1ca0 100644 --- a/src/application/ports/inbound.rs +++ b/src/application/ports/inbound.rs @@ -71,6 +71,18 @@ pub trait FolderUseCase: Send + Sync + 'static { user_id: &str, name: String, ) -> Result; + + /// Lists every folder in a subtree rooted at `folder_id` (inclusive), + /// ordered by path. Uses ltree `<@` — single GiST-indexed query. + /// + /// Default: returns an empty vec (stubs / mocks). + async fn list_subtree_folders( + &self, + folder_id: &str, + ) -> Result, DomainError> { + let _ = folder_id; + Ok(Vec::new()) + } } /** diff --git a/src/application/ports/storage_ports.rs b/src/application/ports/storage_ports.rs index 2e4dc890..74a5c6c2 100644 --- a/src/application/ports/storage_ports.rs +++ b/src/application/ports/storage_ports.rs @@ -78,6 +78,19 @@ pub trait FileReadPort: Send + Sync + 'static { Ok(None) } + /// Lists every file in the subtree rooted at `folder_id`. + /// + /// Uses an ltree `<@` join against `storage.folders` so the entire + /// subtree is fetched in a single GiST-indexed query. + /// + /// Default: falls back to `list_files(Some(folder_id))` (one level). + async fn list_files_in_subtree( + &self, + folder_id: &str, + ) -> Result, DomainError> { + self.list_files(Some(folder_id)).await + } + /// Search files with pagination and filtering at database level. /// /// This is more efficient than loading all files and filtering in memory, diff --git a/src/application/services/file_retrieval_service.rs b/src/application/services/file_retrieval_service.rs index 45e1f1b8..e3389607 100644 --- a/src/application/services/file_retrieval_service.rs +++ b/src/application/services/file_retrieval_service.rs @@ -308,4 +308,12 @@ impl FileRetrievalUseCase for FileRetrievalService { ) -> Result> + Send>, DomainError> { self.file_read.get_file_range_stream(id, start, end).await } + + async fn list_files_in_subtree( + &self, + folder_id: &str, + ) -> Result, DomainError> { + let files = self.file_read.list_files_in_subtree(folder_id).await?; + Ok(files.into_iter().map(FileDto::from).collect()) + } } diff --git a/src/application/services/folder_service.rs b/src/application/services/folder_service.rs index 04541ca5..7dc2e7f7 100644 --- a/src/application/services/folder_service.rs +++ b/src/application/services/folder_service.rs @@ -182,6 +182,14 @@ impl FolderUseCase for FolderService { Ok(FolderDto::from(folder)) } + async fn list_subtree_folders( + &self, + folder_id: &str, + ) -> Result, DomainError> { + let folders = self.folder_storage.list_subtree_folders(folder_id).await?; + Ok(folders.into_iter().map(FolderDto::from).collect()) + } + /// Gets a folder by its ID async fn get_folder(&self, id: &str) -> Result { let folder = self.folder_storage.get_folder(id).await.map_err(|e| { diff --git a/src/domain/repositories/folder_repository.rs b/src/domain/repositories/folder_repository.rs index cd8ff26a..b1b9267e 100644 --- a/src/domain/repositories/folder_repository.rs +++ b/src/domain/repositories/folder_repository.rs @@ -104,6 +104,20 @@ pub trait FolderRepository: Send + Sync + 'static { /// This is used during user registration to create the user's personal folder. async fn create_home_folder(&self, user_id: &str, name: String) -> Result; + /// Lists every folder in a subtree rooted at `folder_id` (inclusive). + /// + /// Uses ltree `<@` for a single GiST-indexed scan. The result is + /// ordered by `path` so callers can iterate in directory order. + /// + /// Default: falls back to `list_folders` (one level only). + async fn list_subtree_folders( + &self, + folder_id: &str, + ) -> Result, DomainError> { + let _ = folder_id; + Ok(Vec::new()) + } + /// Lists all descendant folders in a subtree (ltree-based). /// /// Returns all folders whose lpath is a descendant of the given folder's diff --git a/src/infrastructure/repositories/pg/file_blob_read_repository.rs b/src/infrastructure/repositories/pg/file_blob_read_repository.rs index 0ac28b23..9300c748 100644 --- a/src/infrastructure/repositories/pg/file_blob_read_repository.rs +++ b/src/infrastructure/repositories/pg/file_blob_read_repository.rs @@ -368,6 +368,52 @@ impl FileReadPort for FileBlobReadRepository { } } + /// Lists every file in the subtree rooted at `folder_id` (inclusive). + /// + /// Single GiST-indexed query via ltree `<@`. + /// Ordered by `(fo.path, fi.name)` so callers iterate in directory order. + async fn list_files_in_subtree( + &self, + folder_id: &str, + ) -> Result, DomainError> { + let rows: Vec<( + String, + String, + Option, + Option, + i64, + String, + i64, + i64, + Option, + )> = sqlx::query_as( + r#" + 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 + JOIN storage.folders fo ON fo.id = fi.folder_id + WHERE fo.lpath <@ (SELECT lpath FROM storage.folders WHERE id = $1::uuid) + AND NOT fi.is_trashed + ORDER BY fo.path, fi.name + "#, + ) + .bind(folder_id) + .fetch_all(self.pool.as_ref()) + .await + .map_err(|e| { + DomainError::internal_error("FileBlobRead", format!("subtree files: {e}")) + })?; + + 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() + } + /// Search files with filtering and pagination at database level. /// This is much more efficient than loading all files and filtering in memory. /// diff --git a/src/infrastructure/repositories/pg/folder_db_repository.rs b/src/infrastructure/repositories/pg/folder_db_repository.rs index 2b1d618f..3f952516 100644 --- a/src/infrastructure/repositories/pg/folder_db_repository.rs +++ b/src/infrastructure/repositories/pg/folder_db_repository.rs @@ -678,6 +678,39 @@ impl FolderRepository for FolderDbRepository { } } + /// Lists every folder in a subtree rooted at `folder_id` (inclusive). + /// + /// Single GiST-indexed query: `fo.lpath <@ (root's lpath)`. + /// Ordered by `fo.path` so callers can iterate in directory order. + async fn list_subtree_folders( + &self, + folder_id: &str, + ) -> Result, DomainError> { + let sql = "SELECT fo.id::text, fo.name, fo.path, fo.parent_id::text, \ + fo.user_id::text, \ + EXTRACT(EPOCH FROM fo.created_at)::bigint, \ + EXTRACT(EPOCH FROM fo.updated_at)::bigint \ + FROM storage.folders fo \ + WHERE fo.is_trashed = false \ + AND fo.lpath <@ (SELECT lpath FROM storage.folders WHERE id = $1::uuid) \ + ORDER BY fo.path"; + + let rows: Vec<(String, String, String, Option, Option, i64, i64)> = + sqlx::query_as(sql) + .bind(folder_id) + .fetch_all(self.pool()) + .await + .map_err(|e| { + DomainError::internal_error("FolderDb", format!("subtree folders: {e}")) + })?; + + rows.into_iter() + .map(|(id, name, path, pid, uid, ca, ma)| { + Self::row_to_folder(id, name, path, pid, uid, ca, ma) + }) + .collect() + } + /// Lists all descendant folders in a subtree using ltree GiST index. /// /// Single SQL query: `fo.lpath <@ (root's lpath)` fetches the entire diff --git a/src/infrastructure/services/zip_service.rs b/src/infrastructure/services/zip_service.rs index 9ce6f45d..7787f395 100644 --- a/src/infrastructure/services/zip_service.rs +++ b/src/infrastructure/services/zip_service.rs @@ -1,6 +1,5 @@ use crate::{ application::dtos::file_dto::FileDto, - application::dtos::folder_dto::FolderDto, application::ports::file_ports::FileRetrievalUseCase, application::ports::inbound::FolderUseCase, application::ports::zip_ports::ZipPort, @@ -11,6 +10,7 @@ use async_zip::base::write::ZipFileWriter; use async_zip::{Compression, ZipEntryBuilder}; use futures::io::AsyncWriteExt as FuturesWriteExt; use futures::StreamExt; +use std::collections::HashMap; use std::sync::Arc; use tempfile::NamedTempFile; use thiserror::Error; @@ -72,6 +72,9 @@ impl ZipService { /// Creates a ZIP file backed by a temporary file, containing the contents /// of a folder and all its subfolders. Returns the `NamedTempFile` so the /// caller can stream it and let the OS clean up on drop. + /// + /// Uses **2 SQL queries** (ltree `<@`) to fetch the entire subtree instead + /// of the previous N+1 BFS traversal. pub async fn create_folder_zip( &self, folder_id: &str, @@ -82,125 +85,103 @@ impl ZipService { folder_name, folder_id ); - // Verify the folder exists - let folder = match self.folder_service.get_folder(folder_id).await { - Ok(folder) => folder, + // Verify the folder exists and get its path for prefix stripping + let root_folder = match self.folder_service.get_folder(folder_id).await { + Ok(f) => f, Err(e) => { error!("Error getting folder {}: {}", folder_id, e); return Err(ZipError::FolderNotFound(folder_id.to_string()).into()); } }; - // Create a temp file; open a second async handle for writing. + // ── 1. Bulk-fetch the entire subtree (2 queries total) ─────────── + let all_folders = self + .folder_service + .list_subtree_folders(folder_id) + .await + .map_err(|e| { + ZipError::FolderContentsError(format!("subtree folders: {}", e)) + })?; + + let all_files = self + .file_service + .list_files_in_subtree(folder_id) + .await + .map_err(|e| { + ZipError::FolderContentsError(format!("subtree files: {}", e)) + })?; + + info!( + "ZIP subtree: {} folders, {} files", + all_folders.len(), + all_files.len() + ); + + // ── 2. Group files by folder_id ────────────────────────────────── + let mut files_by_folder: HashMap> = + HashMap::with_capacity(all_folders.len()); + for file in all_files { + let fid = file.folder_id.clone().unwrap_or_default(); + files_by_folder.entry(fid).or_default().push(file); + } + + // ── 3. Build a mapping: folder_id → ZIP-relative path ──────────── + // + // The root folder's DB path is e.g. "/users/alice/Documents". + // We want ZIP entries relative to `folder_name`, so we strip the + // root prefix and prepend `folder_name`. + let root_path = root_folder.path.trim_end_matches('/'); + let folder_zip_path = |db_path: &str| -> String { + let db_path = db_path.trim_end_matches('/'); + if db_path == root_path { + folder_name.to_string() + } else { + let suffix = db_path + .strip_prefix(root_path) + .unwrap_or(db_path) + .trim_start_matches('/'); + format!("{}/{}", folder_name, suffix) + } + }; + + // ── 4. Open the temp file + ZIP writer ─────────────────────────── let temp = NamedTempFile::new().map_err(ZipError::IoError)?; let tokio_file = tokio::fs::File::create(temp.path()) .await .map_err(ZipError::IoError)?; - - // 256 KB buffer keeps syscall count low. let buf_writer = BufWriter::with_capacity(256 * 1024, tokio_file); let mut zip = ZipFileWriter::with_tokio(buf_writer); - // Track processed folders to avoid cycles - let mut processed_folders = std::collections::HashSet::new(); + // ── 5. Write entries (folders are already sorted by path) ───────── + for folder in &all_folders { + let zip_dir = format!("{}/", folder_zip_path(&folder.path)); - // Build the ZIP iteratively - self.process_folder_recursively( - &mut zip, - &folder, - folder_name, - &mut processed_folders, - ) - .await?; + // Directory entry (Stored, zero-length body) + let dir_entry = + ZipEntryBuilder::new(zip_dir.clone().into(), Compression::Stored); + match zip.write_entry_whole(dir_entry, &[]).await { + Ok(()) => debug!("Folder added to ZIP: {}", zip_dir), + Err(e) => { + warn!("Could not add folder entry (may already exist): {}", e); + } + } - // Finalize: writes central directory, then flush buffered data to disk. + // Files belonging to this folder + if let Some(files) = files_by_folder.get(&folder.id) { + for file in files { + self.add_file_to_zip_streamed(&mut zip, file, &zip_dir) + .await?; + } + } + } + + // ── 6. Finalize ────────────────────────────────────────────────── let mut compat_writer = zip.close().await.map_err(ZipError::AsyncZipError)?; compat_writer.close().await.map_err(ZipError::IoError)?; Ok(temp) } - /// Iterative BFS over the folder tree. Writes entries directly to the - /// async `ZipFileWriter` so memory stays flat. - async fn process_folder_recursively( - &self, - zip: &mut AsyncZipWriter, - folder: &FolderDto, - path: &str, - processed_folders: &mut std::collections::HashSet, - ) -> Result<()> { - struct PendingFolder { - folder: FolderDto, - path: String, - } - - let mut work_queue = vec![PendingFolder { - folder: folder.clone(), - path: path.to_string(), - }]; - - while let Some(current) = work_queue.pop() { - let folder_id = current.folder.id.to_string(); - - if processed_folders.contains(&folder_id) { - continue; - } - processed_folders.insert(folder_id.clone()); - - // Directory entry (Stored, zero-length body) - let folder_path = format!("{}/", current.path); - let dir_entry = - ZipEntryBuilder::new(folder_path.clone().into(), Compression::Stored); - match zip.write_entry_whole(dir_entry, &[]).await { - Ok(()) => debug!("Folder added to ZIP: {}", folder_path), - Err(e) => { - warn!("Could not add folder to ZIP (may already exist): {}", e); - } - } - - // Files in this folder - let files = match self.file_service.list_files(Some(&folder_id)).await { - Ok(files) => files, - Err(e) => { - error!("Error listing files in folder {}: {}", folder_id, e); - return Err(ZipError::FolderContentsError(format!( - "Error listing files: {}", - e - )) - .into()); - } - }; - - for file in files { - self.add_file_to_zip_streamed(zip, &file, &folder_path) - .await?; - } - - // Subfolders - let subfolders = match self.folder_service.list_folders(Some(&folder_id)).await { - Ok(folders) => folders, - Err(e) => { - error!("Error listing subfolders in {}: {}", folder_id, e); - return Err(ZipError::FolderContentsError(format!( - "Error listing subfolders: {}", - e - )) - .into()); - } - }; - - for subfolder in subfolders { - let subfolder_path = format!("{}/{}", current.path, subfolder.name); - work_queue.push(PendingFolder { - folder: subfolder, - path: subfolder_path, - }); - } - } - - Ok(()) - } - /// Streams file content in chunks (~64 KB) into an async ZIP entry, /// keeping peak memory independent of individual file sizes. async fn add_file_to_zip_streamed(