aro_fletch/
transaction.rs

1//! Fletch transaction manager adapter.
2
3use std::future::Future;
4use std::sync::Arc;
5
6use aro_core::error::RepoError;
7use aro_core::repository::{BoxFuture, TransactionManager};
8use aro_web::dep::Dep;
9
10use crate::error::from_fletch_err;
11use crate::repo::{FletchDatabase, FletchEntity, FletchTransactionalRepository};
12
13/// Shared transaction state for an open SQLx transaction.
14pub(crate) struct FletchTransactionState<DB: FletchDatabase> {
15    pub(crate) tx: tokio::sync::Mutex<Option<fletch_orm::Transaction<'static, DB>>>,
16}
17
18/// Context passed to transaction closures, carrying the open transaction.
19///
20/// Obtain transaction-scoped repositories via [`repo`](Self::repo) so writes
21/// participate in the same transaction and roll back with it.
22///
23/// Do not clone this context and use it after the [`transaction`](FletchTransactionManager::transaction)
24/// closure returns — the underlying transaction is committed or rolled back on
25/// finish, and further repository calls return
26/// `RepoError::Unknown("transaction is no longer active")`.
27pub struct FletchTransactionContext<DB: FletchDatabase> {
28    state: Arc<FletchTransactionState<DB>>,
29}
30
31impl<DB: FletchDatabase> Clone for FletchTransactionContext<DB> {
32    fn clone(&self) -> Self {
33        Self {
34            state: Arc::clone(&self.state),
35        }
36    }
37}
38
39impl<DB: FletchDatabase> FletchTransactionContext<DB> {
40    /// Return a transaction-scoped repository for `T`.
41    pub fn repo<T>(&self) -> FletchTransactionalRepository<T, DB>
42    where
43        T: FletchEntity<DB>,
44    {
45        FletchTransactionalRepository::new(Arc::clone(&self.state))
46    }
47}
48
49async fn begin_transaction<DB: FletchDatabase>(
50    pool: &fletch_orm::Pool<DB>,
51) -> Result<FletchTransactionContext<DB>, RepoError> {
52    let tx = pool.begin().await.map_err(from_fletch_err)?;
53    Ok(FletchTransactionContext {
54        state: Arc::new(FletchTransactionState {
55            tx: tokio::sync::Mutex::new(Some(tx)),
56        }),
57    })
58}
59
60async fn finish_transaction<DB: FletchDatabase, R: Send>(
61    state: Arc<FletchTransactionState<DB>>,
62    result: Result<R, RepoError>,
63) -> Result<R, RepoError> {
64    let mut guard = state.tx.lock().await;
65    let tx = guard
66        .take()
67        .ok_or_else(|| RepoError::Unknown("transaction is no longer active".into()))?;
68    drop(guard);
69
70    match result {
71        Ok(value) => {
72            tx.commit().await.map_err(from_fletch_err)?;
73            Ok(value)
74        }
75        Err(err) => {
76            if let Err(rollback_err) = tx.rollback().await {
77                return Err(from_fletch_err(rollback_err));
78            }
79            Err(err)
80        }
81    }
82}
83
84/// Fletch-backed [`TransactionManager`] implementation.
85///
86/// Wraps a [`fletch_orm::Pool<DB>`] and drives commit/rollback through SQLx
87/// transactions. Prefer the inherent [`transaction`](Self::transaction)
88/// method, which passes a [`FletchTransactionContext`] so repository writes
89/// use the open transaction.
90pub struct FletchTransactionManager<DB: FletchDatabase> {
91    pool: fletch_orm::Pool<DB>,
92}
93
94impl<DB: FletchDatabase> FletchTransactionManager<DB> {
95    /// Create a new transaction manager from a fletch pool.
96    pub fn new(pool: fletch_orm::Pool<DB>) -> Self {
97        Self { pool }
98    }
99
100    /// Create a new transaction manager by cloning the pool from a [`Dep`].
101    pub fn from_dep(dep: &Dep<fletch_orm::Pool<DB>>) -> Self {
102        Self::new((**dep).clone())
103    }
104
105    /// Return a reference to the underlying pool.
106    pub fn pool(&self) -> &fletch_orm::Pool<DB> {
107        &self.pool
108    }
109
110    /// Execute `f` within a transactional boundary.
111    ///
112    /// The closure receives an owned [`FletchTransactionContext`] with txn-scoped
113    /// repositories. On `Ok`, the transaction is committed; on `Err`, it is
114    /// rolled back.
115    ///
116    /// Do not retain clones of the context beyond this closure — once the
117    /// transaction finishes, cloned contexts cannot execute further queries.
118    pub async fn transaction<F, Fut, R>(&self, f: F) -> Result<R, RepoError>
119    where
120        F: FnOnce(FletchTransactionContext<DB>) -> Fut,
121        Fut: Future<Output = Result<R, RepoError>>,
122        R: Send,
123    {
124        let ctx = begin_transaction(&self.pool).await?;
125        let state = Arc::clone(&ctx.state);
126        let result = f(ctx).await;
127        finish_transaction(state, result).await
128    }
129
130    /// Boxed variant of [`transaction`](Self::transaction) for `'static` callbacks.
131    pub fn transaction_boxed<F, Fut, R>(&self, f: F) -> BoxFuture<'_, Result<R, RepoError>>
132    where
133        F: FnOnce(FletchTransactionContext<DB>) -> Fut + Send + 'static,
134        Fut: Future<Output = Result<R, RepoError>> + Send + 'static,
135        R: Send + 'static,
136    {
137        let pool = self.pool.clone();
138        Box::pin(async move {
139            let manager = FletchTransactionManager { pool };
140            manager.transaction(f).await
141        })
142    }
143}
144
145impl<DB: FletchDatabase> TransactionManager for FletchTransactionManager<DB> {
146    type Context = FletchTransactionContext<DB>;
147
148    fn transaction<F, Fut, R>(&self, f: F) -> BoxFuture<'_, Result<R, RepoError>>
149    where
150        F: FnOnce(FletchTransactionContext<DB>) -> Fut + Send + 'static,
151        Fut: Future<Output = Result<R, RepoError>> + Send + 'static,
152        R: Send + 'static,
153    {
154        self.transaction_boxed(f)
155    }
156}
aro_fletch/transaction.rs

aro_fletch/
transaction.rs
