cratestack_sqlx/query/write/
upsert_do_nothing.rs1use 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 pub fn on_conflict(mut self, target: ConflictTarget) -> Self {
33 self.conflict_target = target;
34 self
35 }
36
37 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 let runtime = self.runtime;
80 let mut tx = runtime
81 .pool()
82 .begin()
83 .await
84 .map_err(cratestack_error_from_sqlx)?;
85 let (outcome, emits_event, audit_event) = run_upsert_do_nothing_in_tx(
86 &mut tx,
87 runtime,
88 self.descriptor,
89 self.input,
90 self.conflict_target,
91 ctx,
92 )
93 .await?;
94 tx.commit().await.map_err(cratestack_error_from_sqlx)?;
95 if emits_event {
96 let _ = runtime.drain_event_outbox().await;
97 }
98 if let Some(event) = &audit_event {
99 dispatch_audit_sink(runtime, std::slice::from_ref(event)).await;
100 }
101 Ok(outcome)
102 }
103
104 pub async fn run_in_tx<'tx>(
112 self,
113 tx: &mut sqlx::Transaction<'tx, sqlx::Postgres>,
114 ctx: &CratestackContext,
115 ) -> Result<RunInTxOutcome<UpsertOutcome<M>>, CratestackError>
116 where
117 for<'r> M: Send + Unpin + sqlx::FromRow<'r, sqlx::postgres::PgRow> + serde::Serialize,
118 PK: Send + sqlx::Type<sqlx::Postgres> + for<'q> sqlx::Encode<'q, sqlx::Postgres>,
119 {
120 let (outcome, _emits_event, audit_event) = run_upsert_do_nothing_in_tx(
121 tx,
122 self.runtime,
123 self.descriptor,
124 self.input,
125 self.conflict_target,
126 ctx,
127 )
128 .await?;
129 Ok(RunInTxOutcome::new(
130 outcome,
131 audit_event.into_iter().collect(),
132 ))
133 }
134}