@@ -38,7 +38,8 @@ import {
3838 normalizeWorkflowEdgeSourceHandle ,
3939 normalizeWorkflowEdgeTargetHandle ,
4040} from '@sim/workflow-types/workflow'
41- import { and , eq , inArray , isNull , or , sql } from 'drizzle-orm'
41+ import { and , eq , inArray , isNull , or , type SQL , sql } from 'drizzle-orm'
42+ import type { PgInsertValue } from 'drizzle-orm/pg-core'
4243import { drizzle } from 'drizzle-orm/postgres-js'
4344import postgres from 'postgres'
4445import { env } from '@/env'
@@ -77,6 +78,45 @@ function canonicalizeEdgeAddCandidate(edge: EdgeAddCandidate): EdgeAddCandidate
7778 }
7879}
7980
81+ type WorkflowScopedTable = typeof workflowBlocks | typeof workflowEdges | typeof workflowSubflows
82+
83+ /**
84+ * Inserts rows into a workflow-scoped table, updating rows that already exist
85+ * in `workflowId`. Row ids are unique across every workflow, so an upsert keyed
86+ * on `id` alone would overwrite another workflow's row: `setWhere` confines
87+ * the update to this workflow, and a conflict it skips leaves a row unreturned,
88+ * which aborts the transaction instead of dropping that row silently.
89+ */
90+ async function upsertWorkflowRows < T extends WorkflowScopedTable > (
91+ tx : DbOrTx ,
92+ table : T ,
93+ workflowId : string ,
94+ values : PgInsertValue < T > [ ] ,
95+ set : Partial < Record < keyof T [ '$inferInsert' ] , SQL > >
96+ ) {
97+ const written = await tx
98+ . insert ( table )
99+ . values ( values )
100+ . onConflictDoUpdate ( { target : table . id , set, setWhere : eq ( table . workflowId , workflowId ) } )
101+ . returning ( { id : table . id } )
102+
103+ if ( written . length !== values . length ) {
104+ throw new Error ( 'One or more ids already belong to another workflow' )
105+ }
106+ }
107+
108+ const SUBFLOW_UPSERT_SET = {
109+ config : sql `excluded.config` ,
110+ updatedAt : sql `now()` ,
111+ } as const
112+
113+ const EDGE_UPSERT_SET = {
114+ sourceBlockId : sql `excluded.source_block_id` ,
115+ targetBlockId : sql `excluded.target_block_id` ,
116+ sourceHandle : sql `excluded.source_handle` ,
117+ targetHandle : sql `excluded.target_handle` ,
118+ } as const
119+
80120interface FilterEdgesForPersistResult < T > {
81121 safeEdges : T [ ]
82122 droppedCounts : Record < string , number >
@@ -949,8 +989,8 @@ async function handleBlocksOperationTx(
949989 workflowId,
950990 type : block . type as string ,
951991 name : block . name as string ,
952- positionX : ( block . position as { x : number ; y : number } ) . x ,
953- positionY : ( block . position as { x : number ; y : number } ) . y ,
992+ positionX : String ( ( block . position as { x : number ; y : number } ) . x ) ,
993+ positionY : String ( ( block . position as { x : number ; y : number } ) . y ) ,
954994 data : ( block . data as Record < string , unknown > | undefined ) || { } ,
955995 subBlocks : mergedSubBlocks ,
956996 outputs : ( block . outputs as Record < string , unknown > ) || { } ,
@@ -960,84 +1000,66 @@ async function handleBlocksOperationTx(
9601000 triggerMode : ( block . triggerMode as boolean ) ?? false ,
9611001 errorEnabled : ( block . errorEnabled as boolean ) ?? false ,
9621002 retry : ( block . retry as Record < string , unknown > | undefined ) ?? null ,
963- height : ( block . height as number ) || 0 ,
1003+ height : String ( ( block . height as number ) || 0 ) ,
9641004 locked : ( block . locked as boolean ) ?? false ,
9651005 }
9661006 } )
9671007
968- await tx
969- . insert ( workflowBlocks )
970- . values ( blockValues )
971- . onConflictDoUpdate ( {
972- target : workflowBlocks . id ,
973- set : {
974- type : sql `excluded.type` ,
975- name : sql `excluded.name` ,
976- positionX : sql `excluded.position_x` ,
977- positionY : sql `excluded.position_y` ,
978- enabled : sql `excluded.enabled` ,
979- horizontalHandles : sql `excluded.horizontal_handles` ,
980- advancedMode : sql `excluded.advanced_mode` ,
981- triggerMode : sql `excluded.trigger_mode` ,
982- errorEnabled : sql `excluded.error_enabled` ,
983- retry : sql `excluded.retry` ,
984- locked : sql `excluded.locked` ,
985- height : sql `excluded.height` ,
986- subBlocks : sql `excluded.sub_blocks` ,
987- outputs : sql `excluded.outputs` ,
988- data : sql `excluded.data` ,
989- updatedAt : sql `now()` ,
990- } ,
991- } )
1008+ await upsertWorkflowRows ( tx , workflowBlocks , workflowId , blockValues , {
1009+ type : sql `excluded.type` ,
1010+ name : sql `excluded.name` ,
1011+ positionX : sql `excluded.position_x` ,
1012+ positionY : sql `excluded.position_y` ,
1013+ enabled : sql `excluded.enabled` ,
1014+ horizontalHandles : sql `excluded.horizontal_handles` ,
1015+ advancedMode : sql `excluded.advanced_mode` ,
1016+ triggerMode : sql `excluded.trigger_mode` ,
1017+ errorEnabled : sql `excluded.error_enabled` ,
1018+ retry : sql `excluded.retry` ,
1019+ locked : sql `excluded.locked` ,
1020+ height : sql `excluded.height` ,
1021+ subBlocks : sql `excluded.sub_blocks` ,
1022+ outputs : sql `excluded.outputs` ,
1023+ data : sql `excluded.data` ,
1024+ updatedAt : sql `now()` ,
1025+ } )
9921026
9931027 // Create subflow entries for loop/parallel blocks (skip if already in payload)
9941028 const loopIds = new Set ( loops ? Object . keys ( loops ) : [ ] )
9951029 const parallelIds = new Set ( parallels ? Object . keys ( parallels ) : [ ] )
1030+ const defaultSubflows = [ ]
9961031 for ( const block of allowedBlocks ) {
9971032 const blockId = block . id as string
9981033 if ( block . type === 'loop' && ! loopIds . has ( blockId ) ) {
999- await tx
1000- . insert ( workflowSubflows )
1001- . values ( {
1002- id : blockId ,
1003- workflowId,
1004- type : 'loop' ,
1005- config : {
1006- loopType : 'for' ,
1007- iterations : DEFAULT_LOOP_ITERATIONS ,
1008- nodes : [ ] ,
1009- } ,
1010- } )
1011- . onConflictDoUpdate ( {
1012- target : workflowSubflows . id ,
1013- set : {
1014- config : sql `excluded.config` ,
1015- updatedAt : sql `now()` ,
1016- } ,
1017- } )
1034+ defaultSubflows . push ( {
1035+ id : blockId ,
1036+ workflowId,
1037+ type : 'loop' ,
1038+ config : { loopType : 'for' , iterations : DEFAULT_LOOP_ITERATIONS , nodes : [ ] } ,
1039+ } )
10181040 } else if ( block . type === 'parallel' && ! parallelIds . has ( blockId ) ) {
1019- await tx
1020- . insert ( workflowSubflows )
1021- . values ( {
1022- id : blockId ,
1023- workflowId,
1024- type : 'parallel' ,
1025- config : {
1026- parallelType : 'count' ,
1027- count : DEFAULT_PARALLEL_COUNT ,
1028- batchSize : DEFAULT_PARALLEL_BATCH_SIZE ,
1029- nodes : [ ] ,
1030- } ,
1031- } )
1032- . onConflictDoUpdate ( {
1033- target : workflowSubflows . id ,
1034- set : {
1035- config : sql `excluded.config` ,
1036- updatedAt : sql `now()` ,
1037- } ,
1038- } )
1041+ defaultSubflows . push ( {
1042+ id : blockId ,
1043+ workflowId,
1044+ type : 'parallel' ,
1045+ config : {
1046+ parallelType : 'count' ,
1047+ count : DEFAULT_PARALLEL_COUNT ,
1048+ batchSize : DEFAULT_PARALLEL_BATCH_SIZE ,
1049+ nodes : [ ] ,
1050+ } ,
1051+ } )
10391052 }
10401053 }
1054+ if ( defaultSubflows . length > 0 ) {
1055+ await upsertWorkflowRows (
1056+ tx ,
1057+ workflowSubflows ,
1058+ workflowId ,
1059+ defaultSubflows ,
1060+ SUBFLOW_UPSERT_SET
1061+ )
1062+ }
10411063
10421064 const parentIds = new Set < string > ( )
10431065 for ( const block of allowedBlocks ) {
@@ -1087,18 +1109,7 @@ async function handleBlocksOperationTx(
10871109 targetHandle : normalizeWorkflowEdgeTargetHandle ( edge . targetHandle ) ,
10881110 } ) )
10891111
1090- await tx
1091- . insert ( workflowEdges )
1092- . values ( edgeValues )
1093- . onConflictDoUpdate ( {
1094- target : workflowEdges . id ,
1095- set : {
1096- sourceBlockId : sql `excluded.source_block_id` ,
1097- targetBlockId : sql `excluded.target_block_id` ,
1098- sourceHandle : sql `excluded.source_handle` ,
1099- targetHandle : sql `excluded.target_handle` ,
1100- } ,
1101- } )
1112+ await upsertWorkflowRows ( tx , workflowEdges , workflowId , edgeValues , EDGE_UPSERT_SET )
11021113 }
11031114 }
11041115
@@ -1110,16 +1121,7 @@ async function handleBlocksOperationTx(
11101121 config : loop as Record < string , unknown > ,
11111122 } ) )
11121123
1113- await tx
1114- . insert ( workflowSubflows )
1115- . values ( loopValues )
1116- . onConflictDoUpdate ( {
1117- target : workflowSubflows . id ,
1118- set : {
1119- config : sql `excluded.config` ,
1120- updatedAt : sql `now()` ,
1121- } ,
1122- } )
1124+ await upsertWorkflowRows ( tx , workflowSubflows , workflowId , loopValues , SUBFLOW_UPSERT_SET )
11231125 }
11241126
11251127 if ( parallels && Object . keys ( parallels ) . length > 0 ) {
@@ -1130,16 +1132,13 @@ async function handleBlocksOperationTx(
11301132 config : parallel as Record < string , unknown > ,
11311133 } ) )
11321134
1133- await tx
1134- . insert ( workflowSubflows )
1135- . values ( parallelValues )
1136- . onConflictDoUpdate ( {
1137- target : workflowSubflows . id ,
1138- set : {
1139- config : sql `excluded.config` ,
1140- updatedAt : sql `now()` ,
1141- } ,
1142- } )
1135+ await upsertWorkflowRows (
1136+ tx ,
1137+ workflowSubflows ,
1138+ workflowId ,
1139+ parallelValues ,
1140+ SUBFLOW_UPSERT_SET
1141+ )
11431142 }
11441143
11451144 logger . info ( `Successfully batch added blocks to workflow ${ workflowId } ` )
@@ -1791,18 +1790,7 @@ async function handleEdgesOperationTx(
17911790 targetHandle : normalizeWorkflowEdgeTargetHandle ( edge . targetHandle ) ,
17921791 } ) )
17931792
1794- await tx
1795- . insert ( workflowEdges )
1796- . values ( edgeValues )
1797- . onConflictDoUpdate ( {
1798- target : workflowEdges . id ,
1799- set : {
1800- sourceBlockId : sql `excluded.source_block_id` ,
1801- targetBlockId : sql `excluded.target_block_id` ,
1802- sourceHandle : sql `excluded.source_handle` ,
1803- targetHandle : sql `excluded.target_handle` ,
1804- } ,
1805- } )
1793+ await upsertWorkflowRows ( tx , workflowEdges , workflowId , edgeValues , EDGE_UPSERT_SET )
18061794
18071795 logger . debug ( `Batch added ${ safeEdges . length } edges to workflow ${ workflowId } ` )
18081796 break
0 commit comments