feat(#113): 100% blob storage model — PostgreSQL metadata + DedupService blobs

BREAKING CHANGE: Storage model completely rewritten. All file/folder
metadata now lives in PostgreSQL (storage schema). File content stored
as content-addressable blobs via DedupService. Filesystem directories
are no longer used for user storage.

New components:
- storage.folders / storage.files / storage.trash_items (PG schema)
- FolderDbRepository: virtual folders backed by PG
- FileBlobReadRepository: file reads via PG metadata + dedup blobs
- FileBlobWriteRepository: file writes via PG metadata + dedup blobs
- TrashDbRepository: soft-delete trash using is_trashed flags

Removed legacy FS components (~5500 lines deleted):
- FolderFsRepository, FileFsReadRepository, FileFsWriteRepository
- CompositeFileRepository, ParallelFileProcessor
- IdMappingService, IdMappingOptimizer, FileMetadataCache
- BufferPool, FileSystemUtils, RepositoryErrors
- TrashFsRepository, FolderFsRepositoryTrash

DI rewired: build_app_state() now requires PgPool (no FS fallback).
FileUploadService.new_with_read() and FileRetrievalService.new_with_cache()
constructors added for blob model (no write-behind needed).

Closes #113
This commit is contained in:
Dionisio
2026-02-14 17:54:25 +01:00
parent f25987e553
commit 3c7c16f07e
27 changed files with 1883 additions and 7429 deletions
@@ -1,157 +0,0 @@
use std::path::PathBuf;
use std::sync::Arc;
use async_trait::async_trait;
use bytes::Bytes;
use futures::Stream;
use crate::application::ports::storage_ports::{FileReadPort, FileWritePort};
use crate::common::errors::DomainError;
use crate::domain::entities::file::File;
use crate::domain::services::path_service::StoragePath;
/// Composite that wraps `Arc<dyn FileReadPort>` + `Arc<dyn FileWritePort>`
/// and delegates each method to the corresponding port.
///
/// Thanks to the blanket impl `impl<T: FileReadPort + FileWritePort> FileStoragePort for T {}`
/// this type automatically gets `FileStoragePort`.
pub struct CompositeFileRepository {
read: Arc<dyn FileReadPort>,
write: Arc<dyn FileWritePort>,
}
impl CompositeFileRepository {
pub fn new(read: Arc<dyn FileReadPort>, write: Arc<dyn FileWritePort>) -> Self {
Self { read, write }
}
}
// ─────────────────────────────────────────────────────
// FileReadPort — delegate to self.read
// ─────────────────────────────────────────────────────
#[async_trait]
impl FileReadPort for CompositeFileRepository {
async fn get_file(&self, id: &str) -> Result<File, DomainError> {
self.read.get_file(id).await
}
async fn list_files(&self, folder_id: Option<&str>) -> Result<Vec<File>, DomainError> {
self.read.list_files(folder_id).await
}
async fn get_file_content(&self, id: &str) -> Result<Vec<u8>, DomainError> {
self.read.get_file_content(id).await
}
async fn get_file_stream(
&self,
id: &str,
) -> Result<Box<dyn Stream<Item = Result<Bytes, std::io::Error>> + Send>, DomainError> {
self.read.get_file_stream(id).await
}
async fn get_file_range_stream(
&self,
id: &str,
start: u64,
end: Option<u64>,
) -> Result<Box<dyn Stream<Item = Result<Bytes, std::io::Error>> + Send>, DomainError> {
self.read.get_file_range_stream(id, start, end).await
}
async fn get_file_mmap(&self, id: &str) -> Result<Bytes, DomainError> {
self.read.get_file_mmap(id).await
}
async fn get_file_path(&self, id: &str) -> Result<StoragePath, DomainError> {
self.read.get_file_path(id).await
}
async fn get_parent_folder_id(&self, path: &str) -> Result<String, DomainError> {
self.read.get_parent_folder_id(path).await
}
}
// ─────────────────────────────────────────────────────
// FileWritePort — delegate to self.write
// ─────────────────────────────────────────────────────
#[async_trait]
impl FileWritePort for CompositeFileRepository {
async fn save_file(
&self,
name: String,
folder_id: Option<String>,
content_type: String,
content: Vec<u8>,
) -> Result<File, DomainError> {
self.write
.save_file(name, folder_id, content_type, content)
.await
}
async fn save_file_from_stream(
&self,
name: String,
folder_id: Option<String>,
content_type: String,
stream: std::pin::Pin<Box<dyn Stream<Item = Result<Bytes, std::io::Error>> + Send>>,
) -> Result<File, DomainError> {
self.write
.save_file_from_stream(name, folder_id, content_type, stream)
.await
}
async fn move_file(
&self,
file_id: &str,
target_folder_id: Option<String>,
) -> Result<File, DomainError> {
self.write.move_file(file_id, target_folder_id).await
}
async fn rename_file(&self, file_id: &str, new_name: &str) -> Result<File, DomainError> {
self.write.rename_file(file_id, new_name).await
}
async fn delete_file(&self, id: &str) -> Result<(), DomainError> {
self.write.delete_file(id).await
}
async fn update_file_content(
&self,
file_id: &str,
content: Vec<u8>,
) -> Result<(), DomainError> {
self.write.update_file_content(file_id, content).await
}
async fn register_file_deferred(
&self,
name: String,
folder_id: Option<String>,
content_type: String,
size: u64,
) -> Result<(File, PathBuf), DomainError> {
self.write
.register_file_deferred(name, folder_id, content_type, size)
.await
}
async fn move_to_trash(&self, file_id: &str) -> Result<(), DomainError> {
self.write.move_to_trash(file_id).await
}
async fn restore_from_trash(
&self,
file_id: &str,
original_path: &str,
) -> Result<(), DomainError> {
self.write.restore_from_trash(file_id, original_path).await
}
async fn delete_file_permanently(&self, file_id: &str) -> Result<(), DomainError> {
self.write.delete_file_permanently(file_id).await
}
}
@@ -1,473 +0,0 @@
use std::path::PathBuf;
use std::sync::Arc;
use async_trait::async_trait;
use bytes::Bytes;
use futures::{Stream, StreamExt};
use mime_guess::from_path;
use tokio::fs::File as TokioFile;
use tokio::task;
use tokio::{fs, time};
use tokio_util::codec::{BytesCodec, FramedRead};
use crate::application::ports::cache_ports::MetadataCachePort;
use crate::application::ports::storage_ports::FileReadPort;
use crate::application::services::storage_mediator::StorageMediator;
use crate::common::config::AppConfig;
use crate::common::errors::DomainError;
use crate::domain::entities::file::File;
use crate::domain::services::path_service::StoragePath;
use crate::infrastructure::repositories::parallel_file_processor::ParallelFileProcessor;
use crate::infrastructure::repositories::repository_errors::{
FileRepositoryError, FileRepositoryResult,
};
use crate::infrastructure::services::path_service::PathService;
/// Repository implementation for file **read** operations.
///
/// Implements `FileReadPort`:
/// get_file, list_files, get_file_content, get_file_stream,
/// get_file_range_stream, get_file_mmap, get_file_path, get_parent_folder_id.
pub struct FileFsReadRepository {
root_path: PathBuf,
storage_mediator: Arc<dyn StorageMediator>,
id_mapping_service: Arc<dyn crate::application::ports::outbound::IdMappingPort>,
path_service: Arc<PathService>,
metadata_cache: Arc<dyn MetadataCachePort>,
config: AppConfig,
parallel_processor: Option<Arc<ParallelFileProcessor>>,
}
impl FileFsReadRepository {
/// Full constructor with all infrastructure dependencies.
pub fn new(
root_path: PathBuf,
storage_mediator: Arc<dyn StorageMediator>,
id_mapping_service: Arc<dyn crate::application::ports::outbound::IdMappingPort>,
path_service: Arc<PathService>,
metadata_cache: Arc<dyn MetadataCachePort>,
config: AppConfig,
parallel_processor: Option<Arc<ParallelFileProcessor>>,
) -> Self {
Self {
root_path,
storage_mediator,
id_mapping_service,
path_service,
metadata_cache,
config,
parallel_processor,
}
}
/// Stub for testing (does not perform real I/O).
pub fn default_stub() -> Self {
Self {
root_path: PathBuf::from("./storage"),
storage_mediator: Arc::new(
crate::application::services::storage_mediator::FileSystemStorageMediator::new_stub(
),
),
id_mapping_service: Arc::new(crate::common::stubs::StubIdMappingPort),
path_service: Arc::new(PathService::new(PathBuf::from("./storage"))),
metadata_cache: Arc::new(
crate::infrastructure::services::file_metadata_cache::FileMetadataCache::default(),
) as Arc<dyn MetadataCachePort>,
config: AppConfig::default(),
parallel_processor: None,
}
}
// ─── internal helpers ─────────────────────────────────────
fn resolve_storage_path(&self, storage_path: &StoragePath) -> PathBuf {
self.path_service.resolve_path(storage_path)
}
async fn get_file_metadata_raw(
&self,
abs_path: &PathBuf,
) -> FileRepositoryResult<(u64, u64, u64)> {
// Cache first
if let Some(cached) = self.metadata_cache.get_metadata(abs_path).await
&& let (Some(s), Some(c), Some(m)) =
(cached.size, cached.created_at, cached.modified_at)
{
return Ok((s, c, m));
}
let metadata = time::timeout(self.config.timeouts.file_timeout(), fs::metadata(abs_path))
.await
.map_err(|_| {
FileRepositoryError::StorageError(format!(
"Timeout metadata: {}",
abs_path.display()
))
})?
.map_err(|e| FileRepositoryError::StorageError(e.to_string()))?;
let size = metadata.len();
let created_at = metadata
.created()
.map(|t| {
t.duration_since(std::time::UNIX_EPOCH)
.unwrap_or_default()
.as_secs()
})
.unwrap_or(0);
let modified_at = metadata
.modified()
.map(|t| {
t.duration_since(std::time::UNIX_EPOCH)
.unwrap_or_default()
.as_secs()
})
.unwrap_or(0);
let _ = self.metadata_cache.refresh_metadata(abs_path).await;
Ok((size, created_at, modified_at))
}
async fn get_file_by_id(&self, id: &str) -> FileRepositoryResult<File> {
let storage_path = self
.id_mapping_service
.get_path_by_id(id)
.await
.map_err(|e| FileRepositoryError::Other(e.to_string()))?;
let abs_path = self.resolve_storage_path(&storage_path);
if !abs_path.exists() || !abs_path.is_file() {
return Err(FileRepositoryError::NotFound(format!(
"File {} not found at {}",
id,
storage_path.to_string()
)));
}
let (size, created_at, modified_at) = self.get_file_metadata_raw(&abs_path).await?;
let name = storage_path
.file_name()
.ok_or_else(|| FileRepositoryError::InvalidPath(storage_path.to_string()))?;
let mime_type = from_path(&abs_path).first_or_octet_stream().to_string();
File::with_timestamps(
id.to_string(),
name,
storage_path,
size,
mime_type,
None,
created_at,
modified_at,
)
.map_err(|e| FileRepositoryError::Other(e.to_string()))
}
}
impl Clone for FileFsReadRepository {
fn clone(&self) -> Self {
Self {
root_path: self.root_path.clone(),
storage_mediator: self.storage_mediator.clone(),
id_mapping_service: self.id_mapping_service.clone(),
path_service: self.path_service.clone(),
metadata_cache: self.metadata_cache.clone(),
config: self.config.clone(),
parallel_processor: self.parallel_processor.clone(),
}
}
}
#[async_trait]
impl FileReadPort for FileFsReadRepository {
async fn get_file(&self, id: &str) -> Result<File, DomainError> {
self.get_file_by_id(id).await.map_err(|e| match e {
FileRepositoryError::NotFound(msg) => DomainError::not_found("File", msg),
FileRepositoryError::StorageError(msg) => DomainError::internal_error("File", msg),
other => DomainError::internal_error("File", other.to_string()),
})
}
async fn list_files(&self, folder_id: Option<&str>) -> Result<Vec<File>, DomainError> {
let folder_storage_path = match folder_id {
Some(id) => match self.storage_mediator.get_folder_path(id).await {
Ok(path) => {
let lossy = path.to_string_lossy().to_string();
let folder_name = path.file_name().and_then(|f| f.to_str()).unwrap_or(&lossy);
StoragePath::from_string(folder_name)
}
Err(_) => return Ok(Vec::new()),
},
None => StoragePath::root(),
};
let abs_folder_path = self.path_service.resolve_path(&folder_storage_path);
if !abs_folder_path.exists() || !abs_folder_path.is_dir() {
return Ok(Vec::new());
}
let mut files_result = Vec::new();
let mut entries = fs::read_dir(&abs_folder_path)
.await
.map_err(|e| DomainError::internal_error("File", e.to_string()))?;
while let Some(entry) = entries
.next_entry()
.await
.map_err(|e| DomainError::internal_error("File", e.to_string()))?
{
let path = entry.path();
if !path.is_file() {
continue;
}
let file_name = entry.file_name().to_string_lossy().to_string();
if file_name.starts_with('.')
|| file_name == "folder_ids.json"
|| file_name == "file_ids.json"
{
continue;
}
let metadata = match fs::metadata(&path).await {
Ok(m) => m,
Err(_) => continue,
};
let file_storage_path = folder_storage_path.join(&file_name);
let id = match self
.id_mapping_service
.get_or_create_id(&file_storage_path)
.await
{
Ok(id) => id,
Err(_) => continue,
};
let size = metadata.len();
let created_at = metadata
.created()
.map(|t| {
t.duration_since(std::time::UNIX_EPOCH)
.unwrap_or_default()
.as_secs()
})
.unwrap_or(0);
let modified_at = metadata
.modified()
.map(|t| {
t.duration_since(std::time::UNIX_EPOCH)
.unwrap_or_default()
.as_secs()
})
.unwrap_or(0);
let mime_type = from_path(&path).first_or_octet_stream().to_string();
match File::with_timestamps(
id,
file_name,
file_storage_path,
size,
mime_type,
folder_id.map(String::from),
created_at,
modified_at,
) {
Ok(file) => files_result.push(file),
Err(_) => continue,
}
}
// Persist any new ID mappings
let _ = self.id_mapping_service.save_changes().await;
Ok(files_result)
}
async fn get_file_content(&self, id: &str) -> Result<Vec<u8>, DomainError> {
let file = self
.get_file_by_id(id)
.await
.map_err(|e| DomainError::internal_error("File", e.to_string()))?;
let abs_path = self.resolve_storage_path(file.storage_path());
let metadata = time::timeout(self.config.timeouts.file_timeout(), fs::metadata(&abs_path))
.await
.map_err(|_| {
DomainError::internal_error(
"File",
format!("Timeout metadata: {}", abs_path.display()),
)
})?
.map_err(|e| DomainError::internal_error("File", e.to_string()))?;
let file_size = metadata.len();
if !self.config.resources.can_load_in_memory(file_size) {
return Err(DomainError::internal_error(
"File",
format!(
"File too large for memory: {} MB",
file_size / (1024 * 1024)
),
));
}
// Parallel read for very large files
if self
.config
.resources
.needs_parallel_processing(file_size, &self.config.concurrency)
{
let content = if let Some(processor) = &self.parallel_processor {
processor.read_file_parallel(&abs_path).await
} else {
let processor = ParallelFileProcessor::new(self.config.clone());
processor.read_file_parallel(&abs_path).await
};
return content.map_err(|e| DomainError::internal_error("File", e.to_string()));
}
// spawn_blocking for large-ish files
if self.config.resources.is_large_file(file_size) {
let abs_clone = abs_path.clone();
let chunk_size = self.config.resources.chunk_size_bytes;
let content = task::spawn_blocking(move || -> std::io::Result<Vec<u8>> {
use std::io::{BufReader, Read};
let file = std::fs::File::open(&abs_clone)?;
let mut reader = BufReader::with_capacity(chunk_size, file);
let mut buf = Vec::with_capacity(file_size as usize);
reader.read_to_end(&mut buf)?;
Ok(buf)
})
.await
.map_err(|e| DomainError::internal_error("File", e.to_string()))?
.map_err(|e| DomainError::internal_error("File", e.to_string()))?;
return Ok(content);
}
// Small files — async read
time::timeout(self.config.timeouts.file_timeout(), fs::read(&abs_path))
.await
.map_err(|_| {
DomainError::internal_error(
"File",
format!("Timeout reading: {}", abs_path.display()),
)
})?
.map_err(|e| DomainError::internal_error("File", e.to_string()))
}
async fn get_file_stream(
&self,
id: &str,
) -> Result<Box<dyn Stream<Item = Result<Bytes, std::io::Error>> + Send>, DomainError> {
let file = self
.get_file_by_id(id)
.await
.map_err(|e| DomainError::internal_error("File", e.to_string()))?;
let abs_path = self.resolve_storage_path(file.storage_path());
let metadata = time::timeout(self.config.timeouts.file_timeout(), fs::metadata(&abs_path))
.await
.map_err(|_| DomainError::internal_error("File", "Timeout getting metadata"))?
.map_err(|e| DomainError::internal_error("File", e.to_string()))?;
let file_size = metadata.len();
let is_large = self.config.resources.is_large_file(file_size);
let fh = time::timeout(
self.config.timeouts.file_timeout(),
TokioFile::open(&abs_path),
)
.await
.map_err(|_| DomainError::internal_error("File", "Timeout opening file"))?
.map_err(|e| DomainError::internal_error("File", e.to_string()))?;
let chunk_size = if is_large {
self.config.resources.chunk_size_bytes
} else {
4096
};
let codec = BytesCodec::new();
let stream =
FramedRead::with_capacity(fh, codec, chunk_size).map(|r| r.map(|bm| bm.freeze()));
Ok(Box::new(stream))
}
async fn get_file_range_stream(
&self,
id: &str,
start: u64,
end: Option<u64>,
) -> Result<Box<dyn Stream<Item = Result<Bytes, std::io::Error>> + Send>, DomainError> {
use tokio::io::AsyncSeekExt;
let file = self
.get_file_by_id(id)
.await
.map_err(|e| DomainError::internal_error("File", e.to_string()))?;
let abs_path = self.resolve_storage_path(file.storage_path());
let metadata = time::timeout(self.config.timeouts.file_timeout(), fs::metadata(&abs_path))
.await
.map_err(|_| DomainError::internal_error("File", "Timeout"))?
.map_err(|e| DomainError::internal_error("File", e.to_string()))?;
let file_size = metadata.len();
if start >= file_size {
return Err(DomainError::internal_error(
"File",
format!("Range start {} beyond file size {}", start, file_size),
));
}
let actual_end = end.map(|e| e.min(file_size - 1)).unwrap_or(file_size - 1);
let range_length = actual_end - start + 1;
let mut fh = time::timeout(
self.config.timeouts.file_timeout(),
TokioFile::open(&abs_path),
)
.await
.map_err(|_| DomainError::internal_error("File", "Timeout opening file"))?
.map_err(|e| DomainError::internal_error("File", e.to_string()))?;
fh.seek(std::io::SeekFrom::Start(start))
.await
.map_err(|e| DomainError::internal_error("File", e.to_string()))?;
let chunk_size = if range_length > 1024 * 1024 {
self.config.resources.chunk_size_bytes
} else {
8192
};
use tokio::io::AsyncReadExt;
let limited = fh.take(range_length);
let codec = BytesCodec::new();
let stream =
FramedRead::with_capacity(limited, codec, chunk_size).map(|r| r.map(|bm| bm.freeze()));
Ok(Box::new(stream))
}
async fn get_file_mmap(&self, id: &str) -> Result<Bytes, DomainError> {
use memmap2::Mmap;
let file = self
.get_file_by_id(id)
.await
.map_err(|e| DomainError::internal_error("File", e.to_string()))?;
let abs_path = self.resolve_storage_path(file.storage_path());
let path_clone = abs_path.clone();
task::spawn_blocking(move || -> Result<Bytes, DomainError> {
let fh = std::fs::File::open(&path_clone)
.map_err(|e| DomainError::internal_error("File", e.to_string()))?;
let mmap = unsafe { Mmap::map(&fh) }
.map_err(|e| DomainError::internal_error("File", e.to_string()))?;
Ok(Bytes::copy_from_slice(&mmap[..]))
})
.await
.map_err(|e| DomainError::internal_error("File", e.to_string()))?
}
async fn get_file_path(&self, id: &str) -> Result<StoragePath, DomainError> {
self.id_mapping_service.get_path_by_id(id).await
}
async fn get_parent_folder_id(&self, path: &str) -> Result<String, DomainError> {
let storage_path = StoragePath::from_string(path);
match storage_path.parent() {
Some(parent) if !parent.is_empty() => {
self.id_mapping_service.get_or_create_id(&parent).await
}
_ => Ok("root".to_string()),
}
}
}
@@ -1,800 +0,0 @@
use async_trait::async_trait;
use bytes::Bytes;
use futures::{Stream, StreamExt};
use mime_guess::from_path;
use std::path::PathBuf;
use std::sync::Arc;
use tokio::fs::File as TokioFile;
use tokio::io::AsyncWriteExt;
use tokio::task;
use tokio::{fs, time};
use crate::application::ports::cache_ports::MetadataCachePort;
use crate::application::ports::storage_ports::FileWritePort;
use crate::application::services::storage_mediator::StorageMediator;
use crate::common::config::AppConfig;
use crate::common::errors::DomainError;
use crate::domain::entities::file::File;
use crate::domain::services::path_service::StoragePath;
use crate::infrastructure::repositories::parallel_file_processor::ParallelFileProcessor;
use crate::infrastructure::repositories::repository_errors::{
FileRepositoryError, FileRepositoryResult,
};
use crate::infrastructure::services::file_system_utils::FileSystemUtils;
use crate::infrastructure::services::path_service::PathService;
/// Repository implementation for file **write** operations.
///
/// Implements `FileWritePort`:
/// save_file, save_file_from_stream, move_file, delete_file,
/// update_file_content, register_file_deferred.
pub struct FileFsWriteRepository {
root_path: PathBuf,
storage_mediator: Arc<dyn StorageMediator>,
id_mapping_service: Arc<dyn crate::application::ports::outbound::IdMappingPort>,
path_service: Arc<PathService>,
metadata_cache: Arc<dyn MetadataCachePort>,
config: AppConfig,
parallel_processor: Option<Arc<ParallelFileProcessor>>,
}
impl FileFsWriteRepository {
/// Full constructor with all dependencies.
pub fn new(
root_path: PathBuf,
storage_mediator: Arc<dyn StorageMediator>,
id_mapping_service: Arc<dyn crate::application::ports::outbound::IdMappingPort>,
path_service: Arc<PathService>,
metadata_cache: Arc<dyn MetadataCachePort>,
config: AppConfig,
parallel_processor: Option<Arc<ParallelFileProcessor>>,
) -> Self {
Self {
root_path,
storage_mediator,
id_mapping_service,
path_service,
metadata_cache,
config,
parallel_processor,
}
}
/// Stub for testing (does not perform real I/O).
pub fn default_stub() -> Self {
Self {
root_path: PathBuf::from("./storage"),
storage_mediator: Arc::new(
crate::application::services::storage_mediator::FileSystemStorageMediator::new_stub(
),
),
id_mapping_service: Arc::new(crate::common::stubs::StubIdMappingPort),
path_service: Arc::new(PathService::new(PathBuf::from("./storage"))),
metadata_cache: Arc::new(
crate::infrastructure::services::file_metadata_cache::FileMetadataCache::default(),
) as Arc<dyn MetadataCachePort>,
config: AppConfig::default(),
parallel_processor: None,
}
}
// ─── helpers ─────────────────────────────────────────────
fn resolve_storage_path(&self, storage_path: &StoragePath) -> PathBuf {
self.path_service.resolve_path(storage_path)
}
async fn ensure_parent_directory(&self, abs_path: &PathBuf) -> FileRepositoryResult<()> {
if let Some(parent) = abs_path.parent() {
time::timeout(
self.config.timeouts.dir_timeout(),
FileSystemUtils::create_dir_with_sync(parent),
)
.await
.map_err(|_| {
FileRepositoryError::StorageError(format!(
"Timeout creating dir: {}",
parent.display()
))
})?
.map_err(|e| FileRepositoryError::StorageError(e.to_string()))?;
}
Ok(())
}
async fn file_exists_at_storage_path(
&self,
storage_path: &StoragePath,
) -> FileRepositoryResult<bool> {
let abs = self.resolve_storage_path(storage_path);
if let Some(is_file) = self.metadata_cache.is_file(&abs).await {
return Ok(is_file);
}
match time::timeout(self.config.timeouts.file_timeout(), fs::metadata(&abs)).await {
Ok(Ok(m)) => {
let _ = self.metadata_cache.refresh_metadata(&abs).await;
Ok(m.is_file())
}
Ok(Err(_)) => Ok(false),
Err(_) => Err(FileRepositoryError::StorageError(format!(
"Timeout: {}",
abs.display()
))),
}
}
async fn get_file_metadata_raw(
&self,
abs_path: &PathBuf,
) -> FileRepositoryResult<(u64, u64, u64)> {
if let Some(cached) = self.metadata_cache.get_metadata(abs_path).await
&& let (Some(s), Some(c), Some(m)) =
(cached.size, cached.created_at, cached.modified_at)
{
return Ok((s, c, m));
}
let meta = time::timeout(self.config.timeouts.file_timeout(), fs::metadata(abs_path))
.await
.map_err(|_| {
FileRepositoryError::StorageError(format!("Timeout: {}", abs_path.display()))
})?
.map_err(|e| FileRepositoryError::StorageError(e.to_string()))?;
let s = meta.len();
let c = meta
.created()
.map(|t| {
t.duration_since(std::time::UNIX_EPOCH)
.unwrap_or_default()
.as_secs()
})
.unwrap_or(0);
let m = meta
.modified()
.map(|t| {
t.duration_since(std::time::UNIX_EPOCH)
.unwrap_or_default()
.as_secs()
})
.unwrap_or(0);
let _ = self.metadata_cache.refresh_metadata(abs_path).await;
Ok((s, c, m))
}
/// Resolve folder to StoragePath
async fn resolve_folder_path(&self, folder_id: &Option<String>) -> StoragePath {
match folder_id {
Some(id) => match self.storage_mediator.get_folder_path(id).await {
Ok(path) => {
let lossy = path.to_string_lossy().to_string();
let folder_name = path.file_name().and_then(|f| f.to_str()).unwrap_or(&lossy);
StoragePath::from_string(folder_name)
}
Err(_) => StoragePath::root(),
},
None => StoragePath::root(),
}
}
/// Generate unique file path avoiding name collisions.
async fn unique_file_path(
&self,
folder_path: &StoragePath,
name: &str,
) -> FileRepositoryResult<(StoragePath, String)> {
let mut file_path = folder_path.join(name);
let mut actual_name = name.to_string();
let mut counter = 1;
while self.file_exists_at_storage_path(&file_path).await? {
let (stem, ext) = if let Some(dot) = name.rfind('.') {
(name[..dot].to_string(), name[dot..].to_string())
} else {
(name.to_string(), String::new())
};
actual_name = format!("{}_{}{}", stem, counter, ext);
file_path = folder_path.join(&actual_name);
counter += 1;
}
Ok((file_path, actual_name))
}
async fn delete_file_non_blocking(&self, abs_path: PathBuf) -> FileRepositoryResult<()> {
let file_size = match fs::metadata(&abs_path).await {
Ok(m) => m.len(),
Err(_) => 0,
};
if self.config.resources.is_large_file(file_size) {
task::spawn_blocking(move || {
let _ = std::fs::remove_file(&abs_path);
})
.await
.map_err(|e| FileRepositoryError::Other(e.to_string()))?;
} else {
time::timeout(
self.config.timeouts.file_timeout(),
fs::remove_file(&abs_path),
)
.await
.map_err(|_| FileRepositoryError::StorageError("Timeout deleting file".into()))?
.map_err(|e| FileRepositoryError::StorageError(e.to_string()))?;
}
Ok(())
}
/// Persist ID mapping with retry + verification.
async fn persist_id_mapping(&self, id: &str, expected_path: &str) -> FileRepositoryResult<()> {
for attempt in 1..=3 {
match self.id_mapping_service.save_changes().await {
Ok(_) => {
if let Ok(verified) = self.id_mapping_service.get_path_by_id(id).await
&& verified.to_string() == expected_path
{
return Ok(());
}
if attempt == 3 {
return Err(FileRepositoryError::Other(
"Failed to verify ID mapping after 3 attempts".into(),
));
}
tokio::time::sleep(tokio::time::Duration::from_millis(100)).await;
}
Err(e) if attempt < 3 => {
tokio::time::sleep(tokio::time::Duration::from_millis(100)).await;
tracing::warn!("ID mapping save retry {}: {}", attempt, e);
}
Err(e) => {
return Err(FileRepositoryError::Other(format!(
"Save ID mapping failed: {}",
e
)));
}
}
}
Ok(())
}
}
impl Clone for FileFsWriteRepository {
fn clone(&self) -> Self {
Self {
root_path: self.root_path.clone(),
storage_mediator: self.storage_mediator.clone(),
id_mapping_service: self.id_mapping_service.clone(),
path_service: self.path_service.clone(),
metadata_cache: self.metadata_cache.clone(),
config: self.config.clone(),
parallel_processor: self.parallel_processor.clone(),
}
}
}
fn map_repo_err(e: FileRepositoryError) -> DomainError {
match e {
FileRepositoryError::NotFound(m) => DomainError::not_found("File", m),
FileRepositoryError::AlreadyExists(m) => DomainError::already_exists("File", m),
FileRepositoryError::StorageError(m) => DomainError::internal_error("File", m),
other => DomainError::internal_error("File", other.to_string()),
}
}
#[async_trait]
impl FileWritePort for FileFsWriteRepository {
async fn save_file(
&self,
name: String,
folder_id: Option<String>,
content_type: String,
content: Vec<u8>,
) -> Result<File, DomainError> {
let folder_path = self.resolve_folder_path(&folder_id).await;
let (file_storage_path, actual_name) = self
.unique_file_path(&folder_path, &name)
.await
.map_err(map_repo_err)?;
let abs_path = self.resolve_storage_path(&file_storage_path);
self.ensure_parent_directory(&abs_path)
.await
.map_err(map_repo_err)?;
let content_size = content.len() as u64;
// Write strategy based on file size
if self
.config
.resources
.needs_parallel_processing(content_size, &self.config.concurrency)
{
if let Some(proc) = &self.parallel_processor {
proc.write_file_parallel(&abs_path, &content)
.await
.map_err(map_repo_err)?;
} else {
let proc = ParallelFileProcessor::new(self.config.clone());
proc.write_file_parallel(&abs_path, &content)
.await
.map_err(map_repo_err)?;
}
} else if content_size > self.config.resources.large_file_threshold_mb * 1024 * 1024 {
let mut fh = time::timeout(
self.config.timeouts.file_timeout(),
TokioFile::create(&abs_path),
)
.await
.map_err(|_| DomainError::internal_error("File", "Timeout creating file"))?
.map_err(|e| DomainError::internal_error("File", e.to_string()))?;
let chunk_size = self.config.resources.chunk_size_bytes;
for chunk in content.chunks(chunk_size) {
fh.write_all(chunk)
.await
.map_err(|e| DomainError::internal_error("File", e.to_string()))?;
}
fh.flush()
.await
.map_err(|e| DomainError::internal_error("File", e.to_string()))?;
} else {
let mut fh = time::timeout(
self.config.timeouts.file_timeout(),
TokioFile::create(&abs_path),
)
.await
.map_err(|_| DomainError::internal_error("File", "Timeout creating file"))?
.map_err(|e| DomainError::internal_error("File", e.to_string()))?;
fh.write_all(&content)
.await
.map_err(|e| DomainError::internal_error("File", e.to_string()))?;
fh.flush()
.await
.map_err(|e| DomainError::internal_error("File", e.to_string()))?;
}
let (size, created_at, modified_at) = self
.get_file_metadata_raw(&abs_path)
.await
.map_err(map_repo_err)?;
let mime = if content_type.is_empty() {
from_path(&abs_path).first_or_octet_stream().to_string()
} else {
content_type
};
let id = self
.id_mapping_service
.get_or_create_id(&file_storage_path)
.await
.map_err(|e| DomainError::internal_error("File", e.to_string()))?;
let path_string = file_storage_path.to_string();
let file = File::with_timestamps(
id.clone(),
actual_name,
file_storage_path,
size,
mime,
folder_id,
created_at,
modified_at,
)
.map_err(|e| DomainError::internal_error("File", e.to_string()))?;
self.persist_id_mapping(&id, &path_string)
.await
.map_err(map_repo_err)?;
if let Some(parent) = abs_path.parent() {
self.metadata_cache.invalidate_directory(parent).await;
}
Ok(file)
}
async fn save_file_from_stream(
&self,
name: String,
folder_id: Option<String>,
content_type: String,
mut stream: std::pin::Pin<Box<dyn Stream<Item = Result<Bytes, std::io::Error>> + Send>>,
) -> Result<File, DomainError> {
let folder_path = self.resolve_folder_path(&folder_id).await;
let (file_storage_path, actual_name) = self
.unique_file_path(&folder_path, &name)
.await
.map_err(map_repo_err)?;
let abs_path = self.resolve_storage_path(&file_storage_path);
self.ensure_parent_directory(&abs_path)
.await
.map_err(map_repo_err)?;
let temp_path = abs_path.with_extension("tmp.upload");
let mut fh = time::timeout(
self.config.timeouts.file_timeout(),
TokioFile::create(&temp_path),
)
.await
.map_err(|_| DomainError::internal_error("File", "Timeout creating temp file"))?
.map_err(|e| DomainError::internal_error("File", e.to_string()))?;
let mut total_bytes: u64 = 0;
while let Some(chunk_result) = stream.next().await {
let chunk =
chunk_result.map_err(|e| DomainError::internal_error("File", e.to_string()))?;
fh.write_all(&chunk)
.await
.map_err(|e| DomainError::internal_error("File", e.to_string()))?;
total_bytes += chunk.len() as u64;
}
fh.flush()
.await
.map_err(|e| DomainError::internal_error("File", e.to_string()))?;
fh.sync_all()
.await
.map_err(|e| DomainError::internal_error("File", e.to_string()))?;
drop(fh);
// Atomic rename
fs::rename(&temp_path, &abs_path)
.await
.map_err(|e| DomainError::internal_error("File", e.to_string()))?;
let (size, created_at, modified_at) = self
.get_file_metadata_raw(&abs_path)
.await
.map_err(map_repo_err)?;
let mime = if content_type.is_empty() {
from_path(&abs_path).first_or_octet_stream().to_string()
} else {
content_type
};
let id = self
.id_mapping_service
.get_or_create_id(&file_storage_path)
.await
.map_err(|e| DomainError::internal_error("File", e.to_string()))?;
let path_string = file_storage_path.to_string();
let log_name = actual_name.clone();
let file = File::with_timestamps(
id.clone(),
actual_name,
file_storage_path,
size,
mime,
folder_id,
created_at,
modified_at,
)
.map_err(|e| DomainError::internal_error("File", e.to_string()))?;
self.persist_id_mapping(&id, &path_string)
.await
.map_err(map_repo_err)?;
if let Some(parent) = abs_path.parent() {
self.metadata_cache.invalidate_directory(parent).await;
}
tracing::info!(
"✅ STREAMING UPLOAD COMPLETE: {} ({} bytes)",
log_name,
total_bytes
);
Ok(file)
}
async fn move_file(
&self,
file_id: &str,
target_folder_id: Option<String>,
) -> Result<File, DomainError> {
// Get original file
let original_path = self.id_mapping_service.get_path_by_id(file_id).await?;
let old_abs = self.resolve_storage_path(&original_path);
if !old_abs.exists() || !old_abs.is_file() {
return Err(DomainError::not_found("File", file_id.to_string()));
}
let (size, created_at, modified_at) = self
.get_file_metadata_raw(&old_abs)
.await
.map_err(map_repo_err)?;
let name = original_path
.file_name()
.ok_or_else(|| DomainError::internal_error("File", "Invalid path"))?;
let mime = from_path(&old_abs).first_or_octet_stream().to_string();
// Build target path
let target_folder_path = self.resolve_folder_path(&target_folder_id).await;
let new_storage_path = target_folder_path.join(&name);
if self
.file_exists_at_storage_path(&new_storage_path)
.await
.map_err(map_repo_err)?
{
return Err(DomainError::already_exists(
"File",
format!("File already exists at {}", new_storage_path.to_string()),
));
}
let new_abs = self.resolve_storage_path(&new_storage_path);
self.ensure_parent_directory(&new_abs)
.await
.map_err(map_repo_err)?;
// Rename
time::timeout(
self.config.timeouts.file_timeout(),
FileSystemUtils::rename_with_sync(&old_abs, &new_abs),
)
.await
.map_err(|_| DomainError::internal_error("File", "Timeout moving file"))?
.map_err(|e| DomainError::internal_error("File", e.to_string()))?;
// Update mapping
self.id_mapping_service
.update_path(file_id, &new_storage_path)
.await?;
let _ = self.id_mapping_service.save_changes().await;
File::with_timestamps(
file_id.to_string(),
name,
new_storage_path,
size,
mime,
target_folder_id,
created_at,
modified_at,
)
.map_err(|e| DomainError::internal_error("File", e.to_string()))
}
async fn rename_file(&self, file_id: &str, new_name: &str) -> Result<File, DomainError> {
// 1. Get current file info
let original_path = self.id_mapping_service.get_path_by_id(file_id).await?;
let old_abs = self.resolve_storage_path(&original_path);
if !old_abs.exists() || !old_abs.is_file() {
return Err(DomainError::not_found("File", file_id.to_string()));
}
let (size, created_at, modified_at) = self
.get_file_metadata_raw(&old_abs)
.await
.map_err(map_repo_err)?;
// 2. Build new path (same parent directory, different filename)
let parent = original_path
.parent()
.unwrap_or_else(|| StoragePath::new(vec![]));
let new_storage_path = parent.join(new_name);
if self
.file_exists_at_storage_path(&new_storage_path)
.await
.map_err(map_repo_err)?
{
return Err(DomainError::already_exists(
"File",
format!("File already exists: {}", new_name),
));
}
let new_abs = self.resolve_storage_path(&new_storage_path);
let mime = from_path(&new_abs).first_or_octet_stream().to_string();
// 3. Rename on disk
time::timeout(
self.config.timeouts.file_timeout(),
FileSystemUtils::rename_with_sync(&old_abs, &new_abs),
)
.await
.map_err(|_| DomainError::internal_error("File", "Timeout renaming file"))?
.map_err(|e| DomainError::internal_error("File", e.to_string()))?;
// 4. Update id→path mapping
self.id_mapping_service
.update_path(file_id, &new_storage_path)
.await?;
let _ = self.id_mapping_service.save_changes().await;
File::with_timestamps(
file_id.to_string(),
new_name.to_string(),
new_storage_path,
size,
mime,
None,
created_at,
modified_at,
)
.map_err(|e| DomainError::internal_error("File", e.to_string()))
}
async fn delete_file(&self, id: &str) -> Result<(), DomainError> {
let storage_path = self.id_mapping_service.get_path_by_id(id).await?;
let abs_path = self.resolve_storage_path(&storage_path);
self.metadata_cache.invalidate(&abs_path).await;
if let Some(parent) = abs_path.parent() {
self.metadata_cache.invalidate_directory(parent).await;
}
self.delete_file_non_blocking(abs_path)
.await
.map_err(map_repo_err)?;
// Clean up the ID mapping so we don't leave orphaned entries
if let Err(e) = self.id_mapping_service.remove_id(id).await {
tracing::warn!("Failed to remove ID mapping for deleted file {}: {}", id, e);
}
let _ = self.id_mapping_service.save_changes().await;
Ok(())
}
async fn update_file_content(
&self,
file_id: &str,
content: Vec<u8>,
) -> Result<(), DomainError> {
let storage_path = self.id_mapping_service.get_path_by_id(file_id).await?;
let physical_path = self.resolve_storage_path(&storage_path);
FileSystemUtils::atomic_write(&physical_path, &content)
.await
.map_err(|e| DomainError::internal_error("File", e.to_string()))?;
// Refresh cache
let _ = self.metadata_cache.refresh_metadata(&physical_path).await;
Ok(())
}
async fn register_file_deferred(
&self,
name: String,
folder_id: Option<String>,
content_type: String,
size: u64,
) -> Result<(File, PathBuf), DomainError> {
let folder_path = self.resolve_folder_path(&folder_id).await;
let (file_storage_path, actual_name) = self
.unique_file_path(&folder_path, &name)
.await
.map_err(map_repo_err)?;
let abs_path = self.resolve_storage_path(&file_storage_path);
self.ensure_parent_directory(&abs_path)
.await
.map_err(map_repo_err)?;
let mime = if content_type.is_empty() {
from_path(&abs_path).first_or_octet_stream().to_string()
} else {
content_type
};
let now = std::time::SystemTime::now()
.duration_since(std::time::UNIX_EPOCH)
.unwrap_or_default()
.as_secs();
let id = self
.id_mapping_service
.get_or_create_id(&file_storage_path)
.await
.map_err(|e| DomainError::internal_error("File", e.to_string()))?;
let _ = self.id_mapping_service.save_changes().await;
let file = File::with_timestamps(
id.clone(),
actual_name,
file_storage_path,
size,
mime,
folder_id,
now,
now,
)
.map_err(|e| DomainError::internal_error("File", e.to_string()))?;
tracing::debug!("⚡ Registered deferred file: {} -> {:?}", id, abs_path);
Ok((file, abs_path))
}
async fn move_to_trash(&self, file_id: &str) -> Result<(), DomainError> {
// Get the file's current path
let storage_path = self.id_mapping_service.get_path_by_id(file_id).await?;
let abs_path = self.resolve_storage_path(&storage_path);
if !abs_path.exists() || !abs_path.is_file() {
return Err(DomainError::not_found("File", file_id.to_string()));
}
// Create trash directory
let trash_dir = self.root_path.join(".trash").join("files");
fs::create_dir_all(&trash_dir).await.map_err(|e| {
DomainError::internal_error("File", format!("Failed to create trash dir: {}", e))
})?;
// Move file to trash
let trash_path = trash_dir.join(file_id);
fs::rename(&abs_path, &trash_path).await.map_err(|e| {
DomainError::internal_error("File", format!("Failed to move file to trash: {}", e))
})?;
// Update mapping to trash location
let trash_storage_path = StoragePath::from_string(&format!(".trash/files/{}", file_id));
self.id_mapping_service
.update_path(file_id, &trash_storage_path)
.await?;
let _ = self.id_mapping_service.save_changes().await;
// Invalidate cache
self.metadata_cache.invalidate(&abs_path).await;
if let Some(parent) = abs_path.parent() {
self.metadata_cache.invalidate_directory(parent).await;
}
tracing::debug!(
"File moved to trash: {} -> {}",
file_id,
trash_path.display()
);
Ok(())
}
async fn restore_from_trash(
&self,
file_id: &str,
original_path: &str,
) -> Result<(), DomainError> {
// Get current path (should be in trash)
let current_storage_path = self.id_mapping_service.get_path_by_id(file_id).await?;
let current_abs_path = self.resolve_storage_path(&current_storage_path);
if !current_abs_path.exists() {
return Err(DomainError::not_found(
"File",
format!("File {} not found in trash", file_id),
));
}
// Ensure parent directory exists for original location
let original_storage_path = StoragePath::from_string(original_path);
let original_abs_path = self.resolve_storage_path(&original_storage_path);
if let Some(parent) = original_abs_path.parent() {
fs::create_dir_all(parent).await.map_err(|e| {
DomainError::internal_error("File", format!("Failed to create parent dir: {}", e))
})?;
}
// Move file back to original location
fs::rename(&current_abs_path, &original_abs_path)
.await
.map_err(|e| {
DomainError::internal_error("File", format!("Failed to restore file: {}", e))
})?;
// Update mapping back to original path
self.id_mapping_service
.update_path(file_id, &original_storage_path)
.await?;
let _ = self.id_mapping_service.save_changes().await;
tracing::debug!(
"File restored from trash: {} -> {}",
file_id,
original_abs_path.display()
);
Ok(())
}
async fn delete_file_permanently(&self, file_id: &str) -> Result<(), DomainError> {
// Get current path (could be in trash or original location)
let storage_path = self.id_mapping_service.get_path_by_id(file_id).await?;
let abs_path = self.resolve_storage_path(&storage_path);
// Delete the physical file if it exists
if abs_path.exists() {
self.delete_file_non_blocking(abs_path.clone())
.await
.map_err(map_repo_err)?;
}
// Remove ID mapping
self.id_mapping_service.remove_id(file_id).await?;
let _ = self.id_mapping_service.save_changes().await;
// Invalidate cache
self.metadata_cache.invalidate(&abs_path).await;
tracing::debug!("File permanently deleted: {}", file_id);
Ok(())
}
}
File diff suppressed because it is too large Load Diff
@@ -1,192 +0,0 @@
use std::path::PathBuf;
use tokio::fs;
use tracing::{debug, error};
use crate::infrastructure::repositories::folder_fs_repository::FolderFsRepository;
use crate::infrastructure::repositories::repository_errors::FolderRepositoryResult;
// This file contains the implementation of trash-related methods
// for the FolderFsRepository folder repository
// Implementation of trash methods for the folder repository
impl FolderFsRepository {
// Gets the full path to the trash directory
fn get_trash_dir(&self) -> PathBuf {
self.get_root_path().join(".trash").join("folders")
}
// Creates a unique path in the trash for the folder
async fn create_trash_folder_path(&self, folder_id: &str) -> FolderRepositoryResult<PathBuf> {
let trash_dir = self.get_trash_dir();
// Ensure the trash directory exists
if !trash_dir.exists() {
fs::create_dir_all(&trash_dir)
.await
.map_err(|e| FolderRepositoryError::StorageError(e.to_string()))?;
}
// Create a unique path for the folder in the trash
Ok(trash_dir.join(folder_id))
}
}
// Implementation of public FolderRepository trait methods related to trash
// Implementation of internal methods for trash functionality
// These will be enabled when the trash feature is re-enabled
impl FolderFsRepository {
/// Helper method that will be used for trash functionality
pub(crate) async fn _trash_move_to_trash(&self, folder_id: &str) -> FolderRepositoryResult<()> {
debug!("Moving folder to trash: {}", folder_id);
// Get the physical path of the folder
let folder_path = match self.get_mapped_folder_path(folder_id).await {
Ok(path) => path,
Err(e) => {
error!("Error getting folder path {}: {:?}", folder_id, e);
return Err(e);
}
};
let folder_path_buf = PathBuf::from(folder_path.to_string());
// Verify the folder exists
if !folder_path_buf.exists() {
return Err(FolderRepositoryError::NotFound(format!(
"Folder not found: {}",
folder_id
)));
}
// Create directory in the trash
let trash_folder_path = self.create_trash_folder_path(folder_id).await?;
// Physically move the folder to the trash
match fs::rename(&folder_path_buf, &trash_folder_path).await {
Ok(_) => {
debug!(
"Folder moved to trash: {} -> {}",
folder_path_buf.display(),
trash_folder_path.display()
);
// Update the mapping to the new path in the trash
if let Err(e) = self
.update_mapped_folder_path(folder_id, &trash_folder_path)
.await
{
error!("Error updating folder mapping in trash: {}", e);
return Err(e);
}
Ok(())
}
Err(e) => {
error!("Error moving folder to trash: {}", e);
Err(FolderRepositoryError::StorageError(e.to_string()))
}
}
}
/// Restores a folder from the trash to its original location
pub(crate) async fn _trash_restore_from_trash(
&self,
folder_id: &str,
original_path: &str,
) -> FolderRepositoryResult<()> {
debug!("Restoring folder {} to {}", folder_id, original_path);
// Get the current path in the trash
let current_path = match self.get_mapped_folder_path(folder_id).await {
Ok(path) => PathBuf::from(path),
Err(e) => {
error!("Error getting current folder path {}: {:?}", folder_id, e);
return Err(e);
}
};
// Convert the original path to PathBuf
let original_path_buf = PathBuf::from(original_path);
// Ensure the destination parent directory exists
if let Some(parent) = original_path_buf.parent()
&& !parent.exists()
{
fs::create_dir_all(parent).await.map_err(|e| {
error!("Error creating parent directory for restoration: {}", e);
FolderRepositoryError::StorageError(e.to_string())
})?;
}
// Move the folder from the trash to its original location
match fs::rename(&current_path, &original_path_buf).await {
Ok(_) => {
debug!(
"Folder restored: {} -> {}",
current_path.display(),
original_path_buf.display()
);
// Update the mapping to the original path
if let Err(e) = self
.update_mapped_folder_path(folder_id, &original_path_buf)
.await
{
error!("Error updating restored folder mapping: {}", e);
return Err(e);
}
Ok(())
}
Err(e) => {
error!("Error restoring folder: {}", e);
Err(FolderRepositoryError::StorageError(e.to_string()))
}
}
}
/// Permanently deletes a folder (used by the trash)
pub(crate) async fn _trash_delete_folder_permanently(
&self,
folder_id: &str,
) -> FolderRepositoryResult<()> {
debug!("Permanently deleting folder: {}", folder_id);
// Similar to delete_folder but without additional validations
let folder_path = match self.get_mapped_folder_path(folder_id).await {
Ok(path) => PathBuf::from(path),
Err(e) => {
error!("Error getting folder path {}: {:?}", folder_id, e);
return Err(e);
}
};
// Delete the folder recursively
if folder_path.exists() {
match fs::remove_dir_all(&folder_path).await {
Ok(_) => {
debug!("Folder permanently deleted: {}", folder_path.display());
}
Err(e) => {
error!("Error permanently deleting folder: {}", e);
// Don't report error if the folder no longer exists
if e.kind() != std::io::ErrorKind::NotFound {
return Err(FolderRepositoryError::StorageError(e.to_string()));
}
}
}
}
// Remove the mapping
if let Err(e) = self.remove_mapped_folder_id(folder_id).await {
error!("Error removing folder mapping: {}", e);
return Err(e);
}
debug!("Folder permanently deleted successfully: {}", folder_id);
Ok(())
}
}
// Re-exports needed by the compiler
use crate::infrastructure::repositories::repository_errors::FolderRepositoryError;
+5 -16
View File
@@ -1,21 +1,10 @@
pub mod folder_fs_repository;
pub mod parallel_file_processor;
pub mod repository_errors;
// Repositorios CQRS (Read/Write) + composite
pub mod composite_file_repository;
pub mod file_fs_read_repository;
pub mod file_fs_write_repository;
pub mod folder_fs_repository_trash;
pub mod share_fs_repository;
pub mod trash_fs_repository;
// Repositorios PostgreSQL
// Repositorios PostgreSQL (blob-storage model)
pub mod pg;
// Re-exportar para facilitar acceso
pub use composite_file_repository::CompositeFileRepository;
pub use file_fs_read_repository::FileFsReadRepository;
pub use file_fs_write_repository::FileFsWriteRepository;
pub use pg::{SessionPgRepository, UserPgRepository};
pub use pg::{
FileBlobReadRepository, FileBlobWriteRepository, FolderDbRepository,
SessionPgRepository, TrashDbRepository, UserPgRepository,
};
@@ -1,571 +0,0 @@
use bytes::{Bytes, BytesMut};
use futures::future::join_all;
use std::io::{self, SeekFrom};
use std::path::PathBuf;
use std::sync::Arc;
use tokio::fs::File;
use tokio::io::{AsyncReadExt, AsyncSeekExt, AsyncWriteExt};
use tokio::sync::{Mutex, Semaphore};
use tokio::task;
use tracing::{debug, error, info};
use crate::common::config::AppConfig;
use crate::infrastructure::repositories::repository_errors::FileRepositoryError;
use crate::infrastructure::services::buffer_pool::BufferPool;
/// Structure for the byte range to process
#[derive(Debug, Clone, Copy)]
pub struct ChunkRange {
/// Chunk index
pub index: usize,
/// Start position in bytes
pub start: u64,
/// Chunk size in bytes
pub size: usize,
}
/// Specific buffer pooling for BytesMut
pub struct BytesBufferPool {
buffers: Mutex<Vec<BytesMut>>,
buffer_size: usize,
max_buffers: usize,
}
impl BytesBufferPool {
pub fn new(buffer_size: usize, max_buffers: usize) -> Self {
Self {
buffers: Mutex::new(Vec::with_capacity(max_buffers)),
buffer_size,
max_buffers,
}
}
/// Get a buffer from the pool or create a new one
pub async fn get_buffer(&self) -> BytesMut {
let mut buffers = self.buffers.lock().await;
if let Some(mut buffer) = buffers.pop() {
// Reuse existing buffer
buffer.clear(); // Keep capacity, clear content
buffer
} else {
// Create new buffer if the pool is empty
BytesMut::with_capacity(self.buffer_size)
}
}
/// Return a buffer to the pool for reuse
pub async fn return_buffer(&self, mut buffer: BytesMut) {
// Reset the buffer for reuse
buffer.clear();
let mut buffers = self.buffers.lock().await;
// Only keep up to max_buffers
if buffers.len() < self.max_buffers {
buffers.push(buffer);
}
// If we already have enough buffers, this one will be discarded
}
}
/// Parallel file processor for IO-intensive operations
pub struct ParallelFileProcessor {
/// Application configuration
config: AppConfig,
/// Semaphore to limit global concurrency
concurrency_limiter: Arc<Semaphore>,
/// Buffer pool to optimize memory
buffer_pool: Option<Arc<BufferPool>>,
/// BytesMut buffer pool for zero-copy operations
bytes_pool: Arc<BytesBufferPool>,
}
impl ParallelFileProcessor {
/// Creates a new processor instance
pub fn new(config: AppConfig) -> Self {
let concurrency_limiter = Arc::new(Semaphore::new(config.concurrency.max_concurrent_io));
// Create BytesMut pool for efficient operations
let chunk_size = config.resources.chunk_size_bytes;
let max_chunks = config.concurrency.max_parallel_chunks;
let bytes_pool = Arc::new(BytesBufferPool::new(chunk_size, max_chunks * 2));
Self {
config,
concurrency_limiter,
buffer_pool: None,
bytes_pool,
}
}
/// Creates a new processor instance with a buffer pool
pub fn new_with_buffer_pool(config: AppConfig, buffer_pool: Arc<BufferPool>) -> Self {
let concurrency_limiter = Arc::new(Semaphore::new(config.concurrency.max_concurrent_io));
// Create BytesMut pool for efficient operations
let chunk_size = config.resources.chunk_size_bytes;
let max_chunks = config.concurrency.max_parallel_chunks;
let bytes_pool = Arc::new(BytesBufferPool::new(chunk_size, max_chunks * 2));
Self {
config,
concurrency_limiter,
buffer_pool: Some(buffer_pool),
bytes_pool,
}
}
/// Divides a file into chunks for parallel processing
pub fn calculate_chunks(&self, file_size: u64) -> Vec<ChunkRange> {
// Determine if the file needs parallel processing
let needs_parallel = self
.config
.resources
.needs_parallel_processing(file_size, &self.config.concurrency);
if !needs_parallel {
// For small files, use a single chunk
return vec![ChunkRange {
index: 0,
start: 0,
size: file_size as usize,
}];
}
// Calculate optimal number of chunks
let chunk_count = self
.config
.resources
.calculate_optimal_chunks(file_size, &self.config.concurrency);
// Calculate size of each chunk
let chunk_size = self
.config
.resources
.calculate_chunk_size(file_size, chunk_count);
// Create chunk ranges
let mut chunks = Vec::with_capacity(chunk_count);
let mut start = 0;
for i in 0..chunk_count {
let current_chunk_size = if i == chunk_count - 1 {
// Last chunk might be smaller
(file_size - start) as usize
} else {
chunk_size
};
chunks.push(ChunkRange {
index: i,
start,
size: current_chunk_size,
});
start += current_chunk_size as u64;
}
debug!(
"File size: {} bytes, divided into {} chunks of ~{} bytes each",
file_size,
chunks.len(),
chunk_size
);
chunks
}
/// Reads a file in parallel and returns the complete content
/// Optimized implementation using BytesMut to reduce memory copies
pub async fn read_file_parallel(
&self,
file_path: &PathBuf,
) -> Result<Vec<u8>, FileRepositoryError> {
// Get file size
let metadata = tokio::fs::metadata(file_path)
.await
.map_err(|e| FileRepositoryError::StorageError(e.to_string()))?;
let file_size = metadata.len();
// Check if the file is too large for memory
if !self.config.resources.can_load_in_memory(file_size) {
return Err(FileRepositoryError::Other(format!(
"File too large to load in memory: {} MB (max: {} MB)",
file_size / (1024 * 1024),
self.config.resources.max_in_memory_file_size_mb
)));
}
// Calculate chunks
let chunks = self.calculate_chunks(file_size);
if chunks.len() == 1 {
// For a single chunk, use simple reading with buffer pool if available
info!(
"Reading file with size {}MB as a single chunk",
file_size / (1024 * 1024)
);
if let Some(pool) = &self.buffer_pool {
// Use buffer from the pool for efficient reading
debug!("Using buffer pool for single chunk read");
let mut buffer = pool.get_buffer().await;
// If the buffer is too small, revert to standard implementation
if buffer.capacity() < file_size as usize {
debug!(
"Buffer from pool too small ({}), using standard read",
buffer.capacity()
);
let content = tokio::fs::read(file_path)
.await
.map_err(|e| FileRepositoryError::StorageError(e.to_string()))?;
return Ok(content);
}
// Use memory buffer from the pool
let mut file = File::open(file_path)
.await
.map_err(|e| FileRepositoryError::StorageError(e.to_string()))?;
let read_size = file
.read(buffer.as_mut_slice())
.await
.map_err(|e| FileRepositoryError::StorageError(e.to_string()))?;
buffer.set_used(read_size);
// Convert to Vec<u8>
let content = buffer.into_vec();
return Ok(content);
} else {
// Standard implementation without pool
let content = tokio::fs::read(file_path)
.await
.map_err(|e| FileRepositoryError::StorageError(e.to_string()))?;
return Ok(content);
}
}
// For multiple chunks, use parallel reading
info!(
"Reading file with size {}MB in {} parallel chunks using BytesMut",
file_size / (1024 * 1024),
chunks.len()
);
// Create final result buffer (pre-allocated)
let mut result = BytesMut::with_capacity(file_size as usize);
result.resize(file_size as usize, 0);
let result_mutex = Arc::new(Mutex::new(result));
// Create tasks for each chunk
let mut tasks = Vec::with_capacity(chunks.len());
// Open file once and share it
let file = Arc::new(
File::open(file_path)
.await
.map_err(|e| FileRepositoryError::StorageError(e.to_string()))?,
);
// Reference to BytesMut pool
let bytes_pool = self.bytes_pool.clone();
// Process chunks in parallel
for chunk in chunks {
let file_clone = file.clone();
let result_clone = result_mutex.clone();
let semaphore_clone = self.concurrency_limiter.clone();
let bytes_pool_clone = bytes_pool.clone();
// Spawn task for this chunk - no need to copy the original data
let task = task::spawn(async move {
// Acquire semaphore permit
let _permit = semaphore_clone.acquire().await.unwrap();
// Get a reusable buffer from the BytesMut pool
let mut chunk_buffer = bytes_pool_clone.get_buffer().await;
// Ensure it has sufficient capacity
if chunk_buffer.capacity() < chunk.size {
chunk_buffer = BytesMut::with_capacity(chunk.size);
}
// Resize to the exact size needed
chunk_buffer.resize(chunk.size, 0);
// Create a duplicate file descriptor for independent use
let mut file_handle = file_clone.try_clone().await?;
// Position and read directly into the BytesMut
file_handle.seek(SeekFrom::Start(chunk.start)).await?;
let bytes_read = file_handle
.read_exact(&mut chunk_buffer[..chunk.size])
.await?;
if bytes_read != chunk.size {
return Err(io::Error::new(
io::ErrorKind::UnexpectedEof,
format!(
"Expected to read {} bytes but got {}",
chunk.size, bytes_read
),
));
}
// Write to final result
let mut result_lock = result_clone.lock().await;
let start_pos = chunk.start as usize;
let end_pos = start_pos + chunk.size;
// Use copy_from_slice to copy from BytesMut to result buffer
result_lock[start_pos..end_pos].copy_from_slice(&chunk_buffer[..chunk.size]);
// Return the buffer to the pool for reuse
bytes_pool_clone.return_buffer(chunk_buffer).await;
// Log progress
debug!(
"Chunk {} processed: {} bytes from offset {}",
chunk.index, chunk.size, chunk.start
);
Ok::<_, io::Error>(())
});
tasks.push(task);
}
// Wait for all tasks to complete
let results = join_all(tasks).await;
// Check for errors
for (i, task_result) in results.into_iter().enumerate() {
match task_result {
Ok(Ok(())) => {}
Ok(Err(e)) => {
error!("Error in chunk {}: {}", i, e);
return Err(FileRepositoryError::StorageError(e.to_string()));
}
Err(e) => {
error!("Task error in chunk {}: {}", i, e);
return Err(FileRepositoryError::Other(format!("Task error: {}", e)));
}
}
}
// Get the final result and convert to Vec<u8>
let result_buffer = result_mutex.lock().await;
let result_vec = result_buffer.to_vec();
info!(
"Successfully read file of {}MB in parallel with optimized BytesMut",
file_size / (1024 * 1024)
);
Ok(result_vec)
}
/// Writes a file in parallel from a buffer
/// Optimized implementation using BytesMut/Bytes to reduce memory copies
pub async fn write_file_parallel(
&self,
file_path: &PathBuf,
content: &[u8],
) -> Result<(), FileRepositoryError> {
let file_size = content.len() as u64;
// Calculate chunks
let chunks = self.calculate_chunks(file_size);
if chunks.len() == 1 {
// For a single chunk, use simple writing
info!(
"Writing file with size {}MB as a single chunk",
file_size / (1024 * 1024)
);
// Standard implementation (buffer pooling offers no advantages for simple writing)
tokio::fs::write(file_path, content)
.await
.map_err(|e| FileRepositoryError::StorageError(e.to_string()))?;
return Ok(());
}
// For multiple chunks, use parallel writing
info!(
"Writing file with size {}MB in {} parallel chunks using Bytes",
file_size / (1024 * 1024),
chunks.len()
);
// Create file (we don't use Mutex to reduce contention)
let file = File::create(file_path)
.await
.map_err(|e| FileRepositoryError::StorageError(e.to_string()))?;
// Convert content to Bytes (single copy step)
let content_bytes = Bytes::copy_from_slice(content);
// Create tasks for each chunk
let mut tasks = Vec::with_capacity(chunks.len());
// Process chunks in parallel
for chunk in chunks {
let file_clone = file
.try_clone()
.await
.map_err(|e| FileRepositoryError::StorageError(e.to_string()))?;
let semaphore_clone = self.concurrency_limiter.clone();
// Create Bytes slice (doesn't copy data, only references)
let start_idx = chunk.start as usize;
let end_idx = start_idx + chunk.size;
let chunk_data = content_bytes.slice(start_idx..end_idx);
// Create and launch task
let task = task::spawn(async move {
// Acquire semaphore permit
let _permit = semaphore_clone.acquire().await.unwrap();
// Position and write
let mut file_handle = file_clone;
file_handle.seek(SeekFrom::Start(chunk.start)).await?;
file_handle.write_all(&chunk_data).await?;
// Log progress
debug!(
"Chunk {} written: {} bytes at offset {}",
chunk.index, chunk.size, chunk.start
);
Ok::<_, io::Error>(())
});
tasks.push(task);
}
// Wait for all tasks to complete
let results = join_all(tasks).await;
// Check for errors
for (i, task_result) in results.into_iter().enumerate() {
match task_result {
Ok(Ok(())) => {}
Ok(Err(e)) => {
error!("Error in chunk {}: {}", i, e);
return Err(FileRepositoryError::StorageError(e.to_string()));
}
Err(e) => {
error!("Task error in chunk {}: {}", i, e);
return Err(FileRepositoryError::Other(format!("Task error: {}", e)));
}
}
}
// Ensure everything has been written correctly
let mut file_handle = file;
file_handle
.flush()
.await
.map_err(|e| FileRepositoryError::StorageError(e.to_string()))?;
info!(
"Successfully wrote file of {}MB in parallel with optimized Bytes",
file_size / (1024 * 1024)
);
Ok(())
}
}
#[cfg(test)]
mod tests {
use super::*;
use bytes::BufMut;
use tempfile::tempdir;
#[tokio::test]
async fn test_parallel_read_write() {
// Create configuration with low threshold for testing
let mut config = AppConfig::default();
config.concurrency.min_size_for_parallel_chunks_mb = 1; // 1MB for testing
config.concurrency.max_parallel_chunks = 4;
let processor = ParallelFileProcessor::new(config);
// Create temporary directory
let temp_dir = tempdir().unwrap();
let file_path = temp_dir.path().join("test_file.bin");
// Create test data (2MB)
let size = 2 * 1024 * 1024;
let mut test_data = Vec::with_capacity(size);
for i in 0..size {
test_data.push((i % 256) as u8);
}
// Write file in parallel
processor
.write_file_parallel(&file_path, &test_data)
.await
.unwrap();
// Read file in parallel
let read_data = processor.read_file_parallel(&file_path).await.unwrap();
// Verify that the data is identical
assert_eq!(test_data.len(), read_data.len());
assert_eq!(test_data, read_data);
}
#[tokio::test]
async fn test_bytesmut_pool() {
// Create pool
let pool = BytesBufferPool::new(1024, 5);
// Get buffer
let mut buffer1 = pool.get_buffer().await;
buffer1.put_slice(b"test data");
assert_eq!(&buffer1[..9], b"test data");
// Return buffer to the pool
pool.return_buffer(buffer1).await;
// Get another buffer (should be the same one)
let buffer2 = pool.get_buffer().await;
assert_eq!(buffer2.capacity(), 1024);
// The buffer should be empty (cleared)
assert_eq!(buffer2.len(), 0);
}
#[test]
fn test_chunk_calculation() {
// Create test configuration
let mut config = AppConfig::default();
config.concurrency.min_size_for_parallel_chunks_mb = 100; // 100MB
config.concurrency.max_parallel_chunks = 4;
config.concurrency.parallel_chunk_size_bytes = 50 * 1024 * 1024; // 50MB
let processor = ParallelFileProcessor::new(config);
// Small file (10MB)
let small_file_size = 10 * 1024 * 1024;
let chunks = processor.calculate_chunks(small_file_size);
assert_eq!(chunks.len(), 1);
assert_eq!(chunks[0].size as u64, small_file_size);
// Large file (300MB)
let large_file_size = 300 * 1024 * 1024;
let chunks = processor.calculate_chunks(large_file_size);
assert_eq!(chunks.len(), 4); // Limited to max_parallel_chunks
// Verify that all chunks add up to the total size
let total_size: u64 = chunks.iter().map(|c| c.size as u64).sum();
assert_eq!(total_size, large_file_size);
}
}
@@ -0,0 +1,271 @@
//! PostgreSQL + Blob-backed file read repository.
//!
//! Implements `FileReadPort` using:
//! - `storage.files` table for metadata lookups
//! - `DedupPort` for reading content-addressable blobs from the filesystem
use async_trait::async_trait;
use bytes::Bytes;
use futures::Stream;
use sqlx::PgPool;
use std::sync::Arc;
use crate::application::ports::dedup_ports::DedupPort;
use crate::application::ports::storage_ports::FileReadPort;
use crate::common::errors::DomainError;
use crate::domain::entities::file::File;
use crate::domain::repositories::folder_repository::FolderRepository;
use crate::domain::services::path_service::StoragePath;
use super::folder_db_repository::FolderDbRepository;
/// File read repository backed by PostgreSQL metadata + blob storage.
pub struct FileBlobReadRepository {
pool: Arc<PgPool>,
dedup: Arc<dyn DedupPort>,
folder_repo: Arc<FolderDbRepository>,
}
impl FileBlobReadRepository {
pub fn new(
pool: Arc<PgPool>,
dedup: Arc<dyn DedupPort>,
folder_repo: Arc<FolderDbRepository>,
) -> Self {
Self {
pool,
dedup,
folder_repo,
}
}
/// Build a virtual StoragePath for a file.
async fn build_file_path(
&self,
folder_id: Option<&str>,
file_name: &str,
) -> Result<StoragePath, DomainError> {
if let Some(fid) = folder_id {
let folder_path = self.folder_repo.get_folder_path(fid).await?;
Ok(folder_path.join(file_name))
} else {
Ok(StoragePath::from_string(file_name))
}
}
/// Convert a database row into a `File` domain entity.
async fn row_to_file(
&self,
id: String,
name: String,
folder_id: Option<String>,
size: i64,
mime_type: String,
created_at: i64,
modified_at: i64,
) -> Result<File, DomainError> {
let storage_path = self
.build_file_path(folder_id.as_deref(), &name)
.await?;
File::with_timestamps(
id,
name,
storage_path,
size as u64,
mime_type,
folder_id,
created_at as u64,
modified_at as u64,
)
.map_err(|e| DomainError::internal_error("FileBlobRead", format!("entity: {e}")))
}
/// Get the blob hash for a file.
async fn get_blob_hash(&self, file_id: &str) -> Result<String, DomainError> {
sqlx::query_scalar::<_, String>(
"SELECT blob_hash FROM storage.files WHERE id = $1::uuid AND NOT is_trashed",
)
.bind(file_id)
.fetch_optional(self.pool.as_ref())
.await
.map_err(|e| DomainError::internal_error("FileBlobRead", format!("hash lookup: {e}")))?
.ok_or_else(|| DomainError::not_found("File", file_id))
}
}
#[async_trait]
impl FileReadPort for FileBlobReadRepository {
async fn get_file(&self, id: &str) -> Result<File, DomainError> {
let row = sqlx::query_as::<_, (String, String, Option<String>, i64, String, i64, i64)>(
r#"
SELECT id::text, name, folder_id::text, size, mime_type,
EXTRACT(EPOCH FROM created_at)::bigint,
EXTRACT(EPOCH FROM updated_at)::bigint
FROM storage.files
WHERE id = $1::uuid AND NOT is_trashed
"#,
)
.bind(id)
.fetch_optional(self.pool.as_ref())
.await
.map_err(|e| DomainError::internal_error("FileBlobRead", format!("get: {e}")))?
.ok_or_else(|| DomainError::not_found("File", id))?;
self.row_to_file(row.0, row.1, row.2, row.3, row.4, row.5, row.6)
.await
}
async fn list_files(
&self,
folder_id: Option<&str>,
) -> Result<Vec<File>, DomainError> {
let rows: Vec<(String, String, Option<String>, i64, String, i64, i64)> =
if let Some(fid) = folder_id {
sqlx::query_as(
r#"
SELECT id::text, name, folder_id::text, size, mime_type,
EXTRACT(EPOCH FROM created_at)::bigint,
EXTRACT(EPOCH FROM updated_at)::bigint
FROM storage.files
WHERE folder_id = $1::uuid AND NOT is_trashed
ORDER BY name
"#,
)
.bind(fid)
.fetch_all(self.pool.as_ref())
.await
} else {
sqlx::query_as(
r#"
SELECT id::text, name, folder_id::text, size, mime_type,
EXTRACT(EPOCH FROM created_at)::bigint,
EXTRACT(EPOCH FROM updated_at)::bigint
FROM storage.files
WHERE folder_id IS NULL AND NOT is_trashed
ORDER BY name
"#,
)
.fetch_all(self.pool.as_ref())
.await
}
.map_err(|e| DomainError::internal_error("FileBlobRead", format!("list: {e}")))?;
let mut files = Vec::with_capacity(rows.len());
for (id, name, fid, size, mime, ca, ma) in rows {
files.push(self.row_to_file(id, name, fid, size, mime, ca, ma).await?);
}
Ok(files)
}
async fn get_file_content(&self, id: &str) -> Result<Vec<u8>, DomainError> {
let blob_hash = self.get_blob_hash(id).await?;
self.dedup.read_blob(&blob_hash).await
}
async fn get_file_stream(
&self,
id: &str,
) -> Result<Box<dyn Stream<Item = Result<Bytes, std::io::Error>> + Send>, DomainError> {
// Read blob as bytes and wrap in a single-chunk stream.
// For very large files, a true streaming implementation from the
// blob file would be better, but DedupPort API currently returns bytes.
let blob_hash = self.get_blob_hash(id).await?;
let content = self.dedup.read_blob_bytes(&blob_hash).await?;
let stream = futures::stream::once(async move { Ok(content) });
Ok(Box::new(stream))
}
async fn get_file_range_stream(
&self,
id: &str,
start: u64,
end: Option<u64>,
) -> Result<Box<dyn Stream<Item = Result<Bytes, std::io::Error>> + Send>, DomainError> {
let blob_hash = self.get_blob_hash(id).await?;
let content = self.dedup.read_blob_bytes(&blob_hash).await?;
let start = start as usize;
let end = end.map_or(content.len(), |e| e as usize).min(content.len());
if start >= content.len() {
return Ok(Box::new(futures::stream::empty()));
}
let slice = content.slice(start..end);
let stream = futures::stream::once(async move { Ok(slice) });
Ok(Box::new(stream))
}
async fn get_file_mmap(&self, id: &str) -> Result<Bytes, DomainError> {
let blob_hash = self.get_blob_hash(id).await?;
self.dedup.read_blob_bytes(&blob_hash).await
}
async fn get_file_path(&self, id: &str) -> Result<StoragePath, DomainError> {
let row = sqlx::query_as::<_, (String, Option<String>)>(
r#"
SELECT name, folder_id::text
FROM storage.files
WHERE id = $1::uuid AND NOT is_trashed
"#,
)
.bind(id)
.fetch_optional(self.pool.as_ref())
.await
.map_err(|e| DomainError::internal_error("FileBlobRead", format!("path: {e}")))?
.ok_or_else(|| DomainError::not_found("File", id))?;
self.build_file_path(row.1.as_deref(), &row.0).await
}
async fn get_parent_folder_id(&self, path: &str) -> Result<String, DomainError> {
// Walk the path to find the parent folder, searching by folder names
let path = path.trim_start_matches('/').trim_end_matches('/');
let segments: Vec<&str> = path.split('/').filter(|s| !s.is_empty()).collect();
if segments.is_empty() {
return Err(DomainError::not_found("Folder", "empty path"));
}
// For path "a/b/c/file.txt", the parent folder path is "a/b/c"
// But we don't know which part is folders vs filename.
// Walk segments trying to find matching folders.
let mut current_parent: Option<String> = None;
for segment in &segments {
let row = if let Some(ref pid) = current_parent {
sqlx::query_as::<_, (String,)>(
r#"
SELECT id::text FROM storage.folders
WHERE name = $1 AND parent_id = $2::uuid AND NOT is_trashed
"#,
)
.bind(segment)
.bind(pid)
.fetch_optional(self.pool.as_ref())
.await
} else {
sqlx::query_as::<_, (String,)>(
r#"
SELECT id::text FROM storage.folders
WHERE name = $1 AND parent_id IS NULL AND NOT is_trashed
"#,
)
.bind(segment)
.fetch_optional(self.pool.as_ref())
.await
}
.map_err(|e| DomainError::internal_error("FileBlobRead", format!("path walk: {e}")))?;
match row {
Some(r) => current_parent = Some(r.0),
None => break, // This segment is not a folder → it's the filename
}
}
current_parent.ok_or_else(|| {
DomainError::not_found("Folder", format!("parent for path: {path}"))
})
}
}
@@ -0,0 +1,435 @@
//! PostgreSQL + Blob-backed file write repository.
//!
//! Implements `FileWritePort` using:
//! - `storage.files` table for metadata
//! - `DedupPort` for content-addressable blob storage on the filesystem
use async_trait::async_trait;
use bytes::Bytes;
use futures::Stream;
use sqlx::PgPool;
use std::path::PathBuf;
use std::pin::Pin;
use std::sync::Arc;
use crate::application::ports::dedup_ports::DedupPort;
use crate::application::ports::storage_ports::FileWritePort;
use crate::common::errors::DomainError;
use crate::domain::entities::file::File;
use crate::domain::repositories::folder_repository::FolderRepository;
use crate::domain::services::path_service::StoragePath;
use super::folder_db_repository::FolderDbRepository;
/// File write repository backed by PostgreSQL metadata + blob storage.
pub struct FileBlobWriteRepository {
pool: Arc<PgPool>,
dedup: Arc<dyn DedupPort>,
folder_repo: Arc<FolderDbRepository>,
}
impl FileBlobWriteRepository {
pub fn new(
pool: Arc<PgPool>,
dedup: Arc<dyn DedupPort>,
folder_repo: Arc<FolderDbRepository>,
) -> Self {
Self {
pool,
dedup,
folder_repo,
}
}
/// Build a virtual StoragePath for a file from its DB metadata.
async fn build_file_path(
&self,
folder_id: Option<&str>,
file_name: &str,
) -> Result<StoragePath, DomainError> {
if let Some(fid) = folder_id {
let folder_path = self.folder_repo.get_folder_path(fid).await?;
Ok(folder_path.join(file_name))
} else {
Ok(StoragePath::from_string(file_name))
}
}
/// Convert a database row into a `File` domain entity.
async fn row_to_file(
&self,
id: String,
name: String,
folder_id: Option<String>,
size: i64,
mime_type: String,
created_at: i64,
modified_at: i64,
) -> Result<File, DomainError> {
let storage_path = self
.build_file_path(folder_id.as_deref(), &name)
.await?;
File::with_timestamps(
id,
name,
storage_path,
size as u64,
mime_type,
folder_id,
created_at as u64,
modified_at as u64,
)
.map_err(|e| DomainError::internal_error("FileBlobWrite", format!("entity: {e}")))
}
/// Derive user_id from the parent folder, or error if folder_id is None.
async fn resolve_user_id(&self, folder_id: Option<&str>) -> Result<String, DomainError> {
match folder_id {
Some(fid) => self.folder_repo.get_folder_user_id(fid).await,
None => Err(DomainError::internal_error(
"FileBlobWrite",
"folder_id is required to determine file owner",
)),
}
}
}
#[async_trait]
impl FileWritePort for FileBlobWriteRepository {
async fn save_file(
&self,
name: String,
folder_id: Option<String>,
content_type: String,
content: Vec<u8>,
) -> Result<File, DomainError> {
let user_id = self.resolve_user_id(folder_id.as_deref()).await?;
let size = content.len() as i64;
// Store content in blob store
let dedup_result = self
.dedup
.store_bytes(&content, Some(content_type.clone()))
.await?;
let blob_hash = dedup_result.hash().to_string();
// Insert file metadata
let row = sqlx::query_as::<_, (String, i64, i64)>(
r#"
INSERT INTO storage.files (name, folder_id, user_id, blob_hash, size, mime_type)
VALUES ($1, $2::uuid, $3, $4, $5, $6)
RETURNING id::text,
EXTRACT(EPOCH FROM created_at)::bigint,
EXTRACT(EPOCH FROM updated_at)::bigint
"#,
)
.bind(&name)
.bind(&folder_id)
.bind(&user_id)
.bind(&blob_hash)
.bind(size)
.bind(&content_type)
.fetch_one(self.pool.as_ref())
.await
.map_err(|e| {
if let sqlx::Error::Database(ref db_err) = e {
if db_err.code().as_deref() == Some("23505") {
return DomainError::already_exists(
"File",
format!("{name} already exists in folder"),
);
}
}
DomainError::internal_error("FileBlobWrite", format!("insert: {e}"))
})?;
tracing::info!(
"💾 BLOB WRITE: {} ({} bytes, hash: {})",
name,
size,
&blob_hash[..12]
);
self.row_to_file(
row.0,
name,
folder_id,
size,
content_type,
row.1,
row.2,
)
.await
}
async fn save_file_from_stream(
&self,
name: String,
folder_id: Option<String>,
content_type: String,
stream: Pin<Box<dyn Stream<Item = Result<Bytes, std::io::Error>> + Send>>,
) -> Result<File, DomainError> {
use futures::StreamExt;
// Collect stream into bytes (blobs are content-addressed, need full content for hash)
let mut content = Vec::new();
let mut stream = stream;
while let Some(chunk) = stream.next().await {
let chunk = chunk.map_err(|e| {
DomainError::internal_error("FileBlobWrite", format!("stream read: {e}"))
})?;
content.extend_from_slice(&chunk);
}
self.save_file(name, folder_id, content_type, content).await
}
async fn move_file(
&self,
file_id: &str,
target_folder_id: Option<String>,
) -> Result<File, DomainError> {
// If moving to a different folder, get the new user_id (must be same user)
let row = sqlx::query_as::<_, (String, String, Option<String>, i64, String, i64, i64)>(
r#"
UPDATE storage.files
SET folder_id = $1::uuid, updated_at = NOW()
WHERE id = $2::uuid AND NOT is_trashed
RETURNING id::text, name, folder_id::text, size, mime_type,
EXTRACT(EPOCH FROM created_at)::bigint,
EXTRACT(EPOCH FROM updated_at)::bigint
"#,
)
.bind(&target_folder_id)
.bind(file_id)
.fetch_optional(self.pool.as_ref())
.await
.map_err(|e| DomainError::internal_error("FileBlobWrite", format!("move: {e}")))?
.ok_or_else(|| DomainError::not_found("File", file_id))?;
self.row_to_file(row.0, row.1, row.2, row.3, row.4, row.5, row.6)
.await
}
async fn rename_file(
&self,
file_id: &str,
new_name: &str,
) -> Result<File, DomainError> {
let row = sqlx::query_as::<_, (String, String, Option<String>, i64, String, i64, i64)>(
r#"
UPDATE storage.files
SET name = $1, updated_at = NOW()
WHERE id = $2::uuid AND NOT is_trashed
RETURNING id::text, name, folder_id::text, size, mime_type,
EXTRACT(EPOCH FROM created_at)::bigint,
EXTRACT(EPOCH FROM updated_at)::bigint
"#,
)
.bind(new_name)
.bind(file_id)
.fetch_optional(self.pool.as_ref())
.await
.map_err(|e| {
if let sqlx::Error::Database(ref db_err) = e {
if db_err.code().as_deref() == Some("23505") {
return DomainError::already_exists(
"File",
format!("{new_name} already exists"),
);
}
}
DomainError::internal_error("FileBlobWrite", format!("rename: {e}"))
})?
.ok_or_else(|| DomainError::not_found("File", file_id))?;
self.row_to_file(row.0, row.1, row.2, row.3, row.4, row.5, row.6)
.await
}
async fn delete_file(&self, id: &str) -> Result<(), DomainError> {
// Get blob_hash before deleting so we can decrement ref
let hash = sqlx::query_scalar::<_, String>(
"SELECT blob_hash FROM storage.files WHERE id = $1::uuid",
)
.bind(id)
.fetch_optional(self.pool.as_ref())
.await
.map_err(|e| DomainError::internal_error("FileBlobWrite", format!("hash lookup: {e}")))?;
let result = sqlx::query("DELETE FROM storage.files WHERE id = $1::uuid")
.bind(id)
.execute(self.pool.as_ref())
.await
.map_err(|e| DomainError::internal_error("FileBlobWrite", format!("delete: {e}")))?;
if result.rows_affected() == 0 {
return Err(DomainError::not_found("File", id));
}
// Decrement blob reference
if let Some(h) = hash {
if let Err(e) = self.dedup.remove_reference(&h).await {
tracing::warn!("Failed to decrement blob ref for {}: {}", &h[..12], e);
}
}
Ok(())
}
async fn update_file_content(
&self,
file_id: &str,
content: Vec<u8>,
) -> Result<(), DomainError> {
// Get old blob hash to decrement ref
let old_hash = sqlx::query_scalar::<_, String>(
"SELECT blob_hash FROM storage.files WHERE id = $1::uuid",
)
.bind(file_id)
.fetch_optional(self.pool.as_ref())
.await
.map_err(|e| DomainError::internal_error("FileBlobWrite", format!("old hash: {e}")))?
.ok_or_else(|| DomainError::not_found("File", file_id))?;
// Store new content
let new_size = content.len() as i64;
let dedup_result = self.dedup.store_bytes(&content, None).await?;
let new_hash = dedup_result.hash().to_string();
// Update file metadata
sqlx::query(
r#"
UPDATE storage.files
SET blob_hash = $1, size = $2, updated_at = NOW()
WHERE id = $3::uuid
"#,
)
.bind(&new_hash)
.bind(new_size)
.bind(file_id)
.execute(self.pool.as_ref())
.await
.map_err(|e| DomainError::internal_error("FileBlobWrite", format!("update: {e}")))?;
// Decrement old blob ref (only if hash changed)
if old_hash != new_hash {
if let Err(e) = self.dedup.remove_reference(&old_hash).await {
tracing::warn!(
"Failed to decrement old blob ref {}: {}",
&old_hash[..12],
e
);
}
}
Ok(())
}
async fn register_file_deferred(
&self,
name: String,
folder_id: Option<String>,
content_type: String,
size: u64,
) -> Result<(File, PathBuf), DomainError> {
let user_id = self.resolve_user_id(folder_id.as_deref()).await?;
// For deferred registration we use a placeholder hash.
// The write-behind cache will call update_file_content later.
let placeholder_hash = "0000000000000000000000000000000000000000000000000000000000000000";
let row = sqlx::query_as::<_, (String, i64, i64)>(
r#"
INSERT INTO storage.files (name, folder_id, user_id, blob_hash, size, mime_type)
VALUES ($1, $2::uuid, $3, $4, $5, $6)
RETURNING id::text,
EXTRACT(EPOCH FROM created_at)::bigint,
EXTRACT(EPOCH FROM updated_at)::bigint
"#,
)
.bind(&name)
.bind(&folder_id)
.bind(&user_id)
.bind(placeholder_hash)
.bind(size as i64)
.bind(&content_type)
.fetch_one(self.pool.as_ref())
.await
.map_err(|e| DomainError::internal_error("FileBlobWrite", format!("deferred: {e}")))?;
let file = self
.row_to_file(
row.0.clone(),
name,
folder_id,
size as i64,
content_type,
row.1,
row.2,
)
.await?;
// The target_path is not meaningful for blob storage (content goes to .blobs/)
// but the WriteBehindCache API requires it. We return a synthetic path.
let target_path = PathBuf::from(format!(".pending/{}", row.0));
Ok((file, target_path))
}
// ── Trash operations ──
async fn move_to_trash(&self, file_id: &str) -> Result<(), DomainError> {
let result = sqlx::query(
r#"
UPDATE storage.files
SET is_trashed = TRUE,
trashed_at = NOW(),
original_folder_id = folder_id,
updated_at = NOW()
WHERE id = $1::uuid AND NOT is_trashed
"#,
)
.bind(file_id)
.execute(self.pool.as_ref())
.await
.map_err(|e| DomainError::internal_error("FileBlobWrite", format!("trash: {e}")))?;
if result.rows_affected() == 0 {
return Err(DomainError::not_found("File", file_id));
}
Ok(())
}
async fn restore_from_trash(
&self,
file_id: &str,
_original_path: &str,
) -> Result<(), DomainError> {
let result = sqlx::query(
r#"
UPDATE storage.files
SET is_trashed = FALSE,
trashed_at = NULL,
folder_id = COALESCE(original_folder_id, folder_id),
original_folder_id = NULL,
updated_at = NOW()
WHERE id = $1::uuid AND is_trashed
"#,
)
.bind(file_id)
.execute(self.pool.as_ref())
.await
.map_err(|e| DomainError::internal_error("FileBlobWrite", format!("restore: {e}")))?;
if result.rows_affected() == 0 {
return Err(DomainError::not_found("File", file_id));
}
Ok(())
}
async fn delete_file_permanently(&self, file_id: &str) -> Result<(), DomainError> {
// Same as delete_file — removes from DB and decrements blob ref
self.delete_file(file_id).await
}
}
@@ -0,0 +1,603 @@
//! PostgreSQL-backed folder repository.
//!
//! Implements `FolderRepository` (and thus `FolderStoragePort`) using the
//! `storage.folders` table. Folders are purely virtual — no physical
//! directories are created on the filesystem.
use async_trait::async_trait;
use sqlx::PgPool;
use std::sync::Arc;
use crate::common::errors::DomainError;
use crate::domain::entities::folder::Folder;
use crate::domain::repositories::folder_repository::FolderRepository;
use crate::domain::services::path_service::StoragePath;
/// PostgreSQL-backed folder repository.
///
/// All folder metadata lives in the `storage.folders` table. The physical
/// filesystem is never touched for folder operations.
pub struct FolderDbRepository {
pool: Option<Arc<PgPool>>,
}
impl FolderDbRepository {
pub fn new(pool: Arc<PgPool>) -> Self {
Self { pool: Some(pool) }
}
/// Creates a stub instance for `AppState::default()`.
/// This is never called in production — only used for route scaffolding.
pub fn new_stub() -> Self {
Self { pool: None }
}
/// Get the pool, panicking if stub.
fn pool(&self) -> &PgPool {
self.pool.as_deref().expect("FolderDbRepository: pool not available (stub instance)")
}
// ── helpers ──────────────────────────────────────────────────
/// Build the full virtual path for a folder by walking up the `parent_id` chain.
async fn build_folder_path(&self, folder_id: &str) -> Result<StoragePath, DomainError> {
// CTE-based recursive query to build path segments
let _rows = sqlx::query_as::<_, (String,)>(
r#"
WITH RECURSIVE ancestors AS (
SELECT id, name, parent_id
FROM storage.folders
WHERE id = $1::uuid
UNION ALL
SELECT f.id, f.name, f.parent_id
FROM storage.folders f
JOIN ancestors a ON f.id = a.parent_id
)
SELECT name FROM ancestors ORDER BY name
"#,
)
.bind(folder_id)
.fetch_all(self.pool())
.await
.map_err(|e| DomainError::internal_error("FolderDb", format!("path query: {e}")))?;
// Actually we need a proper ordering. Let me rewrite with depth tracking.
// Re-query with depth.
let rows = sqlx::query_as::<_, (String, i32)>(
r#"
WITH RECURSIVE ancestors AS (
SELECT id, name, parent_id, 0 AS depth
FROM storage.folders
WHERE id = $1::uuid
UNION ALL
SELECT f.id, f.name, f.parent_id, a.depth + 1
FROM storage.folders f
JOIN ancestors a ON f.id = a.parent_id
)
SELECT name, depth FROM ancestors ORDER BY depth DESC
"#,
)
.bind(folder_id)
.fetch_all(self.pool())
.await
.map_err(|e| DomainError::internal_error("FolderDb", format!("path query: {e}")))?;
let path_parts: Vec<&str> = rows.iter().map(|(name, _)| name.as_str()).collect();
let path_str = path_parts.join("/");
Ok(StoragePath::from_string(&path_str))
}
/// Convert a database row into a `Folder` domain entity.
async fn row_to_folder(
&self,
id: String,
name: String,
parent_id: Option<String>,
created_at: i64,
modified_at: i64,
) -> Result<Folder, DomainError> {
let storage_path = self.build_folder_path(&id).await?;
Folder::with_timestamps(
id,
name,
storage_path,
parent_id,
created_at as u64,
modified_at as u64,
)
.map_err(|e| DomainError::internal_error("FolderDb", format!("entity: {e}")))
}
}
#[async_trait]
impl FolderRepository for FolderDbRepository {
async fn create_folder(
&self,
name: String,
parent_id: Option<String>,
) -> Result<Folder, DomainError> {
// Derive user_id from parent folder. Root-level folders require the
// caller to have set up the home folder beforehand (done during user
// registration).
let user_id: String = if let Some(ref pid) = parent_id {
sqlx::query_scalar::<_, String>(
"SELECT user_id FROM storage.folders WHERE id = $1::uuid",
)
.bind(pid)
.fetch_optional(self.pool())
.await
.map_err(|e| DomainError::internal_error("FolderDb", format!("parent lookup: {e}")))?
.ok_or_else(|| DomainError::not_found("Folder", pid))?
} else {
return Err(DomainError::internal_error(
"FolderDb",
"Cannot create root folder without user_id — use create_home_folder instead",
));
};
let row = sqlx::query_as::<_, (String, i64, i64)>(
r#"
INSERT INTO storage.folders (name, parent_id, user_id)
VALUES ($1, $2::uuid, $3)
RETURNING id::text,
EXTRACT(EPOCH FROM created_at)::bigint,
EXTRACT(EPOCH FROM updated_at)::bigint
"#,
)
.bind(&name)
.bind(&parent_id)
.bind(&user_id)
.fetch_one(self.pool())
.await
.map_err(|e| {
if let sqlx::Error::Database(ref db_err) = e {
if db_err.code().as_deref() == Some("23505") {
return DomainError::already_exists(
"Folder",
format!("{name} already exists in parent"),
);
}
}
DomainError::internal_error("FolderDb", format!("insert: {e}"))
})?;
self.row_to_folder(row.0, name, parent_id, row.1, row.2)
.await
}
async fn get_folder(&self, id: &str) -> Result<Folder, DomainError> {
let row = sqlx::query_as::<_, (String, String, Option<String>, i64, i64)>(
r#"
SELECT id::text, name, parent_id::text,
EXTRACT(EPOCH FROM created_at)::bigint,
EXTRACT(EPOCH FROM updated_at)::bigint
FROM storage.folders
WHERE id = $1::uuid AND NOT is_trashed
"#,
)
.bind(id)
.fetch_optional(self.pool())
.await
.map_err(|e| DomainError::internal_error("FolderDb", format!("get: {e}")))?
.ok_or_else(|| DomainError::not_found("Folder", id))?;
self.row_to_folder(row.0, row.1, row.2, row.3, row.4)
.await
}
async fn get_folder_by_path(
&self,
storage_path: &StoragePath,
) -> Result<Folder, DomainError> {
// Walk the path segments to find the folder.
let path_str = storage_path.to_string();
let segments: Vec<&str> = path_str.split('/').filter(|s| !s.is_empty()).collect();
if segments.is_empty() {
return Err(DomainError::not_found("Folder", "empty path"));
}
let mut current_parent: Option<String> = None;
let mut current_id = String::new();
for segment in &segments {
let row = if let Some(ref pid) = current_parent {
sqlx::query_as::<_, (String,)>(
r#"
SELECT id::text FROM storage.folders
WHERE name = $1 AND parent_id = $2::uuid AND NOT is_trashed
"#,
)
.bind(segment)
.bind(pid)
.fetch_optional(self.pool())
.await
} else {
sqlx::query_as::<_, (String,)>(
r#"
SELECT id::text FROM storage.folders
WHERE name = $1 AND parent_id IS NULL AND NOT is_trashed
"#,
)
.bind(segment)
.fetch_optional(self.pool())
.await
}
.map_err(|e| DomainError::internal_error("FolderDb", format!("path walk: {e}")))?
.ok_or_else(|| {
DomainError::not_found("Folder", format!("segment '{segment}' in path"))
})?;
current_id = row.0;
current_parent = Some(current_id.clone());
}
self.get_folder(&current_id).await
}
async fn list_folders(
&self,
parent_id: Option<&str>,
) -> Result<Vec<Folder>, DomainError> {
let rows: Vec<(String, String, Option<String>, i64, i64)> = if let Some(pid) = parent_id {
sqlx::query_as(
r#"
SELECT id::text, name, parent_id::text,
EXTRACT(EPOCH FROM created_at)::bigint,
EXTRACT(EPOCH FROM updated_at)::bigint
FROM storage.folders
WHERE parent_id = $1::uuid AND NOT is_trashed
ORDER BY name
"#,
)
.bind(pid)
.fetch_all(self.pool())
.await
} else {
sqlx::query_as(
r#"
SELECT id::text, name, parent_id::text,
EXTRACT(EPOCH FROM created_at)::bigint,
EXTRACT(EPOCH FROM updated_at)::bigint
FROM storage.folders
WHERE parent_id IS NULL AND NOT is_trashed
ORDER BY name
"#,
)
.fetch_all(self.pool())
.await
}
.map_err(|e| DomainError::internal_error("FolderDb", format!("list: {e}")))?;
let mut folders = Vec::with_capacity(rows.len());
for (id, name, pid, ca, ma) in rows {
folders.push(self.row_to_folder(id, name, pid, ca, ma).await?);
}
Ok(folders)
}
async fn list_folders_paginated(
&self,
parent_id: Option<&str>,
offset: usize,
limit: usize,
include_total: bool,
) -> Result<(Vec<Folder>, Option<usize>), DomainError> {
let total = if include_total {
let count: i64 = if let Some(pid) = parent_id {
sqlx::query_scalar(
"SELECT COUNT(*) FROM storage.folders WHERE parent_id = $1::uuid AND NOT is_trashed",
)
.bind(pid)
.fetch_one(self.pool())
.await
} else {
sqlx::query_scalar(
"SELECT COUNT(*) FROM storage.folders WHERE parent_id IS NULL AND NOT is_trashed",
)
.fetch_one(self.pool())
.await
}
.map_err(|e| DomainError::internal_error("FolderDb", format!("count: {e}")))?;
Some(count as usize)
} else {
None
};
let rows: Vec<(String, String, Option<String>, i64, i64)> = if let Some(pid) = parent_id {
sqlx::query_as(
r#"
SELECT id::text, name, parent_id::text,
EXTRACT(EPOCH FROM created_at)::bigint,
EXTRACT(EPOCH FROM updated_at)::bigint
FROM storage.folders
WHERE parent_id = $1::uuid AND NOT is_trashed
ORDER BY name
LIMIT $2 OFFSET $3
"#,
)
.bind(pid)
.bind(limit as i64)
.bind(offset as i64)
.fetch_all(self.pool())
.await
} else {
sqlx::query_as(
r#"
SELECT id::text, name, parent_id::text,
EXTRACT(EPOCH FROM created_at)::bigint,
EXTRACT(EPOCH FROM updated_at)::bigint
FROM storage.folders
WHERE parent_id IS NULL AND NOT is_trashed
ORDER BY name
LIMIT $1 OFFSET $2
"#,
)
.bind(limit as i64)
.bind(offset as i64)
.fetch_all(self.pool())
.await
}
.map_err(|e| DomainError::internal_error("FolderDb", format!("paginate: {e}")))?;
let mut folders = Vec::with_capacity(rows.len());
for (id, name, pid, ca, ma) in rows {
folders.push(self.row_to_folder(id, name, pid, ca, ma).await?);
}
Ok((folders, total))
}
async fn rename_folder(
&self,
id: &str,
new_name: String,
) -> Result<Folder, DomainError> {
sqlx::query(
r#"
UPDATE storage.folders
SET name = $1, updated_at = NOW()
WHERE id = $2::uuid AND NOT is_trashed
"#,
)
.bind(&new_name)
.bind(id)
.execute(self.pool())
.await
.map_err(|e| {
if let sqlx::Error::Database(ref db_err) = e {
if db_err.code().as_deref() == Some("23505") {
return DomainError::already_exists(
"Folder",
format!("{new_name} already exists"),
);
}
}
DomainError::internal_error("FolderDb", format!("rename: {e}"))
})?;
self.get_folder(id).await
}
async fn move_folder(
&self,
id: &str,
new_parent_id: Option<&str>,
) -> Result<Folder, DomainError> {
sqlx::query(
r#"
UPDATE storage.folders
SET parent_id = $1::uuid, updated_at = NOW()
WHERE id = $2::uuid AND NOT is_trashed
"#,
)
.bind(new_parent_id)
.bind(id)
.execute(self.pool())
.await
.map_err(|e| DomainError::internal_error("FolderDb", format!("move: {e}")))?;
self.get_folder(id).await
}
async fn delete_folder(&self, id: &str) -> Result<(), DomainError> {
// Hard delete folder and all descendants (CASCADE handles children)
let result = sqlx::query("DELETE FROM storage.folders WHERE id = $1::uuid")
.bind(id)
.execute(self.pool())
.await
.map_err(|e| DomainError::internal_error("FolderDb", format!("delete: {e}")))?;
if result.rows_affected() == 0 {
return Err(DomainError::not_found("Folder", id));
}
Ok(())
}
async fn folder_exists(&self, storage_path: &StoragePath) -> Result<bool, DomainError> {
// Try to find by walking the path
match self.get_folder_by_path(storage_path).await {
Ok(_) => Ok(true),
Err(e) if e.to_string().contains("not found") => Ok(false),
Err(e) => Err(e),
}
}
async fn get_folder_path(&self, id: &str) -> Result<StoragePath, DomainError> {
self.build_folder_path(id).await
}
// ── Trash operations ──
async fn move_to_trash(&self, folder_id: &str) -> Result<(), DomainError> {
// Soft-delete: set is_trashed = true and remember original parent
let result = sqlx::query(
r#"
UPDATE storage.folders
SET is_trashed = TRUE,
trashed_at = NOW(),
original_parent_id = parent_id,
updated_at = NOW()
WHERE id = $1::uuid AND NOT is_trashed
"#,
)
.bind(folder_id)
.execute(self.pool())
.await
.map_err(|e| DomainError::internal_error("FolderDb", format!("trash: {e}")))?;
if result.rows_affected() == 0 {
return Err(DomainError::not_found("Folder", folder_id));
}
// Also trash all files inside the folder (recursively)
sqlx::query(
r#"
WITH RECURSIVE descendants AS (
SELECT id FROM storage.folders WHERE id = $1::uuid
UNION ALL
SELECT f.id FROM storage.folders f JOIN descendants d ON f.parent_id = d.id
)
UPDATE storage.files
SET is_trashed = TRUE, trashed_at = NOW(), original_folder_id = folder_id
WHERE folder_id IN (SELECT id FROM descendants) AND NOT is_trashed
"#,
)
.bind(folder_id)
.execute(self.pool())
.await
.map_err(|e| DomainError::internal_error("FolderDb", format!("trash files: {e}")))?;
Ok(())
}
async fn restore_from_trash(
&self,
folder_id: &str,
_original_path: &str,
) -> Result<(), DomainError> {
// Restore: set is_trashed = false, restore parent_id from original_parent_id
let result = sqlx::query(
r#"
UPDATE storage.folders
SET is_trashed = FALSE,
trashed_at = NULL,
parent_id = COALESCE(original_parent_id, parent_id),
original_parent_id = NULL,
updated_at = NOW()
WHERE id = $1::uuid AND is_trashed
"#,
)
.bind(folder_id)
.execute(self.pool())
.await
.map_err(|e| DomainError::internal_error("FolderDb", format!("restore: {e}")))?;
if result.rows_affected() == 0 {
return Err(DomainError::not_found("Folder", folder_id));
}
// Also restore files that were trashed with this folder
sqlx::query(
r#"
WITH RECURSIVE descendants AS (
SELECT id FROM storage.folders WHERE id = $1::uuid
UNION ALL
SELECT f.id FROM storage.folders f JOIN descendants d ON f.parent_id = d.id
)
UPDATE storage.files
SET is_trashed = FALSE,
trashed_at = NULL,
folder_id = COALESCE(original_folder_id, folder_id),
original_folder_id = NULL
WHERE folder_id IN (SELECT id FROM descendants) AND is_trashed
"#,
)
.bind(folder_id)
.execute(self.pool())
.await
.map_err(|e| DomainError::internal_error("FolderDb", format!("restore files: {e}")))?;
Ok(())
}
async fn delete_folder_permanently(&self, folder_id: &str) -> Result<(), DomainError> {
// Permanently delete — CASCADE handles children
let result = sqlx::query("DELETE FROM storage.folders WHERE id = $1::uuid")
.bind(folder_id)
.execute(self.pool())
.await
.map_err(|e| DomainError::internal_error("FolderDb", format!("perm delete: {e}")))?;
if result.rows_affected() == 0 {
return Err(DomainError::not_found("Folder", folder_id));
}
Ok(())
}
}
// ── Extra helpers for blob-storage bootstrap ──
impl FolderDbRepository {
/// Creates a root-level home folder for a user.
/// This is called during user registration.
pub async fn create_home_folder(
&self,
user_id: &str,
name: &str,
) -> Result<Folder, DomainError> {
let row = sqlx::query_as::<_, (String, i64, i64)>(
r#"
INSERT INTO storage.folders (name, parent_id, user_id)
VALUES ($1, NULL, $2)
ON CONFLICT DO NOTHING
RETURNING id::text,
EXTRACT(EPOCH FROM created_at)::bigint,
EXTRACT(EPOCH FROM updated_at)::bigint
"#,
)
.bind(name)
.bind(user_id)
.fetch_optional(self.pool())
.await
.map_err(|e| DomainError::internal_error("FolderDb", format!("home folder: {e}")))?;
match row {
Some((id, ca, ma)) => {
self.row_to_folder(id, name.to_string(), None, ca, ma).await
}
None => {
// Already exists — fetch it
let existing = sqlx::query_as::<_, (String, i64, i64)>(
r#"
SELECT id::text,
EXTRACT(EPOCH FROM created_at)::bigint,
EXTRACT(EPOCH FROM updated_at)::bigint
FROM storage.folders
WHERE name = $1 AND user_id = $2 AND parent_id IS NULL
"#,
)
.bind(name)
.bind(user_id)
.fetch_one(self.pool())
.await
.map_err(|e| {
DomainError::internal_error("FolderDb", format!("home fetch: {e}"))
})?;
self.row_to_folder(existing.0, name.to_string(), None, existing.1, existing.2)
.await
}
}
}
/// Returns user_id for a given folder. Used by file repositories.
pub async fn get_folder_user_id(&self, folder_id: &str) -> Result<String, DomainError> {
sqlx::query_scalar::<_, String>(
"SELECT user_id FROM storage.folders WHERE id = $1::uuid",
)
.bind(folder_id)
.fetch_optional(self.pool())
.await
.map_err(|e| DomainError::internal_error("FolderDb", format!("user_id lookup: {e}")))?
.ok_or_else(|| DomainError::not_found("Folder", folder_id))
}
}
+10
View File
@@ -11,6 +11,12 @@ mod settings_pg_repository;
mod transaction_utils;
mod user_pg_repository;
// ── Blob-storage repositories ──
pub mod file_blob_read_repository;
pub mod file_blob_write_repository;
pub mod folder_db_repository;
pub mod trash_db_repository;
pub use address_book_pg_repository::AddressBookPgRepository;
pub use calendar_event_pg_repository::CalendarEventPgRepository;
pub use calendar_pg_repository::CalendarPgRepository;
@@ -18,7 +24,11 @@ pub use contact_group_pg_repository::ContactGroupPgRepository;
pub use contact_persistence_dto::*;
pub use contact_pg_repository::ContactPgRepository;
pub use favorites_pg_repository::FavoritesPgRepository;
pub use file_blob_read_repository::FileBlobReadRepository;
pub use file_blob_write_repository::FileBlobWriteRepository;
pub use folder_db_repository::FolderDbRepository;
pub use recent_items_pg_repository::RecentItemsPgRepository;
pub use session_pg_repository::SessionPgRepository;
pub use settings_pg_repository::SettingsPgRepository;
pub use trash_db_repository::TrashDbRepository;
pub use user_pg_repository::UserPgRepository;
@@ -0,0 +1,191 @@
//! PostgreSQL-backed trash repository.
//!
//! Implements `TrashRepository` using soft-delete columns in `storage.files`
//! and `storage.folders`. There is no separate trash table — trashed items
//! are files/folders with `is_trashed = TRUE`.
use async_trait::async_trait;
use chrono::{DateTime, Utc};
use sqlx::PgPool;
use std::sync::Arc;
use uuid::Uuid;
use crate::common::errors::{DomainError, Result};
use crate::domain::entities::trashed_item::{TrashedItem, TrashedItemType};
use crate::domain::repositories::trash_repository::TrashRepository;
/// Default retention period (days) used when computing deletion_date.
const _DEFAULT_RETENTION_DAYS: i64 = 30;
/// PostgreSQL-backed trash repository using soft-delete flags.
pub struct TrashDbRepository {
pool: Arc<PgPool>,
retention_days: i64,
}
impl TrashDbRepository {
pub fn new(pool: Arc<PgPool>, retention_days: u32) -> Self {
Self {
pool,
retention_days: retention_days as i64,
}
}
/// Convert a trash_items view row into a TrashedItem entity.
fn row_to_trashed_item(
&self,
id: Uuid,
name: String,
item_type: String,
user_id: String,
trashed_at: Option<DateTime<Utc>>,
) -> TrashedItem {
let trashed_at = trashed_at.unwrap_or_else(Utc::now);
let deletion_date = trashed_at + chrono::Duration::days(self.retention_days);
let item_type_enum = match item_type.as_str() {
"folder" => TrashedItemType::Folder,
_ => TrashedItemType::File,
};
let user_uuid = Uuid::parse_str(&user_id).unwrap_or_else(|_| Uuid::nil());
// In the soft-delete model, the trash entry ID is the same as the
// original item ID since there is no separate trash table.
TrashedItem::from_raw(
id, // trash entry id (same as original)
id, // original item id
user_uuid, // owner
item_type_enum,
name.clone(),
String::new(), // original_path — not stored separately in soft-delete model
trashed_at,
deletion_date,
)
}
}
#[async_trait]
impl TrashRepository for TrashDbRepository {
async fn add_to_trash(&self, _item: &TrashedItem) -> Result<()> {
// No-op: the actual flagging is done by FileWritePort::move_to_trash
// or FolderRepository::move_to_trash. This method exists for interface
// compatibility with the TrashService.
Ok(())
}
async fn get_trash_items(&self, user_id: &Uuid) -> Result<Vec<TrashedItem>> {
let rows = sqlx::query_as::<_, (Uuid, String, String, String, Option<DateTime<Utc>>)>(
r#"
SELECT id, name, item_type, user_id, trashed_at
FROM storage.trash_items
WHERE user_id = $1
ORDER BY trashed_at DESC
"#,
)
.bind(user_id.to_string())
.fetch_all(self.pool.as_ref())
.await
.map_err(|e| {
DomainError::internal_error("TrashDb", format!("list: {e}"))
})?;
Ok(rows
.into_iter()
.map(|(id, name, item_type, uid, trashed_at)| {
self.row_to_trashed_item(id, name, item_type, uid, trashed_at)
})
.collect())
}
async fn get_trash_item(
&self,
id: &Uuid,
user_id: &Uuid,
) -> Result<Option<TrashedItem>> {
let row = sqlx::query_as::<_, (Uuid, String, String, String, Option<DateTime<Utc>>)>(
r#"
SELECT id, name, item_type, user_id, trashed_at
FROM storage.trash_items
WHERE id = $1 AND user_id = $2
"#,
)
.bind(id)
.bind(user_id.to_string())
.fetch_optional(self.pool.as_ref())
.await
.map_err(|e| {
DomainError::internal_error("TrashDb", format!("get: {e}"))
})?;
Ok(row.map(|(id, name, item_type, uid, trashed_at)| {
self.row_to_trashed_item(id, name, item_type, uid, trashed_at)
}))
}
async fn restore_from_trash(&self, _id: &Uuid, _user_id: &Uuid) -> Result<()> {
// No-op: the actual restore is done by FileWritePort::restore_from_trash
// or FolderRepository::restore_from_trash. The TrashService also removes
// the index entry — which in the soft-delete model means the flag is
// already cleared.
Ok(())
}
async fn delete_permanently(&self, _id: &Uuid, _user_id: &Uuid) -> Result<()> {
// No-op: the actual delete is done by FileWritePort::delete_file_permanently
// or FolderRepository::delete_folder_permanently.
Ok(())
}
async fn clear_trash(&self, user_id: &Uuid) -> Result<()> {
// Delete all trashed files for this user
sqlx::query(
"DELETE FROM storage.files WHERE user_id = $1 AND is_trashed = TRUE",
)
.bind(user_id.to_string())
.execute(self.pool.as_ref())
.await
.map_err(|e| {
DomainError::internal_error("TrashDb", format!("clear files: {e}"))
})?;
// Delete all trashed folders for this user
sqlx::query(
"DELETE FROM storage.folders WHERE user_id = $1 AND is_trashed = TRUE",
)
.bind(user_id.to_string())
.execute(self.pool.as_ref())
.await
.map_err(|e| {
DomainError::internal_error("TrashDb", format!("clear folders: {e}"))
})?;
Ok(())
}
async fn get_expired_items(&self) -> Result<Vec<TrashedItem>> {
let cutoff = Utc::now() - chrono::Duration::days(self.retention_days);
let rows = sqlx::query_as::<_, (Uuid, String, String, String, Option<DateTime<Utc>>)>(
r#"
SELECT id, name, item_type, user_id, trashed_at
FROM storage.trash_items
WHERE trashed_at < $1
ORDER BY trashed_at ASC
"#,
)
.bind(cutoff)
.fetch_all(self.pool.as_ref())
.await
.map_err(|e| {
DomainError::internal_error("TrashDb", format!("expired: {e}"))
})?;
Ok(rows
.into_iter()
.map(|(id, name, item_type, uid, trashed_at)| {
self.row_to_trashed_item(id, name, item_type, uid, trashed_at)
})
.collect())
}
}
@@ -1,110 +0,0 @@
//! Infrastructure-layer error types for file and folder repository operations.
//!
//! These error types are used internally by the filesystem repository implementations
//! (FileFsReadRepository, FileFsWriteRepository, FolderFsRepository, etc.) to represent
//! errors that can occur during storage operations. They are converted to `DomainError`
//! at the port boundary before crossing into the application layer.
use crate::common::errors::DomainError;
/// Error types for file repository operations.
#[derive(Debug, thiserror::Error)]
pub enum FileRepositoryError {
#[error("File not found: {0}")]
NotFound(String),
#[error("File already exists: {0}")]
AlreadyExists(String),
#[error("Invalid file path: {0}")]
InvalidPath(String),
#[error("Operation not supported: {0}")]
OperationNotSupported(String),
#[error("Storage error: {0}")]
StorageError(String),
#[error("Domain error: {0}")]
DomainError(#[from] DomainError),
#[error("Other error: {0}")]
Other(String),
}
pub type FileRepositoryResult<T> = Result<T, FileRepositoryError>;
/// Error types for folder repository operations.
#[derive(Debug, thiserror::Error)]
pub enum FolderRepositoryError {
#[error("Folder not found: {0}")]
NotFound(String),
#[error("Folder already exists: {0}")]
AlreadyExists(String),
#[error("Invalid folder path: {0}")]
InvalidPath(String),
#[error("Operation not supported: {0}")]
OperationNotSupported(String),
#[error("Storage error: {0}")]
StorageError(String),
#[error("Validation error: {0}")]
ValidationError(String),
#[error("Domain error: {0}")]
DomainError(#[from] DomainError),
#[error("Other error: {0}")]
Other(String),
}
pub type FolderRepositoryResult<T> = Result<T, FolderRepositoryError>;
// ── Conversions to DomainError ──
impl From<FileRepositoryError> for DomainError {
fn from(err: FileRepositoryError) -> Self {
match err {
FileRepositoryError::NotFound(id) => DomainError::not_found("File", id),
FileRepositoryError::AlreadyExists(path) => DomainError::already_exists("File", path),
FileRepositoryError::InvalidPath(path) => {
DomainError::validation_error(format!("Invalid path: {}", path))
}
FileRepositoryError::StorageError(msg) => {
DomainError::internal_error("File", format!("Storage error: {}", msg))
}
FileRepositoryError::Other(msg) => DomainError::internal_error("File", msg),
FileRepositoryError::OperationNotSupported(msg) => {
DomainError::operation_not_supported("File", msg)
}
FileRepositoryError::DomainError(e) => e,
}
}
}
impl From<FolderRepositoryError> for DomainError {
fn from(err: FolderRepositoryError) -> Self {
match err {
FolderRepositoryError::NotFound(id) => DomainError::not_found("Folder", id),
FolderRepositoryError::AlreadyExists(path) => {
DomainError::already_exists("Folder", path)
}
FolderRepositoryError::InvalidPath(path) => {
DomainError::validation_error(format!("Invalid path: {}", path))
}
FolderRepositoryError::StorageError(msg) => {
DomainError::internal_error("Folder", format!("Storage error: {}", msg))
}
FolderRepositoryError::ValidationError(msg) => DomainError::validation_error(msg),
FolderRepositoryError::Other(msg) => DomainError::internal_error("Folder", msg),
FolderRepositoryError::OperationNotSupported(msg) => {
DomainError::operation_not_supported("Folder", msg)
}
FolderRepositoryError::DomainError(e) => e,
}
}
}
@@ -1,441 +0,0 @@
use async_trait::async_trait;
use chrono::Utc;
use serde::{Deserialize, Serialize};
use std::path::{Path, PathBuf};
use std::sync::Arc;
use tokio::fs;
use tracing::{debug, error, instrument};
use uuid::Uuid;
use crate::application::ports::outbound::IdMappingPort;
use crate::common::errors::{DomainError, ErrorKind, Result};
use crate::domain::entities::trashed_item::{TrashedItem, TrashedItemType};
use crate::domain::repositories::trash_repository::TrashRepository;
/// Structure for storing trash items in JSON format
#[derive(Debug, Serialize, Deserialize)]
struct TrashedItemEntry {
id: String,
original_id: String,
user_id: String,
item_type: String,
name: String,
original_path: String,
trashed_at: String,
deletion_date: String,
}
/// Trash repository implementation using the file system
pub struct TrashFsRepository {
trash_dir: PathBuf,
trash_index_path: PathBuf,
}
impl TrashFsRepository {
pub fn new(
storage_root: impl AsRef<Path>,
_id_mapping_service: Arc<dyn IdMappingPort>,
) -> Self {
let trash_dir = storage_root.as_ref().join(".trash");
let trash_index_path = trash_dir.join("trash_index.json");
Self {
trash_dir,
trash_index_path,
}
}
/// Ensures the trash directory exists
async fn ensure_trash_dir(&self) -> Result<()> {
debug!(
"Checking if trash directory exists: {}",
self.trash_dir.display()
);
if !self.trash_dir.exists() {
debug!(
"Trash directory does not exist, creating it: {}",
self.trash_dir.display()
);
fs::create_dir_all(&self.trash_dir).await.map_err(|e| {
error!(
"Failed to create trash directory {}: {}",
self.trash_dir.display(),
e
);
DomainError::new(
ErrorKind::InternalError,
"Trash",
format!(
"Failed to create trash directory {}: {}",
self.trash_dir.display(),
e
),
)
})?;
debug!("Trash directory created successfully");
} else {
debug!("Trash directory already exists");
}
// Ensure the files directory exists
let files_dir = self.trash_dir.join("files");
debug!(
"Checking if trash files directory exists: {}",
files_dir.display()
);
if !files_dir.exists() {
debug!(
"Trash files directory does not exist, creating it: {}",
files_dir.display()
);
fs::create_dir_all(&files_dir).await.map_err(|e| {
error!(
"Failed to create trash files directory {}: {}",
files_dir.display(),
e
);
DomainError::new(
ErrorKind::InternalError,
"Trash",
format!(
"Failed to create trash files directory {}: {}",
files_dir.display(),
e
),
)
})?;
debug!("Trash files directory created successfully");
} else {
debug!("Trash files directory already exists");
}
// Also ensure the folders directory exists
let folders_dir = self.trash_dir.join("folders");
debug!(
"Checking if trash folders directory exists: {}",
folders_dir.display()
);
if !folders_dir.exists() {
debug!(
"Trash folders directory does not exist, creating it: {}",
folders_dir.display()
);
fs::create_dir_all(&folders_dir).await.map_err(|e| {
error!(
"Failed to create trash folders directory {}: {}",
folders_dir.display(),
e
);
DomainError::new(
ErrorKind::InternalError,
"Trash",
format!(
"Failed to create trash folders directory {}: {}",
folders_dir.display(),
e
),
)
})?;
debug!("Trash folders directory created successfully");
} else {
debug!("Trash folders directory already exists");
}
Ok(())
}
/// Gets all entries from the trash index
async fn get_trash_entries(&self) -> Result<Vec<TrashedItemEntry>> {
self.ensure_trash_dir().await?;
if !self.trash_index_path.exists() {
return Ok(Vec::new());
}
let content = fs::read_to_string(&self.trash_index_path)
.await
.map_err(|e| {
DomainError::new(
ErrorKind::InternalError,
"Trash",
format!("Failed to read trash index: {}", e),
)
})?;
if content.trim().is_empty() {
return Ok(Vec::new());
}
let entries: Vec<TrashedItemEntry> = serde_json::from_str(&content).map_err(|e| {
DomainError::new(
ErrorKind::InternalError,
"Trash",
format!("Failed to parse trash index: {}", e),
)
})?;
Ok(entries)
}
/// Saves all entries to the trash index
async fn save_trash_entries(&self, entries: Vec<TrashedItemEntry>) -> Result<()> {
self.ensure_trash_dir().await?;
let json = serde_json::to_string_pretty(&entries).map_err(|e| {
DomainError::new(
ErrorKind::InternalError,
"Trash",
format!("Failed to serialize trash index: {}", e),
)
})?;
fs::write(&self.trash_index_path, json).await.map_err(|e| {
DomainError::new(
ErrorKind::InternalError,
"Trash",
format!("Failed to write trash index: {}", e),
)
})?;
Ok(())
}
/// Converts a JSON entry to a TrashedItem entity
fn entry_to_trashed_item(&self, entry: TrashedItemEntry) -> Result<TrashedItem> {
let item_type = match entry.item_type.as_str() {
"file" => TrashedItemType::File,
"folder" => TrashedItemType::Folder,
_ => {
return Err(DomainError::new(
ErrorKind::InvalidInput,
"Trash",
format!("Invalid trashed item type: {}", entry.item_type),
));
}
};
let original_id = Uuid::parse_str(&entry.original_id).map_err(|e| {
DomainError::validation_error(format!("Invalid original ID format: {}", e))
})?;
let id = Uuid::parse_str(&entry.id)
.map_err(|e| DomainError::validation_error(format!("Invalid ID format: {}", e)))?;
let user_id = Uuid::parse_str(&entry.user_id)
.map_err(|e| DomainError::validation_error(format!("Invalid user ID format: {}", e)))?;
let trashed_at = chrono::DateTime::parse_from_rfc3339(&entry.trashed_at)
.map_err(|e| DomainError::validation_error(format!("Invalid trashed_at date: {}", e)))?
.with_timezone(&Utc);
let deletion_date = chrono::DateTime::parse_from_rfc3339(&entry.deletion_date)
.map_err(|e| DomainError::validation_error(format!("Invalid deletion_date: {}", e)))?
.with_timezone(&Utc);
Ok(TrashedItem::from_raw(
id,
original_id,
user_id,
item_type,
entry.name,
entry.original_path,
trashed_at,
deletion_date,
))
}
/// Converts a TrashedItem entity to a JSON entry
fn trashed_item_to_entry(&self, item: &TrashedItem) -> TrashedItemEntry {
TrashedItemEntry {
id: item.id().to_string(),
original_id: item.original_id().to_string(),
user_id: item.user_id().to_string(),
item_type: match item.item_type() {
TrashedItemType::File => "file".to_string(),
TrashedItemType::Folder => "folder".to_string(),
},
name: item.name().to_string(),
original_path: item.original_path().to_string(),
trashed_at: item.trashed_at().to_rfc3339(),
deletion_date: item.deletion_date().to_rfc3339(),
}
}
}
#[async_trait]
impl TrashRepository for TrashFsRepository {
#[instrument(skip(self))]
async fn add_to_trash(&self, item: &TrashedItem) -> Result<()> {
debug!(
"Adding item to trash: id={}, user={}",
item.id(),
item.user_id()
);
// Ensure the trash directory exists for this user
let user_trash_dir = self
.trash_dir
.join("files")
.join(item.user_id().to_string());
debug!("User trash directory path: {}", user_trash_dir.display());
// Create the user-specific trash directory
debug!(
"Creating user trash directory: {}",
user_trash_dir.display()
);
match fs::create_dir_all(&user_trash_dir).await {
Ok(_) => debug!("User trash directory created successfully"),
Err(e) => {
error!(
"Failed to create user trash directory {}: {}",
user_trash_dir.display(),
e
);
return Err(DomainError::new(
ErrorKind::InternalError,
"Trash",
format!("Failed to create user trash directory: {}", e),
));
}
}
// Log the current trash entries before adding the new one
let mut entries = self.get_trash_entries().await?;
debug!("Current trash entries count: {}", entries.len());
// Create the entry for the trash index
let entry = self.trashed_item_to_entry(item);
debug!(
"Created trash entry: id={}, original_id={}, name={}",
entry.id, entry.original_id, entry.name
);
// Add the entry to the index and save
entries.push(entry);
debug!("Saving updated trash index with {} entries", entries.len());
self.save_trash_entries(entries).await?;
debug!("Trash index updated successfully");
Ok(())
}
#[instrument(skip(self))]
async fn get_trash_items(&self, user_id: &Uuid) -> Result<Vec<TrashedItem>> {
debug!("Getting trash items for user: {}", user_id);
let entries = self.get_trash_entries().await?;
let user_id_str = user_id.to_string();
let user_entries = entries
.into_iter()
.filter(|entry| entry.user_id == user_id_str)
.collect::<Vec<_>>();
let mut items = Vec::new();
for entry in user_entries {
match self.entry_to_trashed_item(entry) {
Ok(item) => items.push(item),
Err(e) => error!("Error converting trash entry to item: {}", e),
}
}
Ok(items)
}
#[instrument(skip(self))]
async fn get_trash_item(&self, id: &Uuid, user_id: &Uuid) -> Result<Option<TrashedItem>> {
debug!("Looking for item in trash: id={}, user={}", id, user_id);
let entries = self.get_trash_entries().await?;
let id_str = id.to_string();
let user_id_str = user_id.to_string();
let item_entry = entries
.into_iter()
.find(|entry| entry.id == id_str && entry.user_id == user_id_str);
match item_entry {
Some(entry) => {
let item = self.entry_to_trashed_item(entry)?;
Ok(Some(item))
}
None => Ok(None),
}
}
#[instrument(skip(self))]
async fn restore_from_trash(&self, id: &Uuid, user_id: &Uuid) -> Result<()> {
debug!("Restoring item from trash: id={}, user={}", id, user_id);
let mut entries = self.get_trash_entries().await?;
let id_str = id.to_string();
let user_id_str = user_id.to_string();
let index = entries
.iter()
.position(|entry| entry.id == id_str && entry.user_id == user_id_str);
if let Some(index) = index {
entries.remove(index);
self.save_trash_entries(entries).await?;
Ok(())
} else {
Err(DomainError::not_found("TrashedItem", id.to_string()))
}
}
#[instrument(skip(self))]
async fn delete_permanently(&self, id: &Uuid, user_id: &Uuid) -> Result<()> {
debug!(
"Permanently deleting item from trash: id={}, user={}",
id, user_id
);
// Simply remove the entry from the index
// Physical files will be deleted through the corresponding repository
self.restore_from_trash(id, user_id).await
}
#[instrument(skip(self))]
async fn clear_trash(&self, user_id: &Uuid) -> Result<()> {
debug!("Clearing trash for user: {}", user_id);
let mut entries = self.get_trash_entries().await?;
let user_id_str = user_id.to_string();
entries.retain(|entry| entry.user_id != user_id_str);
self.save_trash_entries(entries).await?;
Ok(())
}
#[instrument(skip(self))]
async fn get_expired_items(&self) -> Result<Vec<TrashedItem>> {
debug!("Looking for expired trash items");
let entries = self.get_trash_entries().await?;
let now = Utc::now();
let mut expired_items = Vec::new();
for entry in entries {
match chrono::DateTime::parse_from_rfc3339(&entry.deletion_date) {
Ok(date) => {
let utc_date = date.with_timezone(&Utc);
if utc_date <= now {
match self.entry_to_trashed_item(entry) {
Ok(item) => expired_items.push(item),
Err(e) => error!("Error converting expired trash entry: {}", e),
}
}
}
Err(e) => error!("Invalid date format in trash entry: {}", e),
}
}
Ok(expired_items)
}
}