| import type { ClassicScheme, LitArea2D } from '@retejs/lit-plugin'; |
| import { NodeEditor } from 'rete'; |
| import { AreaPlugin } from 'rete-area-plugin'; |
| import { DataflowEngine } from 'rete-engine'; |
| import type { DataflowEngineScheme } from 'rete-engine'; |
| import type { BaseWorkflowNode } from './nodes/base-node'; |
| import { WorkflowError } from './types'; |
| import type { ExecutionProgress } from './types'; |
| import { updateNodeDisplay } from './editor'; |
|
|
| type AreaExtra = LitArea2D<ClassicScheme>; |
|
|
| function getUpstreamNodes( |
| nodeId: string, |
| editor: NodeEditor<ClassicScheme> |
| ): Set<string> { |
| const visited = new Set<string>(); |
| const queue = [nodeId]; |
| while (queue.length > 0) { |
| const current = queue.shift()!; |
| if (visited.has(current)) continue; |
| visited.add(current); |
| const incoming = editor |
| .getConnections() |
| .filter((c) => c.target === current); |
| for (const conn of incoming) { |
| queue.push(conn.source); |
| } |
| } |
| return visited; |
| } |
|
|
| function topologicalSort( |
| nodeIds: Set<string>, |
| editor: NodeEditor<ClassicScheme> |
| ): string[] { |
| const inDegree = new Map<string, number>(); |
| const adj = new Map<string, string[]>(); |
| for (const id of nodeIds) { |
| inDegree.set(id, 0); |
| adj.set(id, []); |
| } |
| for (const conn of editor.getConnections()) { |
| if (nodeIds.has(conn.source) && nodeIds.has(conn.target)) { |
| adj.get(conn.source)!.push(conn.target); |
| inDegree.set(conn.target, (inDegree.get(conn.target) || 0) + 1); |
| } |
| } |
| const queue: string[] = []; |
| for (const [id, deg] of inDegree) { |
| if (deg === 0) queue.push(id); |
| } |
| const sorted: string[] = []; |
| while (queue.length > 0) { |
| const current = queue.shift()!; |
| sorted.push(current); |
| for (const next of adj.get(current) || []) { |
| const newDeg = (inDegree.get(next) ?? 0) - 1; |
| inDegree.set(next, newDeg); |
| if (newDeg === 0) queue.push(next); |
| } |
| } |
| if (sorted.length < nodeIds.size) { |
| throw new WorkflowError( |
| 'Circular dependency detected in workflow. Please remove any loops between nodes.' |
| ); |
| } |
|
|
| return sorted; |
| } |
|
|
| function tick(): Promise<void> { |
| return new Promise((r) => setTimeout(r, 0)); |
| } |
|
|
| function validateEncryptOrdering( |
| editor: NodeEditor<ClassicScheme>, |
| pipelineNodes: string[] |
| ): string | null { |
| for (const nodeId of pipelineNodes) { |
| const node = editor.getNode(nodeId) as BaseWorkflowNode; |
| if (node?.label !== 'Encrypt') continue; |
|
|
| const outConns = editor.getConnections().filter((c) => c.source === nodeId); |
| for (const conn of outConns) { |
| const target = editor.getNode(conn.target) as BaseWorkflowNode; |
| if (target && target.category !== 'Output') { |
| return `The Encrypt node feeds into "${target.label}", which may fail on encrypted data. Move Encrypt to just before the output node.`; |
| } |
| } |
| } |
| return null; |
| } |
|
|
| export async function executeWorkflow( |
| editor: NodeEditor<ClassicScheme>, |
| engine: DataflowEngine<DataflowEngineScheme>, |
| area: AreaPlugin<ClassicScheme, AreaExtra>, |
| onProgress: (progress: ExecutionProgress) => void |
| ): Promise<void> { |
| const nodes = editor.getNodes() as BaseWorkflowNode[]; |
| const connections = editor.getConnections(); |
|
|
| const nodesWithOutputConnections = new Set(connections.map((c) => c.source)); |
| const terminalNodes = nodes.filter( |
| (n) => !nodesWithOutputConnections.has(n.id) |
| ); |
|
|
| if (terminalNodes.length === 0) { |
| throw new Error( |
| 'No output nodes found. Add a Download node to complete your workflow.' |
| ); |
| } |
|
|
| const pipelineNodes = new Set<string>(); |
| for (const terminal of terminalNodes) { |
| const upstream = getUpstreamNodes(terminal.id, editor); |
| for (const id of upstream) pipelineNodes.add(id); |
| } |
|
|
| for (const node of nodes) { |
| node.execStatus = 'idle'; |
| updateNodeDisplay(node.id, editor, area); |
| } |
| await tick(); |
|
|
| const sorted = topologicalSort(pipelineNodes, editor); |
|
|
| const encryptWarning = validateEncryptOrdering(editor, sorted); |
| if (encryptWarning) { |
| throw new WorkflowError(encryptWarning, 'Encrypt'); |
| } |
|
|
| engine.reset(); |
|
|
| for (const nodeId of sorted) { |
| const node = editor.getNode(nodeId) as BaseWorkflowNode; |
| if (!node) continue; |
|
|
| node.execStatus = 'running'; |
| updateNodeDisplay(node.id, editor, area); |
| onProgress({ |
| nodeId: node.id, |
| nodeName: node.label, |
| status: 'running', |
| message: `Processing ${node.label}...`, |
| }); |
| await tick(); |
|
|
| try { |
| await engine.fetch(nodeId); |
|
|
| node.execStatus = 'completed'; |
| updateNodeDisplay(node.id, editor, area); |
| onProgress({ |
| nodeId: node.id, |
| nodeName: node.label, |
| status: 'completed', |
| }); |
| await tick(); |
| } catch (error) { |
| node.execStatus = 'error'; |
| updateNodeDisplay(node.id, editor, area); |
|
|
| for (const remainingId of sorted) { |
| const remaining = editor.getNode(remainingId) as BaseWorkflowNode; |
| if (remaining && remaining.execStatus === 'running') { |
| remaining.execStatus = 'idle'; |
| updateNodeDisplay(remaining.id, editor, area); |
| } |
| } |
|
|
| const message = error instanceof Error ? error.message : String(error); |
| const wrapped = |
| error instanceof WorkflowError |
| ? error |
| : new WorkflowError(message, node.label); |
|
|
| onProgress({ |
| nodeId: node.id, |
| nodeName: node.label, |
| status: 'error', |
| message: wrapped.message, |
| }); |
| throw wrapped; |
| } |
| } |
| } |
|
|
