cratestack_sqlx/descriptor/
event_outbox.rs

1//! `cratestack_event_outbox` table primitives: the row shape, its
2//! hand-written `FromRow` (see the comment on the impl below for why
3//! it isn't derived), table bootstrap, and the enqueue helper each
4//! write path calls in-transaction. Split out of `descriptor.rs` to
5//! keep that file under the project's ~200-line-per-file convention.
6
7use cratestack_core::{CratestackError, CratestackEventEnvelope, ModelEventKind};
8
9use crate::error::cratestack_error_from_sqlx;
10use crate::sqlx;
11
12#[derive(Debug, Clone)]
13pub(crate) struct EventOutboxRow {
14    pub(crate) event_id: uuid::Uuid,
15    pub(crate) model: String,
16    pub(crate) operation: String,
17    pub(crate) occurred_at: chrono::DateTime<chrono::Utc>,
18    pub(crate) payload: serde_json::Value,
19    pub(crate) attempts: i64,
20    pub(crate) last_error: Option<String>,
21}
22
23// Hand-written `FromRow` impl. We can't use `#[derive(sqlx::FromRow)]` because
24// the derive macro hardcodes `::sqlx::*` paths that don't resolve through our
25// `crate::sqlx` shim (the shim is module-scoped, not crate-aliased).
26impl<'r> sqlx::FromRow<'r, sqlx::postgres::PgRow> for EventOutboxRow {
27    fn from_row(row: &'r sqlx::postgres::PgRow) -> Result<Self, sqlx::Error> {
28        use sqlx::Row;
29        Ok(Self {
30            event_id: row.try_get("event_id")?,
31            model: row.try_get("model")?,
32            operation: row.try_get("operation")?,
33            occurred_at: row.try_get("occurred_at")?,
34            payload: row.try_get("payload")?,
35            attempts: row.try_get("attempts")?,
36            last_error: row.try_get("last_error")?,
37        })
38    }
39}
40
41impl EventOutboxRow {
42    pub(crate) fn try_into_envelope(self) -> Result<CratestackEventEnvelope, CratestackError> {
43        let _ = self.attempts;
44        let _ = &self.last_error;
45        Ok(CratestackEventEnvelope {
46            event_id: self.event_id,
47            model: self.model,
48            operation: ModelEventKind::parse(&self.operation)?,
49            occurred_at: self.occurred_at,
50            data: self.payload,
51        })
52    }
53}
54
55/// Bootstraps `cratestack_event_outbox` if it doesn't already exist.
56/// `pub` (rather than `pub(crate)`) since cratestack#507 ("option 3"):
57/// `cratestack-studio`'s `[target.db]` write path calls this directly to
58/// route Studio writes through the same outbox the generated server
59/// uses, rather than duplicating the table DDL in a second crate where
60/// it could drift out of sync with this one.
61pub async fn ensure_event_outbox_table<'e, E>(executor: E) -> Result<(), CratestackError>
62where
63    E: sqlx::Executor<'e, Database = sqlx::Postgres>,
64{
65    sqlx::query(
66        "CREATE TABLE IF NOT EXISTS cratestack_event_outbox (\
67            event_id UUID PRIMARY KEY, \
68            model TEXT NOT NULL, \
69            operation TEXT NOT NULL, \
70            occurred_at TIMESTAMPTZ NOT NULL, \
71            payload JSONB NOT NULL, \
72            delivered_at TIMESTAMPTZ, \
73            attempts BIGINT NOT NULL DEFAULT 0, \
74            last_error TEXT\
75        )",
76    )
77    .execute(executor)
78    .await
79    .map_err(cratestack_error_from_sqlx)?;
80
81    Ok(())
82}
83
84/// Inserts one `cratestack_event_outbox` row. `pub` for the same reason
85/// as [`ensure_event_outbox_table`] — see its doc comment.
86///
87/// `model` takes `&str` rather than the `&'static str` every in-crate
88/// caller happens to pass (generated code's model names are always
89/// `&'static str` literals): `cratestack-studio` parses `.cstack`
90/// schemas at runtime, so its model names are owned `String`s with no
91/// `'static` lifetime available. The function only ever borrows `model`
92/// long enough to bind it as a query parameter, so relaxing the bound
93/// costs nothing for the existing callers and is required for the new
94/// one.
95pub async fn enqueue_event_outbox<'e, E, T>(
96    executor: E,
97    model: &str,
98    operation: ModelEventKind,
99    data: &T,
100) -> Result<(), CratestackError>
101where
102    E: sqlx::Executor<'e, Database = sqlx::Postgres>,
103    T: serde::Serialize,
104{
105    let payload = serde_json::to_value(data).map_err(|error| {
106        CratestackError::Codec(format!("failed to encode event payload: {error}"))
107    })?;
108    sqlx::query(
109        "INSERT INTO cratestack_event_outbox (event_id, model, operation, occurred_at, payload) \
110         VALUES ($1, $2, $3, $4, $5)",
111    )
112    .bind(uuid::Uuid::new_v4())
113    .bind(model)
114    .bind(operation.as_str())
115    .bind(chrono::Utc::now())
116    .bind(payload)
117    .execute(executor)
118    .await
119    .map_err(cratestack_error_from_sqlx)?;
120
121    Ok(())
122}
cratestack_sqlx/descriptor/event_outbox.rs

cratestack_sqlx/descriptor/
event_outbox.rs
