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

cratestack_sqlx/query/write/
upsert_do_nothing.rs
