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::{CoolContext, CoolError};
7
8use crate::audit::{RunInTxOutcome, dispatch_audit_sink};
9use crate::{
10    ConflictTarget, ModelDescriptor, SqlxRuntime, UpsertModelInput, cool_error_from_sqlx, sqlx,
11};
12
13use super::upsert_do_nothing_exec::run_upsert_do_nothing_in_tx;
14use super::upsert_outcome::UpsertOutcome;
15
16#[derive(Debug, Clone)]
17pub struct UpsertRecordDoNothing<'a, M: 'static, PK: 'static, I> {
18    pub(crate) runtime: &'a SqlxRuntime,
19    pub(crate) descriptor: &'static ModelDescriptor<M, PK>,
20    pub(crate) input: I,
21    pub(crate) conflict_target: ConflictTarget,
22}
23
24impl<'a, M: 'static, PK: 'static, I> UpsertRecordDoNothing<'a, M, PK, I>
25where
26    I: UpsertModelInput<M>,
27{
28    /// Choose the conflict target. See [`super::upsert::UpsertRecord::on_conflict`];
29    /// works identically here, and can be called either before or
30    /// after `.do_nothing()`.
31    pub fn on_conflict(mut self, target: ConflictTarget) -> Self {
32        self.conflict_target = target;
33        self
34    }
35
36    /// Render an approximate SQL preview of the insert-branch
37    /// statement. The actual call wraps a `SELECT ... FOR UPDATE`
38    /// probe around it and may perform a fallback `SELECT` on a lost
39    /// race — see [`UpsertOutcome`] for the full sequencing.
40    pub fn preview_sql(&self) -> String {
41        let values = self.input.sql_values();
42        let placeholders = (1..=values.len())
43            .map(|index| format!("${index}"))
44            .collect::<Vec<_>>()
45            .join(", ");
46        let columns = values
47            .iter()
48            .map(|value| value.column)
49            .collect::<Vec<_>>()
50            .join(", ");
51        let conflict_tuple = match self.conflict_target {
52            ConflictTarget::PrimaryKey => self.descriptor.primary_key.to_owned(),
53            ConflictTarget::Columns(cols) => cols.join(", "),
54        };
55
56        format!(
57            "INSERT INTO {table} ({columns}) VALUES ({placeholders}) \
58             ON CONFLICT ({conflict_tuple}) DO NOTHING \
59             RETURNING {projection}",
60            table = self.descriptor.table_name,
61            projection = self.descriptor.select_projection(),
62        )
63    }
64
65    pub async fn run(self, ctx: &CoolContext) -> Result<UpsertOutcome<M>, CoolError>
66    where
67        for<'r> M: Send + Unpin + sqlx::FromRow<'r, sqlx::postgres::PgRow> + serde::Serialize,
68        PK: Send + sqlx::Type<sqlx::Postgres> + for<'q> sqlx::Encode<'q, sqlx::Postgres>,
69    {
70        let runtime = self.runtime;
71        let mut tx = runtime.pool().begin().await.map_err(cool_error_from_sqlx)?;
72        let (outcome, emits_event, audit_event) = run_upsert_do_nothing_in_tx(
73            &mut tx,
74            runtime,
75            self.descriptor,
76            self.input,
77            self.conflict_target,
78            ctx,
79        )
80        .await?;
81        tx.commit().await.map_err(cool_error_from_sqlx)?;
82        if emits_event {
83            let _ = runtime.drain_event_outbox().await;
84        }
85        if let Some(event) = &audit_event {
86            dispatch_audit_sink(runtime, std::slice::from_ref(event)).await;
87        }
88        Ok(outcome)
89    }
90
91    /// Like [`Self::run`] but participates in a caller-supplied
92    /// transaction. The conflict probe (and, on the insert branch, the
93    /// `ON CONFLICT DO NOTHING`) run against `tx`, so any row lock is
94    /// held until the caller commits. Neither the event outbox drain
95    /// nor the `AuditSink` fan-out happens here — see `create.rs`'s
96    /// `run_in_tx` doc comment for the full contract and how a caller
97    /// opts into both after their own commit.
98    pub async fn run_in_tx<'tx>(
99        self,
100        tx: &mut sqlx::Transaction<'tx, sqlx::Postgres>,
101        ctx: &CoolContext,
102    ) -> Result<RunInTxOutcome<UpsertOutcome<M>>, CoolError>
103    where
104        for<'r> M: Send + Unpin + sqlx::FromRow<'r, sqlx::postgres::PgRow> + serde::Serialize,
105        PK: Send + sqlx::Type<sqlx::Postgres> + for<'q> sqlx::Encode<'q, sqlx::Postgres>,
106    {
107        let (outcome, _emits_event, audit_event) = run_upsert_do_nothing_in_tx(
108            tx,
109            self.runtime,
110            self.descriptor,
111            self.input,
112            self.conflict_target,
113            ctx,
114        )
115        .await?;
116        Ok(RunInTxOutcome::new(
117            outcome,
118            audit_event.into_iter().collect(),
119        ))
120    }
121}
cratestack_sqlx/query/write/upsert_do_nothing.rs

cratestack_sqlx/query/write/
upsert_do_nothing.rs
