diff --git a/src/application/ports/outbound.rs b/src/application/ports/outbound.rs index e1bd5376..e9ee2c30 100644 --- a/src/application/ports/outbound.rs +++ b/src/application/ports/outbound.rs @@ -48,34 +48,3 @@ pub trait FolderStoragePort: FolderRepository {} /// Blanket implementation: any type that implements FolderRepository /// is automatically a FolderStoragePort. impl FolderStoragePort for T {} - -/// Secondary port for ID mapping -#[async_trait] -pub trait IdMappingPort: Send + Sync + 'static { - /// Gets or creates an ID for a path - async fn get_or_create_id(&self, path: &StoragePath) -> Result; - - /// Gets a path by its ID - async fn get_path_by_id(&self, id: &str) -> Result; - - /// Updates the path for an existing ID - async fn update_path(&self, id: &str, new_path: &StoragePath) -> Result<(), DomainError>; - - /// Removes an ID from the mapping - async fn remove_id(&self, id: &str) -> Result<(), DomainError>; - - /// Saves pending changes - async fn save_changes(&self) -> Result<(), DomainError>; - - /// Gets the file path as a PathBuf - async fn get_file_path(&self, file_id: &str) -> Result { - let storage_path = self.get_path_by_id(file_id).await?; - Ok(PathBuf::from(storage_path.to_string())) - } - - /// Updates a file's path - async fn update_file_path(&self, file_id: &str, new_path: &PathBuf) -> Result<(), DomainError> { - let storage_path = StoragePath::from_string(new_path.to_string_lossy().as_ref()); - self.update_path(file_id, &storage_path).await - } -} diff --git a/src/application/services/mod.rs b/src/application/services/mod.rs index 7f9a3328..f96cf8e3 100644 --- a/src/application/services/mod.rs +++ b/src/application/services/mod.rs @@ -13,7 +13,6 @@ pub mod i18n_application_service; pub mod recent_service; pub mod search_service; pub mod share_service; -pub mod storage_mediator; pub mod storage_usage_service; pub mod trash_service; diff --git a/src/application/services/storage_mediator.rs b/src/application/services/storage_mediator.rs deleted file mode 100644 index d2978224..00000000 --- a/src/application/services/storage_mediator.rs +++ /dev/null @@ -1,326 +0,0 @@ -use async_trait::async_trait; -use std::path::{Path, PathBuf}; -use std::sync::Arc; -use thiserror::Error; - -use crate::application::ports::outbound::{FolderStoragePort, IdMappingPort, StoragePort}; -use crate::domain::entities::folder::Folder; -use crate::domain::services::path_service::StoragePath; - -/// Storage mediator specific errors -#[derive(Debug, Error)] -pub enum StorageMediatorError { - #[error("Entity not found: {0}")] - NotFound(String), - - #[error("Entity already exists: {0}")] - AlreadyExists(String), - - #[error("Invalid path: {0}")] - InvalidPath(String), - - #[error("Access error: {0}")] - AccessError(String), - - #[error("Internal error: {0}")] - InternalError(String), - - #[error("Domain error: {0}")] - DomainError(#[from] crate::common::errors::DomainError), -} - -/// Result type for mediator operations -pub type StorageMediatorResult = Result; - -/// Interface for the mediator service between file and folder repositories -#[async_trait] -pub trait StorageMediator: Send + Sync + 'static { - /// Gets the path of a folder by its ID - async fn get_folder_path(&self, folder_id: &str) -> StorageMediatorResult; - - /// Gets the domain path of a folder by its ID - async fn get_folder_storage_path(&self, folder_id: &str) -> StorageMediatorResult; - - /// Gets all details of a folder by its ID - async fn get_folder(&self, folder_id: &str) -> StorageMediatorResult; - - /// Checks if a file exists at a specific path - async fn file_exists_at_path(&self, path: &Path) -> StorageMediatorResult; - - /// Checks if a file exists at a specific domain path - async fn file_exists_at_storage_path( - &self, - storage_path: &StoragePath, - ) -> StorageMediatorResult; - - /// Checks if a folder exists at a specific path - async fn folder_exists_at_path(&self, path: &Path) -> StorageMediatorResult; - - /// Checks if a folder exists at a specific domain path - async fn folder_exists_at_storage_path( - &self, - storage_path: &StoragePath, - ) -> StorageMediatorResult; - - /// Resolves a relative path to absolute (legacy) - fn resolve_path(&self, relative_path: &Path) -> PathBuf; - - /// Resolves a domain path to an absolute physical path - fn resolve_storage_path(&self, storage_path: &StoragePath) -> PathBuf; - - /// Creates a directory if it does not exist (legacy) - async fn ensure_directory(&self, path: &Path) -> StorageMediatorResult<()>; - - /// Creates a directory if it does not exist - async fn ensure_storage_directory( - &self, - storage_path: &StoragePath, - ) -> StorageMediatorResult<()>; -} - -/// Concrete implementation of the storage mediator -pub struct FileSystemStorageMediator { - pub folder_storage_port: Arc, - pub path_service: Arc, - pub id_mapping: Arc, -} - -impl FileSystemStorageMediator { - pub fn new( - folder_storage_port: Arc, - path_service: Arc, - id_mapping: Arc, - ) -> Self { - Self { - folder_storage_port, - path_service, - id_mapping, - } - } - - /// Creates a stub implementation for initialization bootstrapping - pub fn new_stub() -> StubStorageMediator { - StubStorageMediator::new() - } -} - -/// Stub implementation for initialization dependency issues -/// This is a minimal stub that doesn't require any infrastructure dependencies -pub struct StubStorageMediator; - -impl StubStorageMediator { - pub fn new() -> Self { - Self - } -} - -impl Default for StubStorageMediator { - fn default() -> Self { - Self::new() - } -} - -#[async_trait] -impl StorageMediator for StubStorageMediator { - async fn get_folder_path(&self, _folder_id: &str) -> StorageMediatorResult { - // Return a stub path - Ok(PathBuf::from("/tmp")) - } - - async fn get_folder_storage_path( - &self, - _folder_id: &str, - ) -> StorageMediatorResult { - // Return a stub storage path - Ok(StoragePath::root()) - } - - async fn get_folder(&self, _folder_id: &str) -> StorageMediatorResult { - // This is a stub that should never be called during initialization - Err(StorageMediatorError::NotFound( - "Stub not implemented".to_string(), - )) - } - - async fn file_exists_at_path(&self, _path: &Path) -> StorageMediatorResult { - Ok(false) - } - - async fn file_exists_at_storage_path( - &self, - _storage_path: &StoragePath, - ) -> StorageMediatorResult { - Ok(false) - } - - async fn folder_exists_at_path(&self, _path: &Path) -> StorageMediatorResult { - Ok(false) - } - - async fn folder_exists_at_storage_path( - &self, - _storage_path: &StoragePath, - ) -> StorageMediatorResult { - Ok(false) - } - - fn resolve_path(&self, _relative_path: &Path) -> PathBuf { - PathBuf::from("/tmp") - } - - fn resolve_storage_path(&self, _storage_path: &StoragePath) -> PathBuf { - PathBuf::from("/tmp") - } - - async fn ensure_directory(&self, _path: &Path) -> StorageMediatorResult<()> { - Ok(()) - } - - async fn ensure_storage_directory( - &self, - _storage_path: &StoragePath, - ) -> StorageMediatorResult<()> { - Ok(()) - } -} - -#[async_trait] -impl StorageMediator for FileSystemStorageMediator { - async fn get_folder_path(&self, folder_id: &str) -> StorageMediatorResult { - let folder = self - .folder_storage_port - .get_folder(folder_id) - .await - .map_err(StorageMediatorError::from)?; - - // Need to get the path from folder ID - let storage_path = self - .id_mapping - .get_path_by_id(folder.id()) - .await - .map_err(StorageMediatorError::from)?; - - // Convert StoragePath to PathBuf - let path_buf = self.path_service.resolve_path(&storage_path); - Ok(path_buf) - } - - async fn get_folder_storage_path(&self, folder_id: &str) -> StorageMediatorResult { - let folder = self - .folder_storage_port - .get_folder(folder_id) - .await - .map_err(StorageMediatorError::from)?; - - // Get path by folder ID - will already be a StoragePath - let storage_path = self - .id_mapping - .get_path_by_id(folder.id()) - .await - .map_err(StorageMediatorError::from)?; - - Ok(storage_path) - } - - async fn get_folder(&self, folder_id: &str) -> StorageMediatorResult { - let folder = self - .folder_storage_port - .get_folder(folder_id) - .await - .map_err(StorageMediatorError::from)?; - - Ok(folder) - } - - async fn file_exists_at_path(&self, path: &Path) -> StorageMediatorResult { - let abs_path = self.resolve_path(path); - - // Check if it exists as a file (not as a directory) - let exists = abs_path.exists() && abs_path.is_file(); - - Ok(exists) - } - - async fn file_exists_at_storage_path( - &self, - storage_path: &StoragePath, - ) -> StorageMediatorResult { - let abs_path = self.resolve_storage_path(storage_path); - - // Check if it exists as a file (not as a directory) - let exists = abs_path.exists() && abs_path.is_file(); - - Ok(exists) - } - - async fn folder_exists_at_path(&self, path: &Path) -> StorageMediatorResult { - let abs_path = self.resolve_path(path); - - // Check if it exists as a directory - let exists = abs_path.exists() && abs_path.is_dir(); - - Ok(exists) - } - - async fn folder_exists_at_storage_path( - &self, - storage_path: &StoragePath, - ) -> StorageMediatorResult { - let abs_path = self.resolve_storage_path(storage_path); - - // Check if it exists as a directory - let exists = abs_path.exists() && abs_path.is_dir(); - - Ok(exists) - } - - fn resolve_path(&self, relative_path: &Path) -> PathBuf { - // Legacy method using PathBuf - let path_str = relative_path.to_string_lossy().to_string(); - let storage_path = StoragePath::from_string(&path_str); - self.path_service.resolve_path(&storage_path) - } - - fn resolve_storage_path(&self, storage_path: &StoragePath) -> PathBuf { - self.path_service.resolve_path(storage_path) - } - - async fn ensure_directory(&self, path: &Path) -> StorageMediatorResult<()> { - let abs_path = self.resolve_path(path); - - // Create directories if they don't exist - if !abs_path.exists() { - tokio::fs::create_dir_all(&abs_path).await.map_err(|e| { - StorageMediatorError::AccessError(format!("Could not create directory: {}", e)) - })?; - } else if !abs_path.is_dir() { - return Err(StorageMediatorError::InvalidPath(format!( - "Path exists but is not a directory: {}", - abs_path.display() - ))); - } - - Ok(()) - } - - async fn ensure_storage_directory( - &self, - storage_path: &StoragePath, - ) -> StorageMediatorResult<()> { - let abs_path = self.resolve_storage_path(storage_path); - - // Create directories if they don't exist - if !abs_path.exists() { - tokio::fs::create_dir_all(&abs_path).await.map_err(|e| { - StorageMediatorError::AccessError(format!("Could not create directory: {}", e)) - })?; - } else if !abs_path.is_dir() { - return Err(StorageMediatorError::InvalidPath(format!( - "Path exists but is not a directory: {}", - abs_path.display() - ))); - } - - Ok(()) - } -} diff --git a/src/infrastructure/services/mod.rs b/src/infrastructure/services/mod.rs index db72e32c..002247b2 100644 --- a/src/infrastructure/services/mod.rs +++ b/src/infrastructure/services/mod.rs @@ -10,5 +10,4 @@ pub mod password_hasher; pub mod path_service; pub mod thumbnail_service; pub mod trash_cleanup_service; -pub mod write_behind_cache; pub mod zip_service; diff --git a/src/infrastructure/services/path_service.rs b/src/infrastructure/services/path_service.rs index 9679aa7d..5f1df644 100644 --- a/src/infrastructure/services/path_service.rs +++ b/src/infrastructure/services/path_service.rs @@ -1,7 +1,7 @@ //! PathService - Infrastructure service for storage path management //! //! This service was moved from domain/services because it implements application traits -//! (StoragePort, StorageMediator) and has file system dependencies (tokio::fs). +//! (StoragePort) and has file system dependencies (tokio::fs). //! //! StoragePath (Value Object) remains in domain/services/path_service.rs @@ -10,11 +10,7 @@ use std::path::{Path, PathBuf}; use tokio::fs; use crate::application::ports::outbound::StoragePort; -use crate::application::services::storage_mediator::{ - StorageMediator, StorageMediatorError, StorageMediatorResult, -}; use crate::common::errors::{DomainError, ErrorKind}; -use crate::domain::entities::folder::Folder; use crate::domain::services::path_service::StoragePath; /// Infrastructure service for handling storage path operations @@ -178,109 +174,6 @@ impl StoragePort for PathService { } } -#[async_trait] -impl StorageMediator for PathService { - async fn get_folder_path(&self, folder_id: &str) -> StorageMediatorResult { - // This is a simplified implementation since PathService doesn't have direct - // access to folder repository. It's typically used through a proxy. - Err(StorageMediatorError::NotFound(format!( - "Folder with ID {} not found", - folder_id - ))) - } - - async fn get_folder_storage_path(&self, folder_id: &str) -> StorageMediatorResult { - // Simplified implementation - should be overridden by actual implementations - Err(StorageMediatorError::NotFound(format!( - "Folder with ID {} not found", - folder_id - ))) - } - - async fn get_folder(&self, folder_id: &str) -> StorageMediatorResult { - // Simplified implementation - should be overridden by actual implementations - Err(StorageMediatorError::NotFound(format!( - "Folder with ID {} not found", - folder_id - ))) - } - - async fn file_exists_at_path(&self, path: &Path) -> StorageMediatorResult { - let abs_path = self.resolve_path(&StoragePath::from_string(&path.to_string_lossy())); - Ok(abs_path.exists() && abs_path.is_file()) - } - - async fn file_exists_at_storage_path( - &self, - storage_path: &StoragePath, - ) -> StorageMediatorResult { - let abs_path = self.resolve_path(storage_path); - Ok(abs_path.exists() && abs_path.is_file()) - } - - async fn folder_exists_at_path(&self, path: &Path) -> StorageMediatorResult { - let abs_path = self.resolve_path(&StoragePath::from_string(&path.to_string_lossy())); - Ok(abs_path.exists() && abs_path.is_dir()) - } - - async fn folder_exists_at_storage_path( - &self, - storage_path: &StoragePath, - ) -> StorageMediatorResult { - let abs_path = self.resolve_path(storage_path); - Ok(abs_path.exists() && abs_path.is_dir()) - } - - fn resolve_path(&self, relative_path: &Path) -> PathBuf { - // Convert path to storage path then resolve - let path_str = relative_path.to_string_lossy().to_string(); - let storage_path = StoragePath::from_string(&path_str); - PathService::resolve_path(self, &storage_path) - } - - fn resolve_storage_path(&self, storage_path: &StoragePath) -> PathBuf { - PathService::resolve_path(self, storage_path) - } - - async fn ensure_directory(&self, path: &Path) -> StorageMediatorResult<()> { - let abs_path = - PathService::resolve_path(self, &StoragePath::from_string(&path.to_string_lossy())); - - if !abs_path.exists() { - fs::create_dir_all(&abs_path).await.map_err(|e| { - StorageMediatorError::AccessError(format!("Failed to create directory: {}", e)) - })?; - } else if !abs_path.is_dir() { - return Err(StorageMediatorError::InvalidPath(format!( - "Path exists but is not a directory: {}", - abs_path.display() - ))); - } - - Ok(()) - } - - async fn ensure_storage_directory( - &self, - storage_path: &StoragePath, - ) -> StorageMediatorResult<()> { - let abs_path = PathService::resolve_path(self, storage_path); - - if !abs_path.exists() { - fs::create_dir_all(&abs_path).await.map_err(|e| { - StorageMediatorError::AccessError(format!("Failed to create directory: {}", e)) - })?; - } else if !abs_path.is_dir() { - return Err(StorageMediatorError::InvalidPath(format!( - "Path exists but is not a directory: {}", - abs_path.display() - ))); - } - - Ok(()) - } -} - #[cfg(test)] mod tests { use super::*; diff --git a/src/infrastructure/services/write_behind_cache.rs b/src/infrastructure/services/write_behind_cache.rs deleted file mode 100644 index f1c8751d..00000000 --- a/src/infrastructure/services/write_behind_cache.rs +++ /dev/null @@ -1,501 +0,0 @@ -// ═══════════════════════════════════════════════════════════════════════════════ -// WRITE-BEHIND CACHE - Zero-latency uploads for small files -// ═══════════════════════════════════════════════════════════════════════════════ -// -// Strategy: -// 1. For files < 1MB, store in RAM and respond immediately (201 Created) -// 2. Flush to disk asynchronously in background -// 3. Serve reads from cache while pending flush -// 4. On read miss, check if pending then serve from cache -// -// This gives users perceived ~0ms upload latency for small files -// ═══════════════════════════════════════════════════════════════════════════════ - -use async_trait::async_trait; -use bytes::Bytes; -use std::collections::HashMap; -use std::path::PathBuf; -use std::sync::Arc; -use std::time::{Duration, Instant}; -use tokio::fs; -use tokio::io::AsyncWriteExt; -use tokio::sync::{RwLock, mpsc}; - -use crate::application::ports::cache_ports::{WriteBehindCachePort, WriteBehindStatsDto}; -use crate::domain::errors::DomainError; - -/// Maximum size for write-behind cache (files larger bypass cache) -const WRITE_BEHIND_MAX_SIZE: usize = 1024 * 1024; // 1MB - -/// Maximum total cache size in bytes -const MAX_CACHE_SIZE: usize = 100 * 1024 * 1024; // 100MB total - -/// Maximum time a file can stay pending before forced flush -const MAX_PENDING_DURATION: Duration = Duration::from_secs(30); - -/// Flush check interval -const FLUSH_INTERVAL: Duration = Duration::from_millis(100); - -/// Entry in the write-behind cache -#[derive(Clone)] -pub struct PendingWrite { - /// File content - pub content: Bytes, - /// Target path on disk - pub target_path: PathBuf, - /// When this entry was created - pub created_at: Instant, - /// File ID for tracking - pub file_id: String, -} - -/// Statistics for monitoring -#[derive(Debug, Clone, Default)] -pub struct WriteBehindStats { - pub pending_count: usize, - pub pending_bytes: usize, - pub total_writes: u64, - pub total_bytes_written: u64, - pub cache_hits: u64, - pub avg_flush_time_us: u64, -} - -/// Write-Behind Cache for zero-latency small file uploads -pub struct WriteBehindCache { - /// Pending writes indexed by file ID - pending: Arc>>, - /// Current total size of pending data - current_size: Arc>, - /// Channel to signal flush worker - flush_tx: mpsc::Sender, - /// Statistics - stats: Arc>, -} - -/// Commands for the flush worker -enum FlushCommand { - /// Flush a specific file - FlushFile(String), - /// Flush all pending files - FlushAll, - /// Shutdown the worker - Shutdown, -} - -impl WriteBehindCache { - /// Create a new write-behind cache with background flush worker - pub fn new() -> Arc { - let (flush_tx, flush_rx) = mpsc::channel(1000); - - let cache = Arc::new(Self { - pending: Arc::new(RwLock::new(HashMap::new())), - current_size: Arc::new(RwLock::new(0)), - flush_tx, - stats: Arc::new(RwLock::new(WriteBehindStats::default())), - }); - - // Start the background flush worker - let cache_clone = cache.clone(); - tokio::spawn(async move { - cache_clone.flush_worker(flush_rx).await; - }); - - // Start the periodic flush checker - let cache_clone2 = cache.clone(); - tokio::spawn(async move { - cache_clone2.periodic_flush_checker().await; - }); - - tracing::info!( - "⚡ Write-Behind Cache initialized (max {}MB)", - MAX_CACHE_SIZE / (1024 * 1024) - ); - - cache - } - - /// Check if a file size is eligible for write-behind caching - #[inline] - pub fn is_eligible(size: usize) -> bool { - size <= WRITE_BEHIND_MAX_SIZE - } - - /// Put a file in the pending write cache - /// Returns Ok(true) if cached, Ok(false) if cache is full - pub async fn put_pending( - &self, - file_id: String, - content: Bytes, - target_path: PathBuf, - ) -> Result { - let content_size = content.len(); - - // Check if we have space - { - let current = *self.current_size.read().await; - if current + content_size > MAX_CACHE_SIZE { - tracing::debug!( - "Write-behind cache full ({}/{}MB), bypassing for {}", - current / (1024 * 1024), - MAX_CACHE_SIZE / (1024 * 1024), - file_id - ); - return Ok(false); - } - } - - // Add to pending - let entry = PendingWrite { - content, - target_path, - created_at: Instant::now(), - file_id: file_id.clone(), - }; - - { - let mut pending = self.pending.write().await; - let mut size = self.current_size.write().await; - - // If replacing existing entry, adjust size - if let Some(old) = pending.insert(file_id.clone(), entry) { - *size -= old.content.len(); - } - *size += content_size; - } - - // Update stats - { - let mut stats = self.stats.write().await; - stats.pending_count += 1; - stats.pending_bytes += content_size; - } - - // Signal flush worker (non-blocking) - let _ = self - .flush_tx - .try_send(FlushCommand::FlushFile(file_id.clone())); - - tracing::debug!( - "⚡ Cached pending write: {} ({} bytes)", - file_id, - content_size - ); - - Ok(true) - } - - /// Get content from cache if pending (for reads before flush completes) - pub async fn get_pending(&self, file_id: &str) -> Option { - let pending = self.pending.read().await; - if let Some(entry) = pending.get(file_id) { - // Update cache hit stats - let mut stats = self.stats.write().await; - stats.cache_hits += 1; - - tracing::debug!("⚡ Cache hit for pending file: {}", file_id); - return Some(entry.content.clone()); - } - None - } - - /// Check if a file is pending flush - pub async fn is_pending(&self, file_id: &str) -> bool { - self.pending.read().await.contains_key(file_id) - } - - /// Force immediate flush of a specific file (for critical operations) - pub async fn force_flush(&self, file_id: &str) -> Result<(), std::io::Error> { - let entry = { - let pending = self.pending.read().await; - pending.get(file_id).cloned() - }; - - if let Some(entry) = entry { - self.flush_single(&entry.file_id, &entry).await?; - } - - Ok(()) - } - - /// Flush all pending writes immediately - pub async fn flush_all(&self) -> Result<(), std::io::Error> { - let _ = self.flush_tx.send(FlushCommand::FlushAll).await; - - // Wait a bit for flush to complete - tokio::time::sleep(Duration::from_millis(50)).await; - - Ok(()) - } - - /// Gracefully shutdown the write-behind cache - /// Flushes all pending writes before stopping the background worker - pub async fn shutdown(&self) -> Result<(), std::io::Error> { - tracing::info!("🛑 Shutting down write-behind cache..."); - - // First flush all pending writes - self.flush_all().await?; - - // Then signal the worker to stop - let _ = self.flush_tx.send(FlushCommand::Shutdown).await; - - // Give worker time to process shutdown - tokio::time::sleep(Duration::from_millis(100)).await; - - tracing::info!("✅ Write-behind cache shutdown complete"); - Ok(()) - } - - /// Get current statistics - pub async fn get_stats(&self) -> WriteBehindStats { - self.stats.read().await.clone() - } - - /// Background worker that handles actual disk writes - async fn flush_worker(&self, mut rx: mpsc::Receiver) { - tracing::info!("🔄 Write-behind flush worker started"); - - while let Some(cmd) = rx.recv().await { - match cmd { - FlushCommand::FlushFile(file_id) => { - // Small delay to batch nearby writes - tokio::time::sleep(Duration::from_millis(10)).await; - - let entry = { - let pending = self.pending.read().await; - pending.get(&file_id).cloned() - }; - - if let Some(entry) = entry - && let Err(e) = self.flush_single(&file_id, &entry).await - { - tracing::error!("Failed to flush {}: {}", file_id, e); - // Keep in cache for retry - continue; - } - } - FlushCommand::FlushAll => { - let entries: Vec<_> = { - let pending = self.pending.read().await; - pending - .iter() - .map(|(k, v)| (k.clone(), v.clone())) - .collect() - }; - - for (file_id, entry) in entries { - if let Err(e) = self.flush_single(&file_id, &entry).await { - tracing::error!("Failed to flush {}: {}", file_id, e); - } - } - } - FlushCommand::Shutdown => { - tracing::info!("Write-behind flush worker shutting down"); - break; - } - } - } - } - - /// Flush a single file to disk - async fn flush_single( - &self, - file_id: &str, - entry: &PendingWrite, - ) -> Result<(), std::io::Error> { - let start = Instant::now(); - - // Ensure parent directory exists - if let Some(parent) = entry.target_path.parent() { - fs::create_dir_all(parent).await?; - } - - // Write atomically using temp file + rename - let temp_path = entry.target_path.with_extension("tmp"); - - { - let mut file = fs::File::create(&temp_path).await?; - file.write_all(&entry.content).await?; - file.sync_all().await?; - } - - fs::rename(&temp_path, &entry.target_path).await?; - - let elapsed = start.elapsed(); - let content_len = entry.content.len(); - - // Remove from pending - { - let mut pending = self.pending.write().await; - let mut size = self.current_size.write().await; - - if pending.remove(file_id).is_some() { - *size = size.saturating_sub(content_len); - } - } - - // Update stats - { - let mut stats = self.stats.write().await; - stats.pending_count = stats.pending_count.saturating_sub(1); - stats.pending_bytes = stats.pending_bytes.saturating_sub(content_len); - stats.total_writes += 1; - stats.total_bytes_written += content_len as u64; - - // Running average of flush time - let flush_us = elapsed.as_micros() as u64; - if stats.avg_flush_time_us == 0 { - stats.avg_flush_time_us = flush_us; - } else { - stats.avg_flush_time_us = (stats.avg_flush_time_us * 9 + flush_us) / 10; - } - } - - tracing::debug!( - "💾 Flushed {} to disk ({} bytes in {:?})", - file_id, - content_len, - elapsed - ); - - Ok(()) - } - - /// Periodic checker for stale pending writes - async fn periodic_flush_checker(&self) { - let mut interval = tokio::time::interval(FLUSH_INTERVAL); - - loop { - interval.tick().await; - - let stale_files: Vec = { - let pending = self.pending.read().await; - pending - .iter() - .filter(|(_, entry)| entry.created_at.elapsed() > MAX_PENDING_DURATION) - .map(|(id, _)| id.clone()) - .collect() - }; - - for file_id in stale_files { - tracing::warn!("Forcing flush of stale pending file: {}", file_id); - let _ = self.flush_tx.try_send(FlushCommand::FlushFile(file_id)); - } - } - } -} - -// ─── Port implementation ───────────────────────────────────────────────────── - -#[async_trait] -impl WriteBehindCachePort for WriteBehindCache { - fn is_eligible_size(&self, size: usize) -> bool { - WriteBehindCache::is_eligible(size) - } - - async fn put_pending( - &self, - file_id: String, - content: Bytes, - target_path: PathBuf, - ) -> Result { - self.put_pending(file_id, content, target_path) - .await - .map_err(DomainError::from) - } - - async fn get_pending(&self, file_id: &str) -> Option { - self.get_pending(file_id).await - } - - async fn is_pending(&self, file_id: &str) -> bool { - self.is_pending(file_id).await - } - - async fn force_flush(&self, file_id: &str) -> Result<(), DomainError> { - self.force_flush(file_id).await.map_err(DomainError::from) - } - - async fn flush_all(&self) -> Result<(), DomainError> { - self.flush_all().await.map_err(DomainError::from) - } - - async fn shutdown(&self) -> Result<(), DomainError> { - self.shutdown().await.map_err(DomainError::from) - } - - async fn get_stats(&self) -> WriteBehindStatsDto { - let stats = self.get_stats().await; - WriteBehindStatsDto { - pending_count: stats.pending_count, - pending_bytes: stats.pending_bytes, - total_writes: stats.total_writes, - total_bytes_written: stats.total_bytes_written, - cache_hits: stats.cache_hits, - avg_flush_time_us: stats.avg_flush_time_us, - } - } -} - -impl Default for WriteBehindCache { - fn default() -> Self { - // Note: This creates a non-Arc version, prefer using new() - let (flush_tx, _) = mpsc::channel(1); - Self { - pending: Arc::new(RwLock::new(HashMap::new())), - current_size: Arc::new(RwLock::new(0)), - flush_tx, - stats: Arc::new(RwLock::new(WriteBehindStats::default())), - } - } -} - -#[cfg(test)] -mod tests { - use super::*; - use tempfile::TempDir; - - #[tokio::test] - async fn test_write_behind_basic() { - let cache = WriteBehindCache::new(); - let temp_dir = TempDir::new().unwrap(); - let target = temp_dir.path().join("test.txt"); - - let content = Bytes::from("Hello, World!"); - - // Put in cache - let cached = cache - .put_pending("test-id".to_string(), content.clone(), target.clone()) - .await - .unwrap(); - - assert!(cached); - assert!(cache.is_pending("test-id").await); - - // Should be readable from cache - let cached_content = cache.get_pending("test-id").await.unwrap(); - assert_eq!(cached_content, content); - - // Force flush - cache.force_flush("test-id").await.unwrap(); - - // Should no longer be pending - assert!(!cache.is_pending("test-id").await); - - // File should exist on disk - assert!(target.exists()); - let disk_content = std::fs::read(&target).unwrap(); - assert_eq!(disk_content, content.as_ref()); - } - - #[tokio::test] - async fn test_eligibility() { - // 500KB should be eligible - assert!(WriteBehindCache::is_eligible(500 * 1024)); - - // 1MB exactly should be eligible - assert!(WriteBehindCache::is_eligible(1024 * 1024)); - - // Over 1MB should not be eligible - assert!(!WriteBehindCache::is_eligible(1024 * 1024 + 1)); - } -}