1#![allow(dead_code)]
2
3use 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#[derive(Debug, Clone)]
65pub enum NestedWrite<T: Model> {
66 Create(Vec<NestedCreateData<T>>),
68 CreateOrConnect(Vec<NestedCreateOrConnectData<T>>),
70 Connect(Vec<Filter>),
72 Disconnect(Vec<Filter>),
74 Set(Vec<Filter>),
76 Delete(Vec<Filter>),
78 Update(Vec<NestedUpdateData<T>>),
80 Upsert(Vec<NestedUpsertData<T>>),
82 UpdateMany(NestedUpdateManyData<T>),
84 DeleteMany(Filter),
86}
87
88impl<T: Model> NestedWrite<T> {
89 pub fn create(data: NestedCreateData<T>) -> Self {
91 Self::Create(vec![data])
92 }
93
94 pub fn create_many(data: Vec<NestedCreateData<T>>) -> Self {
96 Self::Create(data)
97 }
98
99 pub fn connect_one(filter: impl Into<Filter>) -> Self {
101 Self::Connect(vec![filter.into()])
102 }
103
104 pub fn connect(filters: Vec<impl Into<Filter>>) -> Self {
106 Self::Connect(filters.into_iter().map(Into::into).collect())
107 }
108
109 pub fn disconnect_one(filter: impl Into<Filter>) -> Self {
111 Self::Disconnect(vec![filter.into()])
112 }
113
114 pub fn disconnect(filters: Vec<impl Into<Filter>>) -> Self {
116 Self::Disconnect(filters.into_iter().map(Into::into).collect())
117 }
118
119 pub fn set(filters: Vec<impl Into<Filter>>) -> Self {
121 Self::Set(filters.into_iter().map(Into::into).collect())
122 }
123
124 pub fn delete(filters: Vec<impl Into<Filter>>) -> Self {
126 Self::Delete(filters.into_iter().map(Into::into).collect())
127 }
128
129 pub fn delete_many(filter: impl Into<Filter>) -> Self {
131 Self::DeleteMany(filter.into())
132 }
133}
134
135#[derive(Debug, Clone)]
137pub struct NestedCreateData<T: Model> {
138 pub data: Vec<(String, FilterValue)>,
140 _model: PhantomData<T>,
142}
143
144impl<T: Model> NestedCreateData<T> {
145 pub fn new(data: Vec<(String, FilterValue)>) -> Self {
147 Self {
148 data,
149 _model: PhantomData,
150 }
151 }
152
153 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#[derive(Debug, Clone)]
174pub struct NestedCreateOrConnectData<T: Model> {
175 pub filter: Filter,
177 pub create: NestedCreateData<T>,
179}
180
181impl<T: Model> NestedCreateOrConnectData<T> {
182 pub fn new(filter: impl Into<Filter>, create: NestedCreateData<T>) -> Self {
184 Self {
185 filter: filter.into(),
186 create,
187 }
188 }
189}
190
191#[derive(Debug, Clone)]
193pub struct NestedUpdateData<T: Model> {
194 pub filter: Filter,
196 pub data: Vec<(String, FilterValue)>,
198 _model: PhantomData<T>,
200}
201
202impl<T: Model> NestedUpdateData<T> {
203 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 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#[derive(Debug, Clone)]
229pub struct NestedUpsertData<T: Model> {
230 pub filter: Filter,
232 pub create: NestedCreateData<T>,
234 pub update: Vec<(String, FilterValue)>,
236 _model: PhantomData<T>,
238}
239
240impl<T: Model> NestedUpsertData<T> {
241 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#[derive(Debug, Clone)]
258pub struct NestedUpdateManyData<T: Model> {
259 pub filter: Filter,
261 pub data: Vec<(String, FilterValue)>,
263 _model: PhantomData<T>,
265}
266
267impl<T: Model> NestedUpdateManyData<T> {
268 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#[derive(Debug)]
285pub struct NestedWriteBuilder {
286 parent_table: String,
288 parent_pk: Vec<String>,
290 related_table: String,
292 foreign_key: String,
294 is_one_to_many: bool,
296 join_table: Option<JoinTableInfo>,
298}
299
300#[derive(Debug, Clone)]
302pub struct JoinTableInfo {
303 pub table_name: String,
305 pub parent_column: String,
307 pub related_column: String,
309}
310
311impl NestedWriteBuilder {
312 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 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(), is_one_to_many: false,
342 join_table: Some(join_table),
343 }
344 }
345
346 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 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 filter in filters {
372 let (where_sql, mut params) = filter.to_sql(0, &crate::dialect::Postgres);
373
374 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 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 let _ = select_sql;
397 }
398 }
399
400 statements
401 }
402
403 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 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 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 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 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 statements.extend(self.build_connect_sql::<T>(parent_id, filters));
476
477 statements
478 }
479
480 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("e_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 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#[derive(Debug, Default)]
551pub struct NestedWriteOperations {
552 pub pre_statements: Vec<(String, Vec<FilterValue>)>,
554 pub post_statements: Vec<(String, Vec<FilterValue>)>,
556}
557
558impl NestedWriteOperations {
559 pub fn new() -> Self {
561 Self::default()
562 }
563
564 pub fn add_pre(&mut self, sql: String, params: Vec<FilterValue>) {
566 self.pre_statements.push((sql, params));
567 }
568
569 pub fn add_post(&mut self, sql: String, params: Vec<FilterValue>) {
571 self.post_statements.push((sql, params));
572 }
573
574 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 pub fn is_empty(&self) -> bool {
582 self.pre_statements.is_empty() && self.post_statements.is_empty()
583 }
584
585 pub fn len(&self) -> usize {
587 self.pre_statements.len() + self.post_statements.len()
588 }
589}
590
591#[derive(Debug, Clone)]
609pub enum NestedWriteOp {
610 Create {
615 relation: &'static str,
616 target_table: &'static str,
617 foreign_key: &'static str,
618 payload: Vec<Vec<(String, FilterValue)>>,
621 },
622 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 {
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 {
656 relation: &'static str,
657 target_table: &'static str,
658 target_pk: &'static str,
659 pk: FilterValue,
660 },
661 DeleteMany {
669 relation: &'static str,
670 target_table: &'static str,
671 foreign_key: &'static str,
672 filter: Filter,
673 },
674 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 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 {
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 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 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
798fn 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 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 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 if update_payload.is_empty() && create_payload.is_empty() {
1018 return Ok(());
1019 }
1020
1021 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 let insert_arity = create_payload.len() + 1; let (probe_set_text, probe_set_params) =
1045 build_writeop_set_clause(&update_payload, dialect, insert_arity + 1);
1046
1047 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 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 let (_, mut values, placeholders, quoted_cols) =
1074 build_insert_columns_and_values();
1075
1076 let _ = pk;
1081
1082 let effective_upsert_clause = if update_payload.is_empty() {
1083 let do_nothing = dialect.upsert_do_nothing_clause(&conflict_cols);
1088 if do_nothing.is_empty() {
1089 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 if update_payload.is_empty() {
1120 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 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 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 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 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 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 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 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 if set_pks.is_empty() {
1292 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 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 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 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 #[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 fn with_affected(seq: Vec<u64>) -> Self {
1392 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 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 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 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 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 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); }
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 assert!(statements.len() >= 2);
1745
1746 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 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 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 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 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 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 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 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 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 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 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 assert!(
2163 sql.contains("VALUES ($1, $2)"),
2164 "INSERT VALUES placeholders: {sql}"
2165 );
2166 assert!(sql.contains("$3"), "got: {sql}");
2167 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 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 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 assert_eq!(update_params.len(), 2);
2209 assert_eq!(update_params[0], FilterValue::Int(1)); assert_eq!(update_params[1], FilterValue::Int(99)); }
2212
2213 #[tokio::test]
2214 async fn nested_op_upsert_two_statement_fallback_on_mssql_insert_path() {
2215 use crate::inputs::WriteOp;
2216 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 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 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 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 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 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 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 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 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 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 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 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 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 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 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 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 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 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 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 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 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}