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    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. Event outbox is *not* drained —
50    /// the outbox row isn't visible to the drain worker until commit.
51    /// Same reasoning applies to the `AuditSink` fan-out (cratestack#473):
52    /// it does not run here either, since this function has no
53    /// visibility into when — or whether — the caller commits `tx`.
54    pub async fn run_in_tx<'tx>(
55        self,
56        tx: &mut sqlx::Transaction<'tx, sqlx::Postgres>,
57        ctx: &CoolContext,
58    ) -> Result<M, CoolError>
59    where
60        for<'r> M: Send + Unpin + sqlx::FromRow<'r, sqlx::postgres::PgRow> + serde::Serialize,
61    {
62        let emits_event = self.descriptor.emits(ModelEventKind::Created);
63        let audit_enabled = self.descriptor.audit_enabled;
64        if emits_event {
65            ensure_event_outbox_table(&mut **tx).await?;
66        }
67        if audit_enabled {
68            ensure_audit_table(self.runtime).await?;
69        }
70        let record = create_record_with_executor(
71            &mut **tx,
72            self.runtime.pool(),
73            self.descriptor,
74            self.input,
75            ctx,
76        )
77        .await?;
78        if emits_event {
79            enqueue_event_outbox(
80                &mut **tx,
81                self.descriptor.schema_name,
82                ModelEventKind::Created,
83                &record,
84            )
85            .await?;
86        }
87        if audit_enabled {
88            let after = serde_json::to_value(&record).ok();
89            let event =
90                build_audit_event(self.descriptor, AuditOperation::Create, None, after, ctx);
91            enqueue_audit_event(&mut **tx, &event).await?;
92        }
93        Ok(record)
94    }
95
96    pub async fn run(self, ctx: &CoolContext) -> Result<M, CoolError>
97    where
98        for<'r> M: Send + Unpin + sqlx::FromRow<'r, sqlx::postgres::PgRow> + serde::Serialize,
99    {
100        let emits_event = self.descriptor.emits(ModelEventKind::Created);
101        let audit_enabled = self.descriptor.audit_enabled;
102        let needs_tx = emits_event || audit_enabled;
103        let mut audit_event = None;
104        let record = if needs_tx {
105            let mut tx = self
106                .runtime
107                .pool()
108                .begin()
109                .await
110                .map_err(cool_error_from_sqlx)?;
111            if emits_event {
112                ensure_event_outbox_table(&mut *tx).await?;
113            }
114            if audit_enabled {
115                ensure_audit_table(self.runtime).await?;
116            }
117            let record = create_record_with_executor(
118                &mut *tx,
119                self.runtime.pool(),
120                self.descriptor,
121                self.input,
122                ctx,
123            )
124            .await?;
125            if emits_event {
126                enqueue_event_outbox(
127                    &mut *tx,
128                    self.descriptor.schema_name,
129                    ModelEventKind::Created,
130                    &record,
131                )
132                .await?;
133            }
134            if audit_enabled {
135                let after = serde_json::to_value(&record).ok();
136                let event =
137                    build_audit_event(self.descriptor, AuditOperation::Create, None, after, ctx);
138                enqueue_audit_event(&mut *tx, &event).await?;
139                audit_event = Some(event);
140            }
141            tx.commit().await.map_err(cool_error_from_sqlx)?;
142            record
143        } else {
144            create_record_with_executor(
145                self.runtime.pool(),
146                self.runtime.pool(),
147                self.descriptor,
148                self.input,
149                ctx,
150            )
151            .await?
152        };
153
154        if emits_event {
155            let _ = self.runtime.drain_event_outbox().await;
156        }
157        if let Some(event) = &audit_event {
158            dispatch_audit_sink(self.runtime, std::slice::from_ref(event)).await;
159        }
160
161        Ok(record)
162    }
163}
cratestack_sqlx/query/write/create.rs

cratestack_sqlx/query/write/
create.rs
