cratestack_sqlx/query/write/
delete_many.rs

1//! Bulk DELETE-by-predicate: one statement tombstones (soft-delete)
2//! or removes (hard-delete) every row matching the filter AND the
3//! delete policy.
4//!
5//! Same shape as `update_many` — refuses to run without ≥1 filter so
6//! callers can't accidentally truncate a table at the typed-builder
7//! level.
8
9use cratestack_core::{BatchSummary, CratestackContext, CratestackError};
10
11use crate::audit::{RunInTxOutcome, dispatch_audit_sink};
12use crate::{FilterExpr, ModelDescriptor, SqlxRuntime, cratestack_error_from_sqlx, sqlx};
13
14use super::delete_many_exec::run_delete_many_in_tx;
15
16#[derive(Debug, Clone)]
17pub struct DeleteMany<'a, M: 'static, PK: 'static> {
18    pub(crate) runtime: &'a SqlxRuntime,
19    pub(crate) descriptor: &'static ModelDescriptor<M, PK>,
20    pub(crate) filters: Vec<FilterExpr>,
21}
22
23impl<'a, M: 'static, PK: 'static> DeleteMany<'a, M, PK> {
24    pub fn where_(mut self, filter: crate::Filter) -> Self {
25        self.filters.push(FilterExpr::from(filter));
26        self
27    }
28
29    pub fn where_expr(mut self, filter: FilterExpr) -> Self {
30        self.filters.push(filter);
31        self
32    }
33
34    pub fn where_any(mut self, filters: impl IntoIterator<Item = FilterExpr>) -> Self {
35        self.filters.push(FilterExpr::any(filters));
36        self
37    }
38
39    /// Conditionally append a filter; `None` is a no-op.
40    pub fn where_optional<F>(mut self, filter: Option<F>) -> Self
41    where
42        F: Into<FilterExpr>,
43    {
44        if let Some(filter) = filter {
45            self.filters.push(filter.into());
46        }
47        self
48    }
49
50    /// Approximate SQL preview. The runtime path interpolates filter
51    /// predicates and the delete policy clause; this returns the rough
52    /// shape for migration tooling and the schema studio.
53    pub fn preview_sql(&self) -> String {
54        let mut sql = match self.descriptor.soft_delete_column {
55            Some(col) => {
56                let mut s = format!("UPDATE {} SET {col} = NOW()", self.descriptor.table_name);
57                if let Some(version_col) = self.descriptor.version_column {
58                    s.push_str(&format!(", {version_col} = {version_col} + 1"));
59                }
60                s.push_str(&format!(" WHERE {col} IS NULL AND "));
61                s
62            }
63            None => format!("DELETE FROM {} WHERE ", self.descriptor.table_name),
64        };
65        sql.push_str("<filters> AND <delete_policy> RETURNING ");
66        sql.push_str(&self.descriptor.select_projection());
67        sql
68    }
69
70    pub async fn run(self, ctx: &CratestackContext) -> Result<BatchSummary, CratestackError>
71    where
72        for<'r> M: Send + Unpin + sqlx::FromRow<'r, sqlx::postgres::PgRow> + serde::Serialize,
73    {
74        // Inside an `@isolation` procedure: run on its transaction and
75        // defer the post-commit fan-out (docs/design/procedure-isolation.md
76        // §4, §6).
77        if let Some(bound) = self.runtime.bound() {
78            let emits = self
79                .descriptor
80                .emits(cratestack_core::ModelEventKind::Deleted);
81            let outcome = crate::bound::in_bound_savepoint!(bound, |sp| self.run_in_tx(sp, ctx))?;
82            return Ok(bound.settle(outcome, emits));
83        }
84        let runtime = self.runtime;
85        let descriptor = self.descriptor;
86        let mut tx = runtime
87            .pool()
88            .begin()
89            .await
90            .map_err(cratestack_error_from_sqlx)?;
91        let (summary, emits_event, audit_events) =
92            run_delete_many_in_tx(&mut tx, runtime, descriptor, &self.filters, ctx).await?;
93        tx.commit().await.map_err(cratestack_error_from_sqlx)?;
94        if emits_event {
95            let _ = runtime.drain_event_outbox().await;
96        }
97        dispatch_audit_sink(runtime, &audit_events).await;
98        Ok(summary)
99    }
100
101    /// Neither the `AuditSink` fan-out nor the event outbox drain
102    /// happens here — see `create.rs`'s `run_in_tx` doc comment for the
103    /// full contract and how a caller opts into both after their own
104    /// commit.
105    pub async fn run_in_tx<'tx>(
106        self,
107        tx: &mut sqlx::Transaction<'tx, sqlx::Postgres>,
108        ctx: &CratestackContext,
109    ) -> Result<RunInTxOutcome<BatchSummary>, CratestackError>
110    where
111        for<'r> M: Send + Unpin + sqlx::FromRow<'r, sqlx::postgres::PgRow> + serde::Serialize,
112    {
113        let (summary, _emits_event, audit_events) =
114            run_delete_many_in_tx(tx, self.runtime, self.descriptor, &self.filters, ctx).await?;
115        Ok(RunInTxOutcome::new(summary, audit_events))
116    }
117}
cratestack_sqlx/query/write/delete_many.rs

cratestack_sqlx/query/write/
delete_many.rs
