atomr_patterns/tx_outbox/
mod.rs

1//! FR-9 — *Transactional* outbox: atomic event + outbox write.
2//!
3//! Distinct from [`crate::outbox`], which is a journal-**tailing relay** that
4//! re-publishes already-committed events at-least-once on a polling loop.
5//! This module instead writes the event row **and** its outbox row inside the
6//! **same** [`sqlx::Transaction`] the caller controls, so the two either
7//! commit together or roll back together — there is no window where an event
8//! exists without its outbox entry (or vice versa).
9//!
10//! The caller owns commit/rollback: [`TxOutbox::persist_with`] only enqueues
11//! both INSERTs on the supplied transaction.
12//!
13//! ```ignore
14//! let mut tx = pool.begin().await?;
15//! TxOutbox.persist_with(&mut tx, event_row, outbox_row).await?;
16//! tx.commit().await?; // both rows land atomically
17//! ```
18
19use thiserror::Error;
20
21/// The event row to append to `event_journal`. A small param struct so we
22/// never need to touch [`atomr_persistence::PersistentRepr`].
23#[derive(Debug, Clone)]
24pub struct EventRow {
25    pub persistence_id: String,
26    pub sequence_nr: u64,
27    pub payload: Vec<u8>,
28    pub manifest: String,
29    pub writer_uuid: String,
30}
31
32/// The outbox row to append to `outbox` in the same transaction.
33#[derive(Debug, Clone)]
34pub struct OutboxRow {
35    pub topic: String,
36    pub payload: Vec<u8>,
37}
38
39/// Errors from the transactional outbox write.
40#[derive(Debug, Error)]
41pub enum OutboxError {
42    #[error("backend error: {0}")]
43    Backend(String),
44}
45
46impl OutboxError {
47    pub fn backend(e: impl std::fmt::Display) -> Self {
48        Self::Backend(e.to_string())
49    }
50}
51
52/// Zero-sized handle for the transactional outbox write.
53pub struct TxOutbox;
54
55impl TxOutbox {
56    /// Enqueue both the event INSERT and the outbox INSERT on `tx`.
57    ///
58    /// Both statements execute against the *same* transaction, so the caller's
59    /// subsequent `tx.commit()` makes both durable atomically, and dropping /
60    /// rolling back `tx` discards both. This method does **not** commit.
61    ///
62    /// The target tables are assumed to exist (the SQL provider's
63    /// `event_journal` plus an `outbox(topic, payload, created_at)` table the
64    /// application provisions); see the crate tests for the reference DDL.
65    pub async fn persist_with(
66        &self,
67        tx: &mut sqlx::Transaction<'_, sqlx::Any>,
68        event: EventRow,
69        outbox: OutboxRow,
70    ) -> Result<(), OutboxError> {
71        let created_at = now_millis();
72
73        sqlx::query(
74            "INSERT INTO event_journal \
75             (persistence_id, sequence_nr, payload, manifest, writer_uuid, deleted, created_at) \
76             VALUES (?, ?, ?, ?, ?, 0, ?)",
77        )
78        .bind(&event.persistence_id)
79        .bind(event.sequence_nr as i64)
80        .bind(event.payload)
81        .bind(&event.manifest)
82        .bind(&event.writer_uuid)
83        .bind(created_at)
84        .execute(&mut **tx)
85        .await
86        .map_err(OutboxError::backend)?;
87
88        sqlx::query("INSERT INTO outbox (topic, payload, created_at) VALUES (?, ?, ?)")
89            .bind(&outbox.topic)
90            .bind(outbox.payload)
91            .bind(created_at)
92            .execute(&mut **tx)
93            .await
94            .map_err(OutboxError::backend)?;
95
96        Ok(())
97    }
98}
99
100fn now_millis() -> i64 {
101    use std::time::{SystemTime, UNIX_EPOCH};
102    SystemTime::now().duration_since(UNIX_EPOCH).map(|d| d.as_millis() as i64).unwrap_or(0)
103}
atomr_patterns/tx_outbox/mod.rs

atomr_patterns/tx_outbox/
mod.rs
