cratestack_sqlx/
descriptor.rs

1use std::fmt::Write;
2use std::marker::PhantomData;
3
4use cratestack_core::{
5    CoolError, CoolEventBus, CoolEventEnvelope, CoolEventFuture, ModelEventKind,
6};
7use cratestack_policy::ReadPolicy;
8
9#[derive(Debug, Clone)]
10pub struct SqlxRuntime {
11    pool: sqlx::PgPool,
12    events: CoolEventBus,
13}
14
15impl SqlxRuntime {
16    pub fn new(pool: sqlx::PgPool) -> Self {
17        Self {
18            pool,
19            events: CoolEventBus::default(),
20        }
21    }
22
23    pub fn pool(&self) -> &sqlx::PgPool {
24        &self.pool
25    }
26
27    #[doc(hidden)]
28    pub fn subscribe<F>(&self, model: &'static str, operation: ModelEventKind, handler: F)
29    where
30        F: Fn(CoolEventEnvelope) -> CoolEventFuture + Send + Sync + 'static,
31    {
32        self.events.subscribe(model, operation, handler);
33    }
34
35    #[doc(hidden)]
36    pub async fn drain_event_outbox(&self) -> Result<usize, CoolError> {
37        ensure_event_outbox_table(&self.pool).await?;
38
39        let rows = sqlx::query_as::<_, EventOutboxRow>(
40            "SELECT event_id, model, operation, occurred_at, payload, attempts, last_error \
41             FROM cratestack_event_outbox \
42             WHERE delivered_at IS NULL \
43             ORDER BY occurred_at ASC, event_id ASC",
44        )
45        .fetch_all(&self.pool)
46        .await
47        .map_err(|error| CoolError::Database(error.to_string()))?;
48
49        let mut delivered = 0usize;
50        for row in rows {
51            let event_id = row.event_id;
52            let envelope = row.try_into_envelope()?;
53            match self.events.emit(envelope).await {
54                Ok(()) => {
55                    sqlx::query(
56                        "UPDATE cratestack_event_outbox \
57                         SET delivered_at = NOW(), last_error = NULL, attempts = attempts + 1 \
58                         WHERE event_id = $1",
59                    )
60                    .bind(event_id)
61                    .execute(&self.pool)
62                    .await
63                    .map_err(|error| CoolError::Database(error.to_string()))?;
64                    delivered += 1;
65                }
66                Err(error) => {
67                    sqlx::query(
68                        "UPDATE cratestack_event_outbox \
69                         SET attempts = attempts + 1, last_error = $2 \
70                         WHERE event_id = $1",
71                    )
72                    .bind(event_id)
73                    .bind(error.to_string())
74                    .execute(&self.pool)
75                    .await
76                    .map_err(|db_error| CoolError::Database(db_error.to_string()))?;
77                }
78            }
79        }
80
81        Ok(delivered)
82    }
83}
84
85#[derive(Debug, Clone, sqlx::FromRow)]
86pub(crate) struct EventOutboxRow {
87    pub(crate) event_id: uuid::Uuid,
88    pub(crate) model: String,
89    pub(crate) operation: String,
90    pub(crate) occurred_at: chrono::DateTime<chrono::Utc>,
91    pub(crate) payload: serde_json::Value,
92    pub(crate) attempts: i64,
93    pub(crate) last_error: Option<String>,
94}
95
96impl EventOutboxRow {
97    pub(crate) fn try_into_envelope(self) -> Result<CoolEventEnvelope, CoolError> {
98        let _ = self.attempts;
99        let _ = &self.last_error;
100        Ok(CoolEventEnvelope {
101            event_id: self.event_id,
102            model: self.model,
103            operation: ModelEventKind::parse(&self.operation)?,
104            occurred_at: self.occurred_at,
105            data: self.payload,
106        })
107    }
108}
109
110pub(crate) async fn ensure_event_outbox_table<'e, E>(executor: E) -> Result<(), CoolError>
111where
112    E: sqlx::Executor<'e, Database = sqlx::Postgres>,
113{
114    sqlx::query(
115        "CREATE TABLE IF NOT EXISTS cratestack_event_outbox (\
116            event_id UUID PRIMARY KEY, \
117            model TEXT NOT NULL, \
118            operation TEXT NOT NULL, \
119            occurred_at TIMESTAMPTZ NOT NULL, \
120            payload JSONB NOT NULL, \
121            delivered_at TIMESTAMPTZ, \
122            attempts BIGINT NOT NULL DEFAULT 0, \
123            last_error TEXT\
124        )",
125    )
126    .execute(executor)
127    .await
128    .map_err(|error| CoolError::Database(error.to_string()))?;
129
130    Ok(())
131}
132
133pub(crate) async fn enqueue_event_outbox<'e, E, T>(
134    executor: E,
135    model: &'static str,
136    operation: ModelEventKind,
137    data: &T,
138) -> Result<(), CoolError>
139where
140    E: sqlx::Executor<'e, Database = sqlx::Postgres>,
141    T: serde::Serialize,
142{
143    let payload = serde_json::to_value(data)
144        .map_err(|error| CoolError::Codec(format!("failed to encode event payload: {error}")))?;
145    sqlx::query(
146        "INSERT INTO cratestack_event_outbox (event_id, model, operation, occurred_at, payload) \
147         VALUES ($1, $2, $3, $4, $5)",
148    )
149    .bind(uuid::Uuid::new_v4())
150    .bind(model)
151    .bind(operation.as_str())
152    .bind(chrono::Utc::now())
153    .bind(payload)
154    .execute(executor)
155    .await
156    .map_err(|error| CoolError::Database(error.to_string()))?;
157
158    Ok(())
159}
160
161#[derive(Debug, Clone, Copy, PartialEq, Eq)]
162pub struct ModelColumn {
163    pub rust_name: &'static str,
164    pub sql_name: &'static str,
165}
166
167#[derive(Debug, Clone, Copy)]
168pub struct ModelDescriptor<M, PK> {
169    pub schema_name: &'static str,
170    pub table_name: &'static str,
171    pub columns: &'static [ModelColumn],
172    pub primary_key: &'static str,
173    pub allowed_fields: &'static [&'static str],
174    pub allowed_includes: &'static [&'static str],
175    pub allowed_sorts: &'static [&'static str],
176    pub read_allow_policies: &'static [ReadPolicy],
177    pub read_deny_policies: &'static [ReadPolicy],
178    pub detail_allow_policies: &'static [ReadPolicy],
179    pub detail_deny_policies: &'static [ReadPolicy],
180    pub create_allow_policies: &'static [ReadPolicy],
181    pub create_deny_policies: &'static [ReadPolicy],
182    pub update_allow_policies: &'static [ReadPolicy],
183    pub update_deny_policies: &'static [ReadPolicy],
184    pub delete_allow_policies: &'static [ReadPolicy],
185    pub delete_deny_policies: &'static [ReadPolicy],
186    pub create_defaults: &'static [CreateDefault],
187    pub emitted_events: &'static [ModelEventKind],
188    _marker: PhantomData<fn() -> (M, PK)>,
189}
190
191#[derive(Debug, Clone, Copy, PartialEq, Eq)]
192pub enum CreateDefaultType {
193    Bool,
194    Int,
195    String,
196}
197
198#[derive(Debug, Clone, Copy, PartialEq, Eq)]
199pub struct CreateDefault {
200    pub column: &'static str,
201    pub auth_field: &'static str,
202    pub ty: CreateDefaultType,
203    pub nullable: bool,
204}
205
206impl<M, PK> ModelDescriptor<M, PK> {
207    pub const fn new(
208        schema_name: &'static str,
209        table_name: &'static str,
210        columns: &'static [ModelColumn],
211        primary_key: &'static str,
212        allowed_fields: &'static [&'static str],
213        allowed_includes: &'static [&'static str],
214        allowed_sorts: &'static [&'static str],
215        read_allow_policies: &'static [ReadPolicy],
216        read_deny_policies: &'static [ReadPolicy],
217        detail_allow_policies: &'static [ReadPolicy],
218        detail_deny_policies: &'static [ReadPolicy],
219        create_allow_policies: &'static [ReadPolicy],
220        create_deny_policies: &'static [ReadPolicy],
221        update_allow_policies: &'static [ReadPolicy],
222        update_deny_policies: &'static [ReadPolicy],
223        delete_allow_policies: &'static [ReadPolicy],
224        delete_deny_policies: &'static [ReadPolicy],
225        create_defaults: &'static [CreateDefault],
226        emitted_events: &'static [ModelEventKind],
227    ) -> Self {
228        Self {
229            schema_name,
230            table_name,
231            columns,
232            primary_key,
233            allowed_fields,
234            allowed_includes,
235            allowed_sorts,
236            read_allow_policies,
237            read_deny_policies,
238            detail_allow_policies,
239            detail_deny_policies,
240            create_allow_policies,
241            create_deny_policies,
242            update_allow_policies,
243            update_deny_policies,
244            delete_allow_policies,
245            delete_deny_policies,
246            create_defaults,
247            emitted_events,
248            _marker: PhantomData,
249        }
250    }
251
252    pub fn emits(&self, operation: ModelEventKind) -> bool {
253        self.emitted_events.contains(&operation)
254    }
255
256    pub fn select_projection(&self) -> String {
257        let mut sql = String::new();
258        for (index, column) in self.columns.iter().enumerate() {
259            if index > 0 {
260                sql.push_str(", ");
261            }
262            let _ = write!(sql, "{} AS \"{}\"", column.sql_name, column.rust_name);
263        }
264        sql
265    }
266}
cratestack_sqlx/descriptor.rs

cratestack_sqlx/
descriptor.rs
