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

cratestack_sqlx/query/batch/
delete.rs
