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

improve postgresql performance

Browse Source
This commit is contained in:
DioCrafts
2025-04-09 00:21:20 +02:00
parent 7473ee438f
commit 8f1d213526
36 changed files with 2191 additions and 233 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/infrastructure/repositories/pg/mod.rs
+2 -1
View File
@@ -1,5 +1,6 @@
mod user_pg_repository;
mod session_pg_repository;
mod transaction_utils;
pub use user_pg_repository::UserPgRepository;
pub use session_pg_repository::SessionPgRepository;
pub use session_pg_repository::SessionPgRepository;
src/infrastructure/repositories/pg/session_pg_repository.rs
+138 -53
View File
@@ -1,12 +1,21 @@
use async_trait::async_trait;
use sqlx::{PgPool, Row};
use sqlx::{PgPool, Row, Executor};
use std::sync::Arc;
use chrono::Utc;
use futures::future::BoxFuture;
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;
use crate::infrastructure::repositories::pg::transaction_utils::with_transaction;
// Implementar From<sqlx::Error> para SessionRepositoryError para permitir conversiones automáticas
impl From<sqlx::Error> for SessionRepositoryError {
fn from(err: sqlx::Error) -> Self {
SessionPgRepository::map_sqlx_error(err)
}
}
pub struct SessionPgRepository {
pool: Arc<PgPool>,
@@ -18,7 +27,7 @@ impl SessionPgRepository {
}
// Método auxiliar para mapear errores SQL a errores de dominio
fn map_sqlx_error(err: sqlx::Error) -> SessionRepositoryError {
pub fn map_sqlx_error(err: sqlx::Error) -> SessionRepositoryError {
match err {
sqlx::Error::RowNotFound => {
SessionRepositoryError::NotFound("Sesión no encontrada".to_string())
@@ -32,30 +41,66 @@ impl SessionPgRepository {
#[async_trait]
impl SessionRepository for SessionPgRepository {
/// Crea una nueva sesión
/// Crea una nueva sesión utilizando una transacció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)?;
// Crear una copia de la sesión para el closure
let session_clone = session.clone();
with_transaction(
&self.pool,
"create_session",
|tx| {
Box::pin(async move {
// Insertar la sesión
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_clone.id())
.bind(session_clone.user_id())
.bind(session_clone.refresh_token())
.bind(session_clone.expires_at())
.bind(&session_clone.ip_address)
.bind(&session_clone.user_agent)
.bind(session_clone.created_at())
.bind(session_clone.is_revoked())
.execute(&mut **tx)
.await
.map_err(Self::map_sqlx_error)?;
// Opcionalmente, actualizar el último login del usuario
// dentro de la misma transacción
sqlx::query(
r#"
UPDATE auth.users
SET last_login_at = NOW(), updated_at = NOW()
WHERE id = $1
"#
)
.bind(session_clone.user_id())
.execute(&mut **tx)
.await
.map_err(|e| {
// Convertimos el error pero sin interrumpir la creación
// de la sesión si falla la actualización
tracing::warn!("No se pudo actualizar last_login_at para usuario {}: {}",
session_clone.user_id(), e);
SessionRepositoryError::DatabaseError(format!(
"Sesión creada pero no se pudo actualizar last_login_at: {}", e
))
})?;
Ok(session_clone)
}) as BoxFuture<'_, SessionRepositoryResult<Session>>
}
).await?;
Ok(session)
}
@@ -150,38 +195,78 @@ impl SessionRepository for SessionPgRepository {
Ok(sessions)
}
/// Revoca una sesión específica
/// Revoca una sesión específica utilizando una transacción
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(())
let id = session_id.to_string(); // Clone para uso en closure
with_transaction(
&self.pool,
"revoke_session",
|tx| {
Box::pin(async move {
// Revocar la sesión
let result = sqlx::query(
r#"
UPDATE auth.sessions
SET revoked = true
WHERE id = $1
RETURNING user_id
"#
)
.bind(&id)
.fetch_optional(&mut **tx)
.await
.map_err(Self::map_sqlx_error)?;
// Si encontramos la sesión, podemos registrar un evento de seguridad
if let Some(row) = result {
let user_id: String = row.try_get("user_id").unwrap_or_default();
// Registrar evento de seguridad (en una tabla de seguridad)
// Esto es opcional pero muestra cómo se puede realizar operaciones
// adicionales en la misma transacción
tracing::info!("Sesión con ID {} del usuario {} revocada", id, user_id);
}
Ok(())
}) as BoxFuture<'_, SessionRepositoryResult<()>>
}
).await
}
/// Revoca todas las sesiones de un usuario
/// Revoca todas las sesiones de un usuario utilizando una transacción
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())
let user_id_clone = user_id.to_string(); // Clone para uso en closure
with_transaction(
&self.pool,
"revoke_all_user_sessions",
|tx| {
Box::pin(async move {
// Revocar todas las sesiones del usuario
let result = sqlx::query(
r#"
UPDATE auth.sessions
SET revoked = true
WHERE user_id = $1 AND revoked = false
"#
)
.bind(&user_id_clone)
.execute(&mut **tx)
.await
.map_err(Self::map_sqlx_error)?;
let affected = result.rows_affected();
// Registrar evento de seguridad
if affected > 0 {
tracing::info!("Revocadas {} sesiones del usuario {}", affected, user_id_clone);
}
Ok(affected)
}) as BoxFuture<'_, SessionRepositoryResult<u64>>
}
).await
}
/// Elimina sesiones expiradas
src/infrastructure/repositories/pg/transaction_utils.rs
+130
View File
@@ -0,0 +1,130 @@
use sqlx::{PgPool, Transaction, Postgres, Error as SqlxError, Executor};
use std::sync::Arc;
use tracing::{debug, error, info};
/// Helper function to execute database operations in a transaction
/// Takes a database pool and a closure that will be executed within a transaction
/// The closure receives a transaction object that should be used for all database operations
/// If the closure returns an error, the transaction is rolled back
/// If the closure returns Ok, the transaction is committed
pub async fn with_transaction<F, T, E>(
pool: &Arc<PgPool>,
operation_name: &str,
operation: F,
) -> Result<T, E>
where
F: for<'c> FnOnce(&'c mut Transaction<'_, Postgres>) -> futures::future::BoxFuture<'c, Result<T, E>>,
E: From<SqlxError> + std::fmt::Display,
{
debug!("Starting database transaction for: {}", operation_name);
// Begin transaction
let mut tx = pool.begin().await.map_err(|e| {
error!("Failed to begin transaction for {}: {}", operation_name, e);
E::from(e)
})?;
// Execute the operation within the transaction
match operation(&mut tx).await {
Ok(result) => {
// If operation succeeds, commit the transaction
match tx.commit().await {
Ok(_) => {
debug!("Transaction committed successfully for: {}", operation_name);
Ok(result)
},
Err(e) => {
error!("Failed to commit transaction for {}: {}", operation_name, e);
Err(E::from(e))
}
}
},
Err(e) => {
// If operation fails, rollback the transaction
if let Err(rollback_err) = tx.rollback().await {
error!("Failed to rollback transaction for {}: {}", operation_name, rollback_err);
// Still return the original error
} else {
info!("Transaction rolled back for {}: {}", operation_name, e);
}
Err(e)
}
}
}
/// Variant that accepts a transaction isolation level
pub async fn with_transaction_isolation<F, T, E>(
pool: &Arc<PgPool>,
operation_name: &str,
isolation_level: TransactionIsolationLevel,
operation: F,
) -> Result<T, E>
where
F: for<'c> FnOnce(&'c mut Transaction<'_, Postgres>) -> futures::future::BoxFuture<'c, Result<T, E>>,
E: From<SqlxError> + std::fmt::Display,
{
debug!("Starting database transaction with isolation level {:?} for: {}",
isolation_level, operation_name);
// Begin transaction with specific isolation level
let mut tx = pool.begin().await.map_err(|e| {
error!("Failed to begin transaction for {}: {}", operation_name, e);
E::from(e)
})?;
// Set isolation level
tx.execute(&format!("SET TRANSACTION ISOLATION LEVEL {}", isolation_level.to_string())[..])
.await
.map_err(|e| {
error!("Failed to set isolation level for {}: {}", operation_name, e);
E::from(e)
})?;
// Execute the operation within the transaction
match operation(&mut tx).await {
Ok(result) => {
// If operation succeeds, commit the transaction
match tx.commit().await {
Ok(_) => {
debug!("Transaction committed successfully for: {}", operation_name);
Ok(result)
},
Err(e) => {
error!("Failed to commit transaction for {}: {}", operation_name, e);
Err(E::from(e))
}
}
},
Err(e) => {
// If operation fails, rollback the transaction
if let Err(rollback_err) = tx.rollback().await {
error!("Failed to rollback transaction for {}: {}", operation_name, rollback_err);
// Still return the original error
} else {
info!("Transaction rolled back for {}: {}", operation_name, e);
}
Err(e)
}
}
}
/// Transaction isolation levels from SQL standard
#[derive(Debug)]
pub enum TransactionIsolationLevel {
/// Read committed isolation level
ReadCommitted,
/// Repeatable read isolation level
RepeatableRead,
/// Serializable isolation level
Serializable,
}
impl ToString for TransactionIsolationLevel {
fn to_string(&self) -> String {
match self {
TransactionIsolationLevel::ReadCommitted => "READ COMMITTED".to_string(),
TransactionIsolationLevel::RepeatableRead => "REPEATABLE READ".to_string(),
TransactionIsolationLevel::Serializable => "SERIALIZABLE".to_string(),
}
}
}
src/infrastructure/repositories/pg/user_pg_repository.rs
+109 -65
View File
@@ -1,11 +1,20 @@
use async_trait::async_trait;
use sqlx::{PgPool, Row};
use sqlx::{PgPool, Row, Executor};
use std::sync::Arc;
use futures::future::BoxFuture;
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;
use crate::infrastructure::repositories::pg::transaction_utils::with_transaction;
// Implementar From<sqlx::Error> para UserRepositoryError para permitir conversiones automáticas
impl From<sqlx::Error> for UserRepositoryError {
fn from(err: sqlx::Error) -> Self {
UserPgRepository::map_sqlx_error(err)
}
}
pub struct UserPgRepository {
pool: Arc<PgPool>,
@@ -17,7 +26,7 @@ impl UserPgRepository {
}
// Método auxiliar para mapear errores SQL a errores de dominio
fn map_sqlx_error(err: sqlx::Error) -> UserRepositoryError {
pub fn map_sqlx_error(err: sqlx::Error) -> UserRepositoryError {
match err {
sqlx::Error::RowNotFound => {
UserRepositoryError::NotFound("Usuario no encontrado".to_string())
@@ -43,40 +52,58 @@ impl UserPgRepository {
#[async_trait]
impl UserRepository for UserPgRepository {
/// Crea un nuevo usuario
/// Crea un nuevo usuario utilizando una transacción
async fn create_user(&self, user: User) -> UserRepositoryResult<User> {
// Usamos los getters para extraer los valores
// Convertimos user.role() a string para pasarlo como texto plano
let role_str = user.role().to_string();
// Creamos una copia del usuario para el closure
let user_clone = user.clone();
with_transaction(
&self.pool,
"create_user",
|tx| {
// Necesitamos mover el closure a un BoxFuture para devolver dentro
// de la llamada with_transaction
Box::pin(async move {
// Usamos los getters para extraer los valores
// Convertimos user.role() a string para pasarlo como texto plano
let role_str = user_clone.role().to_string();
// Modificar el SQL para hacer un cast explícito al tipo auth.userrole
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::auth.userrole, $6, $7, $8, $9, $10, $11
)
RETURNING *
"#
)
.bind(user_clone.id())
.bind(user_clone.username())
.bind(user_clone.email())
.bind(user_clone.password_hash())
.bind(&role_str) // Convertir a string pero con cast explícito en SQL
.bind(user_clone.storage_quota_bytes())
.bind(user_clone.storage_used_bytes())
.bind(user_clone.created_at())
.bind(user_clone.updated_at())
.bind(user_clone.last_login_at())
.bind(user_clone.is_active())
.execute(&mut **tx)
.await
.map_err(Self::map_sqlx_error)?;
// Podríamos realizar operaciones adicionales aquí,
// como configurar permisos, roles, etc.
Ok(user_clone)
}) as BoxFuture<'_, UserRepositoryResult<User>>
}
).await?;
// Modificar el SQL para hacer un cast explícito al tipo auth.userrole
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::auth.userrole, $6, $7, $8, $9, $10, $11
)
RETURNING *
"#
)
.bind(user.id())
.bind(user.username())
.bind(user.email())
.bind(user.password_hash())
.bind(&role_str) // Convertir a string pero con cast explícito en SQL
.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
}
@@ -197,38 +224,55 @@ impl UserRepository for UserPgRepository {
))
}
/// Actualiza un usuario existente
/// Actualiza un usuario existente utilizando una transacción
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::auth.userrole,
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().to_string()) // Esto no usa el cast explícito porque el SQL ya lo tiene
.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)?;
// Creamos una copia del usuario para el closure
let user_clone = user.clone();
with_transaction(
&self.pool,
"update_user",
|tx| {
Box::pin(async move {
// Actualizar el usuario
sqlx::query(
r#"
UPDATE auth.users
SET
username = $2,
email = $3,
password_hash = $4,
role = $5::auth.userrole,
storage_quota_bytes = $6,
storage_used_bytes = $7,
updated_at = $8,
last_login_at = $9,
active = $10
WHERE id = $1
"#
)
.bind(user_clone.id())
.bind(user_clone.username())
.bind(user_clone.email())
.bind(user_clone.password_hash())
.bind(&user_clone.role().to_string())
.bind(user_clone.storage_quota_bytes())
.bind(user_clone.storage_used_bytes())
.bind(user_clone.updated_at())
.bind(user_clone.last_login_at())
.bind(user_clone.is_active())
.execute(&mut **tx)
.await
.map_err(Self::map_sqlx_error)?;
// Podríamos realizar operaciones adicionales aquí dentro
// de la misma transacción, como actualizar permisos, etc.
Ok(user_clone)
}) as BoxFuture<'_, UserRepositoryResult<User>>
}
).await?;
Ok(user)
}