cratestack_sqlx/query/batch/
delete.rs

1//! `batch_delete` — single `DELETE ... RETURNING` (or `UPDATE` for
2//! soft-delete) with the delete policy in the WHERE. Per-item audit
3//! and outbox events fan out from the RETURNING rows.
4
5use std::collections::HashMap;
6use std::hash::Hash;
7
8use cratestack_core::{
9    AuditOperation, BatchResponse, CratestackContext, CratestackError, ModelEventKind,
10};
11
12use crate::audit::{
13    build_audit_event, dispatch_audit_sink, enqueue_audit_event, ensure_audit_table,
14};
15use crate::descriptor::{enqueue_event_outbox, ensure_event_outbox_table};
16use crate::query::support::push_action_policy_query;
17use crate::{ModelDescriptor, ModelPrimaryKey, SqlxRuntime, cratestack_error_from_sqlx, sqlx};
18
19use super::validate::{reject_duplicate_pks, validate_batch_size};
20
21#[derive(Debug, Clone)]
22pub struct BatchDelete<'a, M: 'static, PK: 'static> {
23    pub(crate) runtime: &'a SqlxRuntime,
24    pub(crate) descriptor: &'static ModelDescriptor<M, PK>,
25    pub(crate) ids: Vec<PK>,
26}
27
28impl<'a, M: 'static, PK: 'static> BatchDelete<'a, M, PK> {
29    pub async fn run(self, ctx: &CratestackContext) -> Result<BatchResponse<M>, CratestackError>
30    where
31        for<'r> M: Send
32            + Unpin
33            + sqlx::FromRow<'r, sqlx::postgres::PgRow>
34            + ModelPrimaryKey<PK>
35            + serde::Serialize,
36        PK: Clone
37            + Eq
38            + Hash
39            + Send
40            + sqlx::Type<sqlx::Postgres>
41            + for<'q> sqlx::Encode<'q, sqlx::Postgres>,
42    {
43        validate_batch_size(self.ids.len())?;
44        reject_duplicate_pks(&self.ids)?;
45        if self.ids.is_empty() {
46            return Ok(BatchResponse::from_results(vec![]));
47        }
48
49        let emits_event = self.descriptor.emits(ModelEventKind::Deleted);
50        let audit_enabled = self.descriptor.audit_enabled;
51
52        let mut tx = self
53            .runtime
54            .pool()
55            .begin()
56            .await
57            .map_err(cratestack_error_from_sqlx)?;
58        if emits_event {
59            ensure_event_outbox_table(&mut *tx).await?;
60        }
61        if audit_enabled {
62            ensure_audit_table(self.runtime).await?;
63        }
64
65        let mut query = sqlx::QueryBuilder::<sqlx::Postgres>::new("");
66        match self.descriptor.soft_delete_column {
67            Some(col) => {
68                query.push("UPDATE ").push(self.descriptor.table_name);
69                query.push(" SET ").push(col).push(" = NOW()");
70                if let Some(version_col) = self.descriptor.version_column {
71                    query
72                        .push(", ")
73                        .push(version_col)
74                        .push(" = ")
75                        .push(version_col)
76                        .push(" + 1");
77                }
78                query.push(" WHERE ").push(col).push(" IS NULL AND ");
79            }
80            None => {
81                query.push("DELETE FROM ").push(self.descriptor.table_name);
82                query.push(" WHERE ");
83            }
84        }
85        query.push(self.descriptor.primary_key).push(" IN (");
86        for (index, id) in self.ids.iter().enumerate() {
87            if index > 0 {
88                query.push(", ");
89            }
90            query.push_bind(id.clone());
91        }
92        query.push(") AND ");
93        push_action_policy_query(
94            &mut query,
95            self.descriptor.delete_allow_policies,
96            self.descriptor.delete_deny_policies,
97            ctx,
98        );
99        query
100            .push(" RETURNING ")
101            .push(self.descriptor.select_projection());
102
103        let deleted: Vec<M> = query
104            .build_query_as::<M>()
105            .fetch_all(&mut *tx)
106            .await
107            .map_err(cratestack_error_from_sqlx)?;
108
109        // The RETURNING row IS the "before" snapshot — DELETE/soft-
110        // delete returns the pre-mutation state.
111        let mut audit_events = Vec::new();
112        for record in &deleted {
113            if emits_event {
114                enqueue_event_outbox(
115                    &mut *tx,
116                    self.descriptor.schema_name,
117                    ModelEventKind::Deleted,
118                    record,
119                )
120                .await?;
121            }
122            if audit_enabled {
123                let before = serde_json::to_value(record).ok();
124                let event =
125                    build_audit_event(self.descriptor, AuditOperation::Delete, before, None, ctx);
126                enqueue_audit_event(&mut *tx, &event).await?;
127                audit_events.push(event);
128            }
129        }
130
131        tx.commit().await.map_err(cratestack_error_from_sqlx)?;
132
133        if emits_event {
134            let _ = self.runtime.drain_event_outbox().await;
135        }
136        dispatch_audit_sink(self.runtime, &audit_events).await;
137
138        // Walk-and-match: any input id whose row isn't in `deleted`
139        // failed the WHERE (tombstoned, policy denied, never existed).
140        // All three collapse to NotFound on the wire.
141        let mut by_pk: HashMap<PK, M> = deleted.into_iter().map(|m| (m.primary_key(), m)).collect();
142        let per_item: Vec<Result<M, CratestackError>> = self
143            .ids
144            .into_iter()
145            .map(|id| {
146                by_pk
147                    .remove(&id)
148                    .ok_or_else(|| CratestackError::NotFound("no row matched".to_owned()))
149            })
150            .collect();
151
152        Ok(BatchResponse::from_results(per_item))
153    }
154}
cratestack_sqlx/query/batch/delete.rs

cratestack_sqlx/query/batch/
delete.rs
