cratestack-sqlx 0.14.1

//! `batch_create` driver — opens the outer tx, fans the inputs out to
//! [`super::create_item::run_create_item`] (one savepoint per input),
//! commits, drains the outbox.

use cratestack_core::{BatchResponse, CratestackContext, CratestackError, ModelEventKind};

use crate::audit::{dispatch_audit_sink, ensure_audit_table};
use crate::descriptor::ensure_event_outbox_table;
use crate::{CreateModelInput, ModelDescriptor, SqlxRuntime, sqlx};

use super::create_item::run_create_item;
use super::validate::validate_batch_size;

#[derive(Debug, Clone)]
pub struct BatchCreate<'a, M: 'static, PK: 'static, I> {
    pub(crate) runtime: &'a SqlxRuntime,
    pub(crate) descriptor: &'static ModelDescriptor<M, PK>,
    pub(crate) inputs: Vec<I>,
}

impl<'a, M: 'static, PK: 'static, I> BatchCreate<'a, M, PK, I>
where
    I: CreateModelInput<M> + Send,
{
    pub async fn run(self, ctx: &CratestackContext) -> Result<BatchResponse<M>, CratestackError>
    where
        for<'r> M: Send + Unpin + sqlx::FromRow<'r, sqlx::postgres::PgRow> + serde::Serialize,
    {
        validate_batch_size(self.inputs.len())?;
        // No PK dedup — `CreateModelInput` doesn't expose the PK
        // generically (and server-generated PKs make duplicates
        // impossible). Client-supplied PK collisions trip the DB
        // uniqueness constraint and surface as `CratestackError::Database`.
        // The right primitive for idempotent client-PK ingestion is
        // `.batch_upsert(...)`.
        if self.inputs.is_empty() {
            return Ok(BatchResponse::from_results(vec![]));
        }

        let emits_event = self.descriptor.emits(ModelEventKind::Created);
        let audit_enabled = self.descriptor.audit_enabled;

        // A savepoint of the `@isolation` transaction when there is one
        // (docs/design/procedure-isolation.md §4).
        let (per_item, audit_events) = crate::bound::in_write_tx!(self.runtime, |tx| async {
            if emits_event {
                ensure_event_outbox_table(&mut **tx).await?;
            }
            if audit_enabled {
                ensure_audit_table(self.runtime, &mut **tx).await?;
            }

            let mut per_item: Vec<Result<M, CratestackError>> =
                Vec::with_capacity(self.inputs.len());
            let mut audit_events = Vec::new();
            for input in self.inputs {
                let (outcome, audit_event) = run_create_item(
                    tx,
                    self.runtime,
                    self.descriptor,
                    input,
                    ctx,
                    emits_event,
                    audit_enabled,
                )
                .await?;
                per_item.push(outcome);
                audit_events.extend(audit_event);
            }

            Ok((per_item, audit_events))
        })?;

        if emits_event {
            let _ = self.runtime.drain_event_outbox().await;
        }
        dispatch_audit_sink(self.runtime, &audit_events).await;

        Ok(BatchResponse::from_results(per_item))
    }
}