Function handleEdgeOperationTx

apps/realtime/src/database/operations.ts:1245–1406

(tx: any, workflowId: string, operation: string, payload: any)

}

async function handleEdgeOperationTx(tx: any, workflowId: string, operation: string, payload: any) {
  switch (operation) {
    case EDGE_OPERATIONS.ADD: {
      if (!payload.id || !payload.source || !payload.target) {
        throw new Error('Missing required fields for add edge operation')
      }

      const edgeBlocks = await tx
        .select({
          id: workflowBlocks.id,
          locked: workflowBlocks.locked,
          data: workflowBlocks.data,
        })
        .from(workflowBlocks)
        .where(
          and(
            eq(workflowBlocks.workflowId, workflowId),
            inArray(workflowBlocks.id, [payload.source, payload.target])
          )
        )

      type EdgeBlockRecord = (typeof edgeBlocks)[number]
      const blocksById: Record<string, EdgeBlockRecord> = Object.fromEntries(
        edgeBlocks.map((b: EdgeBlockRecord) => [b.id, b])
      )

      const parentIds = new Set<string>()
      for (const block of edgeBlocks) {
        const parentId = (block.data as Record<string, unknown> | null)?.parentId as
          | string
          | undefined
        if (parentId && !blocksById[parentId]) {
          parentIds.add(parentId)
        }
      }

      // Fetch parent blocks if needed
      if (parentIds.size > 0) {
        const parentBlocks = await tx
          .select({
            id: workflowBlocks.id,
            locked: workflowBlocks.locked,
            data: workflowBlocks.data,
          })
          .from(workflowBlocks)
          .where(
            and(
              eq(workflowBlocks.workflowId, workflowId),
              inArray(workflowBlocks.id, Array.from(parentIds))
            )
          )
        for (const b of parentBlocks) {
          blocksById[b.id] = b
        }
      }

      if (isWorkflowBlockProtected(payload.target, blocksById)) {
        logger.info(`Skipping edge add - target block is protected`)