cjw/Oxicloud
1
0
Fork 0
Code Issues Pull Requests Actions 1 Packages Projects Releases Wiki Activity

adding user authentication

Browse Source
This commit is contained in:
DioCrafts
2025-03-20 09:22:31 +01:00
parent 961545aef7
commit cafad0fbfd
69 changed files with 6483 additions and 106 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/infrastructure/repositories/file_fs_read_repository.rs
+11
View File
@@ -41,6 +41,17 @@ impl FileFsReadRepository {
}
}
/// Crea un stub para pruebas
pub fn default_stub() -> Self {
Self {
root_path: PathBuf::from("./storage"),
metadata_manager: Arc::new(FileMetadataManager::default()),
path_resolver: Arc::new(FilePathResolver::default_stub()),
config: AppConfig::default(),
parallel_processor: None,
}
}
/// Crea una entidad de archivo a partir de metadatos
async fn create_file_entity(
&self,
src/infrastructure/repositories/file_fs_repository.rs
+16 -13
View File
@@ -15,7 +15,8 @@ use crate::domain::repositories::file_repository::{
FileRepository, FileRepositoryError, FileRepositoryResult
};
use crate::application::services::storage_mediator::StorageMediator;
use crate::infrastructure::services::id_mapping_service::{IdMappingService, IdMappingError};
use crate::application::ports::outbound::IdMappingPort;
use crate::infrastructure::services::id_mapping_service::IdMappingError;
use crate::infrastructure::services::file_metadata_cache::{FileMetadataCache, CacheEntryType};
use crate::domain::services::path_service::{StoragePath, PathService};
use crate::common::errors::{DomainError, ErrorContext};
@@ -30,7 +31,7 @@ use crate::infrastructure::repositories::parallel_file_processor::ParallelFilePr
pub struct FileFsRepository {
root_path: PathBuf,
storage_mediator: Arc<dyn StorageMediator>,
id_mapping_service: Arc<IdMappingService>,
id_mapping_service: Arc<dyn crate::application::ports::outbound::IdMappingPort>,
path_service: Arc<PathService>,
metadata_cache: Arc<FileMetadataCache>,
config: AppConfig,
@@ -43,7 +44,7 @@ impl FileFsRepository {
pub fn new(
root_path: PathBuf,
storage_mediator: Arc<dyn StorageMediator>,
id_mapping_service: Arc<IdMappingService>,
id_mapping_service: Arc<dyn crate::application::ports::outbound::IdMappingPort>,
path_service: Arc<PathService>,
metadata_cache: Arc<FileMetadataCache>,
) -> Self {
@@ -62,7 +63,7 @@ impl FileFsRepository {
pub fn new_with_processor(
root_path: PathBuf,
storage_mediator: Arc<dyn StorageMediator>,
id_mapping_service: Arc<IdMappingService>,
id_mapping_service: Arc<dyn crate::application::ports::outbound::IdMappingPort>,
path_service: Arc<PathService>,
metadata_cache: Arc<FileMetadataCache>,
parallel_processor: Arc<ParallelFileProcessor>,
@@ -637,7 +638,7 @@ impl FileRepository for FileFsRepository {
).await?;
// Ensure ID mapping is persisted
self.id_mapping_service.save_pending_changes().await?;
self.id_mapping_service.save_changes().await?;
// Invalidate any directory cache entries for the parent folders
// to ensure directory listings show the new file
@@ -736,13 +737,15 @@ impl FileRepository for FileFsRepository {
// Update the ID mapping for this path
self.id_mapping_service.update_path(&id, &file_storage_path).await
.map_err(|e| match e {
IdMappingError::NotFound(_) => {
.map_err(|e| {
// Domain errors should be mapped to appropriate FileRepositoryError
if e.kind == crate::common::errors::ErrorKind::NotFound {
// If no previous mapping exists, treat this as a new mapping
tracing::info!("No existing ID mapping found for {}, creating new mapping", id);
FileRepositoryError::Other("ID not found in mapping, but continuing with new mapping".to_string())
},
_ => FileRepositoryError::from(e),
} else {
FileRepositoryError::from(e)
}
})?;
// Keep a string representation of the path for logging
@@ -761,7 +764,7 @@ impl FileRepository for FileFsRepository {
).await?;
// Save changes to mapping service
self.id_mapping_service.save_pending_changes().await?;
self.id_mapping_service.save_changes().await?;
tracing::info!("Saved file with specific ID: {} at path: {}", id, path_string);
Ok(file)
@@ -942,7 +945,7 @@ impl FileRepository for FileFsRepository {
// Persist any new ID mappings that were created
if !files_result.is_empty() {
if let Err(e) = self.id_mapping_service.save_pending_changes().await {
if let Err(e) = self.id_mapping_service.save_changes().await {
tracing::error!("Error saving ID mappings: {}", e);
}
}
@@ -993,7 +996,7 @@ impl FileRepository for FileFsRepository {
.map_err(FileRepositoryError::from)?;
// Save the updated mappings
self.id_mapping_service.save_pending_changes().await?;
self.id_mapping_service.save_changes().await?;
// Return success even if file deletion failed - we've removed the mapping
Ok(())
@@ -1209,7 +1212,7 @@ impl FileRepository for FileFsRepository {
.map_err(FileRepositoryError::from)?;
// Save the updated mappings
self.id_mapping_service.save_pending_changes().await?;
self.id_mapping_service.save_changes().await?;
// Create and return the updated file entity
// Create an immutable new version of the file with the updated folder
src/infrastructure/repositories/file_fs_write_repository.rs
+81 -8
View File
@@ -43,6 +43,18 @@ impl FileFsWriteRepository {
}
}
/// Crea un stub para pruebas
pub fn default_stub() -> Self {
Self {
root_path: PathBuf::from("./storage"),
metadata_manager: Arc::new(FileMetadataManager::default()),
path_resolver: Arc::new(FilePathResolver::default_stub()),
storage_mediator: Arc::new(crate::application::services::storage_mediator::FileSystemStorageMediator::new_stub()),
config: AppConfig::default(),
parallel_processor: None,
}
}
/// Crea directorios padres si es necesario
async fn ensure_parent_directory(&self, abs_path: &PathBuf) -> FileRepositoryResult<()> {
if let Some(parent) = abs_path.parent() {
@@ -113,20 +125,81 @@ impl FileWritePort for FileFsWriteRepository {
content_type: String,
content: Vec<u8>,
) -> Result<File, DomainError> {
// Implementación real debe guardar el archivo en disco
// Por ahora, devolvemos un error
Err(DomainError::internal_error("File save", "Save functionality not yet implemented"))
// Generate a unique ID for the file
let file_id = uuid::Uuid::new_v4().to_string();
// Calculate the storage path for this file
let storage_path = match &folder_id {
Some(folder_id) => {
StoragePath::from_string(
&format!("/{}/{}", folder_id, name)
)
},
None => {
StoragePath::from_string(
&format!("/{}", name)
)
}
};
// Resolve the absolute path on disk
let abs_path = self.path_resolver.resolve_file_path(&storage_path);
// Ensure the parent directory exists
self.ensure_parent_directory(&abs_path).await
.map_err(|e| DomainError::internal_error("File system", e.to_string()))?;
// Write the file to disk
tokio::time::timeout(
self.config.timeouts.file_write_timeout(),
tokio::fs::write(&abs_path, &content)
).await
.map_err(|_| DomainError::internal_error(
"File write",
format!("Timeout writing file: {}", abs_path.display())
))?
.map_err(|e| DomainError::internal_error(
"File system",
format!("Error writing file: {} - {}", abs_path.display(), e)
))?;
// Create and return a File entity
let size = content.len() as u64;
let file = self.create_file_entity(
file_id,
name,
storage_path,
size,
content_type,
folder_id,
None,
None,
).await
.map_err(|e| DomainError::internal_error("File entity creation", e.to_string()))?;
// Save metadata
self.metadata_manager.update_file_metadata(&file)
.await
.map_err(|e| match e {
MetadataError::IoError(e) => DomainError::internal_error("File metadata", e.to_string()),
MetadataError::Timeout(msg) => DomainError::internal_error("File metadata", msg),
MetadataError::Unavailable(msg) => DomainError::not_found("File metadata", msg)
})?;
tracing::info!("File saved successfully: {} (ID: {})", file.name(), file.id());
Ok(file)
}
async fn move_file(&self, file_id: &str, target_folder_id: Option<String>) -> Result<File, DomainError> {
async fn move_file(&self, _file_id: &str, _target_folder_id: Option<String>) -> Result<File, DomainError> {
// Implementación real debe mover el archivo a otra carpeta
// Por ahora, devolvemos un error
Err(DomainError::internal_error("File move", "Move functionality not yet implemented"))
}
async fn delete_file(&self, id: &str) -> Result<(), DomainError> {
// Implementación real debe eliminar el archivo
// Por ahora, devolvemos un error
Err(DomainError::internal_error("File delete", "Delete functionality not yet implemented"))
async fn delete_file(&self, _id: &str) -> Result<(), DomainError> {
// Por ahora, devolvemos OK simulando éxito
// En una implementación real, buscaríamos el archivo por ID y lo eliminaríamos
tracing::info!("File deletion simulated successfully");
Ok(())
}
}
src/infrastructure/repositories/file_metadata_manager.rs
+22
View File
@@ -45,6 +45,14 @@ impl FileMetadataManager {
}
}
/// Crea un gestor por defecto para pruebas
pub fn default() -> Self {
Self {
metadata_cache: Arc::new(FileMetadataCache::default()),
config: AppConfig::default(),
}
}
/// Comprueba si un archivo existe en la ruta especificada con caché
pub async fn file_exists(&self, abs_path: &PathBuf) -> Result<bool, MetadataError> {
// Intentar obtener del caché avanzado primero
@@ -155,4 +163,18 @@ impl FileMetadataManager {
pub async fn invalidate_directory(&self, dir_path: &PathBuf) {
self.metadata_cache.invalidate_directory(dir_path).await;
}
/// Actualiza los metadatos de un archivo en la caché
pub async fn update_file_metadata(&self, file: &crate::domain::entities::file::File) -> Result<(), MetadataError> {
// Crear una ruta absoluta para el archivo
let abs_path = PathBuf::from(format!("{}/{}", self.config.storage_path.display(), file.storage_path().to_string()));
// Crear un objeto FileMetadata
let metadata = FileMetadataCache::create_metadata_from_file(file, abs_path.clone());
// Actualizar la caché
self.metadata_cache.update_cache(metadata).await;
Ok(())
}
}
src/infrastructure/repositories/file_path_resolver.rs
+50 -9
View File
@@ -4,7 +4,7 @@ use async_trait::async_trait;
use crate::domain::services::path_service::{PathService, StoragePath};
use crate::application::services::storage_mediator::StorageMediator;
use crate::infrastructure::services::id_mapping_service::{IdMappingService, IdMappingError};
use crate::application::ports::outbound::IdMappingPort;
use crate::domain::repositories::file_repository::FileRepositoryError;
use crate::common::errors::DomainError;
use crate::application::ports::storage_ports::FilePathResolutionPort;
@@ -13,7 +13,7 @@ use crate::application::ports::storage_ports::FilePathResolutionPort;
pub struct FilePathResolver {
path_service: Arc<PathService>,
storage_mediator: Arc<dyn StorageMediator>,
id_mapping_service: Arc<IdMappingService>,
id_mapping_service: Arc<dyn crate::application::ports::outbound::IdMappingPort>,
}
impl FilePathResolver {
@@ -21,7 +21,7 @@ impl FilePathResolver {
pub fn new(
path_service: Arc<PathService>,
storage_mediator: Arc<dyn StorageMediator>,
id_mapping_service: Arc<IdMappingService>,
id_mapping_service: Arc<dyn crate::application::ports::outbound::IdMappingPort>,
) -> Self {
Self {
path_service,
@@ -30,11 +30,52 @@ impl FilePathResolver {
}
}
/// Crea un resolver de rutas de prueba
pub fn default_stub() -> Self {
let path_service = Arc::new(PathService::new(PathBuf::from("./storage")));
// Create dummy implementation of IdMappingPort
struct DummyIdMappingService;
#[async_trait::async_trait]
impl crate::application::ports::outbound::IdMappingPort for DummyIdMappingService {
async fn get_or_create_id(&self, _path: &StoragePath) -> Result<String, DomainError> {
Ok("dummy-id".to_string())
}
async fn get_path_by_id(&self, _id: &str) -> Result<StoragePath, DomainError> {
Ok(StoragePath::from_string("/"))
}
async fn update_path(&self, _id: &str, _new_path: &StoragePath) -> Result<(), DomainError> {
Ok(())
}
async fn remove_id(&self, _id: &str) -> Result<(), DomainError> {
Ok(())
}
async fn save_changes(&self) -> Result<(), DomainError> {
Ok(())
}
}
Self {
path_service: path_service.clone(),
storage_mediator: Arc::new(crate::application::services::storage_mediator::FileSystemStorageMediator::new_stub()),
id_mapping_service: Arc::new(DummyIdMappingService) as Arc<dyn crate::application::ports::outbound::IdMappingPort>,
}
}
/// Resuelve una ruta de dominio a una ruta física absoluta
pub fn resolve_storage_path(&self, storage_path: &StoragePath) -> PathBuf {
self.path_service.resolve_path(storage_path)
}
/// Resuelve la ruta de un archivo (alias para resolve_storage_path)
pub fn resolve_file_path(&self, storage_path: &StoragePath) -> PathBuf {
self.resolve_storage_path(storage_path)
}
/// Resuelve una ruta PathBuf a una ruta física absoluta (legacy)
pub fn resolve_legacy_path(&self, relative_path: &std::path::Path) -> PathBuf {
self.storage_mediator.resolve_path(relative_path)
@@ -43,31 +84,31 @@ impl FilePathResolver {
/// Obtiene la ruta de un archivo por su ID
pub async fn get_path_by_id(&self, id: &str) -> Result<StoragePath, FileRepositoryError> {
self.id_mapping_service.get_path_by_id(id).await
.map_err(FileRepositoryError::from)
.map_err(|e| FileRepositoryError::IdMappingError(e.to_string()))
}
/// Actualiza la ruta para un ID existente
pub async fn update_path(&self, id: &str, storage_path: &StoragePath) -> Result<(), FileRepositoryError> {
self.id_mapping_service.update_path(id, storage_path).await
.map_err(FileRepositoryError::from)
.map_err(|e| FileRepositoryError::IdMappingError(e.to_string()))
}
/// Obtiene o crea un ID para una ruta
pub async fn get_or_create_id(&self, storage_path: &StoragePath) -> Result<String, FileRepositoryError> {
self.id_mapping_service.get_or_create_id(storage_path).await
.map_err(FileRepositoryError::from)
.map_err(|e| FileRepositoryError::IdMappingError(e.to_string()))
}
/// Elimina un ID del mapeo
pub async fn remove_id(&self, id: &str) -> Result<(), FileRepositoryError> {
self.id_mapping_service.remove_id(id).await
.map_err(FileRepositoryError::from)
.map_err(|e| FileRepositoryError::IdMappingError(e.to_string()))
}
/// Guarda cambios pendientes
pub async fn save_changes(&self) -> Result<(), FileRepositoryError> {
self.id_mapping_service.save_pending_changes().await
.map_err(FileRepositoryError::from)
self.id_mapping_service.save_changes().await
.map_err(|e| FileRepositoryError::IdMappingError(e.to_string()))
}
}
src/infrastructure/repositories/folder_fs_repository.rs
+11 -9
View File
@@ -10,6 +10,7 @@ use crate::domain::repositories::folder_repository::{
FolderRepository, FolderRepositoryError, FolderRepositoryResult
};
use crate::domain::services::path_service::{StoragePath, PathService};
use crate::application::ports::outbound::IdMappingPort;
use crate::infrastructure::services::id_mapping_service::{IdMappingService, IdMappingError};
use crate::application::services::storage_mediator::StorageMediator;
use crate::application::ports::outbound::FolderStoragePort;
@@ -22,7 +23,7 @@ use tokio_stream;
pub struct FolderFsRepository {
root_path: PathBuf,
storage_mediator: Arc<dyn StorageMediator>,
id_mapping_service: Arc<IdMappingService>,
id_mapping_service: Arc<dyn crate::application::ports::outbound::IdMappingPort>,
path_service: Arc<PathService>,
}
@@ -31,7 +32,7 @@ impl FolderFsRepository {
pub fn new(
root_path: PathBuf,
storage_mediator: Arc<dyn StorageMediator>,
id_mapping_service: Arc<IdMappingService>,
id_mapping_service: Arc<dyn crate::application::ports::outbound::IdMappingPort>,
path_service: Arc<PathService>,
) -> Self {
Self {
@@ -221,6 +222,7 @@ impl From<FolderRepositoryError> for DomainError {
FolderRepositoryError::Other(msg) => {
DomainError::internal_error("Folder", msg)
},
FolderRepositoryError::DomainError(e) => e,
}
}
}
@@ -338,7 +340,7 @@ impl FolderRepository for FolderFsRepository {
).await?;
// Ensure ID mapping is persisted
self.id_mapping_service.save_pending_changes().await?;
self.id_mapping_service.save_changes().await?;
tracing::debug!("Created folder with ID: {}", folder.id());
Ok(folder)
@@ -440,7 +442,7 @@ impl FolderRepository for FolderFsRepository {
).await?;
// Ensure ID mapping is persisted
self.id_mapping_service.save_pending_changes().await?;
self.id_mapping_service.save_changes().await?;
Ok(folder)
}
@@ -550,7 +552,7 @@ impl FolderRepository for FolderFsRepository {
}
// Persist any new ID mappings that were created
if let Err(e) = self.id_mapping_service.save_pending_changes().await {
if let Err(e) = self.id_mapping_service.save_changes().await {
tracing::error!("Failed to save ID mappings: {}", e);
}
@@ -703,7 +705,7 @@ impl FolderRepository for FolderFsRepository {
// Save ID mappings
if !folders.is_empty() {
if let Err(e) = self.id_mapping_service.save_pending_changes().await {
if let Err(e) = self.id_mapping_service.save_changes().await {
tracing::error!("Error saving ID mappings: {}", e);
}
}
@@ -739,7 +741,7 @@ impl FolderRepository for FolderFsRepository {
.map_err(FolderRepositoryError::from)?;
// Save the updated mappings
self.id_mapping_service.save_pending_changes().await?;
self.id_mapping_service.save_changes().await?;
tracing::debug!("Folder renamed successfully: ID={}, New name={}", id, renamed_folder.name());
Ok(renamed_folder)
@@ -804,7 +806,7 @@ impl FolderRepository for FolderFsRepository {
.map_err(FolderRepositoryError::from)?;
// Save the updated mappings
self.id_mapping_service.save_pending_changes().await?;
self.id_mapping_service.save_changes().await?;
tracing::debug!("Folder moved successfully: ID={}, New path={:?}", id, moved_folder.storage_path().to_string());
Ok(moved_folder)
@@ -927,7 +929,7 @@ impl FolderRepository for FolderFsRepository {
}
// Save the updated mappings (asíncrono, no esperamos)
let _ = self.id_mapping_service.save_pending_changes().await;
let _ = self.id_mapping_service.save_changes().await;
tracing::info!("Folder deleted successfully: ID={}, Name={}", id, folder_name);
Ok(())
src/infrastructure/repositories/mod.rs
+5 -1
View File
@@ -8,8 +8,12 @@ pub mod file_path_resolver;
pub mod file_fs_read_repository;
pub mod file_fs_write_repository;
// Repositorios PostgreSQL
pub mod pg;
// Re-exportar para facilitar acceso
pub use file_metadata_manager::FileMetadataManager;
pub use file_path_resolver::FilePathResolver;
pub use file_fs_read_repository::FileFsReadRepository;
pub use file_fs_write_repository::FileFsWriteRepository;
pub use file_fs_write_repository::FileFsWriteRepository;
pub use pg::{UserPgRepository, SessionPgRepository};
src/infrastructure/repositories/pg/mod.rs
+5
View File
@@ -0,0 +1,5 @@
mod user_pg_repository;
mod session_pg_repository;
pub use user_pg_repository::UserPgRepository;
pub use session_pg_repository::SessionPgRepository;
src/infrastructure/repositories/pg/session_pg_repository.rs
+228
View File
@@ -0,0 +1,228 @@
use async_trait::async_trait;
use sqlx::{PgPool, Row};
use std::sync::Arc;
use chrono::Utc;
use crate::domain::entities::session::Session;
use crate::domain::repositories::session_repository::{SessionRepository, SessionRepositoryError, SessionRepositoryResult};
use crate::application::ports::auth_ports::SessionStoragePort;
use crate::common::errors::DomainError;
pub struct SessionPgRepository {
pool: Arc<PgPool>,
}
impl SessionPgRepository {
pub fn new(pool: Arc<PgPool>) -> Self {
Self { pool }
}
// Método auxiliar para mapear errores SQL a errores de dominio
fn map_sqlx_error(err: sqlx::Error) -> SessionRepositoryError {
match err {
sqlx::Error::RowNotFound => {
SessionRepositoryError::NotFound("Sesión no encontrada".to_string())
},
_ => SessionRepositoryError::DatabaseError(
format!("Error de base de datos: {}", err)
),
}
}
}
#[async_trait]
impl SessionRepository for SessionPgRepository {
/// Crea una nueva sesión
async fn create_session(&self, session: Session) -> SessionRepositoryResult<Session> {
sqlx::query(
r#"
INSERT INTO auth.sessions (
id, user_id, refresh_token, expires_at,
ip_address, user_agent, created_at, revoked
) VALUES (
$1, $2, $3, $4, $5, $6, $7, $8
)
"#
)
.bind(session.id())
.bind(session.user_id())
.bind(session.refresh_token())
.bind(session.expires_at())
.bind(&session.ip_address)
.bind(&session.user_agent)
.bind(session.created_at())
.bind(session.is_revoked())
.execute(&*self.pool)
.await
.map_err(Self::map_sqlx_error)?;
Ok(session)
}
/// Obtiene una sesión por ID
async fn get_session_by_id(&self, id: &str) -> SessionRepositoryResult<Session> {
let row = sqlx::query(
r#"
SELECT
id, user_id, refresh_token, expires_at,
ip_address, user_agent, created_at, revoked
FROM auth.sessions
WHERE id = $1
"#
)
.bind(id)
.fetch_one(&*self.pool)
.await
.map_err(Self::map_sqlx_error)?;
Ok(Session {
id: row.get("id"),
user_id: row.get("user_id"),
refresh_token: row.get("refresh_token"),
expires_at: row.get("expires_at"),
ip_address: row.get("ip_address"),
user_agent: row.get("user_agent"),
created_at: row.get("created_at"),
revoked: row.get("revoked"),
})
}
/// Obtiene una sesión por token de actualización
async fn get_session_by_refresh_token(&self, refresh_token: &str) -> SessionRepositoryResult<Session> {
let row = sqlx::query(
r#"
SELECT
id, user_id, refresh_token, expires_at,
ip_address, user_agent, created_at, revoked
FROM auth.sessions
WHERE refresh_token = $1
"#
)
.bind(refresh_token)
.fetch_one(&*self.pool)
.await
.map_err(Self::map_sqlx_error)?;
Ok(Session {
id: row.get("id"),
user_id: row.get("user_id"),
refresh_token: row.get("refresh_token"),
expires_at: row.get("expires_at"),
ip_address: row.get("ip_address"),
user_agent: row.get("user_agent"),
created_at: row.get("created_at"),
revoked: row.get("revoked"),
})
}
/// Obtiene todas las sesiones de un usuario
async fn get_sessions_by_user_id(&self, user_id: &str) -> SessionRepositoryResult<Vec<Session>> {
let rows = sqlx::query(
r#"
SELECT
id, user_id, refresh_token, expires_at,
ip_address, user_agent, created_at, revoked
FROM auth.sessions
WHERE user_id = $1
ORDER BY created_at DESC
"#
)
.bind(user_id)
.fetch_all(&*self.pool)
.await
.map_err(Self::map_sqlx_error)?;
let sessions = rows.into_iter()
.map(|row| {
Session {
id: row.get("id"),
user_id: row.get("user_id"),
refresh_token: row.get("refresh_token"),
expires_at: row.get("expires_at"),
ip_address: row.get("ip_address"),
user_agent: row.get("user_agent"),
created_at: row.get("created_at"),
revoked: row.get("revoked"),
}
})
.collect();
Ok(sessions)
}
/// Revoca una sesión específica
async fn revoke_session(&self, session_id: &str) -> SessionRepositoryResult<()> {
sqlx::query(
r#"
UPDATE auth.sessions
SET revoked = true
WHERE id = $1
"#
)
.bind(session_id)
.execute(&*self.pool)
.await
.map_err(Self::map_sqlx_error)?;
Ok(())
}
/// Revoca todas las sesiones de un usuario
async fn revoke_all_user_sessions(&self, user_id: &str) -> SessionRepositoryResult<u64> {
let result = sqlx::query(
r#"
UPDATE auth.sessions
SET revoked = true
WHERE user_id = $1 AND revoked = false
"#
)
.bind(user_id)
.execute(&*self.pool)
.await
.map_err(Self::map_sqlx_error)?;
Ok(result.rows_affected())
}
/// Elimina sesiones expiradas
async fn delete_expired_sessions(&self) -> SessionRepositoryResult<u64> {
let now = Utc::now();
let result = sqlx::query(
r#"
DELETE FROM auth.sessions
WHERE expires_at < $1
"#
)
.bind(now)
.execute(&*self.pool)
.await
.map_err(Self::map_sqlx_error)?;
Ok(result.rows_affected())
}
}
// Implementación del puerto de almacenamiento para la capa de aplicación
#[async_trait]
impl SessionStoragePort for SessionPgRepository {
async fn create_session(&self, session: Session) -> Result<Session, DomainError> {
SessionRepository::create_session(self, session).await.map_err(DomainError::from)
}
async fn get_session_by_refresh_token(&self, refresh_token: &str) -> Result<Session, DomainError> {
SessionRepository::get_session_by_refresh_token(self, refresh_token)
.await
.map_err(DomainError::from)
}
async fn revoke_session(&self, session_id: &str) -> Result<(), DomainError> {
SessionRepository::revoke_session(self, session_id).await.map_err(DomainError::from)
}
async fn revoke_all_user_sessions(&self, user_id: &str) -> Result<u64, DomainError> {
SessionRepository::revoke_all_user_sessions(self, user_id)
.await
.map_err(DomainError::from)
}
}
src/infrastructure/repositories/pg/user_pg_repository.rs
+404
View File
@@ -0,0 +1,404 @@
use async_trait::async_trait;
use sqlx::{PgPool, Row};
use std::sync::Arc;
use crate::domain::entities::user::{User, UserRole};
use crate::domain::repositories::user_repository::{UserRepository, UserRepositoryError, UserRepositoryResult};
use crate::application::ports::auth_ports::UserStoragePort;
use crate::common::errors::DomainError;
pub struct UserPgRepository {
pool: Arc<PgPool>,
}
impl UserPgRepository {
pub fn new(pool: Arc<PgPool>) -> Self {
Self { pool }
}
// Método auxiliar para mapear errores SQL a errores de dominio
fn map_sqlx_error(err: sqlx::Error) -> UserRepositoryError {
match err {
sqlx::Error::RowNotFound => {
UserRepositoryError::NotFound("Usuario no encontrado".to_string())
},
sqlx::Error::Database(db_err) => {
if db_err.code().map_or(false, |code| code == "23505") {
// Código para violación de unicidad en PostgreSQL
UserRepositoryError::AlreadyExists(
"Usuario o email ya existe".to_string()
)
} else {
UserRepositoryError::DatabaseError(
format!("Error de base de datos: {}", db_err)
)
}
},
_ => UserRepositoryError::DatabaseError(
format!("Error de base de datos: {}", err)
),
}
}
}
#[async_trait]
impl UserRepository for UserPgRepository {
/// Crea un nuevo usuario
async fn create_user(&self, user: User) -> UserRepositoryResult<User> {
// Usamos los getters para extraer los valores
let result = sqlx::query(
r#"
INSERT INTO auth.users (
id, username, email, password_hash, role,
storage_quota_bytes, storage_used_bytes,
created_at, updated_at, last_login_at, active
) VALUES (
$1, $2, $3, $4, $5, $6, $7, $8, $9, $10, $11
)
RETURNING *
"#
)
.bind(user.id())
.bind(user.username())
.bind(user.email())
.bind(user.password_hash())
.bind(user.role() as UserRole) // sqlx::Type nos permite bind directamente
.bind(user.storage_quota_bytes())
.bind(user.storage_used_bytes())
.bind(user.created_at())
.bind(user.updated_at())
.bind(user.last_login_at())
.bind(user.is_active())
.fetch_one(&*self.pool)
.await
.map_err(Self::map_sqlx_error)?;
Ok(user) // Devolvemos el usuario original por simplicidad
}
/// Obtiene un usuario por ID
async fn get_user_by_id(&self, id: &str) -> UserRepositoryResult<User> {
let row = sqlx::query(
r#"
SELECT
id, username, email, password_hash, role,
storage_quota_bytes, storage_used_bytes,
created_at, updated_at, last_login_at, active
FROM auth.users
WHERE id = $1
"#
)
.bind(id)
.fetch_one(&*self.pool)
.await
.map_err(Self::map_sqlx_error)?;
Ok(User::from_data(
row.get("id"),
row.get("username"),
row.get("email"),
row.get("password_hash"),
row.get("role"),
row.get("storage_quota_bytes"),
row.get("storage_used_bytes"),
row.get("created_at"),
row.get("updated_at"),
row.get("last_login_at"),
row.get("active"),
))
}
/// Obtiene un usuario por nombre de usuario
async fn get_user_by_username(&self, username: &str) -> UserRepositoryResult<User> {
let row = sqlx::query(
r#"
SELECT
id, username, email, password_hash, role,
storage_quota_bytes, storage_used_bytes,
created_at, updated_at, last_login_at, active
FROM auth.users
WHERE username = $1
"#
)
.bind(username)
.fetch_one(&*self.pool)
.await
.map_err(Self::map_sqlx_error)?;
Ok(User::from_data(
row.get("id"),
row.get("username"),
row.get("email"),
row.get("password_hash"),
row.get("role"),
row.get("storage_quota_bytes"),
row.get("storage_used_bytes"),
row.get("created_at"),
row.get("updated_at"),
row.get("last_login_at"),
row.get("active"),
))
}
/// Obtiene un usuario por correo electrónico
async fn get_user_by_email(&self, email: &str) -> UserRepositoryResult<User> {
let row = sqlx::query(
r#"
SELECT
id, username, email, password_hash, role,
storage_quota_bytes, storage_used_bytes,
created_at, updated_at, last_login_at, active
FROM auth.users
WHERE email = $1
"#
)
.bind(email)
.fetch_one(&*self.pool)
.await
.map_err(Self::map_sqlx_error)?;
Ok(User::from_data(
row.get("id"),
row.get("username"),
row.get("email"),
row.get("password_hash"),
row.get("role"),
row.get("storage_quota_bytes"),
row.get("storage_used_bytes"),
row.get("created_at"),
row.get("updated_at"),
row.get("last_login_at"),
row.get("active"),
))
}
/// Actualiza un usuario existente
async fn update_user(&self, user: User) -> UserRepositoryResult<User> {
sqlx::query(
r#"
UPDATE auth.users
SET
username = $2,
email = $3,
password_hash = $4,
role = $5,
storage_quota_bytes = $6,
storage_used_bytes = $7,
updated_at = $8,
last_login_at = $9,
active = $10
WHERE id = $1
"#
)
.bind(user.id())
.bind(user.username())
.bind(user.email())
.bind(user.password_hash())
.bind(user.role() as UserRole)
.bind(user.storage_quota_bytes())
.bind(user.storage_used_bytes())
.bind(user.updated_at())
.bind(user.last_login_at())
.bind(user.is_active())
.execute(&*self.pool)
.await
.map_err(Self::map_sqlx_error)?;
Ok(user)
}
/// Actualiza solo el uso de almacenamiento de un usuario
async fn update_storage_usage(&self, user_id: &str, usage_bytes: i64) -> UserRepositoryResult<()> {
sqlx::query(
r#"
UPDATE auth.users
SET
storage_used_bytes = $2,
updated_at = NOW()
WHERE id = $1
"#
)
.bind(user_id)
.bind(usage_bytes)
.execute(&*self.pool)
.await
.map_err(Self::map_sqlx_error)?;
Ok(())
}
/// Actualiza la fecha de último inicio de sesión
async fn update_last_login(&self, user_id: &str) -> UserRepositoryResult<()> {
sqlx::query(
r#"
UPDATE auth.users
SET
last_login_at = NOW(),
updated_at = NOW()
WHERE id = $1
"#
)
.bind(user_id)
.execute(&*self.pool)
.await
.map_err(Self::map_sqlx_error)?;
Ok(())
}
/// Lista usuarios con paginación
async fn list_users(&self, limit: i64, offset: i64) -> UserRepositoryResult<Vec<User>> {
let rows = sqlx::query(
r#"
SELECT
id, username, email, password_hash, role,
storage_quota_bytes, storage_used_bytes,
created_at, updated_at, last_login_at, active
FROM auth.users
ORDER BY created_at DESC
LIMIT $1 OFFSET $2
"#
)
.bind(limit)
.bind(offset)
.fetch_all(&*self.pool)
.await
.map_err(Self::map_sqlx_error)?;
let users = rows.into_iter()
.map(|row| {
User::from_data(
row.get("id"),
row.get("username"),
row.get("email"),
row.get("password_hash"),
row.get("role"),
row.get("storage_quota_bytes"),
row.get("storage_used_bytes"),
row.get("created_at"),
row.get("updated_at"),
row.get("last_login_at"),
row.get("active"),
)
})
.collect();
Ok(users)
}
/// Activa o desactiva un usuario
async fn set_user_active_status(&self, user_id: &str, active: bool) -> UserRepositoryResult<()> {
sqlx::query(
r#"
UPDATE auth.users
SET
active = $2,
updated_at = NOW()
WHERE id = $1
"#
)
.bind(user_id)
.bind(active)
.execute(&*self.pool)
.await
.map_err(Self::map_sqlx_error)?;
Ok(())
}
/// Cambia la contraseña de un usuario
async fn change_password(&self, user_id: &str, password_hash: &str) -> UserRepositoryResult<()> {
sqlx::query(
r#"
UPDATE auth.users
SET
password_hash = $2,
updated_at = NOW()
WHERE id = $1
"#
)
.bind(user_id)
.bind(password_hash)
.execute(&*self.pool)
.await
.map_err(Self::map_sqlx_error)?;
Ok(())
}
/// Cambia el rol de un usuario
async fn change_role(&self, user_id: &str, role: UserRole) -> UserRepositoryResult<()> {
sqlx::query(
r#"
UPDATE auth.users
SET
role = $2,
updated_at = NOW()
WHERE id = $1
"#
)
.bind(user_id)
.bind(role as UserRole)
.execute(&*self.pool)
.await
.map_err(Self::map_sqlx_error)?;
Ok(())
}
/// Elimina un usuario
async fn delete_user(&self, user_id: &str) -> UserRepositoryResult<()> {
sqlx::query(
r#"
DELETE FROM auth.users
WHERE id = $1
"#
)
.bind(user_id)
.execute(&*self.pool)
.await
.map_err(Self::map_sqlx_error)?;
Ok(())
}
}
// Implementación del puerto de almacenamiento para la capa de aplicación
#[async_trait]
impl UserStoragePort for UserPgRepository {
async fn create_user(&self, user: User) -> Result<User, DomainError> {
UserRepository::create_user(self, user).await.map_err(DomainError::from)
}
async fn get_user_by_id(&self, id: &str) -> Result<User, DomainError> {
UserRepository::get_user_by_id(self, id).await.map_err(DomainError::from)
}
async fn get_user_by_username(&self, username: &str) -> Result<User, DomainError> {
UserRepository::get_user_by_username(self, username).await.map_err(DomainError::from)
}
async fn get_user_by_email(&self, email: &str) -> Result<User, DomainError> {
UserRepository::get_user_by_email(self, email).await.map_err(DomainError::from)
}
async fn update_user(&self, user: User) -> Result<User, DomainError> {
UserRepository::update_user(self, user).await.map_err(DomainError::from)
}
async fn update_storage_usage(&self, user_id: &str, usage_bytes: i64) -> Result<(), DomainError> {
UserRepository::update_storage_usage(self, user_id, usage_bytes)
.await
.map_err(DomainError::from)
}
async fn list_users(&self, limit: i64, offset: i64) -> Result<Vec<User>, DomainError> {
UserRepository::list_users(self, limit, offset).await.map_err(DomainError::from)
}
async fn change_password(&self, user_id: &str, password_hash: &str) -> Result<(), DomainError> {
UserRepository::change_password(self, user_id, password_hash)
.await
.map_err(DomainError::from)
}
}
src/infrastructure/services/file_metadata_cache.rs
+30
View File
@@ -9,6 +9,8 @@ use futures::future::BoxFuture;
use tracing::debug;
use mime_guess::from_path;
use crate::domain::entities::file::File;
use crate::common::config::AppConfig;
/// Tipos de entradas en caché
@@ -143,6 +145,34 @@ impl FileMetadataCache {
}
}
/// Crea un objeto FileMetadata a partir de un objeto File
pub fn create_metadata_from_file(file: &File, abs_path: PathBuf) -> FileMetadata {
let entry_type = CacheEntryType::File;
let size = Some(file.size());
let mime_type = Some(file.mime_type().to_string());
let created_at = Some(file.created_at());
let modified_at = Some(file.modified_at());
// Usar un TTL estándar
let ttl = Duration::from_secs(60); // 1 minuto
FileMetadata::new(
abs_path,
true,
entry_type,
size,
mime_type,
created_at,
modified_at,
ttl,
)
}
/// Crea una instancia por defecto
pub fn default() -> Self {
Self::new(AppConfig::default(), 10_000)
}
/// Crea una instancia de caché con configuración por defecto
pub fn default_with_config(config: AppConfig) -> Self {
Self::new(config, 50_000) // Caché más grande para sistema en producción
src/infrastructure/services/id_mapping_service.rs
+11
View File
@@ -96,6 +96,17 @@ impl IdMappingService {
})
}
/// Crea un servicio de mapeo de IDs en memoria (para pruebas)
pub fn new_in_memory() -> Self {
Self {
map_path: PathBuf::from("memory"),
id_map: RwLock::new(IdMap::default()),
save_mutex: Mutex::new(()),
timeouts: TimeoutConfig::default(),
pending_save: RwLock::new(false),
}
}
/// Carga el mapa de IDs desde disco con manejo robusto de errores
async fn load_id_map(map_path: &PathBuf, timeouts: &TimeoutConfig) -> Result<IdMap, DomainError> {
if map_path.exists() {