aro_fletch/

repo.rs

//! Generic fletch-backed repository adapter.

use std::marker::PhantomData;
use std::sync::Arc;

use aro_core::error::RepoError;
use aro_core::repository::{BoxFuture, Repository};
use aro_web::dep::Dep;
use sqlx::Database;

use crate::error::from_fletch_err;

/// A SQLx database backend supported by aro-fletch.
///
/// Enabled via Cargo features: `sqlite`, `postgres`, `mysql`.
pub trait FletchDatabase: Database + Send + Sync + 'static {}

#[cfg(feature = "sqlite")]
impl FletchDatabase for sqlx::Sqlite {}

#[cfg(feature = "postgres")]
impl FletchDatabase for sqlx::Postgres {}

#[cfg(feature = "mysql")]
impl FletchDatabase for sqlx::MySql {}

/// A fletch entity suitable for the repository adapter.
///
/// This trait combines `aro_core::Entity` with `fletch_orm::Entity` and the
/// `sqlx::FromRow` bound required by fletch's query layer.
pub trait FletchEntity<DB: FletchDatabase>:
    aro_core::entity::Entity
    + fletch_orm::Entity<Id = <Self as aro_core::entity::Entity>::Id>
    + for<'r> sqlx::FromRow<'r, DB::Row>
    + Send
    + Sync
    + Unpin
    + Clone
{
}

/// Blanket implementation for any type meeting all bounds.
impl<T, DB> FletchEntity<DB> for T
where
    DB: FletchDatabase,
    T: aro_core::entity::Entity
        + fletch_orm::Entity<Id = <T as aro_core::entity::Entity>::Id>
        + for<'r> sqlx::FromRow<'r, DB::Row>
        + Send
        + Sync
        + Unpin
        + Clone,
{
}

/// Generic fletch-backed repository for any domain entity implementing [`FletchEntity`].
///
/// All CRUD operations run against the pool held by this repository.
pub struct FletchRepository<T, DB: FletchDatabase> {
    pool: fletch_orm::Pool<DB>,
    _marker: PhantomData<T>,
}

impl<T, DB> FletchRepository<T, DB>
where
    DB: FletchDatabase,
{
    /// Create a new repository backed by the given pool.
    pub fn new(pool: fletch_orm::Pool<DB>) -> Self {
        Self {
            pool,
            _marker: PhantomData,
        }
    }

    /// Create a new repository by cloning the pool from a [`Dep`].
    ///
    /// `fletch_orm::Pool` wraps an `sqlx::Pool` internally, so cloning is
    /// cheap and all clones share the same pool.
    pub fn from_dep(dep: &Dep<fletch_orm::Pool<DB>>) -> Self {
        Self::new((**dep).clone())
    }

    /// Return a reference to the underlying connection pool.
    pub fn pool(&self) -> &fletch_orm::Pool<DB> {
        &self.pool
    }
}

/// Transaction-scoped repository adapter delegating to fletch CRUD on the open transaction.
///
/// Created from [`FletchTransactionContext::repo`](crate::transaction::FletchTransactionContext::repo)
/// so writes participate in the open transaction and roll back on failure.
pub struct FletchTransactionalRepository<T, DB: FletchDatabase> {
    state: Arc<crate::transaction::FletchTransactionState<DB>>,
    _marker: PhantomData<T>,
}

impl<T, DB> FletchTransactionalRepository<T, DB>
where
    DB: FletchDatabase,
{
    pub(crate) fn new(state: Arc<crate::transaction::FletchTransactionState<DB>>) -> Self {
        Self {
            state,
            _marker: PhantomData,
        }
    }
}

/*
 * Transactional create/update differ by backend:
 *
 * - `returning` (SQLite, Postgres): use `fletch_orm::insert` / `fletch_orm::update` with
 *   RETURNING on the open connection.
 * - `mysql`: use [`fletch_orm::Transaction::insert`] / [`fletch_orm::Transaction::update`]
 *   because MySQL has no RETURNING and re-selects via last insert id.
 */
macro_rules! impl_transactional_repository {
    (mysql, $db:ty, $dialect:expr) => {
        impl<T> FletchTransactionalRepository<T, $db>
        where
            T: FletchEntity<$db>,
        {
            /// Find an entity by its unique identifier.
            pub async fn find_by_id(
                &self,
                id: <T as aro_core::entity::Entity>::Id,
            ) -> Result<T, RepoError> {
                let mut guard = self.state.tx.lock().await;
                let tx = guard
                    .as_mut()
                    .ok_or_else(|| RepoError::Unknown("transaction is no longer active".into()))?;
                fletch_orm::find_by_id($dialect, &mut **tx.inner_mut(), id)
                    .await
                    .map_err(from_fletch_err)
            }

            /// Return all entities of this type.
            pub async fn find_all(&self) -> Result<Vec<T>, RepoError> {
                let mut guard = self.state.tx.lock().await;
                let tx = guard
                    .as_mut()
                    .ok_or_else(|| RepoError::Unknown("transaction is no longer active".into()))?;
                fletch_orm::find_all($dialect, &mut **tx.inner_mut())
                    .await
                    .map_err(from_fletch_err)
            }

            /// Persist a new entity and return the created value.
            pub async fn create(&self, entity: T) -> Result<T, RepoError> {
                let mut guard = self.state.tx.lock().await;
                let tx = guard
                    .as_mut()
                    .ok_or_else(|| RepoError::Unknown("transaction is no longer active".into()))?;
                tx.insert(&entity).await.map_err(from_fletch_err)
            }

            /// Update an existing entity and return the updated value.
            pub async fn update(&self, entity: T) -> Result<T, RepoError> {
                let mut guard = self.state.tx.lock().await;
                let tx = guard
                    .as_mut()
                    .ok_or_else(|| RepoError::Unknown("transaction is no longer active".into()))?;
                tx.update(&entity).await.map_err(from_fletch_err)
            }

            /// Delete an entity by its unique identifier.
            pub async fn delete(
                &self,
                id: <T as aro_core::entity::Entity>::Id,
            ) -> Result<(), RepoError> {
                let mut guard = self.state.tx.lock().await;
                let tx = guard
                    .as_mut()
                    .ok_or_else(|| RepoError::Unknown("transaction is no longer active".into()))?;
                fletch_orm::delete::<T, $db, _, _>($dialect, &mut **tx.inner_mut(), id)
                    .await
                    .map_err(from_fletch_err)
            }
        }
    };

    (returning, $db:ty, $dialect:expr) => {
        impl<T> FletchTransactionalRepository<T, $db>
        where
            T: FletchEntity<$db>,
        {
            /// Find an entity by its unique identifier.
            pub async fn find_by_id(
                &self,
                id: <T as aro_core::entity::Entity>::Id,
            ) -> Result<T, RepoError> {
                let mut guard = self.state.tx.lock().await;
                let tx = guard
                    .as_mut()
                    .ok_or_else(|| RepoError::Unknown("transaction is no longer active".into()))?;
                fletch_orm::find_by_id($dialect, &mut **tx.inner_mut(), id)
                    .await
                    .map_err(from_fletch_err)
            }

            /// Return all entities of this type.
            pub async fn find_all(&self) -> Result<Vec<T>, RepoError> {
                let mut guard = self.state.tx.lock().await;
                let tx = guard
                    .as_mut()
                    .ok_or_else(|| RepoError::Unknown("transaction is no longer active".into()))?;
                fletch_orm::find_all($dialect, &mut **tx.inner_mut())
                    .await
                    .map_err(from_fletch_err)
            }

            /// Persist a new entity and return the created value.
            pub async fn create(&self, entity: T) -> Result<T, RepoError> {
                let mut guard = self.state.tx.lock().await;
                let tx = guard
                    .as_mut()
                    .ok_or_else(|| RepoError::Unknown("transaction is no longer active".into()))?;
                fletch_orm::insert($dialect, &mut **tx.inner_mut(), &entity)
                    .await
                    .map_err(from_fletch_err)
            }

            /// Update an existing entity and return the updated value.
            pub async fn update(&self, entity: T) -> Result<T, RepoError> {
                let mut guard = self.state.tx.lock().await;
                let tx = guard
                    .as_mut()
                    .ok_or_else(|| RepoError::Unknown("transaction is no longer active".into()))?;
                fletch_orm::update($dialect, &mut **tx.inner_mut(), &entity)
                    .await
                    .map_err(from_fletch_err)
            }

            /// Delete an entity by its unique identifier.
            pub async fn delete(
                &self,
                id: <T as aro_core::entity::Entity>::Id,
            ) -> Result<(), RepoError> {
                let mut guard = self.state.tx.lock().await;
                let tx = guard
                    .as_mut()
                    .ok_or_else(|| RepoError::Unknown("transaction is no longer active".into()))?;
                fletch_orm::delete::<T, $db, _, _>($dialect, &mut **tx.inner_mut(), id)
                    .await
                    .map_err(from_fletch_err)
            }
        }
    };
}

#[cfg(feature = "sqlite")]
impl_transactional_repository!(returning, sqlx::Sqlite, &fletch_orm::SqliteDialect);

#[cfg(feature = "postgres")]
impl_transactional_repository!(returning, sqlx::Postgres, &fletch_orm::PostgresDialect);

#[cfg(feature = "mysql")]
impl_transactional_repository!(mysql, sqlx::MySql, &fletch_orm::MySqlDialect);

macro_rules! impl_repository {
    ($db:ty) => {
        impl<T> Repository<T> for FletchRepository<T, $db>
        where
            T: FletchEntity<$db>,
            <T as aro_core::entity::Entity>::Id: Into<fletch_orm::Value>,
        {
            fn find_by_id(
                &self,
                id: <T as aro_core::entity::Entity>::Id,
            ) -> BoxFuture<'_, Result<T, RepoError>> {
                Box::pin(
                    async move { self.pool.find_by_id::<T>(id).await.map_err(from_fletch_err) },
                )
            }

            fn find_all(&self) -> BoxFuture<'_, Result<Vec<T>, RepoError>> {
                Box::pin(async move { self.pool.find_all::<T>().await.map_err(from_fletch_err) })
            }

            fn create(&self, entity: T) -> BoxFuture<'_, Result<T, RepoError>> {
                Box::pin(async move {
                    self.pool
                        .insert::<T>(&entity)
                        .await
                        .map_err(from_fletch_err)
                })
            }

            fn update(&self, entity: T) -> BoxFuture<'_, Result<T, RepoError>> {
                Box::pin(async move {
                    self.pool
                        .update::<T>(&entity)
                        .await
                        .map_err(from_fletch_err)
                })
            }

            fn delete(
                &self,
                id: <T as aro_core::entity::Entity>::Id,
            ) -> BoxFuture<'_, Result<(), RepoError>> {
                Box::pin(async move { self.pool.delete::<T>(id).await.map_err(from_fletch_err) })
            }
        }
    };
}

#[cfg(feature = "sqlite")]
impl_repository!(sqlx::Sqlite);

#[cfg(feature = "postgres")]
impl_repository!(sqlx::Postgres);

#[cfg(feature = "mysql")]
impl_repository!(sqlx::MySql);