aro-fletch 1.0.0

//! Fletch transaction manager adapter.

use std::future::Future;
use std::sync::Arc;

use aro_core::error::RepoError;
use aro_core::repository::{BoxFuture, TransactionManager};
use aro_web::dep::Dep;

use crate::error::from_fletch_err;
use crate::repo::{FletchDatabase, FletchEntity, FletchTransactionalRepository};

/// Shared transaction state for an open SQLx transaction.
pub(crate) struct FletchTransactionState<DB: FletchDatabase> {
    pub(crate) tx: tokio::sync::Mutex<Option<fletch_orm::Transaction<'static, DB>>>,
}

/// Context passed to transaction closures, carrying the open transaction.
///
/// Obtain transaction-scoped repositories via [`repo`](Self::repo) so writes
/// participate in the same transaction and roll back with it.
///
/// Do not clone this context and use it after the [`transaction`](FletchTransactionManager::transaction)
/// closure returns — the underlying transaction is committed or rolled back on
/// finish, and further repository calls return
/// `RepoError::Unknown("transaction is no longer active")`.
pub struct FletchTransactionContext<DB: FletchDatabase> {
    state: Arc<FletchTransactionState<DB>>,
}

impl<DB: FletchDatabase> Clone for FletchTransactionContext<DB> {
    fn clone(&self) -> Self {
        Self {
            state: Arc::clone(&self.state),
        }
    }
}

impl<DB: FletchDatabase> FletchTransactionContext<DB> {
    /// Return a transaction-scoped repository for `T`.
    pub fn repo<T>(&self) -> FletchTransactionalRepository<T, DB>
    where
        T: FletchEntity<DB>,
    {
        FletchTransactionalRepository::new(Arc::clone(&self.state))
    }
}

async fn begin_transaction<DB: FletchDatabase>(
    pool: &fletch_orm::Pool<DB>,
) -> Result<FletchTransactionContext<DB>, RepoError> {
    let tx = pool.begin().await.map_err(from_fletch_err)?;
    Ok(FletchTransactionContext {
        state: Arc::new(FletchTransactionState {
            tx: tokio::sync::Mutex::new(Some(tx)),
        }),
    })
}

async fn finish_transaction<DB: FletchDatabase, R: Send>(
    state: Arc<FletchTransactionState<DB>>,
    result: Result<R, RepoError>,
) -> Result<R, RepoError> {
    let mut guard = state.tx.lock().await;
    let tx = guard
        .take()
        .ok_or_else(|| RepoError::Unknown("transaction is no longer active".into()))?;
    drop(guard);

    match result {
        Ok(value) => {
            tx.commit().await.map_err(from_fletch_err)?;
            Ok(value)
        }
        Err(err) => {
            if let Err(rollback_err) = tx.rollback().await {
                return Err(from_fletch_err(rollback_err));
            }
            Err(err)
        }
    }
}

/// Fletch-backed [`TransactionManager`] implementation.
///
/// Wraps a [`fletch_orm::Pool<DB>`] and drives commit/rollback through SQLx
/// transactions. Prefer the inherent [`transaction`](Self::transaction)
/// method, which passes a [`FletchTransactionContext`] so repository writes
/// use the open transaction.
pub struct FletchTransactionManager<DB: FletchDatabase> {
    pool: fletch_orm::Pool<DB>,
}

impl<DB: FletchDatabase> FletchTransactionManager<DB> {
    /// Create a new transaction manager from a fletch pool.
    pub fn new(pool: fletch_orm::Pool<DB>) -> Self {
        Self { pool }
    }

    /// Create a new transaction manager by cloning the pool from a [`Dep`].
    pub fn from_dep(dep: &Dep<fletch_orm::Pool<DB>>) -> Self {
        Self::new((**dep).clone())
    }

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

    /// Execute `f` within a transactional boundary.
    ///
    /// The closure receives an owned [`FletchTransactionContext`] with txn-scoped
    /// repositories. On `Ok`, the transaction is committed; on `Err`, it is
    /// rolled back.
    ///
    /// Do not retain clones of the context beyond this closure — once the
    /// transaction finishes, cloned contexts cannot execute further queries.
    pub async fn transaction<F, Fut, R>(&self, f: F) -> Result<R, RepoError>
    where
        F: FnOnce(FletchTransactionContext<DB>) -> Fut,
        Fut: Future<Output = Result<R, RepoError>>,
        R: Send,
    {
        let ctx = begin_transaction(&self.pool).await?;
        let state = Arc::clone(&ctx.state);
        let result = f(ctx).await;
        finish_transaction(state, result).await
    }

    /// Boxed variant of [`transaction`](Self::transaction) for `'static` callbacks.
    pub fn transaction_boxed<F, Fut, R>(&self, f: F) -> BoxFuture<'_, Result<R, RepoError>>
    where
        F: FnOnce(FletchTransactionContext<DB>) -> Fut + Send + 'static,
        Fut: Future<Output = Result<R, RepoError>> + Send + 'static,
        R: Send + 'static,
    {
        let pool = self.pool.clone();
        Box::pin(async move {
            let manager = FletchTransactionManager { pool };
            manager.transaction(f).await
        })
    }
}

impl<DB: FletchDatabase> TransactionManager for FletchTransactionManager<DB> {
    type Context = FletchTransactionContext<DB>;

    fn transaction<F, Fut, R>(&self, f: F) -> BoxFuture<'_, Result<R, RepoError>>
    where
        F: FnOnce(FletchTransactionContext<DB>) -> Fut + Send + 'static,
        Fut: Future<Output = Result<R, RepoError>> + Send + 'static,
        R: Send + 'static,
    {
        self.transaction_boxed(f)
    }
}