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) and table bootstrap. The enqueue helpers each write
4//! path calls in-transaction are in [`super::event_enqueue`]. Split out of
5//! `descriptor.rs` to keep that file under the project's ~200-line-per-file
6//! convention.
7
8use cratestack_core::{CratestackError, CratestackEventEnvelope, ModelEventKind};
9
10use crate::error::cratestack_error_from_sqlx;
11use crate::sqlx;
12
13#[derive(Debug, Clone)]
14pub(crate) struct EventOutboxRow {
15    pub(crate) event_id: uuid::Uuid,
16    pub(crate) model: String,
17    pub(crate) operation: String,
18    pub(crate) occurred_at: chrono::DateTime<chrono::Utc>,
19    pub(crate) payload: serde_json::Value,
20    pub(crate) before: Option<serde_json::Value>,
21    pub(crate) attempts: i64,
22    pub(crate) last_error: Option<String>,
23}
24
25// Hand-written `FromRow` impl. We can't use `#[derive(sqlx::FromRow)]` because
26// the derive macro hardcodes `::sqlx::*` paths that don't resolve through our
27// `crate::sqlx` shim (the shim is module-scoped, not crate-aliased).
28impl<'r> sqlx::FromRow<'r, sqlx::postgres::PgRow> for EventOutboxRow {
29    fn from_row(row: &'r sqlx::postgres::PgRow) -> Result<Self, sqlx::Error> {
30        use sqlx::Row;
31        Ok(Self {
32            event_id: row.try_get("event_id")?,
33            model: row.try_get("model")?,
34            operation: row.try_get("operation")?,
35            occurred_at: row.try_get("occurred_at")?,
36            payload: row.try_get("payload")?,
37            // Read as `to_jsonb(o) -> 'payload_before'`, so a table without the
38            // column (not yet upgraded) reads as none; a NULL column is JSON null.
39            before: row
40                .try_get::<Option<serde_json::Value>, _>("payload_before")?
41                .filter(|value| !value.is_null()),
42            attempts: row.try_get("attempts")?,
43            last_error: row.try_get("last_error")?,
44        })
45    }
46}
47
48impl EventOutboxRow {
49    pub(crate) fn try_into_envelope(self) -> Result<CratestackEventEnvelope, CratestackError> {
50        let _ = self.attempts;
51        let _ = &self.last_error;
52        Ok(CratestackEventEnvelope {
53            event_id: self.event_id,
54            model: self.model,
55            operation: ModelEventKind::parse(&self.operation)?,
56            occurred_at: self.occurred_at,
57            data: self.payload,
58            before: self.before,
59        })
60    }
61}
62
63/// Bootstraps `cratestack_event_outbox` if it doesn't already exist. A table
64/// an earlier version created lacks `payload_before`; that is added outside
65/// the write transaction, once per runtime (`outbox_upgrade.rs`).
66/// `pub` (rather than `pub(crate)`) since cratestack#507 ("option 3"):
67/// `cratestack-studio`'s `[target.db]` write path calls this directly to
68/// route Studio writes through the same outbox the generated server
69/// uses, rather than duplicating the table DDL in a second crate where
70/// it could drift out of sync with this one.
71pub async fn ensure_event_outbox_table<'e, E>(executor: E) -> Result<(), CratestackError>
72where
73    E: sqlx::Executor<'e, Database = sqlx::Postgres>,
74{
75    sqlx::query(
76        "CREATE TABLE IF NOT EXISTS cratestack_event_outbox (\
77            event_id UUID PRIMARY KEY, \
78            model TEXT NOT NULL, \
79            operation TEXT NOT NULL, \
80            occurred_at TIMESTAMPTZ NOT NULL, \
81            payload JSONB NOT NULL, \
82            payload_before JSONB, \
83            delivered_at TIMESTAMPTZ, \
84            attempts BIGINT NOT NULL DEFAULT 0, \
85            last_error TEXT\
86        )",
87    )
88    .execute(executor)
89    .await
90    .map_err(cratestack_error_from_sqlx)?;
91
92    Ok(())
93}
cratestack_sqlx/descriptor/event_outbox.rs

cratestack_sqlx/descriptor/
event_outbox.rs
