cratestack_sqlx/query/write/
delete.rs

1//! `DeleteRecord` — single-row DELETE (soft or hard) with policy +
2//! audit + event fan-out. For a hard delete, the `RETURNING` row IS
3//! the pre-delete state, so it doubles as the audit "before" snapshot
4//! and "after" stays `None` (the row no longer exists). For a soft
5//! delete, `delete_returning_record` actually runs an `UPDATE ...
6//! RETURNING`, so that row is the *post*-tombstone state — it's
7//! captured as "after", and "before" comes from a separate row-locked
8//! fetch taken ahead of the mutation, mirroring how `update.rs` splits
9//! its own before/after snapshots.
10
11use cratestack_core::{AuditOperation, CoolContext, CoolError, ModelEventKind};
12
13use crate::audit::{
14    RunInTxOutcome, build_audit_event, dispatch_audit_sink, enqueue_audit_event,
15    ensure_audit_table, fetch_for_audit,
16};
17use crate::descriptor::{enqueue_event_outbox, ensure_event_outbox_table};
18use crate::{ModelDescriptor, SqlxRuntime, cool_error_from_sqlx, sqlx};
19
20use super::delete_exec::delete_returning_record;
21
22#[derive(Debug, Clone)]
23pub struct DeleteRecord<'a, M: 'static, PK: 'static> {
24    pub(crate) runtime: &'a SqlxRuntime,
25    pub(crate) descriptor: &'static ModelDescriptor<M, PK>,
26    pub(crate) id: PK,
27    pub(crate) if_match: Option<i64>,
28}
29
30impl<'a, M: 'static, PK: 'static> DeleteRecord<'a, M, PK> {
31    /// Expected version for optimistic locking. Required on models
32    /// that declare `@version`; ignored otherwise. Mirrors
33    /// [`crate::UpdateRecordSet::if_match`] — see that doc comment for
34    /// the rationale of an `Option<i64>` builder step rather than a
35    /// required constructor argument.
36    pub fn if_match(mut self, expected: i64) -> Self {
37        self.if_match = Some(expected);
38        self
39    }
40
41    pub fn preview_sql(&self) -> String {
42        match self.descriptor.version_column {
43            Some(version_col) => format!(
44                "DELETE FROM {} WHERE {} = $1 AND {} = $2 RETURNING {}",
45                self.descriptor.table_name,
46                self.descriptor.primary_key,
47                version_col,
48                self.descriptor.select_projection(),
49            ),
50            None => format!(
51                "DELETE FROM {} WHERE {} = $1 RETURNING {}",
52                self.descriptor.table_name,
53                self.descriptor.primary_key,
54                self.descriptor.select_projection(),
55            ),
56        }
57    }
58
59    /// Like [`Self::run`] but participates in a caller-supplied
60    /// transaction. Neither the `AuditSink` fan-out nor the event
61    /// outbox drain happens here — see `create.rs`'s `run_in_tx` doc
62    /// comment for the full contract and how a caller opts into both
63    /// after their own commit.
64    pub async fn run_in_tx<'tx>(
65        self,
66        tx: &mut sqlx::Transaction<'tx, sqlx::Postgres>,
67        ctx: &CoolContext,
68    ) -> Result<RunInTxOutcome<M>, CoolError>
69    where
70        for<'r> M: Send + Unpin + sqlx::FromRow<'r, sqlx::postgres::PgRow> + serde::Serialize,
71        PK: Send + Clone + sqlx::Type<sqlx::Postgres> + for<'q> sqlx::Encode<'q, sqlx::Postgres>,
72    {
73        if self.descriptor.version_column.is_some() && self.if_match.is_none() {
74            return Err(CoolError::PreconditionFailed(
75                "If-Match header required for versioned model".to_owned(),
76            ));
77        }
78        let emits_event = self.descriptor.emits(ModelEventKind::Deleted);
79        let audit_enabled = self.descriptor.audit_enabled;
80        let soft_delete = self.descriptor.soft_delete_column.is_some();
81        if emits_event {
82            ensure_event_outbox_table(&mut **tx).await?;
83        }
84        if audit_enabled {
85            ensure_audit_table(self.runtime).await?;
86        }
87        // Soft delete is an UPDATE under the hood, so its RETURNING
88        // row is the post-tombstone state — the pre-delete "before"
89        // snapshot has to come from a separate row-locked read taken
90        // ahead of the mutation.
91        let before_record = if audit_enabled && soft_delete {
92            fetch_for_audit(&mut **tx, self.descriptor, self.id.clone()).await?
93        } else {
94            None
95        };
96        let before_snapshot = before_record
97            .as_ref()
98            .and_then(|m| serde_json::to_value(m).ok());
99        let record = delete_returning_record(
100            &mut **tx,
101            self.runtime.pool(),
102            self.descriptor,
103            self.id,
104            ctx,
105            self.if_match,
106        )
107        .await?;
108        if emits_event {
109            enqueue_event_outbox(
110                &mut **tx,
111                self.descriptor.schema_name,
112                ModelEventKind::Deleted,
113                &record,
114            )
115            .await?;
116        }
117        let mut audit_event = None;
118        if audit_enabled {
119            let (before, after) = if soft_delete {
120                (before_snapshot, serde_json::to_value(&record).ok())
121            } else {
122                (serde_json::to_value(&record).ok(), None)
123            };
124            let event =
125                build_audit_event(self.descriptor, AuditOperation::Delete, before, after, ctx);
126            enqueue_audit_event(&mut **tx, &event).await?;
127            audit_event = Some(event);
128        }
129        Ok(RunInTxOutcome::new(
130            record,
131            audit_event.into_iter().collect(),
132        ))
133    }
134
135    pub async fn run(self, ctx: &CoolContext) -> Result<M, CoolError>
136    where
137        for<'r> M: Send + Unpin + sqlx::FromRow<'r, sqlx::postgres::PgRow> + serde::Serialize,
138        PK: Send + Clone + sqlx::Type<sqlx::Postgres> + for<'q> sqlx::Encode<'q, sqlx::Postgres>,
139    {
140        if self.descriptor.version_column.is_some() && self.if_match.is_none() {
141            return Err(CoolError::PreconditionFailed(
142                "If-Match header required for versioned model".to_owned(),
143            ));
144        }
145        let emits_event = self.descriptor.emits(ModelEventKind::Deleted);
146        let audit_enabled = self.descriptor.audit_enabled;
147        let soft_delete = self.descriptor.soft_delete_column.is_some();
148        let needs_tx = emits_event || audit_enabled;
149        let mut audit_event = None;
150        let record = if needs_tx {
151            let mut tx = self
152                .runtime
153                .pool()
154                .begin()
155                .await
156                .map_err(cool_error_from_sqlx)?;
157            if emits_event {
158                ensure_event_outbox_table(&mut *tx).await?;
159            }
160            if audit_enabled {
161                ensure_audit_table(self.runtime).await?;
162            }
163
164            let before_record = if audit_enabled && soft_delete {
165                fetch_for_audit(&mut *tx, self.descriptor, self.id.clone()).await?
166            } else {
167                None
168            };
169            let before_snapshot = before_record
170                .as_ref()
171                .and_then(|m| serde_json::to_value(m).ok());
172            let record = delete_returning_record(
173                &mut *tx,
174                self.runtime.pool(),
175                self.descriptor,
176                self.id,
177                ctx,
178                self.if_match,
179            )
180            .await?;
181            if emits_event {
182                enqueue_event_outbox(
183                    &mut *tx,
184                    self.descriptor.schema_name,
185                    ModelEventKind::Deleted,
186                    &record,
187                )
188                .await?;
189            }
190            if audit_enabled {
191                let (before, after) = if soft_delete {
192                    (before_snapshot, serde_json::to_value(&record).ok())
193                } else {
194                    (serde_json::to_value(&record).ok(), None)
195                };
196                let event =
197                    build_audit_event(self.descriptor, AuditOperation::Delete, before, after, ctx);
198                enqueue_audit_event(&mut *tx, &event).await?;
199                audit_event = Some(event);
200            }
201            tx.commit().await.map_err(cool_error_from_sqlx)?;
202            record
203        } else {
204            delete_returning_record(
205                self.runtime.pool(),
206                self.runtime.pool(),
207                self.descriptor,
208                self.id,
209                ctx,
210                self.if_match,
211            )
212            .await?
213        };
214
215        if emits_event {
216            let _ = self.runtime.drain_event_outbox().await;
217        }
218        if let Some(event) = &audit_event {
219            dispatch_audit_sink(self.runtime, std::slice::from_ref(event)).await;
220        }
221
222        Ok(record)
223    }
224}
cratestack_sqlx/query/write/delete.rs

cratestack_sqlx/query/write/
delete.rs
