cratestack_sqlx/query/write/
create.rs

1//! `CreateRecord` — single-row INSERT with policy + audit + event
2//! fan-out. `run()` opens its own tx only when audit/event capture is
3//! enabled; otherwise it goes straight against the pool.
4
5use cratestack_core::{AuditOperation, CoolContext, CoolError, ModelEventKind};
6
7use crate::audit::{
8    RunInTxOutcome, build_audit_event, dispatch_audit_sink, enqueue_audit_event, ensure_audit_table,
9};
10use crate::descriptor::{enqueue_event_outbox, ensure_event_outbox_table};
11use crate::{CreateModelInput, ModelDescriptor, SqlxRuntime, cool_error_from_sqlx, sqlx};
12
13use super::create_exec::create_record_with_executor;
14
15#[derive(Debug, Clone)]
16pub struct CreateRecord<'a, M: 'static, PK: 'static, I> {
17    pub(crate) runtime: &'a SqlxRuntime,
18    pub(crate) descriptor: &'static ModelDescriptor<M, PK>,
19    pub(crate) input: I,
20}
21
22impl<'a, M: 'static, PK: 'static, I> CreateRecord<'a, M, PK, I>
23where
24    I: CreateModelInput<M>,
25{
26    pub fn preview_sql(&self) -> String {
27        let values = self.input.sql_values();
28        let placeholders = (1..=values.len())
29            .map(|index| format!("${index}"))
30            .collect::<Vec<_>>()
31            .join(", ");
32        let columns = values
33            .iter()
34            .map(|value| value.column)
35            .collect::<Vec<_>>()
36            .join(", ");
37
38        format!(
39            "INSERT INTO {} ({}) VALUES ({}) RETURNING {}",
40            self.descriptor.table_name,
41            columns,
42            placeholders,
43            self.descriptor.select_projection(),
44        )
45    }
46
47    /// Like [`Self::run`] but participates in a caller-supplied
48    /// transaction. The insert + outbox + audit writes all happen
49    /// inside `tx`; caller commits.
50    ///
51    /// Neither the event outbox nor the `AuditSink` fan-out run here —
52    /// both are post-commit, best-effort projections, and this
53    /// function has no visibility into when, or whether, the caller
54    /// commits `tx` (see [`crate::audit::dispatch_audit_sink`]'s doc
55    /// comment for the full reasoning). Unlike before cratestack#534,
56    /// though, neither is a dead end: this returns a
57    /// [`RunInTxOutcome`] carrying the `AuditEvent` this call built (if
58    /// `@@audit`), for the caller to pass to
59    /// `Cratestack::dispatch_audit_sink` once *their* commit succeeds;
60    /// and if the model `@@emit`s, the caller can call the pre-existing
61    /// `Cratestack::events().drain()` (cratestack#390) the same way — it
62    /// re-scans for any undelivered outbox row, so it needs no event
63    /// handed back to find this one.
64    pub async fn run_in_tx<'tx>(
65        self,
66        tx: &mut sqlx::Transaction<'tx, sqlx::Postgres>,
67        ctx: &CoolContext,
68    ) -> Result<RunInTxOutcome<M>, CoolError>
69    where
70        for<'r> M: Send + Unpin + sqlx::FromRow<'r, sqlx::postgres::PgRow> + serde::Serialize,
71    {
72        let emits_event = self.descriptor.emits(ModelEventKind::Created);
73        let audit_enabled = self.descriptor.audit_enabled;
74        if emits_event {
75            ensure_event_outbox_table(&mut **tx).await?;
76        }
77        if audit_enabled {
78            ensure_audit_table(self.runtime).await?;
79        }
80        let record = create_record_with_executor(
81            &mut **tx,
82            self.runtime.pool(),
83            self.descriptor,
84            self.input,
85            ctx,
86        )
87        .await?;
88        if emits_event {
89            enqueue_event_outbox(
90                &mut **tx,
91                self.descriptor.schema_name,
92                ModelEventKind::Created,
93                &record,
94            )
95            .await?;
96        }
97        let mut audit_event = None;
98        if audit_enabled {
99            let after = serde_json::to_value(&record).ok();
100            let event =
101                build_audit_event(self.descriptor, AuditOperation::Create, None, after, ctx);
102            enqueue_audit_event(&mut **tx, &event).await?;
103            audit_event = Some(event);
104        }
105        Ok(RunInTxOutcome::new(
106            record,
107            audit_event.into_iter().collect(),
108        ))
109    }
110
111    pub async fn run(self, ctx: &CoolContext) -> Result<M, CoolError>
112    where
113        for<'r> M: Send + Unpin + sqlx::FromRow<'r, sqlx::postgres::PgRow> + serde::Serialize,
114    {
115        let emits_event = self.descriptor.emits(ModelEventKind::Created);
116        let audit_enabled = self.descriptor.audit_enabled;
117        let needs_tx = emits_event || audit_enabled;
118        let mut audit_event = None;
119        let record = if needs_tx {
120            let mut tx = self
121                .runtime
122                .pool()
123                .begin()
124                .await
125                .map_err(cool_error_from_sqlx)?;
126            if emits_event {
127                ensure_event_outbox_table(&mut *tx).await?;
128            }
129            if audit_enabled {
130                ensure_audit_table(self.runtime).await?;
131            }
132            let record = create_record_with_executor(
133                &mut *tx,
134                self.runtime.pool(),
135                self.descriptor,
136                self.input,
137                ctx,
138            )
139            .await?;
140            if emits_event {
141                enqueue_event_outbox(
142                    &mut *tx,
143                    self.descriptor.schema_name,
144                    ModelEventKind::Created,
145                    &record,
146                )
147                .await?;
148            }
149            if audit_enabled {
150                let after = serde_json::to_value(&record).ok();
151                let event =
152                    build_audit_event(self.descriptor, AuditOperation::Create, None, after, ctx);
153                enqueue_audit_event(&mut *tx, &event).await?;
154                audit_event = Some(event);
155            }
156            tx.commit().await.map_err(cool_error_from_sqlx)?;
157            record
158        } else {
159            create_record_with_executor(
160                self.runtime.pool(),
161                self.runtime.pool(),
162                self.descriptor,
163                self.input,
164                ctx,
165            )
166            .await?
167        };
168
169        if emits_event {
170            let _ = self.runtime.drain_event_outbox().await;
171        }
172        if let Some(event) = &audit_event {
173            dispatch_audit_sink(self.runtime, std::slice::from_ref(event)).await;
174        }
175
176        Ok(record)
177    }
178}
cratestack_sqlx/query/write/create.rs

cratestack_sqlx/query/write/
create.rs
