A small review of the proposed architecture up to branch 3

[frederikhors]

So in the last few days, I tried to change one of my app codebases to the one you are suggesting (the latest example of it right now is the branch 3).

I found a bit of issues which I hope we'll find a way to fix in the subsequent branches:

1. How do I write traits implementation on multiple files?

   This is important to me. I have tons of methods and I cannot have everything in one file only.

2. What if a repository method must be called within the same transaction?

   This is another big issue currently not handled.

3. I lose the ergonomics of writing this in the service method:

   impl Service {
       pub async fn buy_product(&self, customer_id: &str, id: &str) -> Result<()> {
           // This is just a fake code and a fake example
   
           let transaction = self.repo.begin(); // note begin() is custom and not leaks DB details to service layer
   
           let product = transaction.get_product(id).await?; // repo method
   
           let balance = transaction.check_money_balance(customer_id).await?; // repo method
   
           // This is just a simple example, but there can be many business rules here
           if balance < product.price() {
               return Err("not enough money left");
           }
   
           transaction.buy_product(id, customer_id).await?; // repo method
   
           transaction.commit(); // again, DB details not leaked to the service layer here
   
           Ok(())
       }
   }

   and instead, I have to call a specific repo method passing all the details and above all I'm forced to move some logic functions in the repo method:

   impl Service {
       pub async fn buy_product(&self, customer_id: &str, id: &str) -> Result<()> {
           // Since I cannot start transactions in the service anymore
           // I need to call a repo method to do everything in it instead for the DB transaction:
   
           let result = self.repo.buy_product(customer_id, id).await?; // repo method
   
           Ok(())
       }
   }
   
   impl Db {
       pub async fn buy_product(&self, customer_id: &str, id: &str) -> Result<()> {
           // This is the same code used to be in the service method
   
           let transaction = self.repo.begin();
   
           let product = self.get_product(transaction, id).await?; // repo method
   
           let balance = self.check_money_balance(transaction, customer_id).await?; // repo method
   
           // This kind of rule needs to be in the service layer, not in the repo one
           if balance < product.price() {
               return Err("not enough money left");
           }
   
           self.buy_product(transaction, id, customer_id).await?; // repo method
   
           transaction.commit();
   
           Ok(())
       }
   }

4. To be able to fix the previous point I need to create a repo method for each service method (and for the Trait), which is a mess to maintain: many files to write and update, many Trait declarations to write/update: a mess.

5. I can't call methods of other services in the same transaction (but I have to say that this point doesn't interest me much, in fact, it's wrong as you explain very well in your excellent guide).

   The simple example is this:

   impl Service {
       pub async fn buy_product(&self, customer_id: &str, id: &str) -> Result<()> {
           // This is just a fake code and a fake example
   
           let transaction = self.repo.begin(); // note begin() is custom and not leaks DB details to service layer
   
           self.settings_service.get_customer_sell_settings(transaction, customer_id).await?;
   
           // do something else
   
           transaction.commit(); // again, DB details not leaked to the service layer here
   
           Ok(())
       }
   }

   This is useful and usable only if settings_service is within the same server, maybe in the same codebase, you know: if it's very fast to respond.

It's all for now. I'll update this issue if I find something else to suggest to you. Thanks for everything.

[musjj]

Wrt point 3, why do you feel the need to move the service function to your repository layer? IMO, something like buy_product fits pretty perfectly in the service layer. I don't think there's any need to make your repository methods mirror your service methods.

[frederikhors]

@AngusGMorrison are you ok?

[samueljoli]

@frederikhors Hey how's it going? I too have been toiling and moiling over this detail, "How do I manage transactions" using this pattern. I've yet to come across a solution that respects functional boundaries and doesn't have to make concessions like leaking implementation details. Unfortunately I don't come bearing any gifts, but did want to see if you've made any progress on solving this. I too feel like this is the final piece in the puzzle. Secondly have you heard from @AngusGMorrison? 😅 , it's giving Satoshi Nakamoto...drops this gem and then disappears into obscurity.

Also, I'm using axum and I'm currently evaluating whether or not I want to use something like axum_sqlx_tx, which would wrap your routes inside of a transaction and commit upon success (2xx & 3xx response codes) response codes) and rollback upon failure (4xx & 5xx response codes).

[frederikhors]

Unfortunately, I haven't heard from him since.

My solution is to use transactions and not use 100% clean architecture, which I feel is increasingly restrictive these days.

[samueljoli]

@frederikhors Did you happen to use any of the implementations mentioned by TmLev? I think it was either type erasure or associated types?

[samueljoli]

Hey @frederikhors , with the help of claude, I was able to mostly implement the unit of work pattern to keep from leaking infrastructure implementation details into my domain.

Step 1: Define generic transactional contract

use std::future::Future;

pub trait Transactional: Send {
    fn commit(self) -> impl Future<Output = Result<(), RepositoryError>> + Send;
}

pub trait TransactionalFactory: Send + Sync + Clone + 'static {
    type Trx: Transactional;

    fn begin(&self) -> impl Future<Output = Result<Self::Trx, RepositoryError>> + Send;
}

Transactional models "this transaction scope can be committed" & TransactionalFactory models "the application can begin a new transaction scope"

Step 2: Define the use-case-specific transactional operations

pub trait WebauthnRepository: Transactional {
    fn persist_user(
        &mut self,
        user: &User,
    ) -> impl Future<Output = Result<(), RepositoryError>> + Send;

