cratestack_sqlx/
descriptor.rs

1mod event_outbox;
2
3use std::sync::Arc;
4use std::sync::atomic::AtomicBool;
5
6use crate::sqlx;
7
8use cratestack_core::{
9    AuditSink, CoolError, CoolEventBus, CoolEventEnvelope, CoolEventFuture, ModelEventKind,
10    NoopAuditSink, SubscriptionHandle,
11};
12
13use crate::error::cool_error_from_sqlx;
14use event_outbox::EventOutboxRow;
15
16pub(crate) use event_outbox::{enqueue_event_outbox, ensure_event_outbox_table};
17
18#[derive(Clone)]
19pub struct SqlxRuntime {
20    pool: sqlx::PgPool,
21    events: CoolEventBus,
22    // Shared (not per-clone) so every handle onto the same logical
23    // runtime agrees on whether `cratestack_audit` has been
24    // bootstrapped. See `crate::audit::ensure_audit_table` — this is
25    // what lets it skip re-issuing `CREATE INDEX IF NOT EXISTS` after
26    // the first call, which is what self-deadlocked chained
27    // `run_in_tx` audit writes in a caller-managed transaction.
28    audit_table_ensured: Arc<AtomicBool>,
29    // Installation point for cratestack#473: defaults to `NoopAuditSink`
30    // so existing callers of `new()` see no behavior change. Installed
31    // via `with_audit_sink` (mirrors `IdempotencyLayer::new`/
32    // `with_principal_fingerprint`'s builder shape). The DB write in
33    // `crate::audit::enqueue_audit_event` remains the sole source of
34    // truth; this is a best-effort downstream projection dispatched
35    // from `crate::audit::dispatch_audit_sink` after the owning
36    // transaction commits.
37    audit_sink: Arc<dyn AuditSink>,
38}
39
40// `dyn AuditSink` has no `Debug` bound (matching `IdempotencyStore` /
41// `RateLimitStore`, neither of which require one either), so this can't
42// be `#[derive(Debug)]` — same reason `CoolEventBus` hand-rolls its own
43// `Debug` impl instead of deriving one.
44impl std::fmt::Debug for SqlxRuntime {
45    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
46        f.debug_struct("SqlxRuntime")
47            .field("pool", &self.pool)
48            .field("events", &self.events)
49            .field(
50                "audit_table_ensured",
51                &self
52                    .audit_table_ensured
53                    .load(std::sync::atomic::Ordering::Relaxed),
54            )
55            .finish_non_exhaustive()
56    }
57}
58
59impl SqlxRuntime {
60    pub fn new(pool: sqlx::PgPool) -> Self {
61        Self {
62            pool,
63            events: CoolEventBus::default(),
64            audit_table_ensured: Arc::new(AtomicBool::new(false)),
65            audit_sink: Arc::new(NoopAuditSink),
66        }
67    }
68
69    pub fn pool(&self) -> &sqlx::PgPool {
70        &self.pool
71    }
72
73    pub(crate) fn audit_table_ensured(&self) -> &AtomicBool {
74        &self.audit_table_ensured
75    }
76
77    /// Install a custom [`AuditSink`] that every `@@audit` mutation on
78    /// this runtime fans out to, in addition to the in-database
79    /// `cratestack_audit` table row `enqueue_audit_event` always
80    /// writes. Composable via [`cratestack_core::MulticastAuditSink`]
81    /// for more than one downstream (Kafka, Redis pubsub, a webhook —
82    /// this crate ships none of them; see `AuditSink`'s doc comment).
83    pub fn with_audit_sink(mut self, sink: Arc<dyn AuditSink>) -> Self {
84        self.audit_sink = sink;
85        self
86    }
87
88    pub(crate) fn audit_sink(&self) -> &Arc<dyn AuditSink> {
89        &self.audit_sink
90    }
91
92    #[doc(hidden)]
93    pub fn subscribe<F>(
94        &self,
95        model: &'static str,
96        operation: ModelEventKind,
97        handler: F,
98    ) -> SubscriptionHandle
99    where
100        F: Fn(CoolEventEnvelope) -> CoolEventFuture + Send + Sync + 'static,
101    {
102        self.events.subscribe(model, operation, handler)
103    }
104
105    /// An owned, cheaply-cloneable handle onto the underlying
106    /// `CoolEventBus` — needed by callers (e.g. `@@subscribe` SSE
107    /// dispatch, cratestack#390) that outlive the `&SqlxRuntime` borrow
108    /// `subscribe`/`unsubscribe` would otherwise require.
109    #[doc(hidden)]
110    pub fn events_bus(&self) -> CoolEventBus {
111        self.events.clone()
112    }
113
114    #[doc(hidden)]
115    pub async fn drain_event_outbox(&self) -> Result<usize, CoolError> {
116        ensure_event_outbox_table(&self.pool).await?;
117
118        let rows = sqlx::query_as::<_, EventOutboxRow>(
119            "SELECT event_id, model, operation, occurred_at, payload, attempts, last_error \
120             FROM cratestack_event_outbox \
121             WHERE delivered_at IS NULL \
122             ORDER BY occurred_at ASC, event_id ASC",
123        )
124        .fetch_all(&self.pool)
125        .await
126        .map_err(cool_error_from_sqlx)?;
127
128        let mut delivered = 0usize;
129        for row in rows {
130            let event_id = row.event_id;
131            let envelope = row.try_into_envelope()?;
132            match self.events.emit(envelope).await {
133                Ok(()) => {
134                    sqlx::query(
135                        "UPDATE cratestack_event_outbox \
136                         SET delivered_at = NOW(), last_error = NULL, attempts = attempts + 1 \
137                         WHERE event_id = $1",
138                    )
139                    .bind(event_id)
140                    .execute(&self.pool)
141                    .await
142                    .map_err(cool_error_from_sqlx)?;
143                    delivered += 1;
144                }
145                Err(error) => {
146                    sqlx::query(
147                        "UPDATE cratestack_event_outbox \
148                         SET attempts = attempts + 1, last_error = $2 \
149                         WHERE event_id = $1",
150                    )
151                    .bind(event_id)
152                    .bind(error.to_string())
153                    .execute(&self.pool)
154                    .await
155                    .map_err(cool_error_from_sqlx)?;
156                }
157            }
158        }
159
160        Ok(delivered)
161    }
162}
cratestack_sqlx/descriptor.rs

cratestack_sqlx/
descriptor.rs
