cratestack_sqlx/
descriptor.rs

1mod drain;
2mod event_enqueue;
3mod event_outbox;
4mod outbox_read;
5mod outbox_upgrade;
6
7use std::sync::Arc;
8use std::sync::atomic::AtomicBool;
9
10use crate::sqlx;
11
12use cratestack_core::{
13    AuditSink, CratestackEventBus, CratestackEventEnvelope, CratestackEventFuture, ModelEventKind,
14    NoopAuditSink, SubscriptionBackplane, SubscriptionHandle,
15};
16
17pub use event_enqueue::enqueue_event_outbox;
18pub(crate) use event_enqueue::{enqueue_model_event, enqueue_model_update};
19pub use event_outbox::ensure_event_outbox_table;
20pub(crate) use outbox_read::outbox_envelope_json;
21pub(crate) use outbox_upgrade::OutboxColumn;
22
23#[derive(Clone)]
24pub struct SqlxRuntime {
25    pool: sqlx::PgPool,
26    events: CratestackEventBus,
27    // Shared (not per-clone) so every handle onto the same logical
28    // runtime agrees on whether `cratestack_audit` has been
29    // bootstrapped. See `crate::audit::ensure_audit_table` — this is
30    // what lets it skip re-issuing `CREATE INDEX IF NOT EXISTS` after
31    // the first call, which is what self-deadlocked chained
32    // `run_in_tx` audit writes in a caller-managed transaction.
33    audit_table_ensured: Arc<AtomicBool>,
34    // Shared for the same reason: whether the outbox table is known to have
35    // `payload_before` (`outbox_upgrade.rs`).
36    outbox_column: Arc<outbox_upgrade::OutboxColumn>,
37    // Installation point for cratestack#473: defaults to `NoopAuditSink`
38    // so existing callers of `new()` see no behavior change. Installed
39    // via `with_audit_sink` (mirrors `IdempotencyLayer::new`/
40    // `with_principal_fingerprint`'s builder shape). The DB write in
41    // `crate::audit::enqueue_audit_event` remains the sole source of
42    // truth; this is a best-effort downstream projection dispatched
43    // from `crate::audit::dispatch_audit_sink` after the owning
44    // transaction commits.
45    audit_sink: Arc<dyn AuditSink>,
46    // `Some` only on the per-attempt runtime `run_isolated` builds for an
47    // `@isolation` procedure: every executing path then runs on this
48    // transaction instead of the pool (docs/design/procedure-isolation.md §4).
49    pub(crate) bound: Option<Arc<crate::bound::BoundTx>>,
50    pub(crate) isolation_max_retries: u32,
51    // Where model events reach subscribers on every replica
52    // (docs/design/subscriptions-websocket.md §4). `PgNotifyBackplane` by
53    // default; `with_backplane` swaps it (an `InProcessBackplane` for a
54    // single replica behind a pooler that drops `LISTEN`).
55    backplane: Arc<dyn SubscriptionBackplane>,
56}
57
58// `dyn AuditSink` has no `Debug` bound (matching `IdempotencyStore` /
59// `RateLimitStore`, neither of which require one either), so this can't
60// be `#[derive(Debug)]` — same reason `CratestackEventBus` hand-rolls its own
61// `Debug` impl instead of deriving one.
62impl std::fmt::Debug for SqlxRuntime {
63    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
64        f.debug_struct("SqlxRuntime")
65            .field("pool", &self.pool)
66            .field("events", &self.events)
67            .field("bound", &self.bound.is_some())
68            .field("isolation_max_retries", &self.isolation_max_retries)
69            .field(
70                "audit_table_ensured",
71                &self
72                    .audit_table_ensured
73                    .load(std::sync::atomic::Ordering::Relaxed),
74            )
75            .finish_non_exhaustive()
76    }
77}
78
79impl SqlxRuntime {
80    pub fn new(pool: sqlx::PgPool) -> Self {
81        let backplane: Arc<dyn SubscriptionBackplane> =
82            Arc::new(crate::backplane::PgNotifyBackplane::new(pool.clone()));
83        Self {
84            pool,
85            events: CratestackEventBus::default(),
86            audit_table_ensured: Arc::new(AtomicBool::new(false)),
87            outbox_column: Arc::default(),
88            audit_sink: Arc::new(NoopAuditSink),
89            bound: None,
90            isolation_max_retries: crate::isolation::MAX_RETRIES_DEFAULT,
91            backplane,
92        }
93    }
94
95    pub fn pool(&self) -> &sqlx::PgPool {
96        &self.pool
97    }
98
99    pub(crate) fn audit_table_ensured(&self) -> &AtomicBool {
100        &self.audit_table_ensured
101    }
102
103    /// Install a custom [`AuditSink`] that every `@@audit` mutation on
104    /// this runtime fans out to, in addition to the in-database
105    /// `cratestack_audit` table row `enqueue_audit_event` always
106    /// writes. Composable via [`cratestack_core::MulticastAuditSink`]
107    /// for more than one downstream (Kafka, Redis pubsub, a webhook —
108    /// this crate ships none of them; see `AuditSink`'s doc comment).
109    pub fn with_audit_sink(mut self, sink: Arc<dyn AuditSink>) -> Self {
110        self.audit_sink = sink;
111        self
112    }
113
114    /// Replace the subscription backplane (default: `PgNotifyBackplane` on
115    /// this runtime's pool).
116    pub fn with_backplane(mut self, backplane: Arc<dyn SubscriptionBackplane>) -> Self {
117        self.backplane = backplane;
118        self
119    }
120
121    pub fn backplane(&self) -> &Arc<dyn SubscriptionBackplane> {
122        &self.backplane
123    }
124
125    pub(crate) fn audit_sink(&self) -> &Arc<dyn AuditSink> {
126        &self.audit_sink
127    }
128
129    #[doc(hidden)]
130    pub fn subscribe<F>(
131        &self,
132        model: &'static str,
133        operation: ModelEventKind,
134        handler: F,
135    ) -> SubscriptionHandle
136    where
137        F: Fn(CratestackEventEnvelope) -> CratestackEventFuture + Send + Sync + 'static,
138    {
139        self.events.subscribe(model, operation, handler)
140    }
141
142    /// An owned, cheaply-cloneable handle onto the underlying
143    /// `CratestackEventBus` — needed by callers (e.g. `@@subscribe` SSE
144    /// dispatch, cratestack#390) that outlive the `&SqlxRuntime` borrow
145    /// `subscribe`/`unsubscribe` would otherwise require.
146    #[doc(hidden)]
147    pub fn events_bus(&self) -> CratestackEventBus {
148        self.events.clone()
149    }
150}
cratestack_sqlx/descriptor.rs

cratestack_sqlx/
descriptor.rs