    fn persist_passkey(
        &mut self,
        passkey: &PassKey,
    ) -> impl Future<Output = Result<(), RepositoryError>> + Send;

    fn append_user_created(
        &mut self,
        event: &UserCreated,
    ) -> impl Future<Output = Result<(), RepositoryError>> + Send;

    fn delete_registration_ceremony(
        &mut self,
        ceremony_id: &str,
    ) -> impl Future<Output = Result<(), RepositoryError>> + Send;
}

pub trait TransactionalWebauthnRepository: TransactionalFactory<Trx = Self::Repository> {
    type Repository: WebauthnRepository;
}

Step 3: Define the application service that depends on the transactional factory

The application service does not know about SQL transactions directly. It only knows it can begin a coordinated transaction scope.

#[derive(Clone)]
pub struct WebauthnService<F>
where
    F: TransactionalWebauthnRepository,
{
    transactional_repo: F,
}

The service depends on the ability to begin a transaction. It does not depend on sqlx::Transaction or any concrete database type

Step 4: Implement the concrete adapter

#[derive(Clone)]
pub struct TransactionalWebauthnRepository {
    pool: Pool<Postgres>,
}

pub struct WebauthnRepositoryAdapter {
    tx: Transaction<'static, Postgres>,
}

impl TransactionalWebauthnRepository {
    pub fn new(pool: Pool<Postgres>) -> Self {
        Self { pool }
    }
}

impl TransactionalFactory for TransactionalWebauthnRepository {
    type Trx = WebauthnRepositoryAdapter;

    async fn begin(&self) -> Result<Self::Trx, RepositoryError> {
        let tx = self.pool.begin().await?;
        Ok(WebauthnRepositoryAdapter { tx })
    }
}

impl TransactionalWebauthnRepository for TransactionalWebauthnRepository {
    type Repository = WebauthnRepositoryAdapter;
}

impl Transactional for WebauthnRepositoryAdapter {
    async fn commit(self) -> Result<(), RepositoryError> {
        self.tx.commit().await?;
        Ok(())
    }
}

impl WebauthnRepository for WebauthnRepositoryAdapter {
    async fn persist_user(&mut self, user: &User) -> Result<(), RepositoryError> {
        sqlx::query(
            r#"
            insert into users (id, email)
            values ($1, $2)
            "#,
        )
        .bind(&user.id)
        .bind(&user.email)
        .execute(&mut *self.tx)
        .await?;

        Ok(())
    }

    async fn persist_passkey(&mut self, passkey: &PassKey) -> Result<(), RepositoryError> {
        sqlx::query(
            r#"
            insert into passkeys (user_id, credential_id)
            values ($1, $2)
            "#,
        )
        .bind(&passkey.user_id)
        .bind(&passkey.credential_id)
        .execute(&mut *self.tx)
        .await?;

        Ok(())
    }

    async fn append_user_created(&mut self, event: &UserCreated) -> Result<(), RepositoryError> {
        sqlx::query(
            r#"
            insert into outbox_events (event_type, payload)
            values ($1, $2)
            "#,
        )
        .bind("user.created")
        .bind(format!(r#"{{"user_id":"{}","email":"{}"}}"#, event.user_id, event.email))
        .execute(&mut *self.tx)
        .await?;

        Ok(())
    }

    async fn delete_registration_ceremony(
        &mut self,
        ceremony_id: &str,
    ) -> Result<(), RepositoryError> {
        sqlx::query(
            r#"
            delete from registration_ceremonies
            where id = $1
            "#,
        )
        .bind(ceremony_id)
        .execute(&mut *self.tx)
        .await?;

        Ok(())
    }
}

Step 5: Use the concrete adapter in the coordinated transactional operation

impl<F> WebauthnService<F>
where
    F: TransactionalWebauthnRepository,
{
    pub async fn finish_registration(
        &self,
        ceremony_id: &str,
        user: User,
        passkey: PassKey,
    ) -> Result<(), WebauthnError> {
        let event = UserCreated {
            user_id: user.id.clone(),
            email: user.email.clone(),
        };

        let mut uow = self
            .transactional_repo
            .begin()
            .await
            .map_err(|_| WebauthnError::Unknown)?;

        uow.persist_user(&user)
            .await
            .map_err(|_| WebauthnError::DuplicateEmail)?;

        uow.persist_passkey(&passkey)
            .await
            .map_err(|_| WebauthnError::DuplicateCredential)?;

        uow.append_user_created(&event)
            .await
            .map_err(|_| WebauthnError::Unknown)?;

        uow.delete_registration_ceremony(ceremony_id)
            .await
            .map_err(|_| WebauthnError::Unknown)?;

        uow.commit()
            .await
            .map_err(|_| WebauthnError::Unknown)?;

        Ok(())
    }
}

Rule of Thumb

Use a Unit of Work when:

• multiple writes must succeed or fail together
• the transaction boundary belongs to the use case, not one repository method
• you want to avoid leaking database transaction types through your domain-facing interfaces

Do not reach for it when:

• a simple read or single write is enough
• no coordinated commit boundary is needed
• need to coordinate writes across multiple domains. This is a DDD code smell. As I've seen mentioned before you can utilize the transactional outbox pattern. If you're using postgres you can configure this with transactional log tailing for near real time updates.

Hope this helps 👍