cratestack_sqlx/query/write/upsert.rs
1//! `INSERT … ON CONFLICT (<pk>) DO UPDATE …`, but with the
2//! create/update distinction *resolved* rather than merely guessed, so
3//! we can:
4//!
5//! * pick the right policy slot (both must allow at call time)
6//! * emit the correct ModelEventKind (Created vs Updated)
7//! * capture an audit `before` snapshot only on the update branch
8//!
9//! A `SELECT … FOR UPDATE` probe inside the same transaction predicts
10//! the branch — binding when it finds a row (it holds the lock), a
11//! guess when it does not — so the statement itself gets the last word.
12//! See `upsert_resolve`'s module doc for how a mispredicted insert is
13//! recovered into a proper update (cratestack#745).
14//!
15//! The upsert is always transactional regardless of whether the model
16//! emits events or has `@@audit`. One extra round-trip for the
17//! SELECT, in exchange for clean event/audit semantics. Upsert is not
18//! a hot read path — callers who need raw insert/update throughput
19//! should use `.create()` / `.update()` directly.
20
21use cratestack_core::{CratestackContext, CratestackError};
22
23use crate::audit::{RunInTxOutcome, dispatch_audit_sink};
24use crate::{
25 ConflictTarget, ModelDescriptor, SqlxRuntime, UpsertModelInput, cratestack_error_from_sqlx,
26 sqlx,
27};
28
29use super::upsert_do_nothing::UpsertRecordDoNothing;
30use super::upsert_exec::run_upsert_in_tx;
31
32#[derive(Debug, Clone)]
33pub struct UpsertRecord<'a, M: 'static, PK: 'static, I> {
34 pub(crate) runtime: &'a SqlxRuntime,
35 pub(crate) descriptor: &'static ModelDescriptor<M, PK>,
36 pub(crate) input: I,
37 pub(crate) conflict_target: ConflictTarget,
38}
39
40impl<'a, M: 'static, PK: 'static, I> UpsertRecord<'a, M, PK, I>
41where
42 I: UpsertModelInput<M>,
43{
44 /// Choose the conflict target. Defaults to the model's primary
45 /// key; pass [`ConflictTarget::columns`] to upsert on a composite
46 /// unique key instead. The named columns must form a `UNIQUE`
47 /// constraint/index on the target table.
48 pub fn on_conflict(mut self, target: ConflictTarget) -> Self {
49 self.conflict_target = target;
50 self
51 }
52
53 /// Switch this call to `ON CONFLICT ... DO NOTHING` semantics
54 /// (cratestack#487), independent of `descriptor.upsert_update_columns`:
55 /// on conflict, leave the existing row completely untouched instead
56 /// of merging `upsert_update_columns` into it. This is the
57 /// idempotent-insert shape ledger-style writes need — e.g. a
58 /// cash-in claim that inserts a `PENDING` row and treats a
59 /// conflict as "already in flight" must never let a retry's blank
60 /// values overwrite an existing `COMPLETED` row's `transfer_ref`.
61 ///
62 /// Returns a distinct builder type rather than a flag on
63 /// `UpsertRecord` because the return shape genuinely changes: a
64 /// real `DO NOTHING` returns nothing on conflict, so the caller
65 /// needs `Inserted` vs `Existing` distinguishable in the type
66 /// ([`crate::UpsertOutcome`]) rather than collapsed into a plain
67 /// `M` the way `.run()` returns it today. Encoding that as a
68 /// separate type also means existing `.upsert(..).run(..)` callers
69 /// keep their `Result<M, CratestackError>` signature unchanged — this is
70 /// purely additive, not a behavior change for the DO UPDATE path.
71 pub fn do_nothing(self) -> UpsertRecordDoNothing<'a, M, PK, I> {
72 UpsertRecordDoNothing {
73 runtime: self.runtime,
74 descriptor: self.descriptor,
75 input: self.input,
76 conflict_target: self.conflict_target,
77 }
78 }
79
80 /// Render an approximate SQL preview. The actual upsert wraps a
81 /// `SELECT … FOR UPDATE` around the `INSERT … ON CONFLICT`, but
82 /// this preview returns only the conflict-bearing statement.
83 ///
84 /// Deliberately does NOT call [`ConflictTarget::validate`]
85 /// (cratestack#741 finding 3) — every other `preview_sql()` in
86 /// this codebase (`find_many`, `create`, `update`, `delete`, …)
87 /// returns a bare `String`, no `Result`, so it can never fail;
88 /// matching that established shape here means `cratestack-studio`'s
89 /// interactive SQL-preview tooling can keep calling it uniformly
90 /// across every builder without a special case for upsert. The
91 /// combination this would reject (a predicate on
92 /// [`ConflictTarget::PrimaryKey`]) is still caught before any SQL
93 /// *runs* — [`Self::run`]/[`Self::run_in_tx`] call `.validate()` via
94 /// `prepare_upsert_insert` first — so nothing unsafe executes;
95 /// only the preview string itself can show a `WHERE` clause paired
96 /// with a target `.validate()` would reject. A caller that wants to
97 /// know ahead of rendering can call `.validate()` on the same
98 /// `ConflictTarget` value passed to `.on_conflict(..)`.
99 pub fn preview_sql(&self) -> String {
100 let values = self.input.sql_values();
101 let placeholders = (1..=values.len())
102 .map(|index| format!("${index}"))
103 .collect::<Vec<_>>()
104 .join(", ");
105 let columns = values
106 .iter()
107 .map(|value| value.column)
108 .collect::<Vec<_>>()
109 .join(", ");
110 let update_assignments = self
111 .descriptor
112 .upsert_update_columns
113 .iter()
114 .map(|column| format!("{column} = EXCLUDED.{column}"))
115 .collect::<Vec<_>>()
116 .join(", ");
117 let version_bump = match self.descriptor.version_column {
118 Some(col) => format!(
119 ", {col} = {table}.{col} + 1",
120 table = self.descriptor.table_name,
121 col = col
122 ),
123 None => String::new(),
124 };
125 let conflict_tuple = match self.conflict_target.as_columns() {
126 None => self.descriptor.primary_key.to_owned(),
127 Some(cols) => cols.join(", "),
128 };
129 let conflict_predicate = match self.conflict_target.predicate() {
130 Some(predicate) => format!(" WHERE {predicate}"),
131 None => String::new(),
132 };
133
134 format!(
135 "INSERT INTO {table} ({columns}) VALUES ({placeholders}) \
136 ON CONFLICT ({conflict_tuple}){conflict_predicate} DO UPDATE SET {update_assignments}{version_bump} \
137 RETURNING {projection}",
138 table = self.descriptor.table_name,
139 projection = self.descriptor.select_projection(),
140 )
141 }
142
143 pub async fn run(self, ctx: &CratestackContext) -> Result<M, CratestackError>
144 where
145 for<'r> M: Send + Unpin + sqlx::FromRow<'r, sqlx::postgres::PgRow> + serde::Serialize,
146 PK: Send + sqlx::Type<sqlx::Postgres> + for<'q> sqlx::Encode<'q, sqlx::Postgres>,
147 {
148 // Inside an `@isolation` procedure: run on its transaction and
149 // defer the post-commit fan-out (docs/design/procedure-isolation.md
150 // §4, §6).
151 if let Some(bound) = self.runtime.bound() {
152 let emits = self
153 .descriptor
154 .emits(cratestack_core::ModelEventKind::Created)
155 || self
156 .descriptor
157 .emits(cratestack_core::ModelEventKind::Updated);
158 let outcome = crate::bound::in_bound_savepoint!(bound, |sp| self.run_in_tx(sp, ctx))?;
159 return Ok(bound.settle(outcome, emits));
160 }
161 let runtime = self.runtime;
162 let mut tx = runtime
163 .pool()
164 .begin()
165 .await
166 .map_err(cratestack_error_from_sqlx)?;
167 let (record, emits_event, audit_event) = run_upsert_in_tx(
168 &mut tx,
169 runtime,
170 self.descriptor,
171 self.input,
172 self.conflict_target,
173 ctx,
174 )
175 .await?;
176 tx.commit().await.map_err(cratestack_error_from_sqlx)?;
177 if emits_event {
178 let _ = runtime.drain_event_outbox().await;
179 }
180 if let Some(event) = &audit_event {
181 dispatch_audit_sink(runtime, std::slice::from_ref(event)).await;
182 }
183 Ok(record)
184 }
185
186 /// Like [`Self::run`] but participates in a caller-supplied
187 /// transaction. The conflict probe runs against `tx`, so the row
188 /// lock is held until the caller commits. Neither the event outbox
189 /// drain nor the `AuditSink` fan-out happens here — see
190 /// `create.rs`'s `run_in_tx` doc comment for the full contract and
191 /// how a caller opts into both after their own commit.
192 pub async fn run_in_tx<'tx>(
193 self,
194 tx: &mut sqlx::Transaction<'tx, sqlx::Postgres>,
195 ctx: &CratestackContext,
196 ) -> Result<RunInTxOutcome<M>, CratestackError>
197 where
198 for<'r> M: Send + Unpin + sqlx::FromRow<'r, sqlx::postgres::PgRow> + serde::Serialize,
199 PK: Send + sqlx::Type<sqlx::Postgres> + for<'q> sqlx::Encode<'q, sqlx::Postgres>,
200 {
201 let (record, _emits_event, audit_event) = run_upsert_in_tx(
202 tx,
203 self.runtime,
204 self.descriptor,
205 self.input,
206 self.conflict_target,
207 ctx,
208 )
209 .await?;
210 Ok(RunInTxOutcome::new(
211 record,
212 audit_event.into_iter().collect(),
213 ))
214 }
215}