cratestack_sqlx/query/batch/
upsert.rs

1//! `batch_upsert` driver — dedupes inputs by PK (different shape from
2//! `batch_update` because `UpsertModelInput` exposes a PK getter), then
3//! fans out to [`super::upsert_item::run_upsert_item`].
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::{
10    ModelDescriptor, SqlValue, SqlxRuntime, UpsertModelInput, cratestack_error_from_sqlx, sqlx,
11};
12
13use super::upsert_item::run_upsert_item;
14use super::validate::{reject_duplicate_sql_values, validate_batch_size};
15
16#[derive(Debug, Clone)]
17pub struct BatchUpsert<'a, M: 'static, PK: 'static, I> {
18    pub(crate) runtime: &'a SqlxRuntime,
19    pub(crate) descriptor: &'static ModelDescriptor<M, PK>,
20    pub(crate) inputs: Vec<I>,
21}
22
23impl<'a, M: 'static, PK: 'static, I> BatchUpsert<'a, M, PK, I>
24where
25    I: UpsertModelInput<M>,
26{
27    pub async fn run(self, ctx: &CratestackContext) -> Result<BatchResponse<M>, CratestackError>
28    where
29        for<'r> M: Send + Unpin + sqlx::FromRow<'r, sqlx::postgres::PgRow> + serde::Serialize,
30        PK: Send + sqlx::Type<sqlx::Postgres> + for<'q> sqlx::Encode<'q, sqlx::Postgres>,
31    {
32        validate_batch_size(self.inputs.len())?;
33        // Upsert dedup runs on the per-input primary key — keeps two
34        // callers from both producing batches with the same key and
35        // ending up with surprising "second write wins" semantics.
36        let pks: Vec<SqlValue> = self
37            .inputs
38            .iter()
39            .map(UpsertModelInput::primary_key_value)
40            .collect();
41        reject_duplicate_sql_values(&pks)?;
42        if self.inputs.is_empty() {
43            return Ok(BatchResponse::from_results(vec![]));
44        }
45
46        let emits_created = self.descriptor.emits(ModelEventKind::Created);
47        let emits_updated = self.descriptor.emits(ModelEventKind::Updated);
48        let audit_enabled = self.descriptor.audit_enabled;
49
50        let mut tx = self
51            .runtime
52            .pool()
53            .begin()
54            .await
55            .map_err(cratestack_error_from_sqlx)?;
56        if emits_created || emits_updated {
57            ensure_event_outbox_table(&mut *tx).await?;
58        }
59        if audit_enabled {
60            ensure_audit_table(self.runtime).await?;
61        }
62
63        let mut per_item: Vec<Result<M, CratestackError>> = Vec::with_capacity(self.inputs.len());
64        let mut audit_events = Vec::new();
65        for input in self.inputs {
66            let (outcome, audit_event) = run_upsert_item(
67                &mut tx,
68                self.runtime.pool(),
69                self.descriptor,
70                input,
71                ctx,
72                emits_created,
73                emits_updated,
74                audit_enabled,
75            )
76            .await?;
77            per_item.push(outcome);
78            audit_events.extend(audit_event);
79        }
80
81        tx.commit().await.map_err(cratestack_error_from_sqlx)?;
82
83        if emits_created || emits_updated {
84            let _ = self.runtime.drain_event_outbox().await;
85        }
86        dispatch_audit_sink(self.runtime, &audit_events).await;
87
88        Ok(BatchResponse::from_results(per_item))
89    }
90}
cratestack_sqlx/query/batch/upsert.rs

cratestack_sqlx/query/batch/
upsert.rs
