entity_core/

transaction.rs

// SPDX-FileCopyrightText: 2025-2026 RAprogramm <andrey.rozanov.vl@gmail.com>
// SPDX-License-Identifier: MIT

//! Transaction support for entity-derive.
//!
//! This module provides type-safe transaction management with automatic
//! commit/rollback semantics. It uses a fluent builder pattern for composing
//! multiple entity operations into a single transaction.
//!
//! # Overview
//!
//! - [`Transaction`] — Entry point for creating transactions
//! - [`TransactionContext`] — Holds active transaction, provides repo access
//! - [`TransactionError`] — Error wrapper for transaction operations
//!
//! # Example
//!
//! ```rust,ignore
//! use entity_derive::prelude::*;
//!
//! async fn transfer(pool: &PgPool, from: Uuid, to: Uuid, amount: i64) -> Result<(), AppError> {
//!     Transaction::new(pool)
//!         .run(async |ctx| {
//!             let from_acc = ctx.accounts().find_by_id(from).await?.ok_or(AppError::NotFound)?;
//!
//!             ctx.accounts().update(from, UpdateAccount {
//!                 balance: Some(from_acc.balance - amount),
//!                 ..Default::default()
//!             }).await?;
//!
//!             ctx.transfers().create(CreateTransfer { from, to, amount }).await?;
//!             Ok(())
//!         })
//!         .await
//! }
//! ```

#[cfg(feature = "postgres")]
use std::future::Future;
use std::{error::Error as StdError, fmt};

/// Transaction builder for composing multi-entity operations.
///
/// Use [`Transaction::new`] to create a builder, chain `.with_*()` methods
/// to declare which entities you'll use, then call `.run()` to execute.
///
/// # Type Parameters
///
/// - `'p` — Pool lifetime
/// - `DB` — Database pool type (e.g., `PgPool`)
///
/// # Example
///
/// ```rust,ignore
/// Transaction::new(&pool)
///     .run(async |ctx| {
///         let user = ctx.users().find_by_id(id).await?;
///         ctx.orders().create(order).await?;
///         Ok(())
///     })
///     .await?;
/// ```
pub struct Transaction<'p, DB> {
    pool: &'p DB
}

impl<'p, DB> Transaction<'p, DB> {
    /// Create a new transaction builder.
    ///
    /// # Arguments
    ///
    /// * `pool` — Database connection pool
    ///
    /// # Example
    ///
    /// ```rust,ignore
    /// let tx = Transaction::new(&pool);
    /// ```
    pub const fn new(pool: &'p DB) -> Self {
        Self {
            pool
        }
    }

    /// Get reference to the underlying pool.
    #[must_use]
    pub const fn pool(&self) -> &'p DB {
        self.pool
    }
}

/// Active transaction context with repository access.
///
/// This struct holds the database transaction and provides access to
/// entity repositories via extension traits generated by the macro.
///
/// # Automatic Rollback
///
/// If dropped without explicit commit, the transaction is automatically
/// rolled back via the underlying database transaction's Drop impl.
///
/// # Accessing Repositories
///
/// Each entity with `#[entity(transactions)]` generates an extension trait
/// that adds an accessor method:
///
/// ```rust,ignore
/// // For entity BankAccount, use:
/// ctx.bank_accounts().find_by_id(id).await?;
/// ctx.bank_accounts().create(dto).await?;
/// ctx.bank_accounts().update(id, dto).await?;
/// ```
#[cfg(feature = "postgres")]
pub struct TransactionContext {
    tx: sqlx::Transaction<'static, sqlx::Postgres>
}

#[cfg(feature = "postgres")]
impl TransactionContext {
    /// Create a new transaction context.
    ///
    /// # Arguments
    ///
    /// * `tx` — Active database transaction
    #[doc(hidden)]
    #[must_use]
    pub const fn new(tx: sqlx::Transaction<'static, sqlx::Postgres>) -> Self {
        Self {
            tx
        }
    }

