cratestack-sqlx 0.15.3

//! `batch_delete` — single `DELETE ... RETURNING` (or `UPDATE` for
//! soft-delete) with the delete policy in the WHERE. Per-item audit
//! and outbox events fan out from the RETURNING rows.

use std::collections::HashMap;
use std::hash::Hash;

use cratestack_core::{
    AuditOperation, BatchResponse, CratestackContext, CratestackError, ModelEventKind,
};

use crate::audit::{
    build_audit_event, dispatch_audit_sink, enqueue_audit_event, ensure_audit_table,
};
use crate::descriptor::{enqueue_event_outbox, ensure_event_outbox_table};
use crate::query::support::push_action_policy_query;
use crate::{ModelDescriptor, ModelPrimaryKey, SqlxRuntime, cratestack_error_from_sqlx, sqlx};

use super::validate::{reject_duplicate_pks, validate_batch_size};

#[derive(Debug, Clone)]
pub struct BatchDelete<'a, M: 'static, PK: 'static> {
    pub(crate) runtime: &'a SqlxRuntime,
    pub(crate) descriptor: &'static ModelDescriptor<M, PK>,
    pub(crate) ids: Vec<PK>,
}

impl<'a, M: 'static, PK: 'static> BatchDelete<'a, M, PK> {
    pub async fn run(self, ctx: &CratestackContext) -> Result<BatchResponse<M>, CratestackError>
    where
        for<'r> M: Send
            + Unpin
            + sqlx::FromRow<'r, sqlx::postgres::PgRow>
            + ModelPrimaryKey<PK>
            + serde::Serialize,
        PK: Clone
            + Eq
            + Hash
            + Send
            + sqlx::Type<sqlx::Postgres>
            + for<'q> sqlx::Encode<'q, sqlx::Postgres>,
    {
        validate_batch_size(self.ids.len())?;
        reject_duplicate_pks(&self.ids)?;
        if self.ids.is_empty() {
            return Ok(BatchResponse::from_results(vec![]));
        }

        let emits_event = self.descriptor.emits(ModelEventKind::Deleted);
        let audit_enabled = self.descriptor.audit_enabled;

        // A savepoint of the `@isolation` transaction when there is one
        // (docs/design/procedure-isolation.md §4).
        let (deleted, audit_events) = crate::bound::in_write_tx!(self.runtime, |tx| async {
            if emits_event {
                ensure_event_outbox_table(&mut **tx).await?;
            }
            if audit_enabled {
                ensure_audit_table(self.runtime, &mut **tx).await?;
            }

            let mut query = sqlx::QueryBuilder::<sqlx::Postgres>::new("");
            match self.descriptor.soft_delete_column {
                Some(col) => {
                    query.push("UPDATE ").push(self.descriptor.table_name);
                    query.push(" SET ").push(col).push(" = NOW()");
                    if let Some(version_col) = self.descriptor.version_column {
                        query
                            .push(", ")
                            .push(version_col)
                            .push(" = ")
                            .push(version_col)
                            .push(" + 1");
                    }
                    query.push(" WHERE ").push(col).push(" IS NULL AND ");
                }
                None => {
                    query.push("DELETE FROM ").push(self.descriptor.table_name);
                    query.push(" WHERE ");
                }
            }
            query.push(self.descriptor.primary_key).push(" IN (");
            for (index, id) in self.ids.iter().enumerate() {
                if index > 0 {
                    query.push(", ");
                }
                query.push_bind(id.clone());
            }
            query.push(") AND ");
            push_action_policy_query(
                &mut query,
                self.descriptor.delete_allow_policies,
                self.descriptor.delete_deny_policies,
                ctx,
            );
            query
                .push(" RETURNING ")
                .push(self.descriptor.select_projection());

            let deleted: Vec<M> = query
                .build_query_as::<M>()
                .fetch_all(&mut **tx)
                .await
                .map_err(cratestack_error_from_sqlx)?;

            // The RETURNING row IS the "before" snapshot — DELETE/soft-
            // delete returns the pre-mutation state.
            let mut audit_events = Vec::new();
            for record in &deleted {
                if emits_event {
                    enqueue_event_outbox(
                        &mut **tx,
                        self.descriptor.schema_name,
                        ModelEventKind::Deleted,
                        record,
                    )
                    .await?;
                }
                if audit_enabled {
                    let before = serde_json::to_value(record).ok();
                    let event = build_audit_event(
                        self.descriptor,
                        AuditOperation::Delete,
                        before,
                        None,
                        ctx,
                    );
                    enqueue_audit_event(&mut **tx, &event).await?;
                    audit_events.push(event);
                }
            }

            Ok((deleted, audit_events))
        })?;

        if emits_event {
            let _ = self.runtime.drain_event_outbox().await;
        }
        dispatch_audit_sink(self.runtime, &audit_events).await;

        // Walk-and-match: any input id whose row isn't in `deleted`
        // failed the WHERE (tombstoned, policy denied, never existed).
        // All three collapse to NotFound on the wire.
        let mut by_pk: HashMap<PK, M> = deleted.into_iter().map(|m| (m.primary_key(), m)).collect();
        let per_item: Vec<Result<M, CratestackError>> = self
            .ids
            .into_iter()
            .map(|id| {
                by_pk
                    .remove(&id)
                    .ok_or_else(|| CratestackError::NotFound("no row matched".to_owned()))
            })
            .collect();

        Ok(BatchResponse::from_results(per_item))
    }
}