Skip to main content

prax_query/
nested.rs

1#![allow(dead_code)]
2
3//! Nested write operations for managing relations in a single mutation.
4//!
5//! This module provides support for creating, connecting, disconnecting, and updating
6//! related records within a single create or update operation.
7//!
8//! # Example
9//!
10//! ```rust,ignore
11//! use prax_query::nested::*;
12//!
13//! // Create a user with nested posts
14//! let user = client
15//!     .user()
16//!     .create(user::create::Data {
17//!         email: "user@example.com".into(),
18//!         name: Some("John Doe".into()),
19//!         posts: Some(NestedWrite::create_many(vec![
20//!             post::create::Data { title: "First Post".into(), content: None },
21//!             post::create::Data { title: "Second Post".into(), content: None },
22//!         ])),
23//!     })
24//!     .exec()
25//!     .await?;
26//!
27//! // Connect existing posts to a user
28//! let user = client
29//!     .user()
30//!     .update(user::id::equals(1))
31//!     .data(user::update::Data {
32//!         posts: Some(NestedWrite::connect(vec![
33//!             post::id::equals(10),
34//!             post::id::equals(20),
35//!         ])),
36//!         ..Default::default()
37//!     })
38//!     .exec()
39//!     .await?;
40//!
41//! // Disconnect posts from a user
42//! let user = client
43//!     .user()
44//!     .update(user::id::equals(1))
45//!     .data(user::update::Data {
46//!         posts: Some(NestedWrite::disconnect(vec![
47//!             post::id::equals(10),
48//!         ])),
49//!         ..Default::default()
50//!     })
51//!     .exec()
52//!     .await?;
53//! ```
54
55use std::fmt::Debug;
56use std::marker::PhantomData;
57
58use crate::error::QueryResult;
59use crate::filter::{Filter, FilterValue};
60use crate::sql::quote_identifier;
61use crate::traits::{Model, QueryEngine};
62
63/// Represents a nested write operation for relations.
64#[derive(Debug, Clone)]
65pub enum NestedWrite<T: Model> {
66    /// Create new related records.
67    Create(Vec<NestedCreateData<T>>),
68    /// Create new records or connect existing ones.
69    CreateOrConnect(Vec<NestedCreateOrConnectData<T>>),
70    /// Connect existing records by their unique identifier.
71    Connect(Vec<Filter>),
72    /// Disconnect records from the relation.
73    Disconnect(Vec<Filter>),
74    /// Set the relation to exactly these records (disconnect all others).
75    Set(Vec<Filter>),
76    /// Delete related records.
77    Delete(Vec<Filter>),
78    /// Update related records.
79    Update(Vec<NestedUpdateData<T>>),
80    /// Update or create related records.
81    Upsert(Vec<NestedUpsertData<T>>),
82    /// Update many related records matching a filter.
83    UpdateMany(NestedUpdateManyData<T>),
84    /// Delete many related records matching a filter.
85    DeleteMany(Filter),
86}
87
88impl<T: Model> NestedWrite<T> {
89    /// Create a new related record.
90    pub fn create(data: NestedCreateData<T>) -> Self {
91        Self::Create(vec![data])
92    }
93
94    /// Create multiple new related records.
95    pub fn create_many(data: Vec<NestedCreateData<T>>) -> Self {
96        Self::Create(data)
97    }
98
99    /// Connect an existing record by filter.
100    pub fn connect_one(filter: impl Into<Filter>) -> Self {
101        Self::Connect(vec![filter.into()])
102    }
103
104    /// Connect multiple existing records by filters.
105    pub fn connect(filters: Vec<impl Into<Filter>>) -> Self {
106        Self::Connect(filters.into_iter().map(Into::into).collect())
107    }
108
109    /// Disconnect a record by filter.
110    pub fn disconnect_one(filter: impl Into<Filter>) -> Self {
111        Self::Disconnect(vec![filter.into()])
112    }
113
114    /// Disconnect multiple records by filters.
115    pub fn disconnect(filters: Vec<impl Into<Filter>>) -> Self {
116        Self::Disconnect(filters.into_iter().map(Into::into).collect())
117    }
118
119    /// Set the relation to exactly these records.
120    pub fn set(filters: Vec<impl Into<Filter>>) -> Self {
121        Self::Set(filters.into_iter().map(Into::into).collect())
122    }
123
124    /// Delete related records.
125    pub fn delete(filters: Vec<impl Into<Filter>>) -> Self {
126        Self::Delete(filters.into_iter().map(Into::into).collect())
127    }
128
129    /// Delete many related records matching a filter.
130    pub fn delete_many(filter: impl Into<Filter>) -> Self {
131        Self::DeleteMany(filter.into())
132    }
133}
134
135/// Data for creating a nested record.
136#[derive(Debug, Clone)]
137pub struct NestedCreateData<T: Model> {
138    /// The create data fields.
139    pub data: Vec<(String, FilterValue)>,
140    /// Marker for the model type.
141    _model: PhantomData<T>,
142}
143
144impl<T: Model> NestedCreateData<T> {
145    /// Create new nested create data.
146    pub fn new(data: Vec<(String, FilterValue)>) -> Self {
147        Self {
148            data,
149            _model: PhantomData,
150        }
151    }
152
153    /// Create from field-value pairs.
154    pub fn from_pairs(
155        pairs: impl IntoIterator<Item = (impl Into<String>, impl Into<FilterValue>)>,
156    ) -> Self {
157        Self::new(
158            pairs
159                .into_iter()
160                .map(|(k, v)| (k.into(), v.into()))
161                .collect(),
162        )
163    }
164}
165
166impl<T: Model> Default for NestedCreateData<T> {
167    fn default() -> Self {
168        Self::new(Vec::new())
169    }
170}
171
172/// Data for creating or connecting a nested record.
173#[derive(Debug, Clone)]
174pub struct NestedCreateOrConnectData<T: Model> {
175    /// Filter to find existing record.
176    pub filter: Filter,
177    /// Data to create if not found.
178    pub create: NestedCreateData<T>,
179}
180
181impl<T: Model> NestedCreateOrConnectData<T> {
182    /// Create new create-or-connect data.
183    pub fn new(filter: impl Into<Filter>, create: NestedCreateData<T>) -> Self {
184        Self {
185            filter: filter.into(),
186            create,
187        }
188    }
189}
190
191/// Data for updating a nested record.
192#[derive(Debug, Clone)]
193pub struct NestedUpdateData<T: Model> {
194    /// Filter to find the record to update.
195    pub filter: Filter,
196    /// The update data fields.
197    pub data: Vec<(String, FilterValue)>,
198    /// Marker for the model type.
199    _model: PhantomData<T>,
200}
201
202impl<T: Model> NestedUpdateData<T> {
203    /// Create new nested update data.
204    pub fn new(filter: impl Into<Filter>, data: Vec<(String, FilterValue)>) -> Self {
205        Self {
206            filter: filter.into(),
207            data,
208            _model: PhantomData,
209        }
210    }
211
212    /// Create from filter and field-value pairs.
213    pub fn from_pairs(
214        filter: impl Into<Filter>,
215        pairs: impl IntoIterator<Item = (impl Into<String>, impl Into<FilterValue>)>,
216    ) -> Self {
217        Self::new(
218            filter,
219            pairs
220                .into_iter()
221                .map(|(k, v)| (k.into(), v.into()))
222                .collect(),
223        )
224    }
225}
226
227/// Data for upserting a nested record.
228#[derive(Debug, Clone)]
229pub struct NestedUpsertData<T: Model> {
230    /// Filter to find existing record.
231    pub filter: Filter,
232    /// Data to create if not found.
233    pub create: NestedCreateData<T>,
234    /// Data to update if found.
235    pub update: Vec<(String, FilterValue)>,
236    /// Marker for the model type.
237    _model: PhantomData<T>,
238}
239
240impl<T: Model> NestedUpsertData<T> {
241    /// Create new nested upsert data.
242    pub fn new(
243        filter: impl Into<Filter>,
244        create: NestedCreateData<T>,
245        update: Vec<(String, FilterValue)>,
246    ) -> Self {
247        Self {
248            filter: filter.into(),
249            create,
250            update,
251            _model: PhantomData,
252        }
253    }
254}
255
256/// Data for updating many nested records.
257#[derive(Debug, Clone)]
258pub struct NestedUpdateManyData<T: Model> {
259    /// Filter to match records.
260    pub filter: Filter,
261    /// The update data fields.
262    pub data: Vec<(String, FilterValue)>,
263    /// Marker for the model type.
264    _model: PhantomData<T>,
265}
266
267impl<T: Model> NestedUpdateManyData<T> {
268    /// Create new nested update-many data.
269    pub fn new(filter: impl Into<Filter>, data: Vec<(String, FilterValue)>) -> Self {
270        Self {
271            filter: filter.into(),
272            data,
273            _model: PhantomData,
274        }
275    }
276}
277
278/// Builder for nested write SQL operations.
279///
280/// The SQL emitters here currently bake in [`crate::dialect::Postgres`] —
281/// nested writes are not yet wired into a live client, and the placeholder
282/// syntax (`$N`) is Postgres-shaped. When this builder lands on the live
283/// client path the dialect should thread through from the engine.
284#[derive(Debug)]
285pub struct NestedWriteBuilder {
286    /// The parent table name.
287    parent_table: String,
288    /// The parent primary key column(s).
289    parent_pk: Vec<String>,
290    /// The related table name.
291    related_table: String,
292    /// The foreign key column on the related table.
293    foreign_key: String,
294    /// Whether this is a one-to-many (true) or many-to-many (false) relation.
295    is_one_to_many: bool,
296    /// Join table for many-to-many relations.
297    join_table: Option<JoinTableInfo>,
298}
299
300/// Information about a join table for many-to-many relations.
301#[derive(Debug, Clone)]
302pub struct JoinTableInfo {
303    /// The join table name.
304    pub table_name: String,
305    /// Column referencing the parent table.
306    pub parent_column: String,
307    /// Column referencing the related table.
308    pub related_column: String,
309}
310
311impl NestedWriteBuilder {
312    /// Create a builder for a one-to-many relation.
313    pub fn one_to_many(
314        parent_table: impl Into<String>,
315        parent_pk: Vec<String>,
316        related_table: impl Into<String>,
317        foreign_key: impl Into<String>,
318    ) -> Self {
319        Self {
320            parent_table: parent_table.into(),
321            parent_pk,
322            related_table: related_table.into(),
323            foreign_key: foreign_key.into(),
324            is_one_to_many: true,
325            join_table: None,
326        }
327    }
328
329    /// Create a builder for a many-to-many relation.
330    pub fn many_to_many(
331        parent_table: impl Into<String>,
332        parent_pk: Vec<String>,
333        related_table: impl Into<String>,
334        join_table: JoinTableInfo,
335    ) -> Self {
336        Self {
337            parent_table: parent_table.into(),
338            parent_pk,
339            related_table: related_table.into(),
340            foreign_key: String::new(), // Not used for many-to-many
341            is_one_to_many: false,
342            join_table: Some(join_table),
343        }
344    }
345
346    /// Build SQL for connecting records.
347    pub fn build_connect_sql<T: Model>(
348        &self,
349        parent_id: &FilterValue,
350        filters: &[Filter],
351    ) -> Vec<(String, Vec<FilterValue>)> {
352        let mut statements = Vec::new();
353
354        if self.is_one_to_many {
355            // For one-to-many, update the foreign key on related records
356            for filter in filters {
357                let (where_sql, mut params) = filter.to_sql(0, &crate::dialect::Postgres);
358                let sql = format!(
359                    "UPDATE {} SET {} = ${} WHERE {}",
360                    quote_identifier(&self.related_table),
361                    quote_identifier(&self.foreign_key),
362                    params.len() + 1,
363                    where_sql
364                );
365                params.push(parent_id.clone());
366                statements.push((sql, params));
367            }
368        } else if let Some(join) = &self.join_table {
369            // For many-to-many, insert into join table
370            // First, we need to get the IDs of the related records
371            for filter in filters {
372                let (where_sql, mut params) = filter.to_sql(0, &crate::dialect::Postgres);
373
374                // Get the related record ID (assuming single-column PK for now)
375                let select_sql = format!(
376                    "SELECT {} FROM {} WHERE {}",
377                    quote_identifier(T::PRIMARY_KEY.first().unwrap_or(&"id")),
378                    quote_identifier(&self.related_table),
379                    where_sql
380                );
381
382                // Insert into join table
383                let insert_sql = format!(
384                    "INSERT INTO {} ({}, {}) SELECT ${}, {} FROM {} WHERE {} ON CONFLICT DO NOTHING",
385                    quote_identifier(&join.table_name),
386                    quote_identifier(&join.parent_column),
387                    quote_identifier(&join.related_column),
388                    params.len() + 1,
389                    quote_identifier(T::PRIMARY_KEY.first().unwrap_or(&"id")),
390                    quote_identifier(&self.related_table),
391                    where_sql
392                );
393                params.push(parent_id.clone());
394                statements.push((insert_sql, params));
395                // Keep select_sql for potential subquery use
396                let _ = select_sql;
397            }
398        }
399
400        statements
401    }
402
403    /// Build SQL for disconnecting records.
404    pub fn build_disconnect_sql(
405        &self,
406        parent_id: &FilterValue,
407        filters: &[Filter],
408    ) -> Vec<(String, Vec<FilterValue>)> {
409        let mut statements = Vec::new();
410
411        if self.is_one_to_many {
412            // For one-to-many, set the foreign key to NULL
413            for filter in filters {
414                let (where_sql, mut params) = filter.to_sql(0, &crate::dialect::Postgres);
415                let sql = format!(
416                    "UPDATE {} SET {} = NULL WHERE {} AND {} = ${}",
417                    quote_identifier(&self.related_table),
418                    quote_identifier(&self.foreign_key),
419                    where_sql,
420                    quote_identifier(&self.foreign_key),
421                    params.len() + 1
422                );
423                params.push(parent_id.clone());
424                statements.push((sql, params));
425            }
426        } else if let Some(join) = &self.join_table {
427            // For many-to-many, delete from join table
428            for filter in filters {
429                let (where_sql, mut params) = filter.to_sql(1, &crate::dialect::Postgres);
430                let sql = format!(
431                    "DELETE FROM {} WHERE {} = $1 AND {} IN (SELECT id FROM {} WHERE {})",
432                    quote_identifier(&join.table_name),
433                    quote_identifier(&join.parent_column),
434                    quote_identifier(&join.related_column),
435                    quote_identifier(&self.related_table),
436                    where_sql
437                );
438                let mut final_params = vec![parent_id.clone()];
439                final_params.extend(params);
440                params = final_params;
441                statements.push((sql, params));
442            }
443        }
444
445        statements
446    }
447
448    /// Build SQL for setting the relation (disconnect all, then connect specified).
449    pub fn build_set_sql<T: Model>(
450        &self,
451        parent_id: &FilterValue,
452        filters: &[Filter],
453    ) -> Vec<(String, Vec<FilterValue>)> {
454        let mut statements = Vec::new();
455
456        // First, disconnect all existing relations
457        if self.is_one_to_many {
458            let sql = format!(
459                "UPDATE {} SET {} = NULL WHERE {} = $1",
460                quote_identifier(&self.related_table),
461                quote_identifier(&self.foreign_key),
462                quote_identifier(&self.foreign_key)
463            );
464            statements.push((sql, vec![parent_id.clone()]));
465        } else if let Some(join) = &self.join_table {
466            let sql = format!(
467                "DELETE FROM {} WHERE {} = $1",
468                quote_identifier(&join.table_name),
469                quote_identifier(&join.parent_column)
470            );
471            statements.push((sql, vec![parent_id.clone()]));
472        }
473
474        // Then connect the specified records
475        statements.extend(self.build_connect_sql::<T>(parent_id, filters));
476
477        statements
478    }
479
480    /// Build SQL for creating nested records.
481    pub fn build_create_sql<T: Model>(
482        &self,
483        parent_id: &FilterValue,
484        creates: &[NestedCreateData<T>],
485    ) -> Vec<(String, Vec<FilterValue>)> {
486        let mut statements = Vec::with_capacity(creates.len());
487        let quoted_table = quote_identifier(&self.related_table);
488
489        for create in creates {
490            let row_len = create.data.len() + 1;
491            let mut columns: Vec<String> = Vec::with_capacity(row_len);
492            let mut values: Vec<FilterValue> = Vec::with_capacity(row_len);
493            for (k, v) in &create.data {
494                columns.push(k.clone());
495                values.push(v.clone());
496            }
497
498            columns.push(self.foreign_key.clone());
499            values.push(parent_id.clone());
500
501            let mut col_list = String::new();
502            let mut placeholders = String::new();
503            for (i, c) in columns.iter().enumerate() {
504                if i > 0 {
505                    col_list.push_str(", ");
506                    placeholders.push_str(", ");
507                }
508                col_list.push_str(&quote_identifier(c));
509                use std::fmt::Write;
510                let _ = write!(placeholders, "${}", i + 1);
511            }
512
513            let sql = format!(
514                "INSERT INTO {} ({}) VALUES ({}) RETURNING *",
515                quoted_table, col_list, placeholders,
516            );
517
518            statements.push((sql, values));
519        }
520
521        statements
522    }
523
524    /// Build SQL for deleting nested records.
525    pub fn build_delete_sql(
526        &self,
527        parent_id: &FilterValue,
528        filters: &[Filter],
529    ) -> Vec<(String, Vec<FilterValue>)> {
530        let mut statements = Vec::new();
531
532        for filter in filters {
533            let (where_sql, mut params) = filter.to_sql(0, &crate::dialect::Postgres);
534            let sql = format!(
535                "DELETE FROM {} WHERE {} AND {} = ${}",
536                quote_identifier(&self.related_table),
537                where_sql,
538                quote_identifier(&self.foreign_key),
539                params.len() + 1
540            );
541            params.push(parent_id.clone());
542            statements.push((sql, params));
543        }
544
545        statements
546    }
547}
548
549/// A container for collecting all nested write operations to execute.
550#[derive(Debug, Default)]
551pub struct NestedWriteOperations {
552    /// SQL statements to execute before the main operation.
553    pub pre_statements: Vec<(String, Vec<FilterValue>)>,
554    /// SQL statements to execute after the main operation.
555    pub post_statements: Vec<(String, Vec<FilterValue>)>,
556}
557
558impl NestedWriteOperations {
559    /// Create a new empty container.
560    pub fn new() -> Self {
561        Self::default()
562    }
563
564    /// Add a pre-operation statement.
565    pub fn add_pre(&mut self, sql: String, params: Vec<FilterValue>) {
566        self.pre_statements.push((sql, params));
567    }
568
569    /// Add a post-operation statement.
570    pub fn add_post(&mut self, sql: String, params: Vec<FilterValue>) {
571        self.post_statements.push((sql, params));
572    }
573
574    /// Extend with statements from another container.
575    pub fn extend(&mut self, other: Self) {
576        self.pre_statements.extend(other.pre_statements);
577        self.post_statements.extend(other.post_statements);
578    }
579
580    /// Check if there are any operations.
581    pub fn is_empty(&self) -> bool {
582        self.pre_statements.is_empty() && self.post_statements.is_empty()
583    }
584
585    /// Get total number of statements.
586    pub fn len(&self) -> usize {
587        self.pre_statements.len() + self.post_statements.len()
588    }
589}
590
591/// Model-erased nested write op used by `CreateOperation::with(...)`.
592///
593/// The type-parameterized [`NestedWrite`] above is keyed on the parent
594/// model and doesn't compose across heterogeneous child types — a
595/// `CreateOperation<E, User>.with(posts_write)` needs to carry child
596/// writes for a different model (`Post`) than the parent, so `User`'s
597/// `NestedWrite<User>` can't encode them. This sibling enum drops the
598/// model type parameter and carries only the runtime metadata the
599/// execution path actually needs: the target table, the foreign-key
600/// column on that table, and the raw child-column payload.
601///
602/// Emitted by the codegen's per-relation `create()` / `connect()`
603/// helpers on `user::posts::*`. Payloads are a nested
604/// `Vec<Vec<(String, FilterValue)>>` rather than a strongly-typed
605/// `CreateInput` because the derive path doesn't currently emit a
606/// `CreateInput` struct per model — see the task docs for the trade-off
607/// and the upgrade path.
608#[derive(Debug, Clone)]
609pub enum NestedWriteOp {
610    /// Create children whose FK column points at the parent's PK.
611    ///
612    /// `relation` is retained for diagnostics/debugging; the executor
613    /// only needs `target_table`, `foreign_key`, and `payload`.
614    Create {
615        relation: &'static str,
616        target_table: &'static str,
617        foreign_key: &'static str,
618        /// One `Vec<(column, value)>` per child row. The FK column +
619        /// parent PK are appended by [`NestedWriteOp::execute`].
620        payload: Vec<Vec<(String, FilterValue)>>,
621    },
622    /// Connect an existing child row by its primary-key value.
623    ///
624    /// Lowers to
625    /// `UPDATE <target_table> SET <foreign_key> = <parent_pk> WHERE <target_pk> = <pk>`
626    /// at execute time. The identifier fields are `&'static str` because
627    /// they come from codegen-emitted constants on the per-relation
628    /// `RelationMeta` / `Model` types — the type itself enforces the
629    /// SQL-safety boundary (see `.cursor/rules/sql-safety.mdc`). Only
630    /// `parent_pk` and `pk` flow as `$N`-bound parameters.
631    Connect {
632        relation: &'static str,
633        target_table: &'static str,
634        foreign_key: &'static str,
635        target_pk: &'static str,
636        pk: FilterValue,
637    },
638    /// Disconnect a child row by clearing its FK column to `NULL`.
639    ///
640    /// Lowers to `UPDATE <target_table> SET <foreign_key> = NULL WHERE <target_pk> = <pk>`.
641    /// The child row persists; only the FK is cleared. Use
642    /// [`NestedWriteOp::Delete`] to remove the row entirely.
643    Disconnect {
644        relation: &'static str,
645        target_table: &'static str,
646        foreign_key: &'static str,
647        target_pk: &'static str,
648        pk: FilterValue,
649    },
650    /// Delete a child row by its primary key.
651    ///
652    /// Lowers to `DELETE FROM <target_table> WHERE <target_pk> = <pk>`.
653    /// Returns `QueryError::not_found` when the PK doesn't match any row,
654    /// matching the Connect-batch affected-rows contract.
655    Delete {
656        relation: &'static str,
657        target_table: &'static str,
658        target_pk: &'static str,
659        pk: FilterValue,
660    },
661    /// Delete many child rows matching a scalar filter, scoped to the
662    /// parent's children only.
663    ///
664    /// Lowers to `DELETE FROM <target_table> WHERE <foreign_key> = <parent_pk> AND <filter>`.
665    /// The AND-with-parent-FK clause is a safety bound enforced at SQL
666    /// emit time — the user-supplied filter cannot remove rows belonging
667    /// to other parents.
668    DeleteMany {
669        relation: &'static str,
670        target_table: &'static str,
671        foreign_key: &'static str,
672        filter: Filter,
673    },
674    /// Update a child row by its primary key.
675    ///
676    /// Lowers to
677    /// `UPDATE <target_table> SET <writeop-fragments> WHERE <target_pk> = $1`.
678    /// Each entry in `payload` contributes one column assignment whose
679    /// shape is determined by the [`crate::inputs::WriteOp`] variant
680    /// (plain set, atomic increment/decrement/multiply/divide, or
681    /// null-out via Unset). Returns `QueryError::not_found` when the PK
682    /// doesn't match any row, mirroring [`NestedWriteOp::Delete`]'s
683    /// affected-rows contract.
684    Update {
685        relation: &'static str,
686        target_table: &'static str,
687        target_pk: &'static str,
688        pk: FilterValue,
689        payload: Vec<(String, crate::inputs::WriteOp)>,
690    },
691    /// Update many child rows matching a filter, scoped to the parent's
692    /// children only.
693    ///
694    /// Lowers to
695    /// `UPDATE <target_table> SET <writeop-fragments> WHERE <foreign_key> = $1 AND <filter>`.
696    /// The AND-with-parent-FK clause is a safety bound enforced at SQL
697    /// emit time — the user-supplied filter cannot reach rows belonging
698    /// to other parents.
699    UpdateMany {
700        relation: &'static str,
701        target_table: &'static str,
702        foreign_key: &'static str,
703        filter: Filter,
704        payload: Vec<(String, crate::inputs::WriteOp)>,
705    },
706    /// Upsert: update if a row matches `pk`, else insert.
707    ///
708    /// On dialects with native single-statement upsert (Postgres, SQLite,
709    /// DuckDB, MySQL), emits one
710    /// `INSERT INTO <target_table> (<create_cols + fk>) VALUES (...)
711    /// ON CONFLICT (<target_pk>) DO UPDATE SET <update_writeops>`
712    /// (or `ON DUPLICATE KEY UPDATE ...` on MySQL). The `pk` field is
713    /// unused on this path — conflict detection comes from the inserted
714    /// row's PK column, which the codegen guarantees is present in
715    /// `create_payload`.
716    ///
717    /// On MSSQL and CQL, falls back to the two-statement form:
718    /// 1. `UPDATE <target_table> SET <update_writeops> WHERE <target_pk> = $1`
719    /// 2. If `affected_rows == 0`, `INSERT INTO <target_table>
720    ///    (<create_cols + fk>) VALUES (<...>)`.
721    ///
722    /// **Limitations of the fallback path** (MSSQL/CQL only):
723    /// - The `affected_rows == 0` check cannot distinguish "row absent"
724    ///   from "row exists but no columns changed" — a UPDATE that hits an
725    ///   identical row produces a spurious INSERT attempt and likely a PK
726    ///   unique-violation. Wrap the fallback in a transaction (or use a
727    ///   single-statement dialect) for strongest semantics.
728    /// - There is a TOCTOU race between the UPDATE returning 0 and the
729    ///   subsequent INSERT; a concurrent writer can insert the same row
730    ///   first.
731    ///
732    /// Document-store engines (`NotSql` dialect) are rejected at the top
733    /// of `execute` with `QueryError::unsupported(...)`.
734    ///
735    /// **Empty payloads**: an empty `update_payload` on a single-statement
736    /// dialect lowers to `ON CONFLICT (...) DO NOTHING` (PG/SQLite/DuckDB)
737    /// or `INSERT IGNORE INTO` (MySQL — idempotent no-op).
738    /// An empty `create_payload` errors with `QueryError::invalid_input`.
739    Upsert {
740        relation: &'static str,
741        target_table: &'static str,
742        foreign_key: &'static str,
743        target_pk: &'static str,
744        pk: FilterValue,
745        create_payload: Vec<(String, FilterValue)>,
746        update_payload: Vec<(String, crate::inputs::WriteOp)>,
747    },
748    /// Connect an existing child row if a `where` filter matches, else
749    /// insert a new one with the parent's FK spliced in.
750    ///
751    /// Two-statement engine-agnostic lowering:
752    /// 1. `UPDATE <target_table> SET <foreign_key> = $1 WHERE <filter>`
753    ///    (the connect path — points any matching row at the parent).
754    /// 2. If `affected_rows == 0`, emit
755    ///    `INSERT INTO <target_table> (<create_cols + foreign_key>) VALUES (<...>)`
756    ///    (the create path).
757    ///
758    /// If the filter matches multiple rows, every match has its FK pointed
759    /// at the parent — `connect_or_create` is typically used with a unique
760    /// where, but this is not enforced at runtime.
761    ///
762    /// As a safety measure, an empty (`Filter::None`) `where_filter` is
763    /// rejected at execute time — without this guard, the UPDATE would
764    /// lower to `... WHERE TRUE`, rewriting every row in the table.
765    ConnectOrCreate {
766        relation: &'static str,
767        target_table: &'static str,
768        foreign_key: &'static str,
769        where_filter: Filter,
770        create_payload: Vec<(String, FilterValue)>,
771    },
772    /// Replace the relation contents — after execution, exactly the
773    /// listed child rows are connected to the parent. Rows currently
774    /// connected that aren't in `set_pks` get their FK cleared; rows in
775    /// `set_pks` that aren't currently connected (or are connected to a
776    /// different parent) get their FK pointed at this parent.
777    ///
778    /// Two-statement engine-agnostic lowering:
779    /// 1. `UPDATE <target_table> SET <foreign_key> = NULL WHERE <foreign_key> = $parent AND <target_pk> NOT IN (set_pks)`
780    /// 2. `UPDATE <target_table> SET <foreign_key> = $parent WHERE <target_pk> IN (set_pks)`
781    ///
782    /// When `set_pks` is empty, step 1's `NOT IN ()` is invalid SQL —
783    /// the executor special-cases this to `UPDATE <child> SET <fk> = NULL
784    /// WHERE <fk> = $parent` (no NOT IN clause), then skips step 2.
785    ///
786    /// `set:` claims rows for this parent regardless of who they belonged
787    /// to before — pre-existing FK values get overwritten. This matches
788    /// Prisma's relation-replacement semantics.
789    Set {
790        relation: &'static str,
791        target_table: &'static str,
792        foreign_key: &'static str,
793        target_pk: &'static str,
794        set_pks: Vec<FilterValue>,
795    },
796}
797
798/// Shared SET-clause builder for the `Upsert` single-statement and
799/// two-statement fallback paths. `start_ph` lets the caller position
800/// placeholders after any preceding INSERT values; PG-family dialects
801/// use this index numerically, MySQL ignores it (`?` placeholders are
802/// positional via param push order).
803///
804/// `Unset` ops emit `col = NULL` and consume no placeholder slot.
805fn build_writeop_set_clause(
806    payload: &[(String, crate::inputs::WriteOp)],
807    dialect: &dyn crate::dialect::SqlDialect,
808    start_ph: usize,
809) -> (String, Vec<FilterValue>) {
810    let mut fragments: Vec<String> = Vec::with_capacity(payload.len());
811    let mut params: Vec<FilterValue> = Vec::with_capacity(payload.len());
812    let mut next_ph = start_ph;
813    for (col, op) in payload {
814        let (frag, maybe_val) =
815            op.to_set_fragment(&dialect.quote_ident(col), &dialect.placeholder(next_ph));
816        fragments.push(frag);
817        if let Some(val) = maybe_val {
818            params.push(val);
819            next_ph += 1;
820        }
821    }
822    (fragments.join(", "), params)
823}
824
825impl NestedWriteOp {
826    /// Execute this nested write inside `engine`, using `parent_pk`
827    /// as the foreign-key value to splice into each child row.
828    ///
829    /// For `Create`, this emits one `INSERT INTO <target_table> (...)`
830    /// per child, appending the FK column + parent PK to whatever
831    /// columns/values the caller supplied.
832    pub async fn execute<E>(self, engine: &E, parent_pk: &FilterValue) -> QueryResult<()>
833    where
834        E: QueryEngine,
835    {
836        match self {
837            NestedWriteOp::Connect {
838                relation: _,
839                target_table,
840                foreign_key,
841                target_pk,
842                pk,
843            } => {
844                let dialect = engine.dialect();
845                let sql = format!(
846                    "UPDATE {} SET {} = {} WHERE {} = {}",
847                    dialect.quote_ident(target_table),
848                    dialect.quote_ident(foreign_key),
849                    dialect.placeholder(1),
850                    dialect.quote_ident(target_pk),
851                    dialect.placeholder(2),
852                );
853                engine
854                    .execute_raw(&sql, vec![parent_pk.clone(), pk])
855                    .await?;
856                Ok(())
857            }
858            NestedWriteOp::Disconnect {
859                relation: _,
860                target_table,
861                foreign_key,
862                target_pk,
863                pk,
864            } => {
865                let dialect = engine.dialect();
866                let sql = format!(
867                    "UPDATE {} SET {} = NULL WHERE {} = {}",
868                    dialect.quote_ident(target_table),
869                    dialect.quote_ident(foreign_key),
870                    dialect.quote_ident(target_pk),
871                    dialect.placeholder(1),
872                );
873                engine.execute_raw(&sql, vec![pk]).await?;
874                Ok(())
875            }
876            NestedWriteOp::Delete {
877                relation: _,
878                target_table,
879                target_pk,
880                pk,
881            } => {
882                let dialect = engine.dialect();
883                let sql = format!(
884                    "DELETE FROM {} WHERE {} = {}",
885                    dialect.quote_ident(target_table),
886                    dialect.quote_ident(target_pk),
887                    dialect.placeholder(1),
888                );
889                let affected = engine.execute_raw(&sql, vec![pk]).await?;
890                if affected != 1 {
891                    return Err(crate::error::QueryError::not_found(target_table)
892                        .with_context("Nested Delete by PK"));
893                }
894                Ok(())
895            }
896            NestedWriteOp::DeleteMany {
897                relation: _,
898                target_table,
899                foreign_key,
900                filter,
901            } => {
902                let dialect = engine.dialect();
903                let is_unconstrained = matches!(filter, Filter::None);
904                let sql = if is_unconstrained {
905                    format!(
906                        "DELETE FROM {} WHERE {} = {}",
907                        dialect.quote_ident(target_table),
908                        dialect.quote_ident(foreign_key),
909                        dialect.placeholder(1),
910                    )
911                } else {
912                    let (filter_sql, params_tail) = filter.to_sql(1, dialect);
913                    let sql = format!(
914                        "DELETE FROM {} WHERE {} = {} AND ({})",
915                        dialect.quote_ident(target_table),
916                        dialect.quote_ident(foreign_key),
917                        dialect.placeholder(1),
918                        filter_sql,
919                    );
920                    let mut params = Vec::with_capacity(params_tail.len() + 1);
921                    params.push(parent_pk.clone());
922                    params.extend(params_tail);
923                    return engine.execute_raw(&sql, params).await.map(|_| ());
924                };
925                engine.execute_raw(&sql, vec![parent_pk.clone()]).await?;
926                Ok(())
927            }
928            NestedWriteOp::Update {
929                relation: _,
930                target_table,
931                target_pk,
932                pk,
933                payload,
934            } => {
935                if payload.is_empty() {
936                    return Ok(());
937                }
938                let dialect = engine.dialect();
939                let (set_text, mut update_params) = build_writeop_set_clause(&payload, dialect, 1);
940                let next_placeholder = update_params.len() + 1;
941                update_params.push(pk);
942                let sql = format!(
943                    "UPDATE {} SET {} WHERE {} = {}",
944                    dialect.quote_ident(target_table),
945                    set_text,
946                    dialect.quote_ident(target_pk),
947                    dialect.placeholder(next_placeholder),
948                );
949                let affected = engine.execute_raw(&sql, update_params).await?;
950                if affected != 1 {
951                    return Err(crate::error::QueryError::not_found(target_table)
952                        .with_context("Nested Update by PK"));
953                }
954                Ok(())
955            }
956            NestedWriteOp::UpdateMany {
957                relation: _,
958                target_table,
959                foreign_key,
960                filter,
961                payload,
962            } => {
963                if payload.is_empty() {
964                    return Ok(());
965                }
966                let dialect = engine.dialect();
967                let (set_text, mut params) = build_writeop_set_clause(&payload, dialect, 1);
968                let fk_placeholder_idx = params.len() + 1;
969                let fk_placeholder = dialect.placeholder(fk_placeholder_idx);
970                params.push(parent_pk.clone());
971
972                let is_unconstrained = matches!(filter, Filter::None);
973                let sql = if is_unconstrained {
974                    format!(
975                        "UPDATE {} SET {} WHERE {} = {}",
976                        dialect.quote_ident(target_table),
977                        set_text,
978                        dialect.quote_ident(foreign_key),
979                        fk_placeholder,
980                    )
981                } else {
982                    let (filter_sql, filter_params) = filter.to_sql(fk_placeholder_idx, dialect);
983                    params.extend(filter_params);
984                    format!(
985                        "UPDATE {} SET {} WHERE {} = {} AND ({})",
986                        dialect.quote_ident(target_table),
987                        set_text,
988                        dialect.quote_ident(foreign_key),
989                        fk_placeholder,
990                        filter_sql,
991                    )
992                };
993                engine.execute_raw(&sql, params).await?;
994                Ok(())
995            }
996            NestedWriteOp::Upsert {
997                relation: _,
998                target_table,
999                foreign_key,
1000                target_pk,
1001                pk,
1002                create_payload,
1003                update_payload,
1004            } => {
1005                let dialect = engine.dialect();
1006
1007                // Guard: document-store engines (NotSql) panic on any SQL
1008                // emission helper. Short-circuit before touching the dialect.
1009                if !dialect.supports_upsert() {
1010                    return Err(crate::error::QueryError::unsupported(format!(
1011                        "Nested Upsert is not supported by the `{}` engine",
1012                        std::any::type_name::<dyn crate::dialect::SqlDialect>()
1013                    )));
1014                }
1015
1016                // Guard: an upsert with no payloads is a no-op.
1017                if update_payload.is_empty() && create_payload.is_empty() {
1018                    return Ok(());
1019                }
1020
1021                // Guard: create_payload must not be empty — we can't INSERT
1022                // a row with no columns.
1023                if create_payload.is_empty() {
1024                    return Err(crate::error::QueryError::invalid_input(
1025                        "create_payload",
1026                        "Nested Upsert requires at least one create column when no row to update",
1027                    ));
1028                }
1029
1030                // Probe the dialect for single-statement upsert support by
1031                // checking whether upsert_clause returns a non-empty string.
1032                //
1033                // For the normal path (non-empty update_payload): build the
1034                // SET fragment with placeholders positioned AFTER the INSERT's
1035                // column values (insert occupies $1..$N where N =
1036                // create_payload.len() + 1 for the FK).
1037                //
1038                // On PG-family dialects the `start_ph` arg positions SET-clause
1039                // placeholders after the INSERT VALUES placeholders ($N+1, $N+2, ...).
1040                // On MySQL the placeholder text is always `?` — the SET-vs-VALUES split
1041                // is enforced solely by the param-push order: INSERT values are pushed
1042                // first below, then SET params via `values.extend(probe_set_params)`.
1043                let insert_arity = create_payload.len() + 1; // cols + FK
1044                let (probe_set_text, probe_set_params) =
1045                    build_writeop_set_clause(&update_payload, dialect, insert_arity + 1);
1046
1047                // Builds columns + values + placeholders + quoted_cols from
1048                // create_payload + foreign_key + parent_pk. Used by both the
1049                // single-statement and two-statement INSERT phases below.
1050                let build_insert_columns_and_values = || {
1051                    let mut columns: Vec<String> =
1052                        create_payload.iter().map(|(c, _)| c.clone()).collect();
1053                    let mut values: Vec<FilterValue> =
1054                        create_payload.iter().map(|(_, v)| v.clone()).collect();
1055                    columns.push(foreign_key.to_string());
1056                    values.push(parent_pk.clone());
1057                    let placeholders: Vec<String> =
1058                        (1..=values.len()).map(|i| dialect.placeholder(i)).collect();
1059                    let quoted_cols: Vec<String> =
1060                        columns.iter().map(|c| dialect.quote_ident(c)).collect();
1061                    (columns, values, placeholders, quoted_cols)
1062                };
1063
1064                // Pass raw (unquoted) target_pk — upsert_clause implementations
1065                // are responsible for quoting identifiers internally.
1066                let conflict_cols = [target_pk];
1067                let upsert_clause_text = dialect.upsert_clause(&conflict_cols, &probe_set_text);
1068
1069                if !upsert_clause_text.is_empty() {
1070                    // Single-statement path:
1071                    //   INSERT INTO child (cols + fk) VALUES ($1..$N) ON CONFLICT (...) DO UPDATE SET ...
1072                    //   or (empty update_payload): INSERT … ON CONFLICT DO NOTHING / INSERT IGNORE
1073                    let (_, mut values, placeholders, quoted_cols) =
1074                        build_insert_columns_and_values();
1075
1076                    // Drop the `pk` value — the single-statement form never
1077                    // references it; the conflict target is derived from the
1078                    // inserted row's PK column (which the codegen guarantees
1079                    // is present when create_payload is populated by the macro).
1080                    let _ = pk;
1081
1082                    let effective_upsert_clause = if update_payload.is_empty() {
1083                        // Pure-insert semantics: collapse to conditional INSERT.
1084                        // Use upsert_do_nothing_clause for dialect-specific DO NOTHING syntax.
1085                        // MySQL returns the idempotent self-assign form; PG/SQLite/DuckDB
1086                        // return ON CONFLICT (...) DO NOTHING.
1087                        let do_nothing = dialect.upsert_do_nothing_clause(&conflict_cols);
1088                        if do_nothing.is_empty() {
1089                            // Fallback dialect (MSSQL/CQL) with empty upsert_do_nothing_clause
1090                            // but non-empty upsert_clause_text: very rare. Treat as no-op.
1091                            return Ok(());
1092                        }
1093                        do_nothing
1094                    } else {
1095                        values.extend(probe_set_params);
1096                        upsert_clause_text
1097                    };
1098
1099                    let sql = format!(
1100                        "INSERT INTO {} ({}) VALUES ({}){}",
1101                        dialect.quote_ident(target_table),
1102                        quoted_cols.join(", "),
1103                        placeholders.join(", "),
1104                        effective_upsert_clause,
1105                    );
1106                    engine.execute_raw(&sql, values).await?;
1107                    return Ok(());
1108                }
1109
1110                // Two-statement fallback for dialects without native upsert syntax
1111                // (MSSQL, CQL). NotSql engines short-circuit above with QueryError::unsupported.
1112                //
1113                // TODO(upsert-toctou): `affected_rows == 0` cannot distinguish
1114                // "no row matched" from "row matched but no columns changed"
1115                // (MSSQL `@@ROWCOUNT`, CQL row-count semantics). A no-op UPDATE
1116                // would incorrectly trigger the INSERT, producing a PK unique
1117                // violation. Wrapping in an explicit transaction requires plumbing
1118                // not yet available here; tracked for a follow-up PR.
1119                if update_payload.is_empty() {
1120                    // Pure-insert semantics on fallback path: skip the UPDATE
1121                    // entirely and emit a bare INSERT. If the row already exists
1122                    // the engine will surface a PK/unique violation, which is
1123                    // the correct behaviour for an insert-or-fail path on dialects
1124                    // without native upsert syntax.
1125                    let (_, values, placeholders, quoted_cols) = build_insert_columns_and_values();
1126                    let insert_sql = format!(
1127                        "INSERT INTO {} ({}) VALUES ({})",
1128                        dialect.quote_ident(target_table),
1129                        quoted_cols.join(", "),
1130                        placeholders.join(", "),
1131                    );
1132                    engine.execute_raw(&insert_sql, values).await?;
1133                    return Ok(());
1134                }
1135
1136                // Phase 1: attempt UPDATE.
1137                let (set_text, mut update_params) =
1138                    build_writeop_set_clause(&update_payload, dialect, 1);
1139                let next_placeholder = update_params.len() + 1;
1140                update_params.push(pk.clone());
1141                let update_sql = format!(
1142                    "UPDATE {} SET {} WHERE {} = {}",
1143                    dialect.quote_ident(target_table),
1144                    set_text,
1145                    dialect.quote_ident(target_pk),
1146                    dialect.placeholder(next_placeholder),
1147                );
1148                let affected = engine.execute_raw(&update_sql, update_params).await?;
1149                if affected > 0 {
1150                    return Ok(());
1151                }
1152                // Phase 2: row didn't exist — INSERT it with the FK spliced in.
1153                let (_, values, placeholders, quoted_cols) = build_insert_columns_and_values();
1154                let insert_sql = format!(
1155                    "INSERT INTO {} ({}) VALUES ({})",
1156                    dialect.quote_ident(target_table),
1157                    quoted_cols.join(", "),
1158                    placeholders.join(", "),
1159                );
1160                engine.execute_raw(&insert_sql, values).await?;
1161                Ok(())
1162            }
1163            NestedWriteOp::ConnectOrCreate {
1164                relation: _,
1165                target_table,
1166                foreign_key,
1167                where_filter,
1168                create_payload,
1169            } => {
1170                // Safety: reject an empty (`Filter::None`) where. Without
1171                // this guard the UPDATE would lower to `... WHERE TRUE`
1172                // (per `Filter::None::to_sql`), rewriting every row in
1173                // the child table. `connect_or_create` semantically
1174                // requires a unique-where lookup; an empty where is a
1175                // user error, not an "always match every row" command.
1176                if matches!(where_filter, Filter::None) {
1177                    return Err(crate::error::QueryError::not_found(target_table).with_context(
1178                        "Nested ConnectOrCreate: empty `where` block would match every row; supply a unique filter",
1179                    ));
1180                }
1181                let dialect = engine.dialect();
1182                // Phase 1: attempt UPDATE to connect existing row(s).
1183                let (filter_sql, filter_params) = where_filter.to_sql(1, dialect);
1184                let mut update_params: Vec<FilterValue> =
1185                    Vec::with_capacity(filter_params.len() + 1);
1186                update_params.push(parent_pk.clone());
1187                update_params.extend(filter_params);
1188                let update_sql = format!(
1189                    "UPDATE {} SET {} = {} WHERE {}",
1190                    dialect.quote_ident(target_table),
1191                    dialect.quote_ident(foreign_key),
1192                    dialect.placeholder(1),
1193                    filter_sql,
1194                );
1195                let affected = engine.execute_raw(&update_sql, update_params).await?;
1196                if affected > 0 {
1197                    return Ok(());
1198                }
1199                // Phase 2: no row matched — INSERT with FK spliced in.
1200                if create_payload.is_empty() {
1201                    return Err(
1202                        crate::error::QueryError::not_found(target_table).with_context(
1203                            "Nested ConnectOrCreate: no match and create payload empty",
1204                        ),
1205                    );
1206                }
1207                let mut columns: Vec<String> =
1208                    create_payload.iter().map(|(c, _)| c.clone()).collect();
1209                let mut values: Vec<FilterValue> =
1210                    create_payload.into_iter().map(|(_, v)| v).collect();
1211                columns.push(foreign_key.to_string());
1212                values.push(parent_pk.clone());
1213                let placeholders: Vec<String> =
1214                    (1..=values.len()).map(|i| dialect.placeholder(i)).collect();
1215                let quoted_cols: Vec<String> =
1216                    columns.iter().map(|c| dialect.quote_ident(c)).collect();
1217                let insert_sql = format!(
1218                    "INSERT INTO {} ({}) VALUES ({})",
1219                    dialect.quote_ident(target_table),
1220                    quoted_cols.join(", "),
1221                    placeholders.join(", "),
1222                );
1223                engine.execute_raw(&insert_sql, values).await?;
1224                Ok(())
1225            }
1226            NestedWriteOp::Create {
1227                relation: _,
1228                target_table,
1229                foreign_key,
1230                payload,
1231            } => {
1232                if payload.is_empty() {
1233                    return Ok(());
1234                }
1235
1236                let dialect = engine.dialect();
1237
1238                // All rows in a single `Create` op share the same column
1239                // set (codegen guarantee). Derive columns from the first
1240                // row, then append the FK column once. Each row
1241                // contributes its values + the parent PK.
1242                let first = &payload[0];
1243                let mut columns: Vec<String> = first.iter().map(|(c, _)| c.clone()).collect();
1244                columns.push(foreign_key.to_string());
1245                let cols_per_row = columns.len();
1246
1247                let quoted_cols: Vec<String> =
1248                    columns.iter().map(|c| dialect.quote_ident(c)).collect();
1249
1250                let mut values: Vec<FilterValue> = Vec::with_capacity(payload.len() * cols_per_row);
1251                let mut row_placeholders: Vec<String> = Vec::with_capacity(payload.len());
1252                let mut next_placeholder = 1usize;
1253
1254                for child in payload {
1255                    let mut row_phs: Vec<String> = Vec::with_capacity(cols_per_row);
1256                    for (_, v) in child {
1257                        values.push(v);
1258                        row_phs.push(dialect.placeholder(next_placeholder));
1259                        next_placeholder += 1;
1260                    }
1261                    values.push(parent_pk.clone());
1262                    row_phs.push(dialect.placeholder(next_placeholder));
1263                    next_placeholder += 1;
1264                    row_placeholders.push(format!("({})", row_phs.join(", ")));
1265                }
1266
1267                // NOTE: Combining all rows into a single multi-VALUES
1268                // INSERT means any constraint failure rolls back the
1269                // entire batch, not just the failing row. This matches
1270                // typical Prisma semantics for nested writes.
1271                let sql = format!(
1272                    "INSERT INTO {} ({}) VALUES {}",
1273                    dialect.quote_ident(target_table),
1274                    quoted_cols.join(", "),
1275                    row_placeholders.join(", "),
1276                );
1277
1278                engine.execute_raw(&sql, values).await?;
1279                Ok(())
1280            }
1281            NestedWriteOp::Set {
1282                relation: _,
1283                target_table,
1284                foreign_key,
1285                target_pk,
1286                set_pks,
1287            } => {
1288                let dialect = engine.dialect();
1289
1290                // Phase 1: disconnect current children not in set_pks.
1291                if set_pks.is_empty() {
1292                    // No NOT IN clause needed — clear every child of this parent.
1293                    let sql = format!(
1294                        "UPDATE {} SET {} = NULL WHERE {} = {}",
1295                        dialect.quote_ident(target_table),
1296                        dialect.quote_ident(foreign_key),
1297                        dialect.quote_ident(foreign_key),
1298                        dialect.placeholder(1),
1299                    );
1300                    engine.execute_raw(&sql, vec![parent_pk.clone()]).await?;
1301                    return Ok(());
1302                }
1303                // set_pks is non-empty — emit disconnect with NOT IN clause + connect.
1304                let mut disconnect_params: Vec<FilterValue> = Vec::with_capacity(set_pks.len() + 1);
1305                disconnect_params.push(parent_pk.clone());
1306                let mut not_in_placeholders: Vec<String> = Vec::with_capacity(set_pks.len());
1307                for (i, pk) in set_pks.iter().enumerate() {
1308                    disconnect_params.push(pk.clone());
1309                    not_in_placeholders.push(dialect.placeholder(i + 2));
1310                }
1311                let disconnect_sql = format!(
1312                    "UPDATE {} SET {} = NULL WHERE {} = {} AND {} NOT IN ({})",
1313                    dialect.quote_ident(target_table),
1314                    dialect.quote_ident(foreign_key),
1315                    dialect.quote_ident(foreign_key),
1316                    dialect.placeholder(1),
1317                    dialect.quote_ident(target_pk),
1318                    not_in_placeholders.join(", "),
1319                );
1320                engine
1321                    .execute_raw(&disconnect_sql, disconnect_params)
1322                    .await?;
1323
1324                // Phase 2: connect every row in set_pks (idempotent for already-connected).
1325                let mut connect_params: Vec<FilterValue> = Vec::with_capacity(set_pks.len() + 1);
1326                connect_params.push(parent_pk.clone());
1327                let mut in_placeholders: Vec<String> = Vec::with_capacity(set_pks.len());
1328                for (i, pk) in set_pks.iter().enumerate() {
1329                    connect_params.push(pk.clone());
1330                    in_placeholders.push(dialect.placeholder(i + 2));
1331                }
1332                let connect_sql = format!(
1333                    "UPDATE {} SET {} = {} WHERE {} IN ({})",
1334                    dialect.quote_ident(target_table),
1335                    dialect.quote_ident(foreign_key),
1336                    dialect.placeholder(1),
1337                    dialect.quote_ident(target_pk),
1338                    in_placeholders.join(", "),
1339                );
1340                engine.execute_raw(&connect_sql, connect_params).await?;
1341                Ok(())
1342            }
1343        }
1344    }
1345}
1346
1347#[cfg(test)]
1348mod tests {
1349    use super::*;
1350    use std::sync::{Arc, Mutex};
1351
1352    use crate::error::QueryError;
1353    use crate::traits::BoxFuture;
1354
1355    /// Captured (sql, params) entries from the mock engine.
1356    type StatementLog = Arc<Mutex<Vec<(String, Vec<FilterValue>)>>>;
1357
1358    #[derive(Clone, Copy)]
1359    enum DialectKind {
1360        Postgres,
1361        Mysql,
1362        Mssql,
1363        NotSql,
1364    }
1365
1366    /// Recording mock engine for [`NestedWriteOp::execute`] tests.
1367    ///
1368    /// Captures the (sql, params) of every [`QueryEngine::execute_raw`]
1369    /// call so tests can assert the lowered shape. Returns 1 from
1370    /// `execute_raw` by default; tests that need to exercise the
1371    /// zero-affected-rows path (e.g. upsert's insert phase) can supply a
1372    /// sequence of affected-row counts via [`RecordingEngine::with_affected`].
1373    #[derive(Clone)]
1374    struct RecordingEngine {
1375        recorded: StatementLog,
1376        affected: Arc<Mutex<Vec<u64>>>,
1377        dialect_kind: DialectKind,
1378    }
1379
1380    impl RecordingEngine {
1381        fn new() -> Self {
1382            Self {
1383                recorded: Arc::new(Mutex::new(Vec::new())),
1384                affected: Arc::new(Mutex::new(Vec::new())),
1385                dialect_kind: DialectKind::Postgres,
1386            }
1387        }
1388
1389        /// Build an engine that returns each entry of `seq` from successive
1390        /// `execute_raw` calls, in order. Once exhausted, falls back to 1.
1391        fn with_affected(seq: Vec<u64>) -> Self {
1392            // Stored in reverse so we can pop in O(1).
1393            let mut rev = seq;
1394            rev.reverse();
1395            Self {
1396                recorded: Arc::new(Mutex::new(Vec::new())),
1397                affected: Arc::new(Mutex::new(rev)),
1398                dialect_kind: DialectKind::Postgres,
1399            }
1400        }
1401
1402        fn mysql() -> Self {
1403            Self {
1404                dialect_kind: DialectKind::Mysql,
1405                ..Self::new()
1406            }
1407        }
1408
1409        fn mssql() -> Self {
1410            Self {
1411                dialect_kind: DialectKind::Mssql,
1412                ..Self::new()
1413            }
1414        }
1415
1416        fn mssql_with_affected(seq: Vec<u64>) -> Self {
1417            Self {
1418                dialect_kind: DialectKind::Mssql,
1419                ..Self::with_affected(seq)
1420            }
1421        }
1422
1423        fn notsql() -> Self {
1424            Self {
1425                dialect_kind: DialectKind::NotSql,
1426                ..Self::new()
1427            }
1428        }
1429
1430        fn statements(&self) -> Vec<(String, Vec<FilterValue>)> {
1431            self.recorded.lock().unwrap().clone()
1432        }
1433    }
1434
1435    impl crate::traits::QueryEngine for RecordingEngine {
1436        fn dialect(&self) -> &dyn crate::dialect::SqlDialect {
1437            match self.dialect_kind {
1438                DialectKind::Postgres => &crate::dialect::Postgres,
1439                DialectKind::Mysql => &crate::dialect::Mysql,
1440                DialectKind::Mssql => &crate::dialect::Mssql,
1441                DialectKind::NotSql => &crate::dialect::NotSql,
1442            }
1443        }
1444
1445        fn query_many<T: Model + crate::row::FromRow + Send + 'static>(
1446            &self,
1447            _sql: &str,
1448            _params: Vec<FilterValue>,
1449        ) -> BoxFuture<'_, QueryResult<Vec<T>>> {
1450            Box::pin(async { Ok(Vec::new()) })
1451        }
1452
1453        fn query_one<T: Model + crate::row::FromRow + Send + 'static>(
1454            &self,
1455            _sql: &str,
1456            _params: Vec<FilterValue>,
1457        ) -> BoxFuture<'_, QueryResult<T>> {
1458            Box::pin(async { Err(QueryError::not_found("test")) })
1459        }
1460
1461        fn query_optional<T: Model + crate::row::FromRow + Send + 'static>(
1462            &self,
1463            _sql: &str,
1464            _params: Vec<FilterValue>,
1465        ) -> BoxFuture<'_, QueryResult<Option<T>>> {
1466            Box::pin(async { Ok(None) })
1467        }
1468
1469        fn execute_insert<T: Model + crate::row::FromRow + Send + 'static>(
1470            &self,
1471            _sql: &str,
1472            _params: Vec<FilterValue>,
1473        ) -> BoxFuture<'_, QueryResult<T>> {
1474            Box::pin(async { Err(QueryError::not_found("test")) })
1475        }
1476
1477        fn execute_update<T: Model + crate::row::FromRow + Send + 'static>(
1478            &self,
1479            _sql: &str,
1480            _params: Vec<FilterValue>,
1481        ) -> BoxFuture<'_, QueryResult<Vec<T>>> {
1482            Box::pin(async { Ok(Vec::new()) })
1483        }
1484
1485        fn execute_delete(
1486            &self,
1487            _sql: &str,
1488            _params: Vec<FilterValue>,
1489        ) -> BoxFuture<'_, QueryResult<u64>> {
1490            Box::pin(async { Ok(0) })
1491        }
1492
1493        fn execute_raw(
1494            &self,
1495            sql: &str,
1496            params: Vec<FilterValue>,
1497        ) -> BoxFuture<'_, QueryResult<u64>> {
1498            let recorded = self.recorded.clone();
1499            let affected = self.affected.clone();
1500            let sql = sql.to_string();
1501            Box::pin(async move {
1502                recorded.lock().unwrap().push((sql, params));
1503                let next = affected.lock().unwrap().pop().unwrap_or(1);
1504                Ok(next)
1505            })
1506        }
1507
1508        fn count(&self, _sql: &str, _params: Vec<FilterValue>) -> BoxFuture<'_, QueryResult<u64>> {
1509            Box::pin(async { Ok(0) })
1510        }
1511    }
1512
1513    struct TestModel;
1514
1515    impl Model for TestModel {
1516        const MODEL_NAME: &'static str = "Post";
1517        const TABLE_NAME: &'static str = "posts";
1518        const PRIMARY_KEY: &'static [&'static str] = &["id"];
1519        const COLUMNS: &'static [&'static str] = &["id", "title", "user_id"];
1520    }
1521
1522    struct TagModel;
1523
1524    impl Model for TagModel {
1525        const MODEL_NAME: &'static str = "Tag";
1526        const TABLE_NAME: &'static str = "tags";
1527        const PRIMARY_KEY: &'static [&'static str] = &["id"];
1528        const COLUMNS: &'static [&'static str] = &["id", "name"];
1529    }
1530
1531    #[test]
1532    fn test_nested_create_data() {
1533        let data: NestedCreateData<TestModel> =
1534            NestedCreateData::from_pairs([("title", FilterValue::String("Test Post".to_string()))]);
1535
1536        assert_eq!(data.data.len(), 1);
1537        assert_eq!(data.data[0].0, "title");
1538    }
1539
1540    #[test]
1541    fn test_nested_write_create() {
1542        let data: NestedCreateData<TestModel> =
1543            NestedCreateData::from_pairs([("title", FilterValue::String("Test Post".to_string()))]);
1544
1545        let write: NestedWrite<TestModel> = NestedWrite::create(data);
1546
1547        match write {
1548            NestedWrite::Create(creates) => assert_eq!(creates.len(), 1),
1549            _ => panic!("Expected Create variant"),
1550        }
1551    }
1552
1553    #[test]
1554    fn test_nested_write_connect() {
1555        let write: NestedWrite<TestModel> = NestedWrite::connect(vec![
1556            Filter::Equals("id".into(), FilterValue::Int(1)),
1557            Filter::Equals("id".into(), FilterValue::Int(2)),
1558        ]);
1559
1560        match write {
1561            NestedWrite::Connect(filters) => assert_eq!(filters.len(), 2),
1562            _ => panic!("Expected Connect variant"),
1563        }
1564    }
1565
1566    #[test]
1567    fn test_nested_write_disconnect() {
1568        let write: NestedWrite<TestModel> =
1569            NestedWrite::disconnect_one(Filter::Equals("id".into(), FilterValue::Int(1)));
1570
1571        match write {
1572            NestedWrite::Disconnect(filters) => assert_eq!(filters.len(), 1),
1573            _ => panic!("Expected Disconnect variant"),
1574        }
1575    }
1576
1577    #[test]
1578    fn test_nested_write_set() {
1579        let write: NestedWrite<TestModel> =
1580            NestedWrite::set(vec![Filter::Equals("id".into(), FilterValue::Int(1))]);
1581
1582        match write {
1583            NestedWrite::Set(filters) => assert_eq!(filters.len(), 1),
1584            _ => panic!("Expected Set variant"),
1585        }
1586    }
1587
1588    #[test]
1589    fn test_builder_one_to_many_connect() {
1590        let builder =
1591            NestedWriteBuilder::one_to_many("users", vec!["id".to_string()], "posts", "user_id");
1592
1593        let parent_id = FilterValue::Int(1);
1594        let filters = vec![Filter::Equals("id".into(), FilterValue::Int(10))];
1595
1596        let statements = builder.build_connect_sql::<TestModel>(&parent_id, &filters);
1597
1598        assert_eq!(statements.len(), 1);
1599        let (sql, params) = &statements[0];
1600        // Filter binds first ($1); the parent id takes the slot right after
1601        // the filter's params ($2) — no collision, no unreferenced $1.
1602        assert_eq!(sql, "UPDATE posts SET user_id = $2 WHERE \"id\" = $1");
1603        assert_eq!(params.len(), 2);
1604        assert_eq!(params[0], FilterValue::Int(10));
1605        assert_eq!(params[1], FilterValue::Int(1));
1606    }
1607
1608    #[test]
1609    fn test_builder_one_to_many_disconnect() {
1610        let builder =
1611            NestedWriteBuilder::one_to_many("users", vec!["id".to_string()], "posts", "user_id");
1612
1613        let parent_id = FilterValue::Int(1);
1614        let filters = vec![Filter::Equals("id".into(), FilterValue::Int(10))];
1615
1616        let statements = builder.build_disconnect_sql(&parent_id, &filters);
1617
1618        assert_eq!(statements.len(), 1);
1619        let (sql, params) = &statements[0];
1620        // Filter binds first ($1); the parent-FK comparison takes the slot
1621        // right after the filter's params ($2).
1622        assert_eq!(
1623            sql,
1624            "UPDATE posts SET user_id = NULL WHERE \"id\" = $1 AND user_id = $2"
1625        );
1626        assert_eq!(params.len(), 2);
1627        assert_eq!(params[0], FilterValue::Int(10));
1628        assert_eq!(params[1], FilterValue::Int(1));
1629    }
1630
1631    #[test]
1632    fn test_builder_many_to_many_connect() {
1633        let builder = NestedWriteBuilder::many_to_many(
1634            "posts",
1635            vec!["id".to_string()],
1636            "tags",
1637            JoinTableInfo {
1638                table_name: "post_tags".to_string(),
1639                parent_column: "post_id".to_string(),
1640                related_column: "tag_id".to_string(),
1641            },
1642        );
1643
1644        let parent_id = FilterValue::Int(1);
1645        let filters = vec![Filter::Equals("id".into(), FilterValue::Int(10))];
1646
1647        let statements = builder.build_connect_sql::<TagModel>(&parent_id, &filters);
1648
1649        assert_eq!(statements.len(), 1);
1650        let (sql, params) = &statements[0];
1651        // Filter binds first ($1); the parent id in the SELECT list takes the
1652        // slot right after the filter's params ($2).
1653        assert_eq!(
1654            sql,
1655            "INSERT INTO post_tags (post_id, tag_id) SELECT $2, id FROM tags WHERE \"id\" = $1 ON CONFLICT DO NOTHING"
1656        );
1657        assert_eq!(params.len(), 2);
1658        assert_eq!(params[0], FilterValue::Int(10));
1659        assert_eq!(params[1], FilterValue::Int(1));
1660    }
1661
1662    #[test]
1663    fn test_builder_many_to_many_disconnect() {
1664        let builder = NestedWriteBuilder::many_to_many(
1665            "posts",
1666            vec!["id".to_string()],
1667            "tags",
1668            JoinTableInfo {
1669                table_name: "post_tags".to_string(),
1670                parent_column: "post_id".to_string(),
1671                related_column: "tag_id".to_string(),
1672            },
1673        );
1674
1675        let parent_id = FilterValue::Int(1);
1676        let filters = vec![Filter::Equals("id".into(), FilterValue::Int(10))];
1677
1678        let statements = builder.build_disconnect_sql(&parent_id, &filters);
1679
1680        assert_eq!(statements.len(), 1);
1681        let (sql, params) = &statements[0];
1682        // Parent id is bound first ($1); the subquery filter continues at $2.
1683        assert_eq!(
1684            sql,
1685            "DELETE FROM post_tags WHERE post_id = $1 AND tag_id IN (SELECT id FROM tags WHERE \"id\" = $2)"
1686        );
1687        assert_eq!(params.len(), 2);
1688        assert_eq!(params[0], FilterValue::Int(1));
1689        assert_eq!(params[1], FilterValue::Int(10));
1690    }
1691
1692    #[test]
1693    fn test_builder_one_to_many_delete() {
1694        let builder =
1695            NestedWriteBuilder::one_to_many("users", vec!["id".to_string()], "posts", "user_id");
1696
1697        let parent_id = FilterValue::Int(1);
1698        let filters = vec![Filter::Equals("id".into(), FilterValue::Int(10))];
1699
1700        let statements = builder.build_delete_sql(&parent_id, &filters);
1701
1702        assert_eq!(statements.len(), 1);
1703        let (sql, params) = &statements[0];
1704        // Filter binds first ($1); the parent-FK comparison takes the slot
1705        // right after the filter's params ($2).
1706        assert_eq!(sql, "DELETE FROM posts WHERE \"id\" = $1 AND user_id = $2");
1707        assert_eq!(params.len(), 2);
1708        assert_eq!(params[0], FilterValue::Int(10));
1709        assert_eq!(params[1], FilterValue::Int(1));
1710    }
1711
1712    #[test]
1713    fn test_builder_create() {
1714        let builder =
1715            NestedWriteBuilder::one_to_many("users", vec!["id".to_string()], "posts", "user_id");
1716
1717        let parent_id = FilterValue::Int(1);
1718        let creates = vec![NestedCreateData::<TestModel>::from_pairs([(
1719            "title",
1720            FilterValue::String("New Post".to_string()),
1721        )])];
1722
1723        let statements = builder.build_create_sql::<TestModel>(&parent_id, &creates);
1724
1725        assert_eq!(statements.len(), 1);
1726        let (sql, params) = &statements[0];
1727        assert!(sql.contains("INSERT INTO"));
1728        assert!(sql.contains("posts"));
1729        assert!(sql.contains("RETURNING"));
1730        assert_eq!(params.len(), 2); // title + user_id
1731    }
1732
1733    #[test]
1734    fn test_builder_set() {
1735        let builder =
1736            NestedWriteBuilder::one_to_many("users", vec!["id".to_string()], "posts", "user_id");
1737
1738        let parent_id = FilterValue::Int(1);
1739        let filters = vec![Filter::Equals("id".into(), FilterValue::Int(10))];
1740
1741        let statements = builder.build_set_sql::<TestModel>(&parent_id, &filters);
1742
1743        // Should have disconnect all + connect statements
1744        assert!(statements.len() >= 2);
1745
1746        // First statement should disconnect all
1747        let (first_sql, _) = &statements[0];
1748        assert!(first_sql.contains("UPDATE"));
1749        assert!(first_sql.contains("NULL"));
1750    }
1751
1752    #[test]
1753    fn test_nested_write_operations() {
1754        let mut ops = NestedWriteOperations::new();
1755        assert!(ops.is_empty());
1756        assert_eq!(ops.len(), 0);
1757
1758        ops.add_pre("SELECT 1".to_string(), vec![]);
1759        ops.add_post("SELECT 2".to_string(), vec![]);
1760
1761        assert!(!ops.is_empty());
1762        assert_eq!(ops.len(), 2);
1763    }
1764
1765    #[test]
1766    fn test_nested_create_or_connect() {
1767        let create_data: NestedCreateData<TestModel> =
1768            NestedCreateData::from_pairs([("title", FilterValue::String("New Post".to_string()))]);
1769
1770        let create_or_connect = NestedCreateOrConnectData::new(
1771            Filter::Equals("title".into(), FilterValue::String("Existing".to_string())),
1772            create_data,
1773        );
1774
1775        assert!(matches!(create_or_connect.filter, Filter::Equals(..)));
1776        assert_eq!(create_or_connect.create.data.len(), 1);
1777    }
1778
1779    #[test]
1780    fn test_nested_update_data() {
1781        let update: NestedUpdateData<TestModel> = NestedUpdateData::from_pairs(
1782            Filter::Equals("id".into(), FilterValue::Int(1)),
1783            [("title", FilterValue::String("Updated".to_string()))],
1784        );
1785
1786        assert!(matches!(update.filter, Filter::Equals(..)));
1787        assert_eq!(update.data.len(), 1);
1788        assert_eq!(update.data[0].0, "title");
1789    }
1790
1791    #[tokio::test]
1792    async fn nested_op_connect_emits_update_set_where() {
1793        let engine = RecordingEngine::new();
1794        let op = NestedWriteOp::Connect {
1795            relation: "posts",
1796            target_table: "posts",
1797            foreign_key: "author_id",
1798            target_pk: "id",
1799            pk: FilterValue::Int(42),
1800        };
1801        let parent_pk = FilterValue::Int(7);
1802        op.execute(&engine, &parent_pk).await.unwrap();
1803
1804        let stmts = engine.statements();
1805        assert_eq!(stmts.len(), 1, "expected one UPDATE statement");
1806        let (sql, params) = &stmts[0];
1807        // Postgres dialect quotes idents with double quotes.
1808        assert!(sql.contains("UPDATE"), "got: {sql}");
1809        assert!(sql.contains("posts"), "got: {sql}");
1810        assert!(sql.contains("author_id"), "got: {sql}");
1811        assert!(sql.contains("SET"), "got: {sql}");
1812        assert!(sql.contains("WHERE"), "got: {sql}");
1813        assert!(sql.contains("$1"), "got: {sql}");
1814        assert!(sql.contains("$2"), "got: {sql}");
1815        assert_eq!(params, &vec![FilterValue::Int(7), FilterValue::Int(42)]);
1816    }
1817
1818    #[tokio::test]
1819    async fn nested_op_delete_many_with_filter_emits_fk_and_filter_clause() {
1820        let engine = RecordingEngine::new();
1821        let op = NestedWriteOp::DeleteMany {
1822            relation: "posts",
1823            target_table: "posts",
1824            foreign_key: "author_id",
1825            filter: Filter::Equals("published".into(), FilterValue::Bool(false)),
1826        };
1827        op.execute(&engine, &FilterValue::Int(7)).await.unwrap();
1828
1829        let stmts = engine.statements();
1830        assert_eq!(stmts.len(), 1);
1831        let (sql, params) = &stmts[0];
1832        // Filter placeholders must continue the $1 (parent FK) sequence — the
1833        // filter's first placeholder is $2, not $3.
1834        assert_eq!(
1835            sql,
1836            "DELETE FROM \"posts\" WHERE \"author_id\" = $1 AND (\"published\" = $2)"
1837        );
1838        assert!(!sql.contains("$3"), "off-by-one placeholder: {sql}");
1839        assert_eq!(params.len(), 2);
1840        assert!(matches!(params[0], FilterValue::Int(7)));
1841        assert!(matches!(params[1], FilterValue::Bool(false)));
1842    }
1843
1844    #[tokio::test]
1845    async fn nested_op_delete_many_with_filter_mysql_uses_positional_placeholders() {
1846        let engine = RecordingEngine::mysql();
1847        let op = NestedWriteOp::DeleteMany {
1848            relation: "posts",
1849            target_table: "posts",
1850            foreign_key: "author_id",
1851            filter: Filter::Equals("published".into(), FilterValue::Bool(false)),
1852        };
1853        op.execute(&engine, &FilterValue::Int(7)).await.unwrap();
1854
1855        let stmts = engine.statements();
1856        assert_eq!(stmts.len(), 1);
1857        let (sql, params) = &stmts[0];
1858        // MySQL: backtick-quoted idents, `?` placeholders, no $N anywhere.
1859        assert_eq!(
1860            sql,
1861            "DELETE FROM `posts` WHERE `author_id` = ? AND (`published` = ?)"
1862        );
1863        assert!(!sql.contains('$'), "MySQL must not emit $N: {sql}");
1864        assert_eq!(params.len(), 2);
1865        assert!(matches!(params[0], FilterValue::Int(7)));
1866        assert!(matches!(params[1], FilterValue::Bool(false)));
1867    }
1868
1869    #[tokio::test]
1870    async fn nested_op_delete_many_with_empty_filter_omits_and_clause() {
1871        let engine = RecordingEngine::new();
1872        let op = NestedWriteOp::DeleteMany {
1873            relation: "posts",
1874            target_table: "posts",
1875            foreign_key: "author_id",
1876            filter: Filter::None,
1877        };
1878        op.execute(&engine, &FilterValue::Int(7)).await.unwrap();
1879
1880        let stmts = engine.statements();
1881        let (sql, params) = &stmts[0];
1882        assert!(sql.contains("DELETE FROM"), "got: {sql}");
1883        assert!(
1884            !sql.contains("AND"),
1885            "should omit AND when filter empty: {sql}"
1886        );
1887        assert_eq!(params.len(), 1);
1888    }
1889
1890    #[tokio::test]
1891    async fn nested_op_delete_emits_delete_where_pk() {
1892        let engine = RecordingEngine::new();
1893        let op = NestedWriteOp::Delete {
1894            relation: "posts",
1895            target_table: "posts",
1896            target_pk: "id",
1897            pk: FilterValue::Int(42),
1898        };
1899        op.execute(&engine, &FilterValue::Int(7)).await.unwrap();
1900
1901        let stmts = engine.statements();
1902        assert_eq!(stmts.len(), 1);
1903        let (sql, params) = &stmts[0];
1904        assert!(sql.contains("DELETE FROM"), "got: {sql}");
1905        assert!(sql.contains("posts"), "got: {sql}");
1906        assert!(sql.contains("WHERE"), "got: {sql}");
1907        assert_eq!(params, &vec![FilterValue::Int(42)]);
1908    }
1909
1910    #[tokio::test]
1911    async fn nested_op_disconnect_emits_update_set_null() {
1912        let engine = RecordingEngine::new();
1913        let op = NestedWriteOp::Disconnect {
1914            relation: "posts",
1915            target_table: "posts",
1916            foreign_key: "author_id",
1917            target_pk: "id",
1918            pk: FilterValue::Int(42),
1919        };
1920        op.execute(&engine, &FilterValue::Int(7)).await.unwrap();
1921
1922        let stmts = engine.statements();
1923        assert_eq!(stmts.len(), 1);
1924        let (sql, params) = &stmts[0];
1925        assert!(sql.contains("UPDATE"), "got: {sql}");
1926        assert!(sql.contains("posts"), "got: {sql}");
1927        assert!(sql.contains("author_id"), "got: {sql}");
1928        assert!(sql.contains("NULL"), "got: {sql}");
1929        assert!(sql.contains("WHERE"), "got: {sql}");
1930        assert_eq!(params, &vec![FilterValue::Int(42)]);
1931    }
1932
1933    #[tokio::test]
1934    async fn nested_op_update_plain_set() {
1935        use crate::inputs::WriteOp;
1936        let engine = RecordingEngine::new();
1937        let op = NestedWriteOp::Update {
1938            relation: "posts",
1939            target_table: "posts",
1940            target_pk: "id",
1941            pk: FilterValue::Int(42),
1942            payload: vec![(
1943                "title".to_string(),
1944                WriteOp::Set(FilterValue::String("renamed".to_string())),
1945            )],
1946        };
1947        op.execute(&engine, &FilterValue::Int(7)).await.unwrap();
1948
1949        let stmts = engine.statements();
1950        assert_eq!(stmts.len(), 1);
1951        let (sql, params) = &stmts[0];
1952        assert!(sql.contains("UPDATE"), "got: {sql}");
1953        assert!(sql.contains("posts"), "got: {sql}");
1954        assert!(sql.contains("title"), "got: {sql}");
1955        assert!(sql.contains("SET"), "got: {sql}");
1956        assert!(sql.contains("WHERE"), "got: {sql}");
1957        assert!(sql.contains("$1"), "got: {sql}");
1958        assert!(sql.contains("$2"), "got: {sql}");
1959        assert_eq!(params.len(), 2);
1960        assert!(matches!(params[0], FilterValue::String(_)));
1961        assert_eq!(params[1], FilterValue::Int(42));
1962    }
1963
1964    #[tokio::test]
1965    async fn nested_op_update_increment() {
1966        use crate::inputs::WriteOp;
1967        let engine = RecordingEngine::new();
1968        let op = NestedWriteOp::Update {
1969            relation: "posts",
1970            target_table: "posts",
1971            target_pk: "id",
1972            pk: FilterValue::Int(42),
1973            payload: vec![("views".to_string(), WriteOp::Increment(FilterValue::Int(1)))],
1974        };
1975        op.execute(&engine, &FilterValue::Int(7)).await.unwrap();
1976
1977        let stmts = engine.statements();
1978        let (sql, _) = &stmts[0];
1979        // Postgres dialect quotes idents — fragment will read `"views" = "views" + $1`.
1980        assert!(sql.contains("+"), "got: {sql}");
1981        assert!(sql.contains("views"), "got: {sql}");
1982    }
1983
1984    #[tokio::test]
1985    async fn nested_op_update_mixed_set_and_increment() {
1986        use crate::inputs::WriteOp;
1987        let engine = RecordingEngine::new();
1988        let op = NestedWriteOp::Update {
1989            relation: "posts",
1990            target_table: "posts",
1991            target_pk: "id",
1992            pk: FilterValue::Int(42),
1993            payload: vec![
1994                (
1995                    "title".to_string(),
1996                    WriteOp::Set(FilterValue::String("renamed".to_string())),
1997                ),
1998                ("views".to_string(), WriteOp::Increment(FilterValue::Int(1))),
1999            ],
2000        };
2001        op.execute(&engine, &FilterValue::Int(7)).await.unwrap();
2002
2003        let stmts = engine.statements();
2004        let (sql, params) = &stmts[0];
2005        assert!(sql.contains("title"), "got: {sql}");
2006        assert!(sql.contains("views"), "got: {sql}");
2007        assert!(sql.contains("+"), "got: {sql}");
2008        // 2 SET params + 1 PK = 3 placeholders.
2009        assert!(sql.contains("$3"), "got: {sql}");
2010        assert_eq!(params.len(), 3);
2011    }
2012
2013    #[tokio::test]
2014    async fn nested_op_update_empty_payload_is_noop() {
2015        let engine = RecordingEngine::new();
2016        let op = NestedWriteOp::Update {
2017            relation: "posts",
2018            target_table: "posts",
2019            target_pk: "id",
2020            pk: FilterValue::Int(42),
2021            payload: vec![],
2022        };
2023        op.execute(&engine, &FilterValue::Int(7)).await.unwrap();
2024        assert!(
2025            engine.statements().is_empty(),
2026            "empty payload should emit no SQL"
2027        );
2028    }
2029
2030    #[tokio::test]
2031    async fn nested_op_update_many_with_filter() {
2032        use crate::inputs::WriteOp;
2033        let engine = RecordingEngine::new();
2034        let op = NestedWriteOp::UpdateMany {
2035            relation: "posts",
2036            target_table: "posts",
2037            foreign_key: "author_id",
2038            filter: Filter::Equals("published".into(), FilterValue::Bool(false)),
2039            payload: vec![("views".to_string(), WriteOp::Set(FilterValue::Int(0)))],
2040        };
2041        op.execute(&engine, &FilterValue::Int(7)).await.unwrap();
2042
2043        let stmts = engine.statements();
2044        assert_eq!(stmts.len(), 1);
2045        let (sql, params) = &stmts[0];
2046        // SET value is $1, parent FK is $2, filter continues at $3 (no skipped
2047        // slot between the FK placeholder and the filter).
2048        assert_eq!(
2049            sql,
2050            "UPDATE \"posts\" SET \"views\" = $1 WHERE \"author_id\" = $2 AND (\"published\" = $3)"
2051        );
2052        assert!(!sql.contains("$4"), "skipped placeholder slot: {sql}");
2053        // payload value + FK + filter value = 3 params
2054        assert_eq!(params.len(), 3);
2055        assert_eq!(params[0], FilterValue::Int(0));
2056        assert_eq!(params[1], FilterValue::Int(7));
2057        assert_eq!(params[2], FilterValue::Bool(false));
2058    }
2059
2060    #[tokio::test]
2061    async fn nested_op_update_many_with_filter_mysql_uses_positional_placeholders() {
2062        use crate::inputs::WriteOp;
2063        let engine = RecordingEngine::mysql();
2064        let op = NestedWriteOp::UpdateMany {
2065            relation: "posts",
2066            target_table: "posts",
2067            foreign_key: "author_id",
2068            filter: Filter::Equals("published".into(), FilterValue::Bool(false)),
2069            payload: vec![("views".to_string(), WriteOp::Set(FilterValue::Int(0)))],
2070        };
2071        op.execute(&engine, &FilterValue::Int(7)).await.unwrap();
2072
2073        let stmts = engine.statements();
2074        assert_eq!(stmts.len(), 1);
2075        let (sql, params) = &stmts[0];
2076        // MySQL: backtick-quoted idents, `?` placeholders, no $N anywhere.
2077        assert_eq!(
2078            sql,
2079            "UPDATE `posts` SET `views` = ? WHERE `author_id` = ? AND (`published` = ?)"
2080        );
2081        assert!(!sql.contains('$'), "MySQL must not emit $N: {sql}");
2082        assert_eq!(params.len(), 3);
2083        assert_eq!(params[0], FilterValue::Int(0));
2084        assert_eq!(params[1], FilterValue::Int(7));
2085        assert_eq!(params[2], FilterValue::Bool(false));
2086    }
2087
2088    #[tokio::test]
2089    async fn nested_op_update_many_with_empty_filter() {
2090        use crate::inputs::WriteOp;
2091        let engine = RecordingEngine::new();
2092        let op = NestedWriteOp::UpdateMany {
2093            relation: "posts",
2094            target_table: "posts",
2095            foreign_key: "author_id",
2096            filter: Filter::None,
2097            payload: vec![("views".to_string(), WriteOp::Set(FilterValue::Int(0)))],
2098        };
2099        op.execute(&engine, &FilterValue::Int(7)).await.unwrap();
2100
2101        let stmts = engine.statements();
2102        let (sql, params) = &stmts[0];
2103        assert!(sql.contains("UPDATE"), "got: {sql}");
2104        assert!(
2105            !sql.contains("AND"),
2106            "should omit AND when filter empty: {sql}"
2107        );
2108        // payload value + FK = 2 params
2109        assert_eq!(params.len(), 2);
2110    }
2111
2112    #[test]
2113    fn test_nested_upsert_data() {
2114        let create: NestedCreateData<TestModel> =
2115            NestedCreateData::from_pairs([("title", FilterValue::String("New".to_string()))]);
2116
2117        let upsert: NestedUpsertData<TestModel> = NestedUpsertData::new(
2118            Filter::Equals("id".into(), FilterValue::Int(1)),
2119            create,
2120            vec![(
2121                "title".to_string(),
2122                FilterValue::String("Updated".to_string()),
2123            )],
2124        );
2125
2126        assert!(matches!(upsert.filter, Filter::Equals(..)));
2127        assert_eq!(upsert.create.data.len(), 1);
2128        assert_eq!(upsert.update.len(), 1);
2129    }
2130
2131    #[tokio::test]
2132    async fn nested_op_upsert_single_statement_on_postgres() {
2133        use crate::inputs::WriteOp;
2134        // Postgres dialect supports `ON CONFLICT (...) DO UPDATE SET ...`,
2135        // so the executor must collapse the two-statement form into a
2136        // single `INSERT ... ON CONFLICT` statement.
2137        let engine = RecordingEngine::new();
2138        let op = NestedWriteOp::Upsert {
2139            relation: "posts",
2140            target_table: "posts",
2141            foreign_key: "author_id",
2142            target_pk: "id",
2143            pk: FilterValue::Int(99),
2144            create_payload: vec![("title".to_string(), FilterValue::String("new".to_string()))],
2145            update_payload: vec![("views".to_string(), WriteOp::Increment(FilterValue::Int(1)))],
2146        };
2147        op.execute(&engine, &FilterValue::Int(7)).await.unwrap();
2148
2149        let stmts = engine.statements();
2150        assert_eq!(
2151            stmts.len(),
2152            1,
2153            "expected a single-statement upsert; got {stmts:#?}"
2154        );
2155        let (sql, params) = &stmts[0];
2156        assert!(sql.contains("INSERT INTO"), "got: {sql}");
2157        assert!(sql.contains("posts"), "got: {sql}");
2158        assert!(sql.contains("ON CONFLICT (\"id\")"), "got: {sql}");
2159        assert!(sql.contains("DO UPDATE SET"), "got: {sql}");
2160        // INSERT supplies $1 (title), $2 (author_id=parent_pk);
2161        // SET fragment uses $3 (views increment).
2162        assert!(
2163            sql.contains("VALUES ($1, $2)"),
2164            "INSERT VALUES placeholders: {sql}"
2165        );
2166        assert!(sql.contains("$3"), "got: {sql}");
2167        // Two insert values + one SET value.
2168        assert_eq!(params.len(), 3);
2169        assert_eq!(params[0], FilterValue::String("new".to_string()));
2170        assert_eq!(params[1], FilterValue::Int(7));
2171        assert_eq!(params[2], FilterValue::Int(1));
2172    }
2173
2174    #[tokio::test]
2175    async fn nested_op_upsert_two_statement_fallback_on_mssql_update_path() {
2176        use crate::inputs::WriteOp;
2177        // MSSQL's `upsert_clause` returns empty → the executor must
2178        // fall back to the existing two-statement UPDATE-then-INSERT
2179        // form. Default affected-rows = 1 means UPDATE matches a row
2180        // and the INSERT phase must not run.
2181        let engine = RecordingEngine::mssql();
2182        let op = NestedWriteOp::Upsert {
2183            relation: "posts",
2184            target_table: "posts",
2185            foreign_key: "author_id",
2186            target_pk: "id",
2187            pk: FilterValue::Int(99),
2188            create_payload: vec![("title".to_string(), FilterValue::String("new".to_string()))],
2189            update_payload: vec![("views".to_string(), WriteOp::Increment(FilterValue::Int(1)))],
2190        };
2191        op.execute(&engine, &FilterValue::Int(7)).await.unwrap();
2192
2193        let stmts = engine.statements();
2194        assert_eq!(
2195            stmts.len(),
2196            1,
2197            "expected only the UPDATE — INSERT should not have run"
2198        );
2199        let (sql, update_params) = &stmts[0];
2200        assert!(sql.starts_with("UPDATE"), "got: {sql}");
2201        assert!(!sql.contains("ON CONFLICT"), "got: {sql}");
2202        assert!(!sql.contains("ON DUPLICATE"), "got: {sql}");
2203        // MSSQL uses bracket-quoted identifiers and @P-prefixed placeholders.
2204        assert!(sql.contains("[posts]"), "got: {sql}");
2205        assert!(sql.contains("@P1"), "SET clause should use @P1: {sql}");
2206        assert!(sql.contains("@P2"), "WHERE clause should use @P2: {sql}");
2207        // Two params: the increment value ($1) and the pk ($2).
2208        assert_eq!(update_params.len(), 2);
2209        assert_eq!(update_params[0], FilterValue::Int(1)); // increment value
2210        assert_eq!(update_params[1], FilterValue::Int(99)); // pk
2211    }
2212
2213    #[tokio::test]
2214    async fn nested_op_upsert_two_statement_fallback_on_mssql_insert_path() {
2215        use crate::inputs::WriteOp;
2216        // First execute_raw (UPDATE) returns 0; the executor must
2217        // proceed to emit a separate INSERT.
2218        let engine = RecordingEngine::mssql_with_affected(vec![0, 1]);
2219        let op = NestedWriteOp::Upsert {
2220            relation: "posts",
2221            target_table: "posts",
2222            foreign_key: "author_id",
2223            target_pk: "id",
2224            pk: FilterValue::Int(99),
2225            create_payload: vec![("title".to_string(), FilterValue::String("new".to_string()))],
2226            update_payload: vec![("views".to_string(), WriteOp::Increment(FilterValue::Int(1)))],
2227        };
2228        op.execute(&engine, &FilterValue::Int(7)).await.unwrap();
2229
2230        let stmts = engine.statements();
2231        assert_eq!(stmts.len(), 2, "expected UPDATE then INSERT");
2232        let (update_sql, _) = &stmts[0];
2233        assert!(update_sql.starts_with("UPDATE"), "got: {update_sql}");
2234        let (insert_sql, insert_params) = &stmts[1];
2235        assert!(insert_sql.starts_with("INSERT INTO"), "got: {insert_sql}");
2236        assert!(!insert_sql.contains("ON CONFLICT"), "got: {insert_sql}");
2237        assert!(insert_sql.contains("[posts]"), "got: {insert_sql}");
2238        assert!(insert_sql.contains("[author_id]"), "got: {insert_sql}");
2239        assert_eq!(insert_params.len(), 2);
2240        assert_eq!(insert_params[0], FilterValue::String("new".to_string()));
2241        assert_eq!(insert_params[1], FilterValue::Int(7));
2242    }
2243
2244    #[tokio::test]
2245    async fn nested_op_connect_or_create_connect_path_when_affected() {
2246        // Default RecordingEngine returns 1 from execute_raw, so the
2247        // UPDATE matches at least one row and the INSERT phase must not
2248        // run.
2249        let engine = RecordingEngine::with_affected(vec![1]);
2250        let op = NestedWriteOp::ConnectOrCreate {
2251            relation: "posts",
2252            target_table: "posts",
2253            foreign_key: "author_id",
2254            where_filter: Filter::Equals("id".into(), FilterValue::Int(42)),
2255            create_payload: vec![(
2256                "title".to_string(),
2257                FilterValue::String("fallback".to_string()),
2258            )],
2259        };
2260        op.execute(&engine, &FilterValue::Int(7)).await.unwrap();
2261
2262        let stmts = engine.statements();
2263        assert_eq!(
2264            stmts.len(),
2265            1,
2266            "expected only the UPDATE — INSERT should not have run"
2267        );
2268        let (sql, params) = &stmts[0];
2269        // Parent FK is bound at $1; the where filter must continue at $2,
2270        // not $3.
2271        assert_eq!(
2272            sql,
2273            "UPDATE \"posts\" SET \"author_id\" = $1 WHERE \"id\" = $2"
2274        );
2275        assert!(!sql.contains("$3"), "off-by-one placeholder: {sql}");
2276        assert!(!sql.contains("INSERT"), "got: {sql}");
2277        // params: parent_pk ($1) + filter value ($2)
2278        assert_eq!(params.len(), 2);
2279        assert_eq!(params[0], FilterValue::Int(7));
2280        assert_eq!(params[1], FilterValue::Int(42));
2281    }
2282
2283    #[tokio::test]
2284    async fn nested_op_connect_or_create_connect_path_mysql_uses_positional_placeholders() {
2285        let engine = RecordingEngine::mysql();
2286        let op = NestedWriteOp::ConnectOrCreate {
2287            relation: "posts",
2288            target_table: "posts",
2289            foreign_key: "author_id",
2290            where_filter: Filter::Equals("id".into(), FilterValue::Int(42)),
2291            create_payload: vec![(
2292                "title".to_string(),
2293                FilterValue::String("fallback".to_string()),
2294            )],
2295        };
2296        op.execute(&engine, &FilterValue::Int(7)).await.unwrap();
2297
2298        let stmts = engine.statements();
2299        assert_eq!(stmts.len(), 1);
2300        let (sql, params) = &stmts[0];
2301        // MySQL: backtick-quoted idents, `?` placeholders, no $N anywhere.
2302        assert_eq!(sql, "UPDATE `posts` SET `author_id` = ? WHERE `id` = ?");
2303        assert!(!sql.contains('$'), "MySQL must not emit $N: {sql}");
2304        assert_eq!(params.len(), 2);
2305        assert_eq!(params[0], FilterValue::Int(7));
2306        assert_eq!(params[1], FilterValue::Int(42));
2307    }
2308
2309    #[tokio::test]
2310    async fn nested_op_connect_or_create_create_path_when_zero_affected() {
2311        // UPDATE returns 0 (no matching row) → executor must emit the
2312        // INSERT with the FK spliced in. Second call (the INSERT)
2313        // returns 1.
2314        let engine = RecordingEngine::with_affected(vec![0, 1]);
2315        let op = NestedWriteOp::ConnectOrCreate {
2316            relation: "posts",
2317            target_table: "posts",
2318            foreign_key: "author_id",
2319            where_filter: Filter::Equals("id".into(), FilterValue::Int(42)),
2320            create_payload: vec![(
2321                "title".to_string(),
2322                FilterValue::String("fallback".to_string()),
2323            )],
2324        };
2325        op.execute(&engine, &FilterValue::Int(7)).await.unwrap();
2326
2327        let stmts = engine.statements();
2328        assert_eq!(stmts.len(), 2);
2329        let (update_sql, _) = &stmts[0];
2330        assert!(update_sql.contains("UPDATE"), "got: {update_sql}");
2331        let (insert_sql, insert_params) = &stmts[1];
2332        assert!(insert_sql.contains("INSERT INTO"), "got: {insert_sql}");
2333        assert!(insert_sql.contains("posts"), "got: {insert_sql}");
2334        assert!(insert_sql.contains("title"), "got: {insert_sql}");
2335        assert!(insert_sql.contains("author_id"), "got: {insert_sql}");
2336        // create payload value + FK (parent_pk) = 2 params; parent_pk
2337        // must be last so the FK column lines up with $2.
2338        assert_eq!(insert_params.len(), 2);
2339        assert_eq!(
2340            insert_params[0],
2341            FilterValue::String("fallback".to_string())
2342        );
2343        assert_eq!(insert_params[1], FilterValue::Int(7));
2344    }
2345
2346    #[tokio::test]
2347    async fn nested_op_set_with_empty_list_clears_all_children() {
2348        // Empty set_pks → one UPDATE with `WHERE fk = $1`, no NOT IN clause,
2349        // no connect step.
2350        let engine = RecordingEngine::new();
2351        let op = NestedWriteOp::Set {
2352            relation: "posts",
2353            target_table: "posts",
2354            foreign_key: "author_id",
2355            target_pk: "id",
2356            set_pks: vec![],
2357        };
2358        op.execute(&engine, &FilterValue::Int(7)).await.unwrap();
2359
2360        let stmts = engine.statements();
2361        assert_eq!(stmts.len(), 1, "expected only the disconnect-all UPDATE");
2362        let (sql, params) = &stmts[0];
2363        assert!(sql.contains("UPDATE"), "got: {sql}");
2364        assert!(sql.contains("posts"), "got: {sql}");
2365        assert!(sql.contains("author_id"), "got: {sql}");
2366        assert!(sql.contains("= NULL"), "got: {sql}");
2367        assert!(!sql.contains("NOT IN"), "got: {sql}");
2368        assert!(!sql.contains(" IN ("), "got: {sql}");
2369        // Only the parent_pk param.
2370        assert_eq!(params.len(), 1);
2371        assert_eq!(params[0], FilterValue::Int(7));
2372    }
2373
2374    #[tokio::test]
2375    async fn nested_op_set_with_non_empty_list_emits_disconnect_then_connect() {
2376        let engine = RecordingEngine::new();
2377        let op = NestedWriteOp::Set {
2378            relation: "posts",
2379            target_table: "posts",
2380            foreign_key: "author_id",
2381            target_pk: "id",
2382            set_pks: vec![
2383                FilterValue::Int(1),
2384                FilterValue::Int(2),
2385                FilterValue::Int(3),
2386            ],
2387        };
2388        op.execute(&engine, &FilterValue::Int(7)).await.unwrap();
2389
2390        let stmts = engine.statements();
2391        assert_eq!(stmts.len(), 2);
2392        let (disconnect_sql, disconnect_params) = &stmts[0];
2393        assert!(disconnect_sql.contains("UPDATE"), "got: {disconnect_sql}");
2394        assert!(disconnect_sql.contains("posts"), "got: {disconnect_sql}");
2395        assert!(
2396            disconnect_sql.contains("author_id"),
2397            "got: {disconnect_sql}"
2398        );
2399        assert!(disconnect_sql.contains("= NULL"), "got: {disconnect_sql}");
2400        assert!(disconnect_sql.contains("NOT IN"), "got: {disconnect_sql}");
2401        // params: parent_pk ($1) + 3 set pks ($2..$4)
2402        assert_eq!(disconnect_params.len(), 4);
2403        assert_eq!(disconnect_params[0], FilterValue::Int(7));
2404        assert_eq!(disconnect_params[1], FilterValue::Int(1));
2405        assert_eq!(disconnect_params[2], FilterValue::Int(2));
2406        assert_eq!(disconnect_params[3], FilterValue::Int(3));
2407
2408        let (connect_sql, connect_params) = &stmts[1];
2409        assert!(connect_sql.contains("UPDATE"), "got: {connect_sql}");
2410        assert!(connect_sql.contains("posts"), "got: {connect_sql}");
2411        assert!(connect_sql.contains("author_id"), "got: {connect_sql}");
2412        assert!(connect_sql.contains(" IN ("), "got: {connect_sql}");
2413        assert!(!connect_sql.contains("NOT IN"), "got: {connect_sql}");
2414        // params: parent_pk ($1) + 3 set pks ($2..$4)
2415        assert_eq!(connect_params.len(), 4);
2416        assert_eq!(connect_params[0], FilterValue::Int(7));
2417        assert_eq!(connect_params[1], FilterValue::Int(1));
2418        assert_eq!(connect_params[2], FilterValue::Int(2));
2419        assert_eq!(connect_params[3], FilterValue::Int(3));
2420    }
2421
2422    #[tokio::test]
2423    async fn nested_op_set_with_single_element_uses_single_placeholder_in_lists() {
2424        let engine = RecordingEngine::new();
2425        let op = NestedWriteOp::Set {
2426            relation: "posts",
2427            target_table: "posts",
2428            foreign_key: "author_id",
2429            target_pk: "id",
2430            set_pks: vec![FilterValue::Int(5)],
2431        };
2432        op.execute(&engine, &FilterValue::Int(7)).await.unwrap();
2433
2434        let stmts = engine.statements();
2435        assert_eq!(stmts.len(), 2);
2436        let (disconnect_sql, _) = &stmts[0];
2437        // The NOT IN list has one placeholder.
2438        assert!(
2439            disconnect_sql.contains("NOT IN ($2)"),
2440            "got: {disconnect_sql}"
2441        );
2442        let (connect_sql, _) = &stmts[1];
2443        assert!(connect_sql.contains(" IN ($2)"), "got: {connect_sql}");
2444        assert!(!connect_sql.contains("NOT IN"), "got: {connect_sql}");
2445    }
2446
2447    #[tokio::test]
2448    async fn nested_op_set_disconnect_clears_only_current_parents_children() {
2449        // The disconnect UPDATE must be scoped to `WHERE fk = $parent AND ...`
2450        // — without this clause, we'd clear children belonging to other parents.
2451        let engine = RecordingEngine::new();
2452        let op = NestedWriteOp::Set {
2453            relation: "posts",
2454            target_table: "posts",
2455            foreign_key: "author_id",
2456            target_pk: "id",
2457            set_pks: vec![FilterValue::Int(1)],
2458        };
2459        op.execute(&engine, &FilterValue::Int(7)).await.unwrap();
2460
2461        let stmts = engine.statements();
2462        let (disconnect_sql, _) = &stmts[0];
2463        // The disconnect must scope to the parent's children.
2464        assert!(
2465            disconnect_sql.contains("author_id\" = $1"),
2466            "expected `author_id = $1` clause; got: {disconnect_sql}"
2467        );
2468    }
2469
2470    #[tokio::test]
2471    async fn nested_op_connect_or_create_rejects_empty_where() {
2472        // Filter::None would lower to `WHERE TRUE` and rewrite every row.
2473        // The executor must reject this defensively before issuing SQL.
2474        let engine = RecordingEngine::new();
2475        let op = NestedWriteOp::ConnectOrCreate {
2476            relation: "posts",
2477            target_table: "posts",
2478            foreign_key: "author_id",
2479            where_filter: Filter::None,
2480            create_payload: vec![(
2481                "title".to_string(),
2482                FilterValue::String("fallback".to_string()),
2483            )],
2484        };
2485        let err = op
2486            .execute(&engine, &FilterValue::Int(7))
2487            .await
2488            .expect_err("empty where must be rejected");
2489        let op_ctx = err.context.operation.clone().unwrap_or_default();
2490        assert!(op_ctx.contains("ConnectOrCreate"), "got: {op_ctx}");
2491        // No SQL should have been emitted.
2492        assert!(engine.statements().is_empty());
2493    }
2494
2495    #[tokio::test]
2496    async fn nested_op_upsert_single_statement_empty_update_payload_emits_do_nothing() {
2497        // PG dialect → single-statement path. Empty update_payload should emit
2498        // INSERT INTO ... VALUES (...) ON CONFLICT (target_pk) DO NOTHING
2499        // (per the upsert_do_nothing_clause helper added in 8f01cab).
2500        let engine = RecordingEngine::new();
2501        let op = NestedWriteOp::Upsert {
2502            relation: "posts",
2503            target_table: "posts",
2504            foreign_key: "author_id",
2505            target_pk: "id",
2506            pk: FilterValue::Int(99),
2507            create_payload: vec![("title".to_string(), FilterValue::String("new".into()))],
2508            update_payload: vec![],
2509        };
2510        op.execute(&engine, &FilterValue::Int(7)).await.unwrap();
2511
2512        let stmts = engine.statements();
2513        assert_eq!(
2514            stmts.len(),
2515            1,
2516            "expected one INSERT ... DO NOTHING; got {stmts:#?}"
2517        );
2518        let (sql, params) = &stmts[0];
2519        assert!(sql.starts_with("INSERT INTO"), "got: {sql}");
2520        assert!(sql.contains("posts"), "got: {sql}");
2521        assert!(sql.contains("ON CONFLICT (\"id\")"), "got: {sql}");
2522        assert!(sql.contains("DO NOTHING"), "got: {sql}");
2523        // Two insert values (title + FK); no SET params.
2524        assert_eq!(params.len(), 2);
2525        assert_eq!(params[0], FilterValue::String("new".into()));
2526        assert_eq!(params[1], FilterValue::Int(7));
2527    }
2528
2529    #[tokio::test]
2530    async fn nested_op_upsert_two_statement_fallback_empty_update_payload_emits_bare_insert() {
2531        // MSSQL dialect → two-statement fallback. On the fallback path,
2532        // empty update_payload skips the UPDATE entirely and emits a bare INSERT
2533        // (if the row already exists the engine surfaces a PK unique violation,
2534        // which is correct insert-or-fail semantics on dialects without native
2535        // upsert syntax). See production code comment at the
2536        // `if update_payload.is_empty()` guard in the two-statement fallback path.
2537        let engine = RecordingEngine::mssql();
2538        let op = NestedWriteOp::Upsert {
2539            relation: "posts",
2540            target_table: "posts",
2541            foreign_key: "author_id",
2542            target_pk: "id",
2543            pk: FilterValue::Int(99),
2544            create_payload: vec![("title".to_string(), FilterValue::String("new".into()))],
2545            update_payload: vec![],
2546        };
2547        op.execute(&engine, &FilterValue::Int(7)).await.unwrap();
2548
2549        let stmts = engine.statements();
2550        assert_eq!(stmts.len(), 1, "expected one bare INSERT; got {stmts:#?}");
2551        let (sql, params) = &stmts[0];
2552        assert!(sql.starts_with("INSERT INTO"), "got: {sql}");
2553        assert!(sql.contains("[posts]"), "got: {sql}");
2554        assert!(!sql.contains("ON CONFLICT"), "got: {sql}");
2555        assert_eq!(params.len(), 2);
2556        assert_eq!(params[0], FilterValue::String("new".into()));
2557        assert_eq!(params[1], FilterValue::Int(7));
2558    }
2559
2560    #[tokio::test]
2561    async fn nested_op_upsert_single_statement_all_unset_update_payload() {
2562        use crate::inputs::WriteOp;
2563        // Unset emits `col = NULL` literal (no placeholder consumed).
2564        // The placeholder accounting must remain consistent: only INSERT
2565        // values consume placeholders.
2566        let engine = RecordingEngine::new();
2567        let op = NestedWriteOp::Upsert {
2568            relation: "posts",
2569            target_table: "posts",
2570            foreign_key: "author_id",
2571            target_pk: "id",
2572            pk: FilterValue::Int(99),
2573            create_payload: vec![("title".to_string(), FilterValue::String("new".into()))],
2574            update_payload: vec![("deleted_at".to_string(), WriteOp::Unset)],
2575        };
2576        op.execute(&engine, &FilterValue::Int(7)).await.unwrap();
2577
2578        let stmts = engine.statements();
2579        assert_eq!(
2580            stmts.len(),
2581            1,
2582            "expected one INSERT...ON CONFLICT statement; got {stmts:#?}"
2583        );
2584        let (sql, params) = &stmts[0];
2585        assert!(sql.starts_with("INSERT INTO"), "got: {sql}");
2586        assert!(
2587            sql.contains("ON CONFLICT (\"id\") DO UPDATE SET"),
2588            "got: {sql}"
2589        );
2590        assert!(sql.contains("\"deleted_at\" = NULL"), "got: {sql}");
2591        // Only INSERT params: title + FK. No SET param (Unset consumes none).
2592        assert_eq!(params.len(), 2);
2593        assert_eq!(params[0], FilterValue::String("new".into()));
2594        assert_eq!(params[1], FilterValue::Int(7));
2595    }
2596
2597    #[tokio::test]
2598    async fn nested_op_upsert_empty_create_payload_returns_invalid_input() {
2599        use crate::inputs::WriteOp;
2600        let engine = RecordingEngine::new();
2601        let op = NestedWriteOp::Upsert {
2602            relation: "posts",
2603            target_table: "posts",
2604            foreign_key: "author_id",
2605            target_pk: "id",
2606            pk: FilterValue::Int(99),
2607            create_payload: vec![],
2608            update_payload: vec![("views".to_string(), WriteOp::Increment(FilterValue::Int(1)))],
2609        };
2610        let err = op.execute(&engine, &FilterValue::Int(7)).await.unwrap_err();
2611        assert_eq!(
2612            err.code,
2613            crate::error::ErrorCode::InvalidParameter,
2614            "expected InvalidParameter (invalid_input), got: {err:?}"
2615        );
2616        let msg = format!("{err}");
2617        assert!(
2618            msg.contains("create_payload") || msg.contains("create column"),
2619            "msg: {msg}"
2620        );
2621    }
2622
2623    #[tokio::test]
2624    async fn nested_op_upsert_on_notsql_returns_unsupported_not_panic() {
2625        use crate::inputs::WriteOp;
2626        let engine = RecordingEngine::notsql();
2627        let op = NestedWriteOp::Upsert {
2628            relation: "posts",
2629            target_table: "posts",
2630            foreign_key: "author_id",
2631            target_pk: "id",
2632            pk: FilterValue::Int(99),
2633            create_payload: vec![("title".to_string(), FilterValue::String("x".into()))],
2634            update_payload: vec![("views".to_string(), WriteOp::Increment(FilterValue::Int(1)))],
2635        };
2636        let err = op.execute(&engine, &FilterValue::Int(7)).await.unwrap_err();
2637        let msg = format!("{err}");
2638        assert!(
2639            msg.contains("Upsert is not supported")
2640                || msg.contains("unsupported")
2641                || msg.contains("Unsupported"),
2642            "msg: {msg}"
2643        );
2644        assert_eq!(
2645            engine.statements().len(),
2646            0,
2647            "no SQL should have been emitted"
2648        );
2649    }
2650}