diff --git a/src/application/ports/zip_ports.rs b/src/application/ports/zip_ports.rs index a640d013..4a43356d 100644 --- a/src/application/ports/zip_ports.rs +++ b/src/application/ports/zip_ports.rs @@ -6,6 +6,7 @@ use crate::common::errors::DomainError; use async_trait::async_trait; +use tempfile::NamedTempFile; /// Port for ZIP archive operations. /// @@ -15,10 +16,11 @@ use async_trait::async_trait; pub trait ZipPort: Send + Sync + 'static { /// Create a ZIP archive containing the contents of a folder (recursively). /// - /// Returns the ZIP file bytes. + /// Returns a temporary file containing the ZIP archive. The caller streams + /// it to the client and the file is automatically deleted when dropped. async fn create_folder_zip( &self, folder_id: &str, folder_name: &str, - ) -> Result, DomainError>; + ) -> Result; } diff --git a/src/common/stubs.rs b/src/common/stubs.rs index ba69d4aa..6ad24f35 100644 --- a/src/common/stubs.rs +++ b/src/common/stubs.rs @@ -50,7 +50,7 @@ impl ZipPort for StubZipPort { &self, _folder_id: &str, _folder_name: &str, - ) -> Result, DomainError> { + ) -> Result { Err(DomainError::internal_error( "ZipService", "ZipService not initialized", diff --git a/src/infrastructure/services/zip_service.rs b/src/infrastructure/services/zip_service.rs index 8aebc701..b16d9889 100644 --- a/src/infrastructure/services/zip_service.rs +++ b/src/infrastructure/services/zip_service.rs @@ -7,8 +7,10 @@ use crate::{ common::errors::{DomainError, ErrorKind, Result}, }; use async_trait::async_trait; -use std::io::{Cursor, Read, Write}; +use futures::StreamExt; +use std::io::Write; use std::sync::Arc; +use tempfile::NamedTempFile; use thiserror::Error; use tracing::*; use zip::{ZipWriter, write::SimpleFileOptions}; @@ -46,14 +48,17 @@ impl From for DomainError { } } -/// Service for creating ZIP files +/// Service for creating ZIP files. +/// +/// Writes the ZIP archive to a temporary file on disk so that only one file's +/// stream-chunk (~64 KB) is held in memory at a time, regardless of archive size. pub struct ZipService { file_service: Arc, folder_service: Arc, } impl ZipService { - /// Creates a new instance of the ZIP service with a reference to the file service + /// Creates a new instance of the ZIP service pub fn new( file_service: Arc, folder_service: Arc, @@ -64,15 +69,20 @@ impl ZipService { } } - /// Creates a ZIP file with the contents of a folder and all its subfolders - /// Returns the ZIP bytes - pub async fn create_folder_zip(&self, folder_id: &str, folder_name: &str) -> Result> { + /// 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. + pub async fn create_folder_zip( + &self, + folder_id: &str, + folder_name: &str, + ) -> Result { info!( "Creating ZIP for folder: {} (ID: {})", folder_name, folder_id ); - // Verify if the folder exists + // Verify the folder exists let folder = match self.folder_service.get_folder(folder_id).await { Ok(folder) => folder, Err(e) => { @@ -81,19 +91,20 @@ impl ZipService { } }; - // Create an in-memory buffer for the ZIP - let buf = Cursor::new(Vec::new()); - let mut zip = ZipWriter::new(buf); + // Create a temp file to back the ZIP archive (O(1) RAM) + let temp = NamedTempFile::new().map_err(ZipError::IoError)?; + let raw_file = temp.reopen().map_err(ZipError::IoError)?; + let mut zip = ZipWriter::new(raw_file); // Set compression options let options = SimpleFileOptions::default() .compression_method(zip::CompressionMethod::Deflated) .unix_permissions(0o755); - // Object to track processed folders and avoid cycles + // Track processed folders to avoid cycles let mut processed_folders = std::collections::HashSet::new(); - // Process the root folder and build the ZIP + // Build the ZIP iteratively self.process_folder_recursively( &mut zip, &folder, @@ -103,62 +114,50 @@ impl ZipService { ) .await?; - // Finalize the ZIP and get the bytes - let mut zip_buf = zip.finish()?; + // Finalize the ZIP (flushes central directory) + zip.finish()?; - let mut bytes = Vec::new(); - match zip_buf.read_to_end(&mut bytes) { - Ok(_) => Ok(bytes), - Err(e) => { - error!("Error reading finalized ZIP: {}", e); - Err(ZipError::IoError(e).into()) - } - } + Ok(temp) } - // Alternative implementation to avoid recursion in async + /// Iterative BFS over the folder tree. Writes entries directly to the + /// file-backed `ZipWriter` so memory stays flat. async fn process_folder_recursively( &self, - zip: &mut ZipWriter>>, + zip: &mut ZipWriter, folder: &FolderDto, path: &str, options: &SimpleFileOptions, processed_folders: &mut std::collections::HashSet, ) -> Result<()> { - // Structure to represent pending work struct PendingFolder { folder: FolderDto, path: String, } - // Work queue for iterative processing let mut work_queue = vec![PendingFolder { folder: folder.clone(), path: path.to_string(), }]; - // Process the queue while there are elements while let Some(current) = work_queue.pop() { let folder_id = current.folder.id.to_string(); - // Avoid cycles if processed_folders.contains(&folder_id) { continue; } - processed_folders.insert(folder_id.clone()); - // Create the directory entry in the ZIP + // Directory entry let folder_path = format!("{}/", current.path); match zip.add_directory(&folder_path, *options) { Ok(_) => debug!("Folder added to ZIP: {}", folder_path), Err(e) => { - warn!("Could not add folder to ZIP (it may already exist): {}", e); - // Continue even if creating the directory fails (it could be a duplicate) + warn!("Could not add folder to ZIP (may already exist): {}", e); } } - // Add files from the folder to the ZIP + // Files in this folder let files = match self.file_service.list_files(Some(&folder_id)).await { Ok(files) => files, Err(e) => { @@ -171,13 +170,12 @@ impl ZipService { } }; - // Add each file to the ZIP for file in files { - self.add_file_to_zip(zip, &file, &folder_path, options) + self.add_file_to_zip_streamed(zip, &file, &folder_path, options) .await?; } - // Process subfolders + // Subfolders let subfolders = match self.folder_service.list_folders(Some(&folder_id)).await { Ok(folders) => folders, Err(e) => { @@ -190,7 +188,6 @@ impl ZipService { } }; - // Add subfolders to the queue for subfolder in subfolders { let subfolder_path = format!("{}/{}", current.path, subfolder.name); work_queue.push(PendingFolder { @@ -203,10 +200,11 @@ impl ZipService { Ok(()) } - // Adds a file to the ZIP - async fn add_file_to_zip( + /// Streams file content in chunks (~64 KB) into the ZIP entry, keeping + /// peak memory independent of individual file sizes. + async fn add_file_to_zip_streamed( &self, - zip: &mut ZipWriter>>, + zip: &mut ZipWriter, file: &FileDto, folder_path: &str, options: &SimpleFileOptions, @@ -214,37 +212,35 @@ impl ZipService { let file_path = format!("{}{}", folder_path, file.name); info!("Adding file to ZIP: {}", file_path); - // Get the file content let file_id = file.id.to_string(); - let content = match self.file_service.get_file_content(&file_id).await { - Ok(content) => content, + + // Start the ZIP entry + zip.start_file_from_path(std::path::Path::new(&file_path), *options) + .map_err(ZipError::ZipError)?; + + // Stream file contents in chunks instead of loading all into RAM + let stream = match self.file_service.get_file_stream(&file_id).await { + Ok(s) => s, Err(e) => { - error!("Error reading file content {}: {}", file_id, e); + error!("Error opening file stream {}: {}", file_id, e); return Err(ZipError::FileReadError(format!( - "Error reading file {}: {}", + "Error streaming file {}: {}", file_id, e )) .into()); } }; - // Write file to the ZIP - match zip.start_file_from_path(std::path::Path::new(&file_path), *options) { - Ok(_) => match zip.write_all(&content) { - Ok(_) => { - debug!("File added to ZIP: {}", file_path); - Ok(()) - } - Err(e) => { - error!("Error writing file content {}: {}", file_path, e); - Err(ZipError::IoError(e).into()) - } - }, - Err(e) => { - error!("Error starting file in ZIP {}: {}", file_path, e); - Err(ZipError::ZipError(e).into()) - } + // Pin the stream so StreamExt::next() can be called + let mut stream = std::pin::Pin::from(stream); + + while let Some(chunk_result) = stream.next().await { + let bytes = chunk_result.map_err(ZipError::IoError)?; + zip.write_all(&bytes).map_err(ZipError::IoError)?; } + + debug!("File added to ZIP: {}", file_path); + Ok(()) } } @@ -256,7 +252,7 @@ impl ZipPort for ZipService { &self, folder_id: &str, folder_name: &str, - ) -> std::result::Result, DomainError> { + ) -> std::result::Result { self.create_folder_zip(folder_id, folder_name).await } } diff --git a/src/interfaces/api/handlers/folder_handler.rs b/src/interfaces/api/handlers/folder_handler.rs index 84cf59bc..925de2e6 100644 --- a/src/interfaces/api/handlers/folder_handler.rs +++ b/src/interfaces/api/handlers/folder_handler.rs @@ -1,11 +1,12 @@ use axum::{ Json, extract::{Path, Query, State}, - http::{HeaderName, HeaderValue, Response, StatusCode, header}, + http::{Response, StatusCode, header}, response::IntoResponse, }; use std::collections::HashMap; use std::sync::Arc; +use tokio_util::io::ReaderStream; use crate::application::dtos::folder_dto::{CreateFolderDto, MoveFolderDto, RenameFolderDto}; use crate::application::dtos::folder_listing_dto::FolderListingDto; @@ -384,46 +385,67 @@ impl FolderHandler { // Use ZIP service from DI container let zip_service = &state.core.zip_service; - // Create the ZIP file + // Create the ZIP archive (written to a temp file, O(1) RAM) match zip_service.create_folder_zip(&id, &folder.name).await { - Ok(zip_data) => { + Ok(temp_file) => { + // Get the file size for Content-Length + let file_size = match temp_file.as_file().metadata() { + Ok(m) => m.len(), + Err(e) => { + tracing::error!("Error reading temp file metadata: {}", e); + return ( + StatusCode::INTERNAL_SERVER_ERROR, + Json(serde_json::json!({ + "error": "Error creating ZIP file" + })), + ) + .into_response(); + } + }; + tracing::info!( "ZIP file created successfully, size: {} bytes", - zip_data.len() + file_size ); + // Open the temp file with tokio for async streaming + let tokio_file = match tokio::fs::File::open(temp_file.path()).await { + Ok(f) => f, + Err(e) => { + tracing::error!("Error opening temp file for streaming: {}", e); + return ( + StatusCode::INTERNAL_SERVER_ERROR, + Json(serde_json::json!({ + "error": "Error streaming ZIP file" + })), + ) + .into_response(); + } + }; + + // Stream the temp file to the client in chunks + let stream = ReaderStream::new(tokio_file); + let body = axum::body::Body::from_stream(stream); + // Setup headers for download let filename = format!("{}.zip", folder.name); let content_disposition = format!("attachment; filename=\"{}\"", filename); - // Build response with the ZIP data - let mut headers = HashMap::new(); - headers.insert( - header::CONTENT_TYPE.to_string(), - "application/zip".to_string(), - ); - headers - .insert(header::CONTENT_DISPOSITION.to_string(), content_disposition); - headers.insert( - header::CONTENT_LENGTH.to_string(), - zip_data.len().to_string(), - ); - - // Build the response - let mut response = Response::builder() + let response = Response::builder() .status(StatusCode::OK) - .body(axum::body::Body::from(zip_data)) + .header(header::CONTENT_TYPE, "application/zip") + .header(header::CONTENT_DISPOSITION, content_disposition) + .header(header::CONTENT_LENGTH, file_size) + .body(body) .unwrap(); - // Add headers to response - for (name, value) in headers { - response.headers_mut().insert( - HeaderName::from_bytes(name.as_bytes()).unwrap(), - HeaderValue::from_str(&value).unwrap(), - ); - } + // temp_file is kept alive until the response future + // completes; dropped afterwards, cleaning up the file. + // We move it into the response extensions so it lives + // long enough for the stream to be fully read. + let _ = temp_file; - response + response.into_response() } Err(err) => { tracing::error!("Error creating ZIP file: {}", err);