cratestack_sqlx/query/batch/
create.rs

1//! `batch_create` driver — opens the outer tx, fans the inputs out to
2//! [`super::create_item::run_create_item`] (one savepoint per input),
3//! commits, drains the outbox.
4
5use cratestack_core::{BatchResponse, CratestackContext, CratestackError, ModelEventKind};
6
7use crate::audit::{dispatch_audit_sink, ensure_audit_table};
8use crate::descriptor::ensure_event_outbox_table;
9use crate::{CreateModelInput, ModelDescriptor, SqlxRuntime, sqlx};
10
11use super::create_item::run_create_item;
12use super::validate::validate_batch_size;
13
14#[derive(Debug, Clone)]
15pub struct BatchCreate<'a, M: 'static, PK: 'static, I> {
16    pub(crate) runtime: &'a SqlxRuntime,
17    pub(crate) descriptor: &'static ModelDescriptor<M, PK>,
18    pub(crate) inputs: Vec<I>,
19}
20
21impl<'a, M: 'static, PK: 'static, I> BatchCreate<'a, M, PK, I>
22where
23    I: CreateModelInput<M> + Send,
24{
25    pub async fn run(self, ctx: &CratestackContext) -> Result<BatchResponse<M>, CratestackError>
26    where
27        for<'r> M: Send + Unpin + sqlx::FromRow<'r, sqlx::postgres::PgRow> + serde::Serialize,
28    {
29        validate_batch_size(self.inputs.len())?;
30        // No PK dedup — `CreateModelInput` doesn't expose the PK
31        // generically (and server-generated PKs make duplicates
32        // impossible). Client-supplied PK collisions trip the DB
33        // uniqueness constraint and surface as `CratestackError::Database`.
34        // The right primitive for idempotent client-PK ingestion is
35        // `.batch_upsert(...)`.
36        if self.inputs.is_empty() {
37            return Ok(BatchResponse::from_results(vec![]));
38        }
39
40        let emits_event = self.descriptor.emits(ModelEventKind::Created);
41        let audit_enabled = self.descriptor.audit_enabled;
42
43        // A savepoint of the `@isolation` transaction when there is one
44        // (docs/design/procedure-isolation.md §4).
45        let (per_item, audit_events) = crate::bound::in_write_tx!(self.runtime, |tx| async {
46            if emits_event {
47                ensure_event_outbox_table(&mut **tx).await?;
48            }
49            if audit_enabled {
50                ensure_audit_table(self.runtime, &mut **tx).await?;
51            }
52
53            let mut per_item: Vec<Result<M, CratestackError>> =
54                Vec::with_capacity(self.inputs.len());
55            let mut audit_events = Vec::new();
56            for input in self.inputs {
57                let (outcome, audit_event) = run_create_item(
58                    tx,
59                    self.runtime,
60                    self.descriptor,
61                    input,
62                    ctx,
63                    emits_event,
64                    audit_enabled,
65                )
66                .await?;
67                per_item.push(outcome);
68                audit_events.extend(audit_event);
69            }
70
71            Ok((per_item, audit_events))
72        })?;
73
74        if emits_event {
75            let _ = self.runtime.drain_event_outbox().await;
76        }
77        dispatch_audit_sink(self.runtime, &audit_events).await;
78
79        Ok(BatchResponse::from_results(per_item))
80    }
81}
cratestack_sqlx/query/batch/create.rs

cratestack_sqlx/query/batch/
create.rs