    /// Get mutable reference to the underlying transaction.
    ///
    /// Use this for custom queries within the transaction or
    /// for repository adapters to execute queries.
    pub const fn transaction(&mut self) -> &mut sqlx::Transaction<'static, sqlx::Postgres> {
        &mut self.tx
    }

    /// Commit the transaction.
    ///
    /// Consumes self and commits all changes.
    ///
    /// # Errors
    ///
    /// Propagates any `sqlx::Error` from the database transaction.
    pub async fn commit(self) -> Result<(), sqlx::Error> {
        self.tx.commit().await
    }

    /// Rollback the transaction.
    ///
    /// Consumes self and rolls back all changes.
    ///
    /// # Errors
    ///
    /// Propagates any `sqlx::Error` from the database transaction.
    pub async fn rollback(self) -> Result<(), sqlx::Error> {
        self.tx.rollback().await
    }
}

/// Error type for transaction operations.
///
/// Wraps database errors and provides context about the transaction state.
#[derive(Debug)]
pub enum TransactionError<E> {
    /// Failed to begin transaction.
    Begin(E),

    /// Failed to commit transaction.
    Commit(E),

    /// Failed to rollback transaction.
    Rollback(E),

    /// Operation within transaction failed.
    Operation(E)
}

impl<E: fmt::Display> fmt::Display for TransactionError<E> {
    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
        match self {
            Self::Begin(e) => write!(f, "failed to begin transaction: {e}"),
            Self::Commit(e) => write!(f, "failed to commit transaction: {e}"),
            Self::Rollback(e) => write!(f, "failed to rollback transaction: {e}"),
            Self::Operation(e) => write!(f, "transaction operation failed: {e}")
        }
    }
}

impl<E: StdError + 'static> StdError for TransactionError<E> {
    fn source(&self) -> Option<&(dyn StdError + 'static)> {
        match self {
            Self::Begin(e) | Self::Commit(e) | Self::Rollback(e) | Self::Operation(e) => Some(e)
        }
    }
}

impl<E> TransactionError<E> {
    /// Check if this is a begin error.
    pub const fn is_begin(&self) -> bool {
        matches!(self, Self::Begin(_))
    }

    /// Check if this is a commit error.
    pub const fn is_commit(&self) -> bool {
        matches!(self, Self::Commit(_))
    }

    /// Check if this is a rollback error.
    pub const fn is_rollback(&self) -> bool {
        matches!(self, Self::Rollback(_))
    }

    /// Check if this is an operation error.
    pub const fn is_operation(&self) -> bool {
        matches!(self, Self::Operation(_))
    }

    /// Get the inner error.
    pub fn into_inner(self) -> E {
        match self {
            Self::Begin(e) | Self::Commit(e) | Self::Rollback(e) | Self::Operation(e) => e
        }
    }
}

#[cfg(feature = "postgres")]
impl From<TransactionError<Self>> for sqlx::Error {
    fn from(err: TransactionError<Self>) -> Self {
        err.into_inner()
    }
}

/// Finalize a transaction lifecycle: commit on `Ok`, drop (rollback) on `Err`.
///
/// Backend-agnostic helper extracted so the commit/rollback decision can be
/// unit-tested without a live database connection. Tests provide a mock
/// `ctx` and a tracking `commit_fn` to assert that:
///
/// - `commit_fn` runs exactly once when `result` is `Ok`
/// - `commit_fn` does **not** run when `result` is `Err`
/// - Errors from `commit_fn` propagate via `E::from`
///
/// # Errors
///
/// Returns the closure's original error on `Err`, or the converted commit
/// error if `commit_fn` fails on `Ok`.
#[cfg(any(feature = "postgres", test))]
async fn finalize_with_commit<C, T, E, CommitErr, Cf, Fut>(
    ctx: C,
    result: Result<T, E>,
    commit_fn: Cf
) -> Result<T, E>
where
    Cf: FnOnce(C) -> Fut,
    Fut: core::future::Future<Output = Result<(), CommitErr>>,
    E: From<CommitErr>
{
    match result {
        Ok(value) => {
            commit_fn(ctx).await.map_err(E::from)?;
            Ok(value)
        }
        Err(e) => Err(e)
    }
}

// PostgreSQL implementation
#[cfg(feature = "postgres")]
impl Transaction<'_, sqlx::PgPool> {
    /// Execute a closure within a `PostgreSQL` transaction.
    ///
    /// Commits the transaction explicitly when the closure returns `Ok`.
    /// On `Err`, the transaction context is dropped and `sqlx` rolls back
    /// automatically via its `Drop` implementation.
    ///
    /// The closure receives `&mut TransactionContext` (not by value) so that
    /// `run` retains ownership and can invoke `commit().await` on success.
    ///
    /// # Type Parameters
    ///
    /// - `F` — Async closure
    /// - `T` — Success type
    /// - `E` — Error type (must be convertible from `sqlx::Error`)
    ///
    /// # Example
    ///
    /// ```rust,ignore
    /// Transaction::new(&pool)
    ///     .run(async |ctx| {
    ///         let user = ctx.users().create(dto).await?;
    ///         Ok(user)
    ///     })
    ///     .await?;
    /// ```
    ///
    /// # Errors
    ///
    /// Propagates any error from the closure, from `begin`, or from `commit`.
    #[cfg_attr(
        feature = "tracing",
        ::tracing::instrument(skip_all, fields(op = "tx.run"), err(Debug))
    )]
    pub async fn run<F, T, E>(self, f: F) -> Result<T, E>
    where
        F: AsyncFnOnce(&mut TransactionContext) -> Result<T, E>,
        E: From<sqlx::Error> + core::fmt::Debug
    {
        let tx = self.pool.begin().await.map_err(E::from)?;
        let mut ctx = TransactionContext::new(tx);
        let result = f(&mut ctx).await;
        finalize_with_commit(ctx, result, |c| c.commit()).await
    }

    /// Execute a closure within a transaction with explicit commit.
    ///
    /// # ⚠️ The closure MUST call `ctx.commit().await` on every successful path
    ///
    /// `run_with_commit` hands ownership of [`TransactionContext`] to the
    /// closure. There is no type-level guarantee that the closure commits;
    /// if it returns `Ok(...)` without calling
    /// [`commit`][TransactionContext::commit], `ctx` is dropped and the
    /// underlying `sqlx::Transaction::Drop` **silently rolls back**. The
    /// caller observes `Ok` and assumes the writes persisted — they did
    /// not. This is the same failure mode that affected the old `run()`
    /// implementation; here it is preserved on purpose so callers can
    /// implement conditional commit logic, at the cost of moving the
    /// responsibility onto the closure.
    ///
    /// **Prefer [`run`](Self::run) unless you genuinely need to decide
    /// commit-or-rollback inside the closure.** `run` performs the commit
    /// automatically on `Ok` and rolls back on `Err`, eliminating this
    /// footgun.
    ///
    /// # Examples
    ///
    /// **Correct** — closure commits on the success path:
    ///
    /// ```rust,ignore
    /// Transaction::new(&pool)
    ///     .run_with_commit(|mut ctx| async move {
    ///         let user = ctx.users().create(dto).await?;
    ///         ctx.commit().await?;     // <-- required
    ///         Ok(user)
    ///     })
    ///     .await?;
    /// ```
    ///
    /// **Wrong** — closure returns `Ok` without committing; the write is
    /// rolled back when `ctx` drops, but the caller sees `Ok(user)`:
    ///
    /// ```rust,ignore
    /// Transaction::new(&pool)
    ///     .run_with_commit(|mut ctx| async move {
    ///         let user = ctx.users().create(dto).await?;
    ///         // BUG: forgot `ctx.commit().await?` — the row is rolled back.
    ///         Ok(user)
    ///     })
    ///     .await?;
    /// ```
    ///
    /// **Conditional commit** — the intended use case:
    ///
    /// ```rust,ignore
    /// Transaction::new(&pool)
    ///     .run_with_commit(|mut ctx| async move {
    ///         let user = ctx.users().create(dto).await?;
    ///         if user.flagged {
    ///             ctx.rollback().await?;
    ///             return Ok(None);
    ///         }
    ///         ctx.commit().await?;
    ///         Ok(Some(user))
    ///     })
    ///     .await?;
    /// ```
    ///
    /// # Errors
    ///
    /// Propagates any error from the closure or database transaction.
    #[cfg_attr(
        feature = "tracing",
        ::tracing::instrument(skip_all, fields(op = "tx.run_with_commit"), err(Debug))
    )]
    pub async fn run_with_commit<F, Fut, T, E>(self, f: F) -> Result<T, E>
    where
        F: FnOnce(TransactionContext) -> Fut + Send,
        Fut: Future<Output = Result<T, E>> + Send,
        E: From<sqlx::Error> + core::fmt::Debug
    {
        let tx = self.pool.begin().await.map_err(E::from)?;
        let ctx = TransactionContext::new(tx);
        f(ctx).await
    }
}

#[cfg(test)]
#[allow(clippy::uninlined_format_args)]
mod tests {
    use std::error::Error;

    use super::*;

    #[test]
    fn transaction_error_display_begin() {
        let err: TransactionError<std::io::Error> =
            TransactionError::Begin(std::io::Error::other("test"));
        assert!(err.to_string().contains("begin"));
        assert!(err.to_string().contains("test"));
    }

    #[test]
    fn transaction_error_display_commit() {
        let err: TransactionError<std::io::Error> =
            TransactionError::Commit(std::io::Error::other("test"));
        assert!(err.to_string().contains("commit"));
    }

    #[test]
    fn transaction_error_display_rollback() {
        let err: TransactionError<std::io::Error> =
            TransactionError::Rollback(std::io::Error::other("test"));
        assert!(err.to_string().contains("rollback"));
    }

    #[test]
    fn transaction_error_display_operation() {
        let err: TransactionError<std::io::Error> =
            TransactionError::Operation(std::io::Error::other("test"));
        assert!(err.to_string().contains("operation"));
    }

    #[test]
    fn transaction_error_is_methods() {
        let begin: TransactionError<&str> = TransactionError::Begin("e");
        let commit: TransactionError<&str> = TransactionError::Commit("e");
        let rollback: TransactionError<&str> = TransactionError::Rollback("e");
        let operation: TransactionError<&str> = TransactionError::Operation("e");

        assert!(begin.is_begin());
        assert!(!begin.is_commit());
        assert!(!begin.is_rollback());
        assert!(!begin.is_operation());

        assert!(!commit.is_begin());
        assert!(commit.is_commit());
        assert!(!commit.is_rollback());
        assert!(!commit.is_operation());

        assert!(!rollback.is_begin());
        assert!(!rollback.is_commit());
        assert!(rollback.is_rollback());
        assert!(!rollback.is_operation());

        assert!(!operation.is_begin());
        assert!(!operation.is_commit());
        assert!(!operation.is_rollback());
        assert!(operation.is_operation());
    }

    #[test]
    fn transaction_error_into_inner() {
        let err: TransactionError<&str> = TransactionError::Operation("test");
        assert_eq!(err.into_inner(), "test");
    }

    #[test]
    fn transaction_error_into_inner_begin() {
        let err: TransactionError<&str> = TransactionError::Begin("begin_err");
        assert_eq!(err.into_inner(), "begin_err");
    }

    #[test]
    fn transaction_error_into_inner_commit() {
        let err: TransactionError<&str> = TransactionError::Commit("commit_err");
        assert_eq!(err.into_inner(), "commit_err");
    }

    #[test]
    fn transaction_error_into_inner_rollback() {
        let err: TransactionError<&str> = TransactionError::Rollback("rollback_err");
        assert_eq!(err.into_inner(), "rollback_err");
    }

    #[test]
    fn transaction_error_source_begin() {
        let err: TransactionError<std::io::Error> =
            TransactionError::Begin(std::io::Error::other("src"));
        assert!(err.source().is_some());
    }

    #[test]
    fn transaction_error_source_commit() {
        let err: TransactionError<std::io::Error> =
            TransactionError::Commit(std::io::Error::other("src"));
        assert!(err.source().is_some());
    }

    #[test]
    fn transaction_error_source_rollback() {
        let err: TransactionError<std::io::Error> =
            TransactionError::Rollback(std::io::Error::other("src"));
        assert!(err.source().is_some());
    }

    #[test]
    fn transaction_error_source_operation() {
        let err: TransactionError<std::io::Error> =
            TransactionError::Operation(std::io::Error::other("src"));
        assert!(err.source().is_some());
    }

    #[test]
    fn transaction_builder_new() {
        struct MockPool;
        let pool = MockPool;
        let tx = Transaction::new(&pool);
        let _ = tx.pool();
    }

    #[test]
    fn transaction_builder_pool_accessor() {
        struct MockPool {
            id: u32
        }
        let pool = MockPool {
            id: 42
        };
        let tx = Transaction::new(&pool);
        assert_eq!(tx.pool().id, 42);
    }

    #[test]
    fn transaction_error_debug() {
        let err: TransactionError<&str> = TransactionError::Begin("test");
        let debug_str = format!("{:?}", err);
        assert!(debug_str.contains("Begin"));
        assert!(debug_str.contains("test"));
    }

    #[test]
    fn transaction_error_into_inner_all_variants() {
        let begin: TransactionError<String> = TransactionError::Begin("begin".to_string());
        let commit: TransactionError<String> = TransactionError::Commit("commit".to_string());
        let rollback: TransactionError<String> =
            TransactionError::Rollback("rollback".to_string());
        let operation: TransactionError<String> = TransactionError::Operation("op".to_string());

        assert_eq!(begin.into_inner(), "begin");
        assert_eq!(commit.into_inner(), "commit");
        assert_eq!(rollback.into_inner(), "rollback");
        assert_eq!(operation.into_inner(), "op");
    }

    #[test]
    fn transaction_error_source_all_variants() {
        let begin: TransactionError<std::io::Error> =
            TransactionError::Begin(std::io::Error::other("src"));
        let commit: TransactionError<std::io::Error> =
            TransactionError::Commit(std::io::Error::other("src"));
        let rollback: TransactionError<std::io::Error> =
            TransactionError::Rollback(std::io::Error::other("src"));
        let operation: TransactionError<std::io::Error> =
            TransactionError::Operation(std::io::Error::other("src"));

        assert!(begin.source().is_some());
        assert!(commit.source().is_some());
        assert!(rollback.source().is_some());
        assert!(operation.source().is_some());
    }

    #[test]
    fn transaction_error_display_all_variants() {
        let begin: TransactionError<std::io::Error> =
            TransactionError::Begin(std::io::Error::other("msg"));
        let commit: TransactionError<std::io::Error> =
            TransactionError::Commit(std::io::Error::other("msg"));
        let rollback: TransactionError<std::io::Error> =
            TransactionError::Rollback(std::io::Error::other("msg"));
        let operation: TransactionError<std::io::Error> =
            TransactionError::Operation(std::io::Error::other("msg"));

        let begin_str = begin.to_string();
        let commit_str = commit.to_string();
        let rollback_str = rollback.to_string();
        let operation_str = operation.to_string();

        assert!(begin_str.contains("begin"));
        assert!(commit_str.contains("commit"));
        assert!(rollback_str.contains("rollback"));
        assert!(operation_str.contains("operation"));
    }

    #[test]
    fn transaction_error_is_all_variants() {
        let begin: TransactionError<&str> = TransactionError::Begin("e");
        let commit: TransactionError<&str> = TransactionError::Commit("e");
        let rollback: TransactionError<&str> = TransactionError::Rollback("e");
        let operation: TransactionError<&str> = TransactionError::Operation("e");

        assert!(begin.is_begin());
        assert!(commit.is_commit());
        assert!(rollback.is_rollback());
        assert!(operation.is_operation());

        assert!(!begin.is_commit());
        assert!(!begin.is_rollback());
        assert!(!begin.is_operation());

        assert!(!commit.is_begin());
        assert!(!commit.is_rollback());
        assert!(!commit.is_operation());

        assert!(!rollback.is_begin());
        assert!(!rollback.is_commit());
        assert!(!rollback.is_operation());

        assert!(!operation.is_begin());
        assert!(!operation.is_commit());
        assert!(!operation.is_rollback());
    }

    #[test]
    fn transaction_builder_new_const() {
        struct MockPool;
        let pool = MockPool;
        let tx = Transaction::new(&pool);
        let _ = tx;
    }

