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::{ModelDescriptor, SqlValue, SqlxRuntime, UpsertModelInput, sqlx};
10
11use super::upsert_item::run_upsert_item;
12use super::validate::{reject_duplicate_sql_values, validate_batch_size};
13
14#[derive(Debug, Clone)]
15pub struct BatchUpsert<'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> BatchUpsert<'a, M, PK, I>
22where
23    I: UpsertModelInput<M>,
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        PK: Send + sqlx::Type<sqlx::Postgres> + for<'q> sqlx::Encode<'q, sqlx::Postgres>,
29    {
30        validate_batch_size(self.inputs.len())?;
31        // Upsert dedup runs on the per-input primary key — keeps two
32        // callers from both producing batches with the same key and
33        // ending up with surprising "second write wins" semantics.
34        let pks: Vec<SqlValue> = self
35            .inputs
36            .iter()
37            .map(UpsertModelInput::primary_key_value)
38            .collect();
39        reject_duplicate_sql_values(&pks)?;
40        if self.inputs.is_empty() {
41            return Ok(BatchResponse::from_results(vec![]));
42        }
43
44        let emits_created = self.descriptor.emits(ModelEventKind::Created);
45        let emits_updated = self.descriptor.emits(ModelEventKind::Updated);
46        let audit_enabled = self.descriptor.audit_enabled;
47
48        // Before the transaction takes a connection (`outbox_upgrade.rs`).
49        self.runtime
50            .upgrade_event_outbox(self.descriptor, emits_updated)
51            .await;
52
53        // A savepoint of the `@isolation` transaction when there is one
54        // (docs/design/procedure-isolation.md §4).
55        let (per_item, audit_events) = crate::bound::in_write_tx!(self.runtime, |tx| async {
56            if emits_created || emits_updated {
57                ensure_event_outbox_table(&mut **tx).await?;
58            }
59            let keep = self
60                .runtime
61                .keeps_before(&mut **tx, self.descriptor, emits_updated)
62                .await?;
63            if audit_enabled {
64                ensure_audit_table(self.runtime, &mut **tx).await?;
65            }
66
67            let mut per_item: Vec<Result<M, CratestackError>> =
68                Vec::with_capacity(self.inputs.len());
69            let mut audit_events = Vec::new();
70            for input in self.inputs {
71                let (outcome, audit_event) = run_upsert_item(
72                    tx,
73                    self.descriptor,
74                    input,
75                    ctx,
76                    emits_created,
77                    (emits_updated, keep),
78                    audit_enabled,
79                )
80                .await?;
81                per_item.push(outcome);
82                audit_events.extend(audit_event);
83            }
84
85            Ok((per_item, audit_events))
86        })?;
87
88        if emits_created || emits_updated {
89            let _ = self.runtime.drain_event_outbox().await;
90        }
91        dispatch_audit_sink(self.runtime, &audit_events).await;
92
93        Ok(BatchResponse::from_results(per_item))
94    }
95}
cratestack_sqlx/query/batch/upsert.rs

cratestack_sqlx/query/batch/
upsert.rs
