cratestack_sqlx/
descriptor.rs

1use std::sync::Arc;
2use std::sync::atomic::AtomicBool;
3
4use crate::sqlx;
5
6use cratestack_core::{
7    CoolError, CoolEventBus, CoolEventEnvelope, CoolEventFuture, ModelEventKind,
8};
9
10use crate::error::cool_error_from_sqlx;
11
12#[derive(Debug, Clone)]
13pub struct SqlxRuntime {
14    pool: sqlx::PgPool,
15    events: CoolEventBus,
16    // Shared (not per-clone) so every handle onto the same logical
17    // runtime agrees on whether `cratestack_audit` has been
18    // bootstrapped. See `crate::audit::ensure_audit_table` — this is
19    // what lets it skip re-issuing `CREATE INDEX IF NOT EXISTS` after
20    // the first call, which is what self-deadlocked chained
21    // `run_in_tx` audit writes in a caller-managed transaction.
22    audit_table_ensured: Arc<AtomicBool>,
23}
24
25impl SqlxRuntime {
26    pub fn new(pool: sqlx::PgPool) -> Self {
27        Self {
28            pool,
29            events: CoolEventBus::default(),
30            audit_table_ensured: Arc::new(AtomicBool::new(false)),
31        }
32    }
33
34    pub fn pool(&self) -> &sqlx::PgPool {
35        &self.pool
36    }
37
38    pub(crate) fn audit_table_ensured(&self) -> &AtomicBool {
39        &self.audit_table_ensured
40    }
41
42    #[doc(hidden)]
43    pub fn subscribe<F>(&self, model: &'static str, operation: ModelEventKind, handler: F)
44    where
45        F: Fn(CoolEventEnvelope) -> CoolEventFuture + Send + Sync + 'static,
46    {
47        self.events.subscribe(model, operation, handler);
48    }
49
50    #[doc(hidden)]
51    pub async fn drain_event_outbox(&self) -> Result<usize, CoolError> {
52        ensure_event_outbox_table(&self.pool).await?;
53
54        let rows = sqlx::query_as::<_, EventOutboxRow>(
55            "SELECT event_id, model, operation, occurred_at, payload, attempts, last_error \
56             FROM cratestack_event_outbox \
57             WHERE delivered_at IS NULL \
58             ORDER BY occurred_at ASC, event_id ASC",
59        )
60        .fetch_all(&self.pool)
61        .await
62        .map_err(cool_error_from_sqlx)?;
63
64        let mut delivered = 0usize;
65        for row in rows {
66            let event_id = row.event_id;
67            let envelope = row.try_into_envelope()?;
68            match self.events.emit(envelope).await {
69                Ok(()) => {
70                    sqlx::query(
71                        "UPDATE cratestack_event_outbox \
72                         SET delivered_at = NOW(), last_error = NULL, attempts = attempts + 1 \
73                         WHERE event_id = $1",
74                    )
75                    .bind(event_id)
76                    .execute(&self.pool)
77                    .await
78                    .map_err(cool_error_from_sqlx)?;
79                    delivered += 1;
80                }
81                Err(error) => {
82                    sqlx::query(
83                        "UPDATE cratestack_event_outbox \
84                         SET attempts = attempts + 1, last_error = $2 \
85                         WHERE event_id = $1",
86                    )
87                    .bind(event_id)
88                    .bind(error.to_string())
89                    .execute(&self.pool)
90                    .await
91                    .map_err(cool_error_from_sqlx)?;
92                }
93            }
94        }
95
96        Ok(delivered)
97    }
98}
99
100#[derive(Debug, Clone)]
101pub(crate) struct EventOutboxRow {
102    pub(crate) event_id: uuid::Uuid,
103    pub(crate) model: String,
104    pub(crate) operation: String,
105    pub(crate) occurred_at: chrono::DateTime<chrono::Utc>,
106    pub(crate) payload: serde_json::Value,
107    pub(crate) attempts: i64,
108    pub(crate) last_error: Option<String>,
109}
110
111// Hand-written `FromRow` impl. We can't use `#[derive(sqlx::FromRow)]` because
112// the derive macro hardcodes `::sqlx::*` paths that don't resolve through our
113// `crate::sqlx` shim (the shim is module-scoped, not crate-aliased).
114impl<'r> sqlx::FromRow<'r, sqlx::postgres::PgRow> for EventOutboxRow {
115    fn from_row(row: &'r sqlx::postgres::PgRow) -> Result<Self, sqlx::Error> {
116        use sqlx::Row;
117        Ok(Self {
118            event_id: row.try_get("event_id")?,
119            model: row.try_get("model")?,
120            operation: row.try_get("operation")?,
121            occurred_at: row.try_get("occurred_at")?,
122            payload: row.try_get("payload")?,
123            attempts: row.try_get("attempts")?,
124            last_error: row.try_get("last_error")?,
125        })
126    }
127}
128
129impl EventOutboxRow {
130    pub(crate) fn try_into_envelope(self) -> Result<CoolEventEnvelope, CoolError> {
131        let _ = self.attempts;
132        let _ = &self.last_error;
133        Ok(CoolEventEnvelope {
134            event_id: self.event_id,
135            model: self.model,
136            operation: ModelEventKind::parse(&self.operation)?,
137            occurred_at: self.occurred_at,
138            data: self.payload,
139        })
140    }
141}
142
143pub(crate) async fn ensure_event_outbox_table<'e, E>(executor: E) -> Result<(), CoolError>
144where
145    E: sqlx::Executor<'e, Database = sqlx::Postgres>,
146{
147    sqlx::query(
148        "CREATE TABLE IF NOT EXISTS cratestack_event_outbox (\
149            event_id UUID PRIMARY KEY, \
150            model TEXT NOT NULL, \
151            operation TEXT NOT NULL, \
152            occurred_at TIMESTAMPTZ NOT NULL, \
153            payload JSONB NOT NULL, \
154            delivered_at TIMESTAMPTZ, \
155            attempts BIGINT NOT NULL DEFAULT 0, \
156            last_error TEXT\
157        )",
158    )
159    .execute(executor)
160    .await
161    .map_err(cool_error_from_sqlx)?;
162
163    Ok(())
164}
165
166pub(crate) async fn enqueue_event_outbox<'e, E, T>(
167    executor: E,
168    model: &'static str,
169    operation: ModelEventKind,
170    data: &T,
171) -> Result<(), CoolError>
172where
173    E: sqlx::Executor<'e, Database = sqlx::Postgres>,
174    T: serde::Serialize,
175{
176    let payload = serde_json::to_value(data)
177        .map_err(|error| CoolError::Codec(format!("failed to encode event payload: {error}")))?;
178    sqlx::query(
179        "INSERT INTO cratestack_event_outbox (event_id, model, operation, occurred_at, payload) \
180         VALUES ($1, $2, $3, $4, $5)",
181    )
182    .bind(uuid::Uuid::new_v4())
183    .bind(model)
184    .bind(operation.as_str())
185    .bind(chrono::Utc::now())
186    .bind(payload)
187    .execute(executor)
188    .await
189    .map_err(cool_error_from_sqlx)?;
190
191    Ok(())
192}
cratestack_sqlx/descriptor.rs

cratestack_sqlx/
descriptor.rs
