cratestack_sqlx/query/write/
upsert_do_nothing.rs

1//! `UpsertRecordDoNothing` — the `.upsert(..).do_nothing()` builder
2//! (cratestack#487). Produced by [`super::upsert::UpsertRecord::do_nothing`];
3//! see that method's doc comment for why this is a distinct type
4//! rather than a flag on `UpsertRecord` itself.
5
6use cratestack_core::{CratestackContext, CratestackError};
7
8use crate::audit::{RunInTxOutcome, dispatch_audit_sink};
9use crate::{
10    ConflictTarget, ModelDescriptor, SqlxRuntime, UpsertModelInput, cratestack_error_from_sqlx,
11    sqlx,
12};
13
14use super::upsert_do_nothing_exec::run_upsert_do_nothing_in_tx;
15use super::upsert_outcome::UpsertOutcome;
16
17#[derive(Debug, Clone)]
18pub struct UpsertRecordDoNothing<'a, M: 'static, PK: 'static, I> {
19    pub(crate) runtime: &'a SqlxRuntime,
20    pub(crate) descriptor: &'static ModelDescriptor<M, PK>,
21    pub(crate) input: I,
22    pub(crate) conflict_target: ConflictTarget,
23}
24
25impl<'a, M: 'static, PK: 'static, I> UpsertRecordDoNothing<'a, M, PK, I>
26where
27    I: UpsertModelInput<M>,
28{
29    /// Choose the conflict target. See [`super::upsert::UpsertRecord::on_conflict`];
30    /// works identically here, and can be called either before or
31    /// after `.do_nothing()`.
32    pub fn on_conflict(mut self, target: ConflictTarget) -> Self {
33        self.conflict_target = target;
34        self
35    }
36
37    /// Render an approximate SQL preview of the insert-branch
38    /// statement. The actual call wraps a `SELECT ... FOR UPDATE`
39    /// probe around it and may perform a fallback `SELECT` on a lost
40    /// race — see [`UpsertOutcome`] for the full sequencing.
41    ///
42    /// Deliberately does NOT call `ConflictTarget::validate` — see
43    /// `UpsertRecord::preview_sql`'s doc comment (cratestack#741
44    /// finding 3) for the full reasoning; the same applies here.
45    pub fn preview_sql(&self) -> String {
46        let values = self.input.sql_values();
47        let placeholders = (1..=values.len())
48            .map(|index| format!("${index}"))
49            .collect::<Vec<_>>()
50            .join(", ");
51        let columns = values
52            .iter()
53            .map(|value| value.column)
54            .collect::<Vec<_>>()
55            .join(", ");
56        let conflict_tuple = match self.conflict_target.as_columns() {
57            None => self.descriptor.primary_key.to_owned(),
58            Some(cols) => cols.join(", "),
59        };
60        let conflict_predicate = match self.conflict_target.predicate() {
61            Some(predicate) => format!(" WHERE {predicate}"),
62            None => String::new(),
63        };
64
65        format!(
66            "INSERT INTO {table} ({columns}) VALUES ({placeholders}) \
67             ON CONFLICT ({conflict_tuple}){conflict_predicate} DO NOTHING \
68             RETURNING {projection}",
69            table = self.descriptor.table_name,
70            projection = self.descriptor.select_projection(),
71        )
72    }
73
74    pub async fn run(self, ctx: &CratestackContext) -> Result<UpsertOutcome<M>, CratestackError>
75    where
76        for<'r> M: Send + Unpin + sqlx::FromRow<'r, sqlx::postgres::PgRow> + serde::Serialize,
77        PK: Send + sqlx::Type<sqlx::Postgres> + for<'q> sqlx::Encode<'q, sqlx::Postgres>,
78    {
79        // Inside an `@isolation` procedure: run on its transaction and
80        // defer the post-commit fan-out (docs/design/procedure-isolation.md
81        // §4, §6).
82        if let Some(bound) = self.runtime.bound() {
83            let emits = self
84                .descriptor
85                .emits(cratestack_core::ModelEventKind::Created);
86            let outcome = crate::bound::in_bound_savepoint!(bound, |sp| self.run_in_tx(sp, ctx))?;
87            return Ok(bound.settle(outcome, emits));
88        }
89        let runtime = self.runtime;
90        let mut tx = runtime
91            .pool()
92            .begin()
93            .await
94            .map_err(cratestack_error_from_sqlx)?;
95        let (outcome, emits_event, audit_event) = run_upsert_do_nothing_in_tx(
96            &mut tx,
97            runtime,
98            self.descriptor,
99            self.input,
100            self.conflict_target,
101            ctx,
102        )
103        .await?;
104        tx.commit().await.map_err(cratestack_error_from_sqlx)?;
105        if emits_event {
106            let _ = runtime.drain_event_outbox().await;
107        }
108        if let Some(event) = &audit_event {
109            dispatch_audit_sink(runtime, std::slice::from_ref(event)).await;
110        }
111        Ok(outcome)
112    }
113
114    /// Like [`Self::run`] but participates in a caller-supplied
115    /// transaction. The conflict probe (and, on the insert branch, the
116    /// `ON CONFLICT DO NOTHING`) run against `tx`, so any row lock is
117    /// held until the caller commits. Neither the event outbox drain
118    /// nor the `AuditSink` fan-out happens here — see `create.rs`'s
119    /// `run_in_tx` doc comment for the full contract and how a caller
120    /// opts into both after their own commit.
121    pub async fn run_in_tx<'tx>(
122        self,
123        tx: &mut sqlx::Transaction<'tx, sqlx::Postgres>,
124        ctx: &CratestackContext,
125    ) -> Result<RunInTxOutcome<UpsertOutcome<M>>, CratestackError>
126    where
127        for<'r> M: Send + Unpin + sqlx::FromRow<'r, sqlx::postgres::PgRow> + serde::Serialize,
128        PK: Send + sqlx::Type<sqlx::Postgres> + for<'q> sqlx::Encode<'q, sqlx::Postgres>,
129    {
130        let (outcome, _emits_event, audit_event) = run_upsert_do_nothing_in_tx(
131            tx,
132            self.runtime,
133            self.descriptor,
134            self.input,
135            self.conflict_target,
136            ctx,
137        )
138        .await?;
139        Ok(RunInTxOutcome::new(
140            outcome,
141            audit_event.into_iter().collect(),
142        ))
143    }
144}
cratestack_sqlx/query/write/upsert_do_nothing.rs

cratestack_sqlx/query/write/
upsert_do_nothing.rs
