puerto_sqlx/
uow.rs

1//! The Postgres side of a generated unit of work (ADR-0018 § 2): one transaction that saves an
2//! aggregate and records its events in the outbox. A generated `Pg{Entity}UnitOfWork` wraps a
3//! [`PgTransaction`] and only writes its entity's upsert, through [`PgTransaction::tx`].
4
5use puerto_core::{DomainEvent, EventEnvelope, RepositoryError};
6use sqlx::{PgConnection, PgPool, Postgres, Transaction};
7
8use crate::map_sqlx_error;
9
10/// An open transaction. Dropping it without [`PgTransaction::commit`] rolls it back — an early
11/// `?` in a use case is the rollback.
12pub struct PgTransaction {
13    tx: Transaction<'static, Postgres>,
14}
15
16impl PgTransaction {
17    pub async fn begin(pool: &PgPool) -> Result<Self, RepositoryError> {
18        let tx = pool.begin().await.map_err(|err| {
19            tracing::error!(error = %err, "unit_of_work: begin failed");
20            map_sqlx_error(&err)
21        })?;
22        Ok(Self { tx })
23    }
24
25    /// The connection the entity's own writes run on: `query(…).execute(uow.tx())`.
26    pub fn tx(&mut self) -> &mut PgConnection {
27        &mut self.tx
28    }
29
30    /// Records a new event in the outbox, inside this transaction.
31    pub async fn record(&mut self, event: &impl DomainEvent) -> Result<(), RepositoryError> {
32        self.record_envelope(&EventEnvelope::new(event)).await
33    }
34
35    /// Records an envelope built by the caller — one carrying headers (`correlation_id`).
36    pub async fn record_envelope(&mut self, event: &EventEnvelope) -> Result<(), RepositoryError> {
37        let headers = serde_json::to_value(&event.headers).map_err(|err| {
38            tracing::error!(error = %err, event_type = %event.event_type, "unit_of_work: headers not serialisable");
39            RepositoryError::InvalidData
40        })?;
41        sqlx::query(
42            "INSERT INTO outbox \
43             (id, aggregate_type, aggregate_id, event_type, schema_version, payload, headers, occurred_at) \
44             VALUES ($1, $2, $3, $4, $5, $6, $7, $8)",
45        )
46        .bind(event.id)
47        .bind(&event.aggregate_type)
48        .bind(event.aggregate_id)
49        .bind(&event.event_type)
50        .bind(i32::from(event.schema_version))
51        .bind(&event.payload)
52        .bind(headers)
53        .bind(event.occurred_at)
54        .execute(&mut *self.tx)
55        .await
56        .map_err(|err| {
57            tracing::error!(error = %err, event_type = %event.event_type, "unit_of_work: outbox insert failed");
58            map_sqlx_error(&err)
59        })?;
60        Ok(())
61    }
62
63    pub async fn commit(self) -> Result<(), RepositoryError> {
64        self.tx.commit().await.map_err(|err| {
65            tracing::error!(error = %err, "unit_of_work: commit failed");
66            map_sqlx_error(&err)
67        })
68    }
69}
puerto_sqlx/uow.rs

puerto_sqlx/
uow.rs
