cratestack_sqlx/query/write/
update_many.rs

1//! Bulk UPDATE-by-predicate: emit one statement that mutates every
2//! row the filter matches AND the update policy admits, in one
3//! round-trip.
4//!
5//! Differences from per-row `.update(id).set(input)`:
6//!   * No `if_match` slot — bulk updates aren't an optimistic-locking
7//!     idiom. `@version` is auto-incremented for every matched row;
8//!     the caller does NOT supply an expected version.
9//!   * Requires at least one filter — predicate-less bulk updates
10//!     should be raw SQL so the intent is obvious at review.
11
12use cratestack_core::{BatchSummary, CratestackContext, CratestackError};
13
14use crate::audit::{RunInTxOutcome, dispatch_audit_sink};
15use crate::{
16    FilterExpr, ModelDescriptor, SqlxRuntime, UpdateModelInput, cratestack_error_from_sqlx, sqlx,
17};
18
19use super::preview::render_update_many_preview_sql;
20use super::update_many_exec::run_update_many_in_tx;
21
22#[derive(Debug, Clone)]
23pub struct UpdateMany<'a, M: 'static, PK: 'static> {
24    pub(crate) runtime: &'a SqlxRuntime,
25    pub(crate) descriptor: &'static ModelDescriptor<M, PK>,
26    pub(crate) filters: Vec<FilterExpr>,
27}
28
29impl<'a, M: 'static, PK: 'static> UpdateMany<'a, M, PK> {
30    pub fn where_(mut self, filter: crate::Filter) -> Self {
31        self.filters.push(FilterExpr::from(filter));
32        self
33    }
34
35    pub fn where_expr(mut self, filter: FilterExpr) -> Self {
36        self.filters.push(filter);
37        self
38    }
39
40    pub fn where_any(mut self, filters: impl IntoIterator<Item = FilterExpr>) -> Self {
41        self.filters.push(FilterExpr::any(filters));
42        self
43    }
44
45    /// Conditionally append a filter — `None` is a no-op.
46    pub fn where_optional<F>(mut self, filter: Option<F>) -> Self
47    where
48        F: Into<FilterExpr>,
49    {
50        if let Some(filter) = filter {
51            self.filters.push(filter.into());
52        }
53        self
54    }
55
56    /// Supply the patch values. Returns a builder ready to `.run(ctx)`.
57    pub fn set<I>(self, input: I) -> UpdateManySet<'a, M, PK, I> {
58        UpdateManySet {
59            runtime: self.runtime,
60            descriptor: self.descriptor,
61            filters: self.filters,
62            input,
63        }
64    }
65}
66
67#[derive(Debug, Clone)]
68pub struct UpdateManySet<'a, M: 'static, PK: 'static, I> {
69    pub(crate) runtime: &'a SqlxRuntime,
70    pub(crate) descriptor: &'static ModelDescriptor<M, PK>,
71    pub(crate) filters: Vec<FilterExpr>,
72    pub(crate) input: I,
73}
74
75impl<'a, M: 'static, PK: 'static, I> UpdateManySet<'a, M, PK, I>
76where
77    I: UpdateModelInput<M>,
78{
79    pub fn preview_sql(&self) -> String {
80        let values = self.input.sql_values();
81        let columns: Vec<&str> = values.iter().map(|v| v.column).collect();
82        render_update_many_preview_sql(
83            self.descriptor.table_name,
84            self.descriptor.soft_delete_column.is_some(),
85            self.descriptor.version_column,
86            &columns,
87            &self.descriptor.select_projection(),
88        )
89    }
90
91    /// Returns `BatchSummary { total, ok, err }` where
92    /// `total = ok = rows actually updated` and `err = 0`.
93    /// Statement-level failures surface as the outer `Err`.
94    pub async fn run(self, ctx: &CratestackContext) -> Result<BatchSummary, CratestackError>
95    where
96        for<'r> M: Send + Unpin + sqlx::FromRow<'r, sqlx::postgres::PgRow> + serde::Serialize,
97    {
98        // Inside an `@isolation` procedure: run on its transaction and
99        // defer the post-commit fan-out (docs/design/procedure-isolation.md
100        // §4, §6).
101        if let Some(bound) = self.runtime.bound() {
102            let emits = self
103                .descriptor
104                .emits(cratestack_core::ModelEventKind::Updated);
105            let outcome = crate::bound::in_bound_savepoint!(bound, |sp| self.run_in_tx(sp, ctx))?;
106            return Ok(bound.settle(outcome, emits));
107        }
108        let runtime = self.runtime;
109        let descriptor = self.descriptor;
110        let mut tx = runtime
111            .pool()
112            .begin()
113            .await
114            .map_err(cratestack_error_from_sqlx)?;
115        let (summary, emits_event, audit_events) =
116            run_update_many_in_tx(&mut tx, runtime, descriptor, &self.filters, self.input, ctx)
117                .await?;
118        tx.commit().await.map_err(cratestack_error_from_sqlx)?;
119        if emits_event {
120            let _ = runtime.drain_event_outbox().await;
121        }
122        dispatch_audit_sink(runtime, &audit_events).await;
123        Ok(summary)
124    }
125
126    /// Run inside a caller-supplied transaction. Audit + outbox
127    /// writes land in `tx`; caller commits. Neither the `AuditSink`
128    /// fan-out nor the event outbox drain happens here — see
129    /// `create.rs`'s `run_in_tx` doc comment for the full contract and
130    /// how a caller opts into both after their own commit.
131    pub async fn run_in_tx<'tx>(
132        self,
133        tx: &mut sqlx::Transaction<'tx, sqlx::Postgres>,
134        ctx: &CratestackContext,
135    ) -> Result<RunInTxOutcome<BatchSummary>, CratestackError>
136    where
137        for<'r> M: Send + Unpin + sqlx::FromRow<'r, sqlx::postgres::PgRow> + serde::Serialize,
138    {
139        let (summary, _emits_event, audit_events) = run_update_many_in_tx(
140            tx,
141            self.runtime,
142            self.descriptor,
143            &self.filters,
144            self.input,
145            ctx,
146        )
147        .await?;
148        Ok(RunInTxOutcome::new(summary, audit_events))
149    }
150}
cratestack_sqlx/query/write/update_many.rs

cratestack_sqlx/query/write/
update_many.rs
