prax_query/operations/
upsert.rs

1//! Upsert operation for creating or updating records.
2
3use std::marker::PhantomData;
4
5use crate::error::QueryResult;
6use crate::filter::{Filter, FilterValue};
7use crate::inputs::WriteOp;
8use crate::nested::NestedWriteOp;
9use crate::traits::{Model, ModelWithPk, QueryEngine};
10use crate::types::Select;
11
12/// An upsert (insert or update) operation.
13///
14/// # Example
15///
16/// ```rust,ignore
17/// let user = client
18///     .user()
19///     .upsert()
20///     .r#where(user::email::equals("test@example.com"))
21///     .create(user::Create { email: "test@example.com".into(), name: Some("Test".into()) })
22///     .update(user::Update { name: Some("Updated".into()), ..Default::default() })
23///     .exec()
24///     .await?;
25/// ```
26pub struct UpsertOperation<E: QueryEngine, M: Model> {
27    engine: E,
28    filter: Filter,
29    create_columns: Vec<String>,
30    create_values: Vec<FilterValue>,
31    update_columns: Vec<String>,
32    update_values: Vec<FilterValue>,
33    /// Update-path entries pushed via [`Self::with_update_input`] or
34    /// [`Self::update_set_op`]. When non-empty these take precedence
35    /// over `update_columns`/`update_values` because they carry atomic
36    /// operators (`Increment`/`Decrement`/`Unset`) the flat
37    /// column/value pair can't express.
38    update_ops: Vec<(String, WriteOp)>,
39    conflict_columns: Vec<String>,
40    select: Select,
41    /// Nested-write ops to run when the *create* branch fires (the
42    /// row didn't previously exist). Empty on the fast path.
43    create_nested: Vec<NestedWriteOp>,
44    /// Nested-write ops to run when the *update* branch fires (the
45    /// row already existed). Empty on the fast path.
46    update_nested: Vec<NestedWriteOp>,
47    _model: PhantomData<M>,
48}
49
50impl<E: QueryEngine, M: Model + crate::row::FromRow> UpsertOperation<E, M> {
51    /// Create a new Upsert operation.
52    pub fn new(engine: E) -> Self {
53        Self {
54            engine,
55            filter: Filter::None,
56            create_columns: Vec::new(),
57            create_values: Vec::new(),
58            update_columns: Vec::new(),
59            update_values: Vec::new(),
60            update_ops: Vec::new(),
61            conflict_columns: Vec::new(),
62            select: Select::All,
63            create_nested: Vec::new(),
64            update_nested: Vec::new(),
65            _model: PhantomData,
66        }
67    }
68
69    /// Add a filter condition (identifies the record to upsert).
70    ///
71    /// On the single-statement fast path the filter doubles as the
72    /// conflict target: when [`Self::on_conflict`] was not called, a
73    /// simple equality filter (`col = value`, or an AND of equalities
74    /// for composite keys) supplies the `ON CONFLICT (col)` columns.
75    /// Explicit [`Self::on_conflict`] columns take precedence when both
76    /// are set. A filter of any other shape cannot name a conflict
77    /// target and is rejected by [`Self::exec`] with an
78    /// `invalid_input` error unless [`Self::on_conflict`] is used. On
79    /// the nested-write slow path the filter is instead used directly
80    /// as the update branch's WHERE clause.
81    pub fn r#where(mut self, filter: impl Into<Filter>) -> Self {
82        self.filter = filter.into();
83        self
84    }
85
86    /// Set the columns to check for conflict.
87    ///
88    /// Takes precedence over the conflict target derived from the
89    /// [`Self::r#where`] filter when both are set.
90    pub fn on_conflict(mut self, columns: impl IntoIterator<Item = impl Into<String>>) -> Self {
91        self.conflict_columns = columns.into_iter().map(Into::into).collect();
92        self
93    }
94
95    /// Set the create data.
96    pub fn create(
97        mut self,
98        values: impl IntoIterator<Item = (impl Into<String>, impl Into<FilterValue>)>,
99    ) -> Self {
100        for (col, val) in values {
101            self.create_columns.push(col.into());
102            self.create_values.push(val.into());
103        }
104        self
105    }
106
107    /// Set a single create column.
108    pub fn create_set(mut self, column: impl Into<String>, value: impl Into<FilterValue>) -> Self {
109        self.create_columns.push(column.into());
110        self.create_values.push(value.into());
111        self
112    }
113
114    /// Set the update data.
115    pub fn update(
116        mut self,
117        values: impl IntoIterator<Item = (impl Into<String>, impl Into<FilterValue>)>,
118    ) -> Self {
119        for (col, val) in values {
120            self.update_columns.push(col.into());
121            self.update_values.push(val.into());
122        }
123        self
124    }
125
126    /// Set a single update column.
127    pub fn update_set(mut self, column: impl Into<String>, value: impl Into<FilterValue>) -> Self {
128        self.update_columns.push(column.into());
129        self.update_values.push(value.into());
130        self
131    }
132
133    /// Select specific fields to return.
134    pub fn select(mut self, select: impl Into<Select>) -> Self {
135        self.select = select.into();
136        self
137    }
138
139    /// Apply a typed `WhereUniqueInput`. Overwrites the existing filter.
140    pub fn with_where_input<W: crate::inputs::WhereUniqueInput<Model = M>>(mut self, w: W) -> Self {
141        self.filter = w.into_ir();
142        self
143    }
144
145    /// Apply a typed `SelectInput`.
146    pub fn with_select_input<S: crate::inputs::SelectInput<Model = M>>(mut self, s: S) -> Self {
147        self.select = s.into_ir();
148        self
149    }
150
151    /// Apply a typed `CreateInput` to the upsert's create path.
152    ///
153    /// The columns / values produced by the input are appended to the
154    /// existing `create_columns` / `create_values` lists. Phase 5a's
155    /// codegen ensures every `<Model>CreateInput` carries the
156    /// model's `@unique` conflict column, so [`Self::on_conflict`]
157    /// remains useful with this method.
158    pub fn with_create_input<I>(mut self, input: I) -> Self
159    where
160        I: crate::inputs::CreateInput<Model = M, Data = crate::inputs::CreatePayload>,
161    {
162        let data: crate::inputs::CreatePayload = input.into_ir();
163        for (col, val) in data {
164            self.create_columns.push(col);
165            self.create_values.push(val);
166        }
167        self
168    }
169
170    /// Apply a typed `UpdateInput` to the upsert's update path.
171    ///
172    /// Atomic operators are preserved — when the update branch fires,
173    /// `Increment(n)` emits `col = col + $n` in the `DO UPDATE SET`
174    /// clause, etc. Setting any input via this method overrides any
175    /// flat update columns recorded by [`Self::update`] or
176    /// [`Self::update_set`].
177    pub fn with_update_input<I>(mut self, input: I) -> Self
178    where
179        I: crate::inputs::UpdateInput<Model = M, Data = crate::inputs::UpdatePayload>,
180    {
181        let data: crate::inputs::UpdatePayload = input.into_ir();
182        for (col, op) in data {
183            self.update_ops.push((col, op));
184        }
185        self
186    }
187
188    /// Build the SQL query.
189    ///
190    /// Conflict-target precedence: [`Self::on_conflict`] columns win;
191    /// otherwise a `where` filter that is a simple equality (or an AND
192    /// of equalities, for composite keys) supplies the conflict
193    /// column(s). Because this method returns SQL unconditionally, the
194    /// combinations that would emit invalid SQL (a non-derivable
195    /// `where` filter with no `.on_conflict(...)`, or an update branch
196    /// with no conflict target at all) are rejected by [`Self::exec`]
197    /// instead — direct callers must uphold the same contract.
198    pub fn build_sql(
199        &self,
200        dialect: &dyn crate::dialect::SqlDialect,
201    ) -> (String, Vec<FilterValue>) {
202        let mut sql = String::new();
203        let mut params = Vec::new();
204        let mut param_idx = 1;
205
206        // Conflict-target precedence: explicit `.on_conflict(...)`
207        // columns win; otherwise a simple equality `where` filter
208        // supplies the target column(s) so the filter isn't silently
209        // dropped on the single-statement path. `exec` rejects the
210        // remaining invalid combinations (a non-derivable filter, or
211        // an update branch with no target at all). Resolved before the
212        // INSERT keyword so the empty-target do-nothing form can pick
213        // the dialect's INSERT prefix (see below).
214        let derived_cols;
215        let conflict_cols: Vec<&str> = if !self.conflict_columns.is_empty() {
216            self.conflict_columns.iter().map(|s| s.as_str()).collect()
217        } else {
218            derived_cols = conflict_cols_from_filter(&self.filter).unwrap_or_default();
219            derived_cols.iter().map(|s| s.as_str()).collect()
220        };
221
222        // Classify the dialect's empty-conflict-target do-nothing form
223        // before writing the INSERT keyword — the same classification
224        // create.rs uses for skip_duplicates: MySQL has no trailing
225        // DO NOTHING clause and expresses the semantics as an
226        // `INSERT IGNORE` prefix; MSSQL/CQL have no single-statement
227        // equivalent at all; Postgres/SQLite take the target-less
228        // `ON CONFLICT DO NOTHING` suffix.
229        let targetless_do_nothing = dialect.upsert_do_nothing_clause(&[]);
230        let do_nothing_no_target = self.update_ops.is_empty()
231            && self.update_columns.is_empty()
232            && conflict_cols.is_empty();
233        let insert_ignore =
234            do_nothing_no_target && targetless_do_nothing.starts_with(" ON DUPLICATE KEY");
235
236        // INSERT INTO clause
237        if insert_ignore {
238            sql.push_str("INSERT IGNORE INTO ");
239        } else {
240            sql.push_str("INSERT INTO ");
241        }
242        sql.push_str(M::TABLE_NAME);
243
244        // Columns
245        sql.push_str(" (");
246        sql.push_str(&self.create_columns.join(", "));
247        sql.push(')');
248
249        // VALUES
250        sql.push_str(" VALUES (");
251        let placeholders: Vec<_> = self
252            .create_values
253            .iter()
254            .map(|v| {
255                params.push(v.clone());
256                let p = dialect.placeholder(param_idx);
257                param_idx += 1;
258                p
259            })
260            .collect();
261        sql.push_str(&placeholders.join(", "));
262        sql.push(')');
263
264        // Upsert clause (ON CONFLICT / ON DUPLICATE KEY)
265        // Build update SET clause. When `update_ops` is non-empty it
266        // supplants the legacy flat column/value list — typed inputs
267        // can carry atomic operators (`col = col + $n`) the flat
268        // pair can't represent.
269        let update_set = if !self.update_ops.is_empty() {
270            let update_parts: Vec<String> = self
271                .update_ops
272                .iter()
273                .map(|(col, op)| {
274                    let placeholder = dialect.placeholder(param_idx);
275                    let (fragment, value) = op.to_set_fragment(col, &placeholder);
276                    if let Some(v) = value {
277                        params.push(v);
278                        param_idx += 1;
279                    }
280                    fragment
281                })
282                .collect();
283            update_parts.join(", ")
284        } else if !self.update_columns.is_empty() {
285            let update_parts: Vec<_> = self
286                .update_columns
287                .iter()
288                .zip(self.update_values.iter())
289                .map(|(col, val)| {
290                    params.push(val.clone());
291                    let part = format!("{} = {}", col, dialect.placeholder(param_idx));
292                    param_idx += 1;
293                    part
294                })
295                .collect();
296            update_parts.join(", ")
297        } else {
298            String::new()
299        };
300
301        if update_set.is_empty() {
302            // DO NOTHING variant — routed through the dialect so each
303            // backend emits its own syntax (MySQL renders a no-op
304            // `ON DUPLICATE KEY UPDATE col = col` self-assign; MSSQL/CQL
305            // render no clause). The dialect hook needs at least one
306            // target column to produce a well-formed clause, so the
307            // target-less form is spelled per dialect: an `INSERT
308            // IGNORE` prefix on MySQL (written above), no clause at all
309            // on MSSQL/CQL, and the bare `ON CONFLICT DO NOTHING` on
310            // Postgres/SQLite — the parenthesized
311            // `ON CONFLICT () DO NOTHING` would be invalid, same as
312            // createMany's skip_duplicates path in create.rs.
313            if conflict_cols.is_empty() {
314                if insert_ignore {
315                    // MySQL: the INSERT IGNORE prefix above carries the
316                    // do-nothing semantics; no trailing clause exists.
317                } else if targetless_do_nothing.is_empty() {
318                    // MSSQL/CQL: no single-statement equivalent — emit a
319                    // plain INSERT (create.rs makes the same fallback).
320                    tracing::warn!(
321                        table = M::TABLE_NAME,
322                        "upsert do-nothing has no single-statement equivalent on this \
323                         dialect without a conflict target; emitting a plain INSERT"
324                    );
325                } else {
326                    sql.push_str(" ON CONFLICT DO NOTHING");
327                }
328            } else {
329                sql.push_str(&dialect.upsert_do_nothing_clause(&conflict_cols));
330            }
331        } else {
332            // Use dialect's upsert_clause for DO UPDATE SET
333            sql.push_str(&dialect.upsert_clause(&conflict_cols, &update_set));
334        }
335
336        // RETURNING clause
337        sql.push_str(&dialect.returning_clause(&self.select.to_sql()));
338
339        (sql, params)
340    }
341
342    /// Enforce the single-statement upsert contract before SQL goes out.
343    ///
344    /// `build_sql` returns SQL unconditionally (its signature predates
345    /// fallible builders in this crate), so the two combinations that
346    /// would emit invalid or filter-dropping SQL are rejected here:
347    ///
348    /// - a `where` filter without `.on_conflict(...)` whose conflict
349    ///   column(s) can't be derived (the filter isn't a simple equality
350    ///   or AND of equalities) — the filter would be silently ignored;
351    /// - an update branch with no conflict target at all — Postgres
352    ///   rejects `ON CONFLICT () DO UPDATE`.
353    fn validate_fast_path(&self) -> QueryResult<()> {
354        if self.conflict_columns.is_empty() {
355            if !self.filter.is_none() && conflict_cols_from_filter(&self.filter).is_none() {
356                return Err(crate::error::QueryError::invalid_input(
357                    "where",
358                    "upsert `where` filter must be an equality on the conflict column(s) \
359                     to serve as the ON CONFLICT target on the single-statement path",
360                )
361                .with_help(
362                    "use `.on_conflict([...])` to name the conflict column(s) explicitly, \
363                     or simplify the filter to `col = value` (an AND of equalities for \
364                     composite keys)",
365                ));
366            }
367            let has_update = !self.update_ops.is_empty() || !self.update_columns.is_empty();
368            if has_update && self.filter.is_none() {
369                return Err(crate::error::QueryError::invalid_input(
370                    "on_conflict",
371                    "upsert with an update branch requires `.on_conflict(...)` or a \
372                     `where` filter to name the conflict target",
373                )
374                .with_help(
375                    "add `.on_conflict([\"<unique-column>\"])` (or a `where` equality on \
376                     the unique column) — Postgres rejects `ON CONFLICT () DO UPDATE`",
377                ));
378            }
379        }
380        Ok(())
381    }
382
383    /// Queue a nested write to fire when the *create* branch runs
384    /// (i.e. no existing row matched).
385    pub fn with_create_nested(mut self, nw: NestedWriteOp) -> Self
386    where
387        E: crate::capabilities::SupportsNestedWrites,
388    {
389        self.create_nested.push(nw);
390        self
391    }
392
393    /// Queue a nested write to fire when the *update* branch runs
394    /// (i.e. an existing row was found and updated).
395    pub fn with_update_nested(mut self, nw: NestedWriteOp) -> Self
396    where
397        E: crate::capabilities::SupportsNestedWrites,
398    {
399        self.update_nested.push(nw);
400        self
401    }
402
403    /// Execute the upsert and return the record.
404    ///
405    /// Fast path (no nested writes queued): runs a single
406    /// vendor-specific upsert (`INSERT ... ON CONFLICT DO UPDATE` on
407    /// Postgres, the dialect's equivalent elsewhere). Enforces the
408    /// conflict-target contract first: a `where` filter that can't
409    /// supply the target (with no `.on_conflict(...)`), or an update
410    /// branch with no target at all, fails with an `invalid_input`
411    /// error before any SQL is sent.
412    ///
413    /// Slow path (nested writes queued via `with_create_nested` /
414    /// `with_update_nested`): runs a two-statement
415    /// engine-agnostic upsert inside a transaction so we can tell which
416    /// branch fired:
417    /// 1. `UPDATE` the row by primary key. If `affected > 0`, the
418    ///    update branch ran — fire `update_nested` with the PK we
419    ///    already have from `where:`.
420    /// 2. Otherwise `INSERT` the row, take the PK from the inserted
421    ///    model, and fire `create_nested`.
422    pub async fn exec(self) -> QueryResult<M>
423    where
424        M: Send + 'static + ModelWithPk,
425    {
426        // Fast path: single-statement vendor-specific upsert.
427        if self.create_nested.is_empty() && self.update_nested.is_empty() {
428            self.validate_fast_path()?;
429            let dialect = self.engine.dialect();
430            let (sql, params) = self.build_sql(dialect);
431            return self.engine.execute_insert::<M>(&sql, params).await;
432        }
433
434        // Nested writes are queued — the existing where-unique must
435        // equal-match the primary key column. This is the same
436        // restriction as `update!`'s nested-write path.
437        let parent_pk =
438            crate::operations::update::extract_pk_from_filter(&self.filter, M::PRIMARY_KEY[0])
439                .ok_or_else(|| {
440                    crate::error::QueryError::invalid_input(
441                        "where",
442                        "nested writes inside `upsert!` require the `where:` clause to equal-match \
443                 the primary-key column",
444                    )
445                    .with_help(format!(
446                        "expected `where: {{ {pk}: <value> }}` on `{table}` — non-PK unique \
447                 columns are not yet supported for nested writes inside upsert!. \
448                 Lift this restriction by running the nested ops in a separate operation \
449                 after looking up the row's PK.",
450                        pk = M::PRIMARY_KEY[0],
451                        table = M::TABLE_NAME,
452                    ))
453                })?;
454
455        let UpsertOperation {
456            engine,
457            filter,
458            create_columns,
459            create_values,
460            update_columns,
461            update_values,
462            update_ops,
463            conflict_columns: _,
464            select,
465            create_nested,
466            update_nested,
467            _model,
468        } = self;
469
470        engine
471            .transaction(move |tx| async move {
472                let dialect = tx.dialect();
473
474                // Phase 1: try UPDATE first.
475                let (update_sql, update_params) = build_update_sql::<M>(
476                    &filter,
477                    &update_columns,
478                    &update_values,
479                    &update_ops,
480                    dialect,
481                );
482                let affected = tx.execute_raw(&update_sql, update_params).await?;
483
484                let (row, fired_nested): (M, Vec<NestedWriteOp>) = if affected > 0 {
485                    // Update branch — fetch the row back via SELECT so
486                    // the caller sees the freshly-updated columns. We
487                    // know the PK from the where filter.
488                    let (sel_sql, sel_params) =
489                        build_select_by_pk_sql::<M>(parent_pk.clone(), &select, dialect);
490                    let fetched: M = tx.query_one::<M>(&sel_sql, sel_params).await?;
491                    (fetched, update_nested)
492                } else {
493                    // Create branch — INSERT and capture the returned row.
494                    let (ins_sql, ins_params) =
495                        build_insert_sql::<M>(&create_columns, &create_values, &select, dialect);
496                    let inserted: M = tx.execute_insert::<M>(&ins_sql, ins_params).await?;
497                    (inserted, create_nested)
498                };
499
500                // Dispatch the chosen nested-op vec, sharing the same
501                // partition-by-target-Connect batching as create.rs.
502                let parent_pk_for_nested = if affected > 0 {
503                    parent_pk
504                } else {
505                    row.pk_value()
506                };
507                run_nested_ops(&tx, dialect, fired_nested, &parent_pk_for_nested).await?;
508
509                Ok(row)
510            })
511            .await
512    }
513}
514
515/// Derive `ON CONFLICT` target columns from a where-unique filter.
516///
517/// Handles the shapes codegen produces for `upsert!`: a single
518/// equality (`col = value`) or an AND of equalities (composite unique
519/// keys). Any other filter shape (OR, ranges, `Contains`, ...) has no
520/// single conflict target, so this returns `None` and the caller
521/// either falls back to explicit `.on_conflict(...)` columns or errors.
522fn conflict_cols_from_filter(filter: &Filter) -> Option<Vec<String>> {
523    match filter {
524        Filter::Equals(name, _) => Some(vec![name.to_string()]),
525        Filter::And(parts) => {
526            let mut cols = Vec::with_capacity(parts.len());
527            for part in parts.iter() {
528                match part {
529                    Filter::Equals(name, _) => cols.push(name.to_string()),
530                    _ => return None,
531                }
532            }
533            if cols.is_empty() { None } else { Some(cols) }
534        }
535        _ => None,
536    }
537}
538
539/// Build a two-statement-style UPDATE for the upsert's "update branch".
540///
541/// Uses `update_ops` when populated (carries atomic operators), else
542/// falls back to the legacy flat `update_columns`/`update_values` pair.
543/// Always emits a `WHERE` clause from `filter` (the where-unique).
544fn build_update_sql<M: Model>(
545    filter: &Filter,
546    update_columns: &[String],
547    update_values: &[FilterValue],
548    update_ops: &[(String, WriteOp)],
549    dialect: &dyn crate::dialect::SqlDialect,
550) -> (String, Vec<FilterValue>) {
551    let mut sql = String::new();
552    let mut params = Vec::new();
553    let mut param_idx = 1;
554
555    sql.push_str("UPDATE ");
556    sql.push_str(M::TABLE_NAME);
557    sql.push_str(" SET ");
558
559    let set_parts: Vec<String> = if !update_ops.is_empty() {
560        update_ops
561            .iter()
562            .map(|(col, op)| {
563                let placeholder = dialect.placeholder(param_idx);
564                let (fragment, value) = op.to_set_fragment(col, &placeholder);
565                if let Some(v) = value {
566                    params.push(v);
567                    param_idx += 1;
568                }
569                fragment
570            })
571            .collect()
572    } else {
573        update_columns
574            .iter()
575            .zip(update_values.iter())
576            .map(|(col, val)| {
577                params.push(val.clone());
578                let part = format!("{} = {}", col, dialect.placeholder(param_idx));
579                param_idx += 1;
580                part
581            })
582            .collect()
583    };
584    sql.push_str(&set_parts.join(", "));
585
586    if !filter.is_none() {
587        let (where_sql, where_params) = filter.to_sql(param_idx - 1, dialect);
588        sql.push_str(" WHERE ");
589        sql.push_str(&where_sql);
590        params.extend(where_params);
591    }
592
593    (sql, params)
594}
595
596/// Build the create-branch INSERT used by the two-statement upsert.
597fn build_insert_sql<M: Model>(
598    columns: &[String],
599    values: &[FilterValue],
600    select: &Select,
601    dialect: &dyn crate::dialect::SqlDialect,
602) -> (String, Vec<FilterValue>) {
603    let mut sql = String::new();
604    sql.push_str("INSERT INTO ");
605    sql.push_str(M::TABLE_NAME);
606    sql.push_str(" (");
607    sql.push_str(&columns.join(", "));
608    sql.push(')');
609    sql.push_str(" VALUES (");
610    let placeholders: Vec<_> = (1..=values.len()).map(|i| dialect.placeholder(i)).collect();
611    sql.push_str(&placeholders.join(", "));
612    sql.push(')');
613    sql.push_str(&dialect.returning_clause(&select.to_sql()));
614    (sql, values.to_vec())
615}
616
617/// Build the SELECT-by-pk used to re-fetch the row after the update
618/// branch ran.
619fn build_select_by_pk_sql<M: Model>(
620    pk: FilterValue,
621    select: &Select,
622    dialect: &dyn crate::dialect::SqlDialect,
623) -> (String, Vec<FilterValue>) {
624    let cols = select.to_sql();
625    let sql = format!(
626        "SELECT {} FROM {} WHERE {} = {}",
627        if cols.is_empty() || cols == "*" {
628            "*".to_string()
629        } else {
630            cols
631        },
632        M::TABLE_NAME,
633        dialect.quote_ident(M::PRIMARY_KEY[0]),
634        dialect.placeholder(1),
635    );
636    (sql, vec![pk])
637}
638
639/// Iterate `nested` against `tx`, batching consecutive Connect ops with
640/// the same target — mirrors the create.rs partition loop.
641async fn run_nested_ops<E: QueryEngine>(
642    tx: &E,
643    dialect: &dyn crate::dialect::SqlDialect,
644    nested: Vec<NestedWriteOp>,
645    parent_pk: &FilterValue,
646) -> QueryResult<()> {
647    let mut idx = 0;
648    while idx < nested.len() {
649        if let NestedWriteOp::Connect {
650            target_table: run_table,
651            foreign_key: run_fk,
652            target_pk: run_target_pk,
653            ..
654        } = &nested[idx]
655        {
656            let run_table = *run_table;
657            let run_fk = *run_fk;
658            let run_target_pk = *run_target_pk;
659            let mut end = idx + 1;
660            while end < nested.len() {
661                match &nested[end] {
662                    NestedWriteOp::Connect {
663                        target_table,
664                        foreign_key,
665                        target_pk,
666                        ..
667                    } if *target_table == run_table
668                        && *foreign_key == run_fk
669                        && *target_pk == run_target_pk =>
670                    {
671                        end += 1;
672                    }
673                    _ => break,
674                }
675            }
676
677            if end - idx == 1 {
678                let op = nested[idx].clone();
679                op.execute(tx, parent_pk).await?;
680            } else {
681                let expected = (end - idx) as u64;
682                let mut pks: Vec<FilterValue> = Vec::with_capacity(end - idx + 1);
683                pks.push(parent_pk.clone());
684                for op in &nested[idx..end] {
685                    if let NestedWriteOp::Connect { pk, .. } = op {
686                        pks.push(pk.clone());
687                    }
688                }
689                let placeholders: Vec<String> =
690                    (2..=pks.len()).map(|i| dialect.placeholder(i)).collect();
691                let sql = format!(
692                    "UPDATE {} SET {} = {} WHERE {} IN ({})",
693                    dialect.quote_ident(run_table),
694                    dialect.quote_ident(run_fk),
695                    dialect.placeholder(1),
696                    dialect.quote_ident(run_target_pk),
697                    placeholders.join(", "),
698                );
699                let affected = tx.execute_raw(&sql, pks).await?;
700                if affected != expected {
701                    return Err(crate::error::QueryError::not_found(run_table)
702                        .with_context("Nested Connect batch")
703                        .with_help(format!(
704                            "Expected {} matching rows but UPDATE affected {}",
705                            expected, affected
706                        )));
707                }
708            }
709            idx = end;
710        } else {
711            let op = nested[idx].clone();
712            op.execute(tx, parent_pk).await?;
713            idx += 1;
714        }
715    }
716    Ok(())
717}
718
719#[cfg(test)]
720mod tests {
721    use super::*;
722    use crate::error::QueryError;
723
724    #[derive(Debug)]
725    struct TestModel;
726
727    impl Model for TestModel {
728        const MODEL_NAME: &'static str = "TestModel";
729        const TABLE_NAME: &'static str = "test_models";
730        const PRIMARY_KEY: &'static [&'static str] = &["id"];
731        const COLUMNS: &'static [&'static str] = &["id", "name", "email"];
732    }
733
734    impl crate::row::FromRow for TestModel {
735        fn from_row(_row: &impl crate::row::RowRef) -> Result<Self, crate::row::RowError> {
736            Ok(TestModel)
737        }
738    }
739
740    // Phase-5c slow-path nested-write wiring requires `ModelWithPk` on
741    // the return type. The constant PK is fine because the legacy
742    // single-statement tests never exercise the slow path.
743    impl crate::traits::ModelWithPk for TestModel {
744        fn pk_value(&self) -> FilterValue {
745            FilterValue::Int(0)
746        }
747        fn get_column_value(&self, _column: &str) -> Option<FilterValue> {
748            None
749        }
750    }
751
752    #[derive(Clone)]
753    struct MockEngine;
754
755    impl QueryEngine for MockEngine {
756        fn dialect(&self) -> &dyn crate::dialect::SqlDialect {
757            &crate::dialect::Postgres
758        }
759
760        fn query_many<T: Model + crate::row::FromRow + Send + 'static>(
761            &self,
762            _sql: &str,
763            _params: Vec<FilterValue>,
764        ) -> crate::traits::BoxFuture<'_, QueryResult<Vec<T>>> {
765            Box::pin(async { Ok(Vec::new()) })
766        }
767
768        fn query_one<T: Model + crate::row::FromRow + Send + 'static>(
769            &self,
770            _sql: &str,
771            _params: Vec<FilterValue>,
772        ) -> crate::traits::BoxFuture<'_, QueryResult<T>> {
773            Box::pin(async { Err(QueryError::not_found("test")) })
774        }
775
776        fn query_optional<T: Model + crate::row::FromRow + Send + 'static>(
777            &self,
778            _sql: &str,
779            _params: Vec<FilterValue>,
780        ) -> crate::traits::BoxFuture<'_, QueryResult<Option<T>>> {
781            Box::pin(async { Ok(None) })
782        }
783
784        fn execute_insert<T: Model + crate::row::FromRow + Send + 'static>(
785            &self,
786            _sql: &str,
787            _params: Vec<FilterValue>,
788        ) -> crate::traits::BoxFuture<'_, QueryResult<T>> {
789            Box::pin(async { Err(QueryError::not_found("test")) })
790        }
791
792        fn execute_update<T: Model + crate::row::FromRow + Send + 'static>(
793            &self,
794            _sql: &str,
795            _params: Vec<FilterValue>,
796        ) -> crate::traits::BoxFuture<'_, QueryResult<Vec<T>>> {
797            Box::pin(async { Ok(Vec::new()) })
798        }
799
800        fn execute_delete(
801            &self,
802            _sql: &str,
803            _params: Vec<FilterValue>,
804        ) -> crate::traits::BoxFuture<'_, QueryResult<u64>> {
805            Box::pin(async { Ok(0) })
806        }
807
808        fn execute_raw(
809            &self,
810            _sql: &str,
811            _params: Vec<FilterValue>,
812        ) -> crate::traits::BoxFuture<'_, QueryResult<u64>> {
813            Box::pin(async { Ok(0) })
814        }
815
816        fn count(
817            &self,
818            _sql: &str,
819            _params: Vec<FilterValue>,
820        ) -> crate::traits::BoxFuture<'_, QueryResult<u64>> {
821            Box::pin(async { Ok(0) })
822        }
823    }
824
825    // ========== Construction Tests ==========
826
827    #[test]
828    fn test_upsert_new() {
829        let op = UpsertOperation::<MockEngine, TestModel>::new(MockEngine);
830        let (sql, params) = op.build_sql(&crate::dialect::Postgres);
831
832        assert!(sql.contains("INSERT INTO test_models"));
833        assert!(sql.contains("ON CONFLICT"));
834        assert!(sql.contains("RETURNING *"));
835        assert!(params.is_empty());
836    }
837
838    #[test]
839    fn test_upsert_basic() {
840        let op = UpsertOperation::<MockEngine, TestModel>::new(MockEngine)
841            .on_conflict(["email"])
842            .create_set("email", "test@example.com")
843            .create_set("name", "Test")
844            .update_set("name", "Updated");
845
846        let (sql, params) = op.build_sql(&crate::dialect::Postgres);
847
848        assert!(sql.contains("INSERT INTO test_models"));
849        assert!(sql.contains("ON CONFLICT (\"email\")"));
850        assert!(sql.contains("DO UPDATE SET"));
851        assert!(sql.contains("RETURNING *"));
852        assert_eq!(params.len(), 3); // 2 create + 1 update
853    }
854
855    // ========== Conflict Column Tests ==========
856
857    #[test]
858    fn test_upsert_single_conflict_column() {
859        let op = UpsertOperation::<MockEngine, TestModel>::new(MockEngine)
860            .on_conflict(["id"])
861            .create_set("id", FilterValue::Int(1));
862
863        let (sql, _) = op.build_sql(&crate::dialect::Postgres);
864
865        assert!(sql.contains("ON CONFLICT (\"id\")"));
866    }
867
868    #[test]
869    fn test_upsert_multiple_conflict_columns() {
870        let op = UpsertOperation::<MockEngine, TestModel>::new(MockEngine)
871            .on_conflict(["tenant_id", "email"])
872            .create_set("email", "test@example.com")
873            .create_set("tenant_id", FilterValue::Int(1));
874
875        let (sql, _) = op.build_sql(&crate::dialect::Postgres);
876
877        assert!(sql.contains("ON CONFLICT (\"tenant_id\", \"email\")"));
878    }
879
880    #[test]
881    fn test_upsert_without_conflict_columns() {
882        let op = UpsertOperation::<MockEngine, TestModel>::new(MockEngine)
883            .create_set("email", "test@example.com");
884
885        let (sql, _) = op.build_sql(&crate::dialect::Postgres);
886
887        assert!(sql.contains("ON CONFLICT"));
888        assert!(!sql.contains("ON CONFLICT ("));
889    }
890
891    // ========== Create Tests ==========
892
893    #[test]
894    fn test_upsert_create_with_set() {
895        let op = UpsertOperation::<MockEngine, TestModel>::new(MockEngine)
896            .on_conflict(["email"])
897            .create_set("email", "test@example.com")
898            .create_set("name", "Test User");
899
900        let (sql, params) = op.build_sql(&crate::dialect::Postgres);
901
902        assert!(sql.contains("(email, name)"));
903        assert!(sql.contains("VALUES ($1, $2)"));
904        assert_eq!(params.len(), 2);
905    }
906
907    #[test]
908    fn test_upsert_create_with_iterator() {
909        let create_data = vec![
910            ("email", FilterValue::String("test@example.com".to_string())),
911            ("name", FilterValue::String("Test User".to_string())),
912            ("age", FilterValue::Int(25)),
913        ];
914        let op = UpsertOperation::<MockEngine, TestModel>::new(MockEngine)
915            .on_conflict(["email"])
916            .create(create_data);
917
918        let (sql, params) = op.build_sql(&crate::dialect::Postgres);
919
920        assert!(sql.contains("(email, name, age)"));
921        assert!(sql.contains("VALUES ($1, $2, $3)"));
922        assert_eq!(params.len(), 3);
923    }
924
925    // ========== Update Tests ==========
926
927    #[test]
928    fn test_upsert_update_with_set() {
929        let op = UpsertOperation::<MockEngine, TestModel>::new(MockEngine)
930            .on_conflict(["email"])
931            .create_set("email", "test@example.com")
932            .update_set("name", "Updated Name")
933            .update_set("updated_at", "2024-01-01");
934
935        let (sql, params) = op.build_sql(&crate::dialect::Postgres);
936
937        assert!(sql.contains("DO UPDATE SET"));
938        assert!(sql.contains("name = $"));
939        assert!(sql.contains("updated_at = $"));
940        assert_eq!(params.len(), 3); // 1 create + 2 update
941    }
942
943    #[test]
944    fn test_upsert_update_with_iterator() {
945        let update_data = vec![
946            ("name", FilterValue::String("Updated".to_string())),
947            ("status", FilterValue::String("active".to_string())),
948        ];
949        let op = UpsertOperation::<MockEngine, TestModel>::new(MockEngine)
950            .on_conflict(["id"])
951            .create_set("id", FilterValue::Int(1))
952            .update(update_data);
953
954        let (sql, params) = op.build_sql(&crate::dialect::Postgres);
955
956        assert!(sql.contains("DO UPDATE SET"));
957        assert_eq!(params.len(), 3); // 1 create + 2 update
958    }
959
960    // ========== Do Nothing Tests ==========
961
962    #[test]
963    fn test_upsert_do_nothing() {
964        let op = UpsertOperation::<MockEngine, TestModel>::new(MockEngine)
965            .on_conflict(["email"])
966            .create_set("email", "test@example.com");
967
968        let (sql, _) = op.build_sql(&crate::dialect::Postgres);
969
970        assert!(sql.contains("DO NOTHING"));
971        assert!(!sql.contains("DO UPDATE"));
972    }
973
974    #[test]
975    fn test_upsert_do_nothing_multiple_create() {
976        let op = UpsertOperation::<MockEngine, TestModel>::new(MockEngine)
977            .on_conflict(["email"])
978            .create_set("email", "test@example.com")
979            .create_set("name", "Test");
980
981        let (sql, params) = op.build_sql(&crate::dialect::Postgres);
982
983        assert!(sql.contains("DO NOTHING"));
984        assert_eq!(params.len(), 2);
985    }
986
987    #[test]
988    fn test_upsert_do_nothing_mysql() {
989        let op = UpsertOperation::<MockEngine, TestModel>::new(MockEngine)
990            .on_conflict(["email"])
991            .create_set("email", "test@example.com");
992
993        let (sql, _) = op.build_sql(&crate::dialect::Mysql);
994
995        // MySQL has no ON CONFLICT — the dialect renders a no-op
996        // self-assign on the first conflict column instead.
997        assert!(
998            sql.contains("ON DUPLICATE KEY UPDATE `email` = `email`"),
999            "got: {sql}"
1000        );
1001        assert!(!sql.contains("ON CONFLICT"), "got: {sql}");
1002    }
1003
1004    #[test]
1005    fn test_upsert_do_nothing_mysql_no_conflict_target() {
1006        // MySQL has no ON CONFLICT DO NOTHING; an empty conflict target
1007        // must come out as an INSERT IGNORE prefix (the canonical MySQL
1008        // form — mirrors create.rs's skip_duplicates), never the Postgres
1009        // target-less spelling.
1010        let op = UpsertOperation::<MockEngine, TestModel>::new(MockEngine)
1011            .create_set("email", "test@example.com");
1012
1013        let (sql, _) = op.build_sql(&crate::dialect::Mysql);
1014
1015        assert!(
1016            sql.starts_with("INSERT IGNORE INTO test_models"),
1017            "expected INSERT IGNORE prefix, got: {sql}"
1018        );
1019        assert!(!sql.contains("ON CONFLICT"), "got: {sql}");
1020        assert!(
1021            !sql.contains("ON DUPLICATE KEY"),
1022            "INSERT IGNORE replaces the self-assign suffix: {sql}"
1023        );
1024    }
1025
1026    // ========== Select Tests ==========
1027
1028    #[test]
1029    fn test_upsert_with_select() {
1030        let op = UpsertOperation::<MockEngine, TestModel>::new(MockEngine)
1031            .on_conflict(["email"])
1032            .create_set("email", "test@example.com")
1033            .update_set("name", "Updated")
1034            .select(Select::fields(["id", "email"]));
1035
1036        let (sql, _) = op.build_sql(&crate::dialect::Postgres);
1037
1038        assert!(sql.contains("RETURNING id, email"));
1039        assert!(!sql.contains("RETURNING *"));
1040    }
1041
1042    #[test]
1043    fn test_upsert_select_all() {
1044        let op = UpsertOperation::<MockEngine, TestModel>::new(MockEngine)
1045            .on_conflict(["email"])
1046            .create_set("email", "test@example.com")
1047            .select(Select::All);
1048
1049        let (sql, _) = op.build_sql(&crate::dialect::Postgres);
1050
1051        assert!(sql.contains("RETURNING *"));
1052    }
1053
1054    // ========== Where Filter Tests ==========
1055
1056    #[test]
1057    fn test_upsert_with_where() {
1058        let op = UpsertOperation::<MockEngine, TestModel>::new(MockEngine)
1059            .r#where(Filter::Equals(
1060                "email".into(),
1061                FilterValue::String("test@example.com".to_string()),
1062            ))
1063            .on_conflict(["email"])
1064            .create_set("email", "test@example.com");
1065
1066        let (_, _) = op.build_sql(&crate::dialect::Postgres);
1067        // where_ sets the filter but doesn't affect upsert SQL directly
1068    }
1069
1070    // ========== Conflict-Target Validation Tests ==========
1071
1072    #[test]
1073    fn test_upsert_where_derives_conflict_target() {
1074        // No explicit .on_conflict(...) — the where equality supplies
1075        // the ON CONFLICT column so the filter isn't silently dropped.
1076        let op = UpsertOperation::<MockEngine, TestModel>::new(MockEngine)
1077            .r#where(Filter::Equals(
1078                "email".into(),
1079                FilterValue::String("test@example.com".to_string()),
1080            ))
1081            .create_set("email", "test@example.com")
1082            .update_set("name", "Updated");
1083
1084        let (sql, _) = op.build_sql(&crate::dialect::Postgres);
1085
1086        assert!(sql.contains("ON CONFLICT (\"email\")"), "got: {sql}");
1087        assert!(sql.contains("DO UPDATE SET"), "got: {sql}");
1088    }
1089
1090    #[tokio::test]
1091    async fn test_upsert_update_without_conflict_target_errors() {
1092        // An update branch with neither .on_conflict(...) nor a where
1093        // filter would render `ON CONFLICT () DO UPDATE` — rejected
1094        // before any SQL goes out.
1095        let op = UpsertOperation::<MockEngine, TestModel>::new(MockEngine)
1096            .create_set("email", "test@example.com")
1097            .update_set("name", "Updated");
1098
1099        let err = op.exec().await.unwrap_err();
1100
1101        assert_eq!(err.code, crate::error::ErrorCode::InvalidParameter);
1102        let msg = format!("{err}");
1103        assert!(msg.contains("on_conflict"), "msg: {msg}");
1104    }
1105
1106    #[tokio::test]
1107    async fn test_upsert_non_derivable_where_without_conflict_errors() {
1108        // A range filter can't name a conflict target; with no
1109        // .on_conflict(...) the filter would be silently dropped.
1110        let op = UpsertOperation::<MockEngine, TestModel>::new(MockEngine)
1111            .r#where(Filter::Gt("id".into(), FilterValue::Int(3)))
1112            .create_set("email", "test@example.com");
1113
1114        let err = op.exec().await.unwrap_err();
1115
1116        assert_eq!(err.code, crate::error::ErrorCode::InvalidParameter);
1117        let msg = format!("{err}");
1118        assert!(msg.contains("where"), "msg: {msg}");
1119    }
1120
1121    // ========== SQL Structure Tests ==========
1122
1123    #[test]
1124    fn test_upsert_sql_structure() {
1125        let op = UpsertOperation::<MockEngine, TestModel>::new(MockEngine)
1126            .on_conflict(["email"])
1127            .create_set("email", "test@example.com")
1128            .update_set("name", "Updated")
1129            .select(Select::fields(["id"]));
1130
1131        let (sql, _) = op.build_sql(&crate::dialect::Postgres);
1132
1133        let insert_pos = sql.find("INSERT INTO").unwrap();
1134        let values_pos = sql.find("VALUES").unwrap();
1135        let conflict_pos = sql.find("ON CONFLICT").unwrap();
1136        let update_pos = sql.find("DO UPDATE SET").unwrap();
1137        let returning_pos = sql.find("RETURNING").unwrap();
1138
1139        assert!(insert_pos < values_pos);
1140        assert!(values_pos < conflict_pos);
1141        assert!(conflict_pos < update_pos);
1142        assert!(update_pos < returning_pos);
1143    }
1144
1145    #[test]
1146    fn test_upsert_table_name() {
1147        let op = UpsertOperation::<MockEngine, TestModel>::new(MockEngine);
1148        let (sql, _) = op.build_sql(&crate::dialect::Postgres);
1149
1150        assert!(sql.contains("test_models"));
1151    }
1152
1153    // ========== Param Ordering Tests ==========
1154
1155    #[test]
1156    fn test_upsert_param_ordering() {
1157        let op = UpsertOperation::<MockEngine, TestModel>::new(MockEngine)
1158            .on_conflict(["email"])
1159            .create_set("email", "create@test.com")
1160            .create_set("name", "Create Name")
1161            .update_set("name", "Update Name");
1162
1163        let (sql, params) = op.build_sql(&crate::dialect::Postgres);
1164
1165        // Create params first, then update params
1166        assert!(sql.contains("VALUES ($1, $2)"));
1167        assert!(sql.contains("name = $3"));
1168        assert_eq!(params.len(), 3);
1169        assert_eq!(
1170            params[0],
1171            FilterValue::String("create@test.com".to_string())
1172        );
1173        assert_eq!(params[1], FilterValue::String("Create Name".to_string()));
1174        assert_eq!(params[2], FilterValue::String("Update Name".to_string()));
1175    }
1176
1177    // ========== Async Execution Tests ==========
1178
1179    #[tokio::test]
1180    async fn test_upsert_exec() {
1181        let op = UpsertOperation::<MockEngine, TestModel>::new(MockEngine)
1182            .on_conflict(["email"])
1183            .create_set("email", "test@example.com");
1184
1185        let result = op.exec().await;
1186
1187        // MockEngine returns not_found for execute_insert
1188        assert!(result.is_err());
1189    }
1190
1191    // ========== Method Chaining Tests ==========
1192
1193    #[test]
1194    fn test_upsert_full_chain() {
1195        let op = UpsertOperation::<MockEngine, TestModel>::new(MockEngine)
1196            .r#where(Filter::Equals(
1197                "email".into(),
1198                FilterValue::String("test@example.com".to_string()),
1199            ))
1200            .on_conflict(["email"])
1201            .create_set("email", "test@example.com")
1202            .create_set("name", "Test User")
1203            .update_set("name", "Updated User")
1204            .select(Select::fields(["id", "name", "email"]));
1205
1206        let (sql, params) = op.build_sql(&crate::dialect::Postgres);
1207
1208        assert!(sql.contains("INSERT INTO test_models"));
1209        assert!(sql.contains("ON CONFLICT (\"email\")"));
1210        assert!(sql.contains("DO UPDATE SET"));
1211        assert!(sql.contains("RETURNING id, name, email"));
1212        assert_eq!(params.len(), 3);
1213    }
1214
1215    // ========== Value Type Tests ==========
1216
1217    #[test]
1218    fn test_upsert_with_null_value() {
1219        let op = UpsertOperation::<MockEngine, TestModel>::new(MockEngine)
1220            .on_conflict(["id"])
1221            .create_set("id", FilterValue::Int(1))
1222            .create_set("nickname", FilterValue::Null);
1223
1224        let (_, params) = op.build_sql(&crate::dialect::Postgres);
1225
1226        assert_eq!(params[1], FilterValue::Null);
1227    }
1228
1229    #[test]
1230    fn test_upsert_with_boolean_value() {
1231        let op = UpsertOperation::<MockEngine, TestModel>::new(MockEngine)
1232            .on_conflict(["id"])
1233            .create_set("id", FilterValue::Int(1))
1234            .create_set("active", FilterValue::Bool(true))
1235            .update_set("active", FilterValue::Bool(false));
1236
1237        let (_, params) = op.build_sql(&crate::dialect::Postgres);
1238
1239        assert_eq!(params[1], FilterValue::Bool(true));
1240        assert_eq!(params[2], FilterValue::Bool(false));
1241    }
1242
1243    #[test]
1244    fn test_upsert_with_numeric_values() {
1245        let op = UpsertOperation::<MockEngine, TestModel>::new(MockEngine)
1246            .on_conflict(["id"])
1247            .create_set("id", FilterValue::Int(1))
1248            .create_set("score", FilterValue::Float(99.5));
1249
1250        let (_, params) = op.build_sql(&crate::dialect::Postgres);
1251
1252        assert_eq!(params[0], FilterValue::Int(1));
1253        assert_eq!(params[1], FilterValue::Float(99.5));
1254    }
1255
1256    #[test]
1257    fn test_upsert_with_json_value() {
1258        let json = serde_json::json!({"key": "value"});
1259        let op = UpsertOperation::<MockEngine, TestModel>::new(MockEngine)
1260            .on_conflict(["id"])
1261            .create_set("id", FilterValue::Int(1))
1262            .create_set("metadata", FilterValue::Json(json.clone()));
1263
1264        let (_, params) = op.build_sql(&crate::dialect::Postgres);
1265
1266        assert_eq!(params[1], FilterValue::Json(json));
1267    }
1268
1269    // ========== Phase 5a: typed-input wiring ==========
1270
1271    struct MockCreateInput(Vec<(String, FilterValue)>);
1272
1273    impl crate::inputs::CreateInput for MockCreateInput {
1274        type Model = TestModel;
1275        type Data = crate::inputs::CreatePayload;
1276        fn into_ir(self) -> Self::Data {
1277            self.0
1278        }
1279    }
1280
1281    struct MockUpdateInput(Vec<(String, WriteOp)>);
1282
1283    impl crate::inputs::UpdateInput for MockUpdateInput {
1284        type Model = TestModel;
1285        type Data = crate::inputs::UpdatePayload;
1286        fn into_ir(self) -> Self::Data {
1287            self.0
1288        }
1289    }
1290
1291    #[test]
1292    fn upsert_with_create_input_appends_create_columns() {
1293        let input = MockCreateInput(vec![
1294            ("email".into(), FilterValue::String("a@x.com".into())),
1295            ("name".into(), FilterValue::String("Alice".into())),
1296        ]);
1297        let op = UpsertOperation::<MockEngine, TestModel>::new(MockEngine)
1298            .on_conflict(["email"])
1299            .with_create_input(input)
1300            .update_set("name", "Updated");
1301
1302        let (sql, params) = op.build_sql(&crate::dialect::Postgres);
1303        assert!(sql.contains("(email, name)"), "got: {sql}");
1304        assert!(sql.contains("VALUES ($1, $2)"), "got: {sql}");
1305        assert!(sql.contains("ON CONFLICT (\"email\")"));
1306        // 2 create params + 1 update param.
1307        assert_eq!(params.len(), 3);
1308    }
1309
1310    #[test]
1311    fn upsert_with_update_input_uses_atomic_ops() {
1312        let create = MockCreateInput(vec![("id".into(), FilterValue::Int(1))]);
1313        let update = MockUpdateInput(vec![
1314            (
1315                "name".into(),
1316                WriteOp::Set(FilterValue::String("Renamed".into())),
1317            ),
1318            (
1319                "login_count".into(),
1320                WriteOp::Increment(FilterValue::Int(1)),
1321            ),
1322        ]);
1323        let op = UpsertOperation::<MockEngine, TestModel>::new(MockEngine)
1324            .on_conflict(["id"])
1325            .with_create_input(create)
1326            .with_update_input(update);
1327
1328        let (sql, params) = op.build_sql(&crate::dialect::Postgres);
1329        // Atomic operator must round-trip through the upsert path.
1330        assert!(sql.contains("login_count = login_count + $"), "got: {sql}");
1331        // 1 create + 2 update params.
1332        assert_eq!(params.len(), 3);
1333    }
1334
1335    // ========== Phase 5c: nested-write wiring on upsert! ==========
1336
1337    use std::sync::{Arc, Mutex};
1338
1339    type StatementLog = Arc<Mutex<Vec<(String, Vec<FilterValue>)>>>;
1340
1341    /// Recording engine that exposes a settable `affected` sequence and
1342    /// returns a default `TestModel` from `execute_insert` / `query_one`
1343    /// (the two paths the nested-upsert exec consumes).
1344    #[derive(Clone)]
1345    struct RecordingEngine {
1346        recorded: StatementLog,
1347        affected: Arc<Mutex<Vec<u64>>>,
1348    }
1349
1350    impl RecordingEngine {
1351        fn with_affected(seq: Vec<u64>) -> Self {
1352            let mut rev = seq;
1353            rev.reverse();
1354            Self {
1355                recorded: Arc::new(Mutex::new(Vec::new())),
1356                affected: Arc::new(Mutex::new(rev)),
1357            }
1358        }
1359
1360        fn statements(&self) -> Vec<(String, Vec<FilterValue>)> {
1361            self.recorded.lock().unwrap().clone()
1362        }
1363    }
1364
1365    impl crate::capabilities::SupportsNestedWrites for RecordingEngine {}
1366
1367    impl QueryEngine for RecordingEngine {
1368        fn dialect(&self) -> &dyn crate::dialect::SqlDialect {
1369            &crate::dialect::Postgres
1370        }
1371
1372        fn query_many<T: Model + crate::row::FromRow + Send + 'static>(
1373            &self,
1374            _sql: &str,
1375            _params: Vec<FilterValue>,
1376        ) -> crate::traits::BoxFuture<'_, QueryResult<Vec<T>>> {
1377            Box::pin(async { Ok(Vec::new()) })
1378        }
1379
1380        fn query_one<T: Model + crate::row::FromRow + Send + 'static>(
1381            &self,
1382            sql: &str,
1383            params: Vec<FilterValue>,
1384        ) -> crate::traits::BoxFuture<'_, QueryResult<T>> {
1385            let recorded = self.recorded.clone();
1386            let sql = sql.to_string();
1387            Box::pin(async move {
1388                recorded.lock().unwrap().push((sql, params));
1389                T::from_row(&CannedRow).map_err(|e| QueryError::internal(e.to_string()))
1390            })
1391        }
1392
1393        fn query_optional<T: Model + crate::row::FromRow + Send + 'static>(
1394            &self,
1395            _sql: &str,
1396            _params: Vec<FilterValue>,
1397        ) -> crate::traits::BoxFuture<'_, QueryResult<Option<T>>> {
1398            Box::pin(async { Ok(None) })
1399        }
1400
1401        fn execute_insert<T: Model + crate::row::FromRow + Send + 'static>(
1402            &self,
1403            sql: &str,
1404            params: Vec<FilterValue>,
1405        ) -> crate::traits::BoxFuture<'_, QueryResult<T>> {
1406            let recorded = self.recorded.clone();
1407            let sql = sql.to_string();
1408            Box::pin(async move {
1409                recorded.lock().unwrap().push((sql, params));
1410                T::from_row(&CannedRow).map_err(|e| QueryError::internal(e.to_string()))
1411            })
1412        }
1413
1414        fn execute_update<T: Model + crate::row::FromRow + Send + 'static>(
1415            &self,
1416            _sql: &str,
1417            _params: Vec<FilterValue>,
1418        ) -> crate::traits::BoxFuture<'_, QueryResult<Vec<T>>> {
1419            Box::pin(async { Ok(Vec::new()) })
1420        }
1421
1422        fn execute_delete(
1423            &self,
1424            _sql: &str,
1425            _params: Vec<FilterValue>,
1426        ) -> crate::traits::BoxFuture<'_, QueryResult<u64>> {
1427            Box::pin(async { Ok(0) })
1428        }
1429
1430        fn execute_raw(
1431            &self,
1432            sql: &str,
1433            params: Vec<FilterValue>,
1434        ) -> crate::traits::BoxFuture<'_, QueryResult<u64>> {
1435            let recorded = self.recorded.clone();
1436            let affected = self.affected.clone();
1437            let sql_string = sql.to_string();
1438            let default = if sql.contains(" IN (") {
1439                (params.len() as u64).saturating_sub(1)
1440            } else {
1441                1
1442            };
1443            Box::pin(async move {
1444                recorded.lock().unwrap().push((sql_string, params));
1445                Ok(affected.lock().unwrap().pop().unwrap_or(default))
1446            })
1447        }
1448
1449        fn count(
1450            &self,
1451            _sql: &str,
1452            _params: Vec<FilterValue>,
1453        ) -> crate::traits::BoxFuture<'_, QueryResult<u64>> {
1454            Box::pin(async { Ok(0) })
1455        }
1456    }
1457
1458    /// Stand-in RowRef so `execute_insert` / `query_one` can synthesise
1459    /// a `TestModel` value without a live database.
1460    struct CannedRow;
1461
1462    impl crate::row::RowRef for CannedRow {
1463        fn get_i32(&self, _column: &str) -> Result<i32, crate::row::RowError> {
1464            Ok(0)
1465        }
1466        fn get_i32_opt(&self, _column: &str) -> Result<Option<i32>, crate::row::RowError> {
1467            Ok(Some(0))
1468        }
1469        fn get_i64(&self, _column: &str) -> Result<i64, crate::row::RowError> {
1470            Ok(0)
1471        }
1472        fn get_i64_opt(&self, _column: &str) -> Result<Option<i64>, crate::row::RowError> {
1473            Ok(None)
1474        }
1475        fn get_f64(&self, _column: &str) -> Result<f64, crate::row::RowError> {
1476            Ok(0.0)
1477        }
1478        fn get_f64_opt(&self, _column: &str) -> Result<Option<f64>, crate::row::RowError> {
1479            Ok(None)
1480        }
1481        fn get_bool(&self, _column: &str) -> Result<bool, crate::row::RowError> {
1482            Ok(false)
1483        }
1484        fn get_bool_opt(&self, _column: &str) -> Result<Option<bool>, crate::row::RowError> {
1485            Ok(None)
1486        }
1487        fn get_str(&self, _column: &str) -> Result<&str, crate::row::RowError> {
1488            Ok("canned")
1489        }
1490        fn get_str_opt(&self, _column: &str) -> Result<Option<&str>, crate::row::RowError> {
1491            Ok(Some("canned"))
1492        }
1493        fn get_bytes(&self, _column: &str) -> Result<&[u8], crate::row::RowError> {
1494            Ok(b"")
1495        }
1496        fn get_bytes_opt(&self, _column: &str) -> Result<Option<&[u8]>, crate::row::RowError> {
1497            Ok(None)
1498        }
1499    }
1500
1501    #[tokio::test]
1502    async fn upsert_with_nested_in_update_branch_fires_update_nested_only() {
1503        // affected=1 on the UPDATE → update branch.
1504        let engine = RecordingEngine::with_affected(vec![1]);
1505        let op = UpsertOperation::<RecordingEngine, TestModel>::new(engine.clone())
1506            .r#where(Filter::Equals("id".into(), FilterValue::Int(7)))
1507            .create_set("id", FilterValue::Int(7))
1508            .create_set("email", "new@x.com")
1509            .update_set("name", "Renamed")
1510            .with_update_nested(NestedWriteOp::Disconnect {
1511                relation: "posts",
1512                target_table: "posts",
1513                foreign_key: "author_id",
1514                target_pk: "id",
1515                pk: FilterValue::Int(42),
1516            })
1517            .with_create_nested(NestedWriteOp::Create {
1518                relation: "posts",
1519                target_table: "posts",
1520                foreign_key: "author_id",
1521                payload: vec![vec![("title".into(), FilterValue::String("p1".into()))]],
1522            });
1523
1524        let _ = op.exec().await.expect("upsert update branch");
1525
1526        let stmts = engine.statements();
1527        // Expect: UPDATE (affected=1) + SELECT (re-fetch) + nested Disconnect UPDATE
1528        assert_eq!(
1529            stmts.len(),
1530            3,
1531            "UPDATE + SELECT + nested disconnect; got {stmts:#?}"
1532        );
1533        assert!(
1534            stmts[0].0.starts_with("UPDATE test_models"),
1535            "first stmt should be parent UPDATE: {}",
1536            stmts[0].0
1537        );
1538        assert!(
1539            stmts[1].0.starts_with("SELECT"),
1540            "second stmt should re-fetch the row: {}",
1541            stmts[1].0
1542        );
1543        // Third stmt is the nested Disconnect — UPDATE child + NULL.
1544        assert!(stmts[2].0.contains("UPDATE"), "got: {}", stmts[2].0);
1545        assert!(stmts[2].0.contains("posts"), "got: {}", stmts[2].0);
1546        assert!(stmts[2].0.contains("NULL"), "got: {}", stmts[2].0);
1547        // Verify no INSERT (no create branch) and no nested Create (FK splicing).
1548        assert!(
1549            !stmts.iter().any(|(s, _)| s.starts_with("INSERT INTO")),
1550            "no INSERT must fire on update branch: {stmts:#?}"
1551        );
1552    }
1553
1554    #[tokio::test]
1555    async fn upsert_with_nested_in_create_branch_fires_create_nested_only() {
1556        // affected=0 on UPDATE → create branch (INSERT runs).
1557        let engine = RecordingEngine::with_affected(vec![0]);
1558        let op = UpsertOperation::<RecordingEngine, TestModel>::new(engine.clone())
1559            .r#where(Filter::Equals("id".into(), FilterValue::Int(7)))
1560            .create_set("id", FilterValue::Int(7))
1561            .create_set("email", "new@x.com")
1562            .update_set("name", "Renamed")
1563            .with_update_nested(NestedWriteOp::Disconnect {
1564                relation: "posts",
1565                target_table: "posts",
1566                foreign_key: "author_id",
1567                target_pk: "id",
1568                pk: FilterValue::Int(42),
1569            })
1570            .with_create_nested(NestedWriteOp::Create {
1571                relation: "posts",
1572                target_table: "posts",
1573                foreign_key: "author_id",
1574                payload: vec![vec![("title".into(), FilterValue::String("p1".into()))]],
1575            });
1576
1577        let _ = op.exec().await.expect("upsert create branch");
1578
1579        let stmts = engine.statements();
1580        // Expect: UPDATE (affected=0) + INSERT + nested Create child INSERT
1581        assert_eq!(
1582            stmts.len(),
1583            3,
1584            "UPDATE + INSERT + nested child INSERT; got {stmts:#?}"
1585        );
1586        assert!(
1587            stmts[0].0.starts_with("UPDATE test_models"),
1588            "first stmt should be parent UPDATE: {}",
1589            stmts[0].0
1590        );
1591        assert!(
1592            stmts[1].0.contains("INSERT INTO test_models"),
1593            "second stmt should be the create-branch INSERT: {}",
1594            stmts[1].0
1595        );
1596        assert!(
1597            stmts[2].0.contains("INSERT INTO"),
1598            "third stmt should be the nested Create child INSERT: {}",
1599            stmts[2].0
1600        );
1601        assert!(
1602            stmts[2].0.contains("posts"),
1603            "nested INSERT targets posts: {}",
1604            stmts[2].0
1605        );
1606        // No SELECT (we got the row directly from the INSERT) and no
1607        // nested Disconnect UPDATE on posts table.
1608        assert!(
1609            !stmts.iter().any(|(s, _)| s.starts_with("SELECT")),
1610            "no SELECT must fire on create branch: {stmts:#?}"
1611        );
1612        assert!(
1613            !stmts
1614                .iter()
1615                .any(|(s, _)| s.contains("UPDATE \"posts\"") || s.contains("UPDATE posts")),
1616            "no nested Disconnect on update_nested must fire: {stmts:#?}"
1617        );
1618    }
1619
1620    #[tokio::test]
1621    async fn upsert_both_branches_carry_nested_only_one_fires() {
1622        // Run twice with two engines — once for each branch — and
1623        // confirm only the branch-appropriate nested ops execute.
1624        // Branch 1: update.
1625        let engine_u = RecordingEngine::with_affected(vec![1]);
1626        let _ = UpsertOperation::<RecordingEngine, TestModel>::new(engine_u.clone())
1627            .r#where(Filter::Equals("id".into(), FilterValue::Int(7)))
1628            .create_set("id", FilterValue::Int(7))
1629            .update_set("name", "Renamed")
1630            .with_update_nested(NestedWriteOp::Disconnect {
1631                relation: "posts",
1632                target_table: "posts",
1633                foreign_key: "author_id",
1634                target_pk: "id",
1635                pk: FilterValue::Int(42),
1636            })
1637            .with_create_nested(NestedWriteOp::Create {
1638                relation: "posts",
1639                target_table: "posts",
1640                foreign_key: "author_id",
1641                payload: vec![vec![("title".into(), FilterValue::String("p".into()))]],
1642            })
1643            .exec()
1644            .await
1645            .expect("update branch");
1646        let u_stmts = engine_u.statements();
1647        // Disconnect must fire, child INSERT (nested Create) must not.
1648        assert!(
1649            u_stmts.iter().any(|(s, _)| s.contains("NULL")),
1650            "expected nested Disconnect: {u_stmts:#?}"
1651        );
1652        assert!(
1653            !u_stmts
1654                .iter()
1655                .any(|(s, _)| s.contains("INSERT INTO") && s.contains("posts")),
1656            "no nested Create child INSERT on update branch: {u_stmts:#?}"
1657        );
1658
1659        // Branch 2: create.
1660        let engine_c = RecordingEngine::with_affected(vec![0]);
1661        let _ = UpsertOperation::<RecordingEngine, TestModel>::new(engine_c.clone())
1662            .r#where(Filter::Equals("id".into(), FilterValue::Int(7)))
1663            .create_set("id", FilterValue::Int(7))
1664            .update_set("name", "Renamed")
1665            .with_update_nested(NestedWriteOp::Disconnect {
1666                relation: "posts",
1667                target_table: "posts",
1668                foreign_key: "author_id",
1669                target_pk: "id",
1670                pk: FilterValue::Int(42),
1671            })
1672            .with_create_nested(NestedWriteOp::Create {
1673                relation: "posts",
1674                target_table: "posts",
1675                foreign_key: "author_id",
1676                payload: vec![vec![("title".into(), FilterValue::String("p".into()))]],
1677            })
1678            .exec()
1679            .await
1680            .expect("create branch");
1681        let c_stmts = engine_c.statements();
1682        // Child INSERT (nested Create on posts) must fire, Disconnect must not.
1683        assert!(
1684            c_stmts
1685                .iter()
1686                .any(|(s, _)| s.contains("INSERT INTO") && s.contains("posts")),
1687            "expected nested Create child INSERT: {c_stmts:#?}"
1688        );
1689        assert!(
1690            !c_stmts.iter().any(|(s, _)| s.contains("NULL")),
1691            "no nested Disconnect on create branch: {c_stmts:#?}"
1692        );
1693    }
1694}
prax_query/operations/upsert.rs

prax_query/operations/
upsert.rs