    // Regression tests for `finalize_with_commit`.
    //
    // Prior to this fix, `Transaction::run` consumed `TransactionContext` and
    // dropped it before commit was ever called, so successful runs silently
    // rolled back. The fix is to keep ownership of `ctx` in `run` and call
    // commit on Ok. The backend-agnostic decision lives in
    // `finalize_with_commit`, which these tests cover end-to-end with a mock
    // context and a tracking commit closure — no database required.

    #[derive(Debug, PartialEq, Eq)]
    struct MockCtx;

    #[derive(Debug, PartialEq, Eq)]
    struct CommitErr(&'static str);

    #[derive(Debug, PartialEq, Eq)]
    enum AppErr {
        Closure(&'static str),
        Commit(&'static str)
    }

    impl From<CommitErr> for AppErr {
        fn from(e: CommitErr) -> Self {
            Self::Commit(e.0)
        }
    }

    #[tokio::test]
    async fn finalize_commits_on_ok() {
        let committed = std::sync::Arc::new(std::sync::atomic::AtomicBool::new(false));
        let flag = committed.clone();

        let result: Result<i32, AppErr> = finalize_with_commit::<_, _, _, CommitErr, _, _>(
            MockCtx,
            Ok::<i32, AppErr>(42),
            move |_ctx| {
                let flag = flag.clone();
                async move {
                    flag.store(true, std::sync::atomic::Ordering::SeqCst);
                    Ok::<(), CommitErr>(())
                }
            }
        )
        .await;

        assert_eq!(result, Ok(42));
        assert!(
            committed.load(std::sync::atomic::Ordering::SeqCst),
            "commit_fn must run on Ok"
        );
    }

    #[tokio::test]
    async fn finalize_skips_commit_on_err() {
        let committed = std::sync::Arc::new(std::sync::atomic::AtomicBool::new(false));
        let flag = committed.clone();

        let result: Result<i32, AppErr> = finalize_with_commit::<_, _, _, CommitErr, _, _>(
            MockCtx,
            Err::<i32, AppErr>(AppErr::Closure("nope")),
            move |_ctx| {
                let flag = flag.clone();
                async move {
                    flag.store(true, std::sync::atomic::Ordering::SeqCst);
                    Ok::<(), CommitErr>(())
                }
            }
        )
        .await;

        assert_eq!(result, Err(AppErr::Closure("nope")));
        assert!(
            !committed.load(std::sync::atomic::Ordering::SeqCst),
            "commit_fn must NOT run on Err"
        );
    }

    #[tokio::test]
    async fn finalize_propagates_commit_error_on_ok() {
        let result: Result<i32, AppErr> = finalize_with_commit::<_, _, _, CommitErr, _, _>(
            MockCtx,
            Ok::<i32, AppErr>(42),
            |_ctx| async { Err::<(), CommitErr>(CommitErr("commit failed")) }
        )
        .await;

        assert_eq!(result, Err(AppErr::Commit("commit failed")));
    }

    #[tokio::test]
    async fn finalize_preserves_closure_value_on_ok() {
        // Confirms the Ok payload survives the commit step (return type
        // matches the closure's success type, not the commit_fn result).
        let result: Result<String, AppErr> = finalize_with_commit::<_, _, _, CommitErr, _, _>(
            MockCtx,
            Ok::<String, AppErr>("payload".to_string()),
            |_ctx| async { Ok::<(), CommitErr>(()) }
        )
        .await;

        assert_eq!(result, Ok("payload".to_string()));
    }

    #[tokio::test]
    async fn finalize_does_not_swallow_closure_error_when_commit_also_would_fail() {
        // On Err, commit_fn is never called, so a faulty commit_fn cannot
        // hide the closure's original error.
        let result: Result<(), AppErr> = finalize_with_commit::<_, _, _, CommitErr, _, _>(
            MockCtx,
            Err::<(), AppErr>(AppErr::Closure("original")),
            |_ctx| async { Err::<(), CommitErr>(CommitErr("never reached")) }
        )
        .await;

        assert_eq!(result, Err(AppErr::Closure("original")));
    }
}