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, CoolContext, CoolError};
13
14use crate::audit::dispatch_audit_sink;
15use crate::{
16    FilterExpr, ModelDescriptor, SqlxRuntime, UpdateModelInput, cool_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: &CoolContext) -> Result<BatchSummary, CoolError>
95    where
96        for<'r> M: Send + Unpin + sqlx::FromRow<'r, sqlx::postgres::PgRow> + serde::Serialize,
97    {
98        let runtime = self.runtime;
99        let descriptor = self.descriptor;
100        let mut tx = runtime.pool().begin().await.map_err(cool_error_from_sqlx)?;
101        let (summary, emits_event, audit_events) =
102            run_update_many_in_tx(&mut tx, runtime, descriptor, &self.filters, self.input, ctx)
103                .await?;
104        tx.commit().await.map_err(cool_error_from_sqlx)?;
105        if emits_event {
106            let _ = runtime.drain_event_outbox().await;
107        }
108        dispatch_audit_sink(runtime, &audit_events).await;
109        Ok(summary)
110    }
111
112    /// Run inside a caller-supplied transaction. Audit + outbox
113    /// writes land in `tx`; caller commits. No `AuditSink` fan-out
114    /// happens here for the same reason the event outbox isn't
115    /// drained here — see `create.rs`'s `run_in_tx` doc comment.
116    pub async fn run_in_tx<'tx>(
117        self,
118        tx: &mut sqlx::Transaction<'tx, sqlx::Postgres>,
119        ctx: &CoolContext,
120    ) -> Result<BatchSummary, CoolError>
121    where
122        for<'r> M: Send + Unpin + sqlx::FromRow<'r, sqlx::postgres::PgRow> + serde::Serialize,
123    {
124        let (summary, ..) = run_update_many_in_tx(
125            tx,
126            self.runtime,
127            self.descriptor,
128            &self.filters,
129            self.input,
130            ctx,
131        )
132        .await?;
133        Ok(summary)
134    }
135}
cratestack_sqlx/query/write/update_many.rs

cratestack_sqlx/query/write/
update_many.rs
