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::{
15    FilterExpr, ModelDescriptor, SqlxRuntime, UpdateModelInput, cool_error_from_sqlx, sqlx,
16};
17
18use super::preview::render_update_many_preview_sql;
19use super::update_many_exec::run_update_many_in_tx;
20
21#[derive(Debug, Clone)]
22pub struct UpdateMany<'a, M: 'static, PK: 'static> {
23    pub(crate) runtime: &'a SqlxRuntime,
24    pub(crate) descriptor: &'static ModelDescriptor<M, PK>,
25    pub(crate) filters: Vec<FilterExpr>,
26}
27
28impl<'a, M: 'static, PK: 'static> UpdateMany<'a, M, PK> {
29    pub fn where_(mut self, filter: crate::Filter) -> Self {
30        self.filters.push(FilterExpr::from(filter));
31        self
32    }
33
34    pub fn where_expr(mut self, filter: FilterExpr) -> Self {
35        self.filters.push(filter);
36        self
37    }
38
39    pub fn where_any(mut self, filters: impl IntoIterator<Item = FilterExpr>) -> Self {
40        self.filters.push(FilterExpr::any(filters));
41        self
42    }
43
44    /// Conditionally append a filter — `None` is a no-op.
45    pub fn where_optional<F>(mut self, filter: Option<F>) -> Self
46    where
47        F: Into<FilterExpr>,
48    {
49        if let Some(filter) = filter {
50            self.filters.push(filter.into());
51        }
52        self
53    }
54
55    /// Supply the patch values. Returns a builder ready to `.run(ctx)`.
56    pub fn set<I>(self, input: I) -> UpdateManySet<'a, M, PK, I> {
57        UpdateManySet {
58            runtime: self.runtime,
59            descriptor: self.descriptor,
60            filters: self.filters,
61            input,
62        }
63    }
64}
65
66#[derive(Debug, Clone)]
67pub struct UpdateManySet<'a, M: 'static, PK: 'static, I> {
68    pub(crate) runtime: &'a SqlxRuntime,
69    pub(crate) descriptor: &'static ModelDescriptor<M, PK>,
70    pub(crate) filters: Vec<FilterExpr>,
71    pub(crate) input: I,
72}
73
74impl<'a, M: 'static, PK: 'static, I> UpdateManySet<'a, M, PK, I>
75where
76    I: UpdateModelInput<M>,
77{
78    pub fn preview_sql(&self) -> String {
79        let values = self.input.sql_values();
80        let columns: Vec<&str> = values.iter().map(|v| v.column).collect();
81        render_update_many_preview_sql(
82            self.descriptor.table_name,
83            self.descriptor.soft_delete_column.is_some(),
84            self.descriptor.version_column,
85            &columns,
86            &self.descriptor.select_projection(),
87        )
88    }
89
90    /// Returns `BatchSummary { total, ok, err }` where
91    /// `total = ok = rows actually updated` and `err = 0`.
92    /// Statement-level failures surface as the outer `Err`.
93    pub async fn run(self, ctx: &CoolContext) -> Result<BatchSummary, CoolError>
94    where
95        for<'r> M: Send + Unpin + sqlx::FromRow<'r, sqlx::postgres::PgRow> + serde::Serialize,
96    {
97        let runtime = self.runtime;
98        let descriptor = self.descriptor;
99        let mut tx = runtime.pool().begin().await.map_err(cool_error_from_sqlx)?;
100        let (summary, emits_event) =
101            run_update_many_in_tx(&mut tx, runtime, descriptor, &self.filters, self.input, ctx)
102                .await?;
103        tx.commit().await.map_err(cool_error_from_sqlx)?;
104        if emits_event {
105            let _ = runtime.drain_event_outbox().await;
106        }
107        Ok(summary)
108    }
109
110    /// Run inside a caller-supplied transaction. Audit + outbox
111    /// writes land in `tx`; caller commits.
112    pub async fn run_in_tx<'tx>(
113        self,
114        tx: &mut sqlx::Transaction<'tx, sqlx::Postgres>,
115        ctx: &CoolContext,
116    ) -> Result<BatchSummary, CoolError>
117    where
118        for<'r> M: Send + Unpin + sqlx::FromRow<'r, sqlx::postgres::PgRow> + serde::Serialize,
119    {
120        let (summary, _) = run_update_many_in_tx(
121            tx,
122            self.runtime,
123            self.descriptor,
124            &self.filters,
125            self.input,
126            ctx,
127        )
128        .await?;
129        Ok(summary)
130    }
131}
cratestack_sqlx/query/write/update_many.rs

cratestack_sqlx/query/write/
update_many.rs
