noema-actix-webapi 0.1.0

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

use futures::future::BoxFuture;
use noema::core::{Container, Injectable};
use noema::resolve;

use crate::db::{PgPool, pool};

/// Opaque transaction token. Application code only passes this into repository ports.
pub struct Tx {
    inner: sqlx::Transaction<'static, sqlx::Postgres>,
}

impl Tx {
    pub(crate) fn inner_mut(&mut self) -> &mut sqlx::Transaction<'static, sqlx::Postgres> {
        &mut self.inner
    }
}

pub type BoxDynError = Box<dyn std::error::Error + Send + Sync>;

/// Object-safe unit of work. Use [`with_transaction!`] or `resolve::<dyn UnitOfWork + Send + Sync>()`.
#[async_trait::async_trait]
pub trait UnitOfWork: Send + Sync {
    async fn transaction(
        &self,
        f: Box<dyn for<'a> FnOnce(&'a mut Tx) -> BoxFuture<'a, Result<(), BoxDynError>> + Send>,
    ) -> Result<(), BoxDynError>;
}

pub struct PsqlUnitOfWork {
    pool: Arc<PgPool>,
}

impl Injectable<Container> for PsqlUnitOfWork {
    fn inject(_: &Container) -> Self {
        Self {
            pool: resolve::<PgPool>(),
        }
    }
}

#[async_trait::async_trait]
impl UnitOfWork for PsqlUnitOfWork {
    async fn transaction(
        &self,
        f: Box<dyn for<'a> FnOnce(&'a mut Tx) -> BoxFuture<'a, Result<(), BoxDynError>> + Send>,
    ) -> Result<(), BoxDynError> {
        let _ = self.pool.connection();
        let mut tx = Tx {
            inner: pool().begin().await?,
        };
        match f(&mut tx).await {
            Ok(()) => {
                tx.inner.commit().await?;
                Ok(())
            }
            Err(e) => {
                let _ = tx.inner.rollback().await;
                Err(e)
            }
        }
    }
}

/// Same as `resolve::<dyn UnitOfWork + Send + Sync>().transaction(...)`.
pub async fn run_transaction<F>(f: F) -> Result<(), BoxDynError>
where
    F: for<'a> FnOnce(
            &'a mut Tx,
        )
            -> Pin<Box<dyn Future<Output = Result<(), BoxDynError>> + Send + 'a>>
        + Send
        + 'static,
{
    let uow = resolve::<dyn UnitOfWork + Send + Sync>();
    uow.transaction(Box::new(f)).await
}

/// Run `body` inside a unit of work. `$tx` is `&mut Tx`.
///
/// ```ignore
/// with_transaction!(tx, {
///     users.save(tx, &user).await?;
///     Ok(())
/// })
/// ```
#[macro_export]
macro_rules! with_transaction {
    ($tx:ident, $body:block) => {
        $crate::uow::run_transaction(move |$tx| {
            Box::pin(async move $body)
        })
        .await
    };
}