import { Background, BackgroundVariant, type Connection, type Node as FlowCanvasNode, ReactFlow, ReactFlowProvider, useNodesState, useReactFlow, useUpdateNodeInternals, } from "@xyflow/react" import "@xyflow/react/dist/style.css" import { useMutation, useQuery, useQueryClient, useSuspenseQuery, } from "@tanstack/react-query" import { useNavigate } from "@tanstack/react-router" import { Workflow } from "lucide-react" import { useCallback, useEffect, useMemo, useRef, useState } from "react" import { type FlowDef_Input, FlowsService, type MessageSpec, type NodeDef_Input, } from "@/client" import { Button } from "@/components/ui/button" import { Dialog, DialogContent, DialogDescription, DialogFooter, DialogHeader, DialogTitle, } from "@/components/ui/dialog" import useCustomToast from "@/hooks/useCustomToast" import { CommandPalette } from "./CommandPalette" import { bindingsKey, deriveEdges, portOf, qualify } from "./deriveEdges" import { EdgeInspector, type InspectedEdge } from "./EdgeInspector" import { FlowDock } from "./FlowDock" import { FlowNode, type FlowNodeData } from "./FlowNode" import { FlowTabs } from "./FlowTabs" import { LiveEdge } from "./LiveEdge" import { NodePanel } from "./NodePanel" import "./flow.css" import { liveStore } from "./liveStore" import { flowQueryOptions, flowsQueryOptions, nodeTypesQueryOptions, useAutosave, } from "./queries" import { useFlowSocket } from "./useFlowSocket" const nodeTypes = { flow: FlowNode } const edgeTypes = { live: LiveEdge } type Rebind = { nodeId: string nodeLabel: string port: string from: string to: string dtype: MessageSpec["dtype"] } /** Step a new node off any node already sitting at that spot. */ function freePosition( nodes: NodeDef_Input[], start: { x: number; y: number }, ): { x: number; y: number } { const position = { ...start } // Roughly a node's footprint, so a nudged node clears the one below it. const occupied = () => nodes.some( (node) => Math.abs((node.position?.x ?? 0) - position.x) < 220 && Math.abs((node.position?.y ?? 0) - position.y) < 80, ) while (occupied()) { position.x += 48 position.y += 96 } return position } /** A name that does not collide with the nodes already on the canvas. */ function uniqueNodeId(existing: NodeDef_Input[], type: string): string { const taken = new Set(existing.map((node) => node.id)) for (let index = 1; ; index += 1) { const candidate = index === 1 ? type : `${type}_${index}` if (!taken.has(candidate)) return candidate } } function FlowEditorInner({ flowName }: { flowName: string }) { const queryClient = useQueryClient() const { showErrorToast } = useCustomToast() const { screenToFlowPosition, fitView } = useReactFlow() const updateNodeInternals = useUpdateNodeInternals() const { data: flows } = useSuspenseQuery(flowsQueryOptions()) const { data: detail } = useSuspenseQuery(flowQueryOptions(flowName)) const { data: nodeTypeInfo } = useQuery(nodeTypesQueryOptions()) const { save, flush, mutation: saving } = useAutosave(flowName) const [definitions, setDefinitions] = useState( () => detail.definition.nodes ?? [], ) const [canvasNodes, setCanvasNodes, onNodesChange] = useUnpositionedNodes( detail.definition.nodes ?? [], ) const [selectedId, setSelectedId] = useState(null) const [paletteOpen, setPaletteOpen] = useState(false) const [inspected, setInspected] = useState(null) const [rebind, setRebind] = useState(null) const issues = detail.issues ?? [] // Keep the latest document in a ref so autosave never captures a stale copy. const latest = useRef(detail.definition) const commit = useCallback( (nodes: NodeDef_Input[], positions?: FlowCanvasNode[]) => { const placed = nodes.map((node) => { const canvas = (positions ?? canvasNodes).find((n) => n.id === node.id) return canvas ? { ...node, position: canvas.position } : node }) setDefinitions(placed) const next: FlowDef_Input = { ...detail.definition, nodes: placed } latest.current = next save(next) }, [canvasNodes, detail.definition, save], ) const typeLabels = useMemo( () => new Map((nodeTypeInfo ?? []).map((info) => [info.type, info.title])), [nodeTypeInfo], ) const issuesByNode = useMemo(() => { const map = new Map() for (const issue of issues) { if (!issue.node) continue const list = map.get(issue.node) ?? [] list.push(issue.message) map.set(issue.node, list) } return map }, [issues]) // Canvas nodes carry the definition so the node component can render it. const renderedNodes = useMemo( () => canvasNodes.map((node) => { const definition = definitions.find((entry) => entry.id === node.id) const nodeIssues = issuesByNode.get(`${flowName}.${node.id}`) ?? [] return { ...node, selected: node.id === selectedId, data: { definition: definition ?? { id: node.id }, flow: flowName, typeLabel: typeLabels.get(definition?.type ?? "") ?? definition?.type ?? "", issueText: nodeIssues.join("\n"), } satisfies FlowNodeData, } }), [canvasNodes, definitions, flowName, issuesByNode, selectedId, typeLabels], ) // A cheap fingerprint of the wiring: it changes when a name does, but not // when a node merely moves. const key = bindingsKey(definitions) // Offer the names already in play: everything published is worth reading, // and an input nobody provides yet is worth publishing. // biome-ignore lint/correctness/useExhaustiveDependencies: the bindings key is what changes names. const suggestions = useMemo(() => { const provided = new Set() const consumed = new Set() for (const node of definitions) { for (const spec of node.provides ?? []) { if (spec.name) provided.add(spec.name) } for (const spec of node.requires ?? []) { if (spec.name) consumed.add(spec.name) } } return { consumes: [...provided].sort(), provides: [...consumed].filter((name) => !provided.has(name)).sort(), } }, [key]) // Edges follow from the name bindings, so they are derived, never stored. // biome-ignore lint/correctness/useExhaustiveDependencies: the key is the dependency; the array identity changes on every drag frame. const edges = useMemo( () => deriveEdges(definitions, flowName), [key, flowName], ) // Editing ports adds and removes handles. React Flow measures those once, so // it has to be told, or an edge to a brand-new handle never gets drawn. // biome-ignore lint/correctness/useExhaustiveDependencies: the bindings key is what changes handles. useEffect(() => { updateNodeInternals(definitions.map((node) => node.id)) }, [key, updateNodeInternals]) const runMutation = useMutation({ mutationFn: () => FlowsService.runFlow({ name: flowName, requestBody: { inputs: {} } }), onSuccess: (state) => { liveStore.setValues( Object.fromEntries( Object.entries(state.values ?? {}).map(([name, value]) => [ name, { value: value.value, ts: value.ts ?? null }, ]), ), ) }, onError: () => showErrorToast("The flow could not run. Check the node errors."), }) const sourceMutation = useMutation({ mutationFn: ({ nodeId, code }: { nodeId: string; code: string }) => FlowsService.saveNodeSource({ name: flowName, nodeId, requestBody: { code }, }), onSuccess: () => { queryClient.invalidateQueries({ queryKey: ["flows", flowName] }) }, }) const addNode = useCallback( (type: string) => { const id = uniqueNodeId(definitions, type) // Drop it where the user is looking, but never on top of another node. const position = freePosition( definitions, screenToFlowPosition({ x: window.innerWidth / 2, y: window.innerHeight / 2, }), ) const node: NodeDef_Input = { id, type, position, params: {}, requires: [], provides: [], } const nextDefinitions = [...definitions, node] const nextCanvas = [ ...canvasNodes, { id, type: "flow", position, data: {} } as FlowCanvasNode, ] setCanvasNodes(nextCanvas) commit(nextDefinitions, nextCanvas) setSelectedId(id) }, [canvasNodes, commit, definitions, screenToFlowPosition, setCanvasNodes], ) const updateNode = useCallback( (next: NodeDef_Input) => { commit(definitions.map((node) => (node.id === next.id ? next : node))) }, [commit, definitions], ) const deleteNodes = useCallback( (ids: string[]) => { const remaining = definitions.filter((node) => !ids.includes(node.id)) setCanvasNodes(canvasNodes.filter((node) => !ids.includes(node.id))) commit(remaining) if (selectedId && ids.includes(selectedId)) setSelectedId(null) }, [canvasNodes, commit, definitions, selectedId, setCanvasNodes], ) const applyBinding = useCallback( (nodeId: string, port: string, message: string) => { commit( definitions.map((node) => node.id === nodeId ? { ...node, requires: (node.requires ?? []).map((spec) => portOf(spec) === port ? { ...spec, name: message, port } : spec, ), } : node, ), ) }, [commit, definitions], ) /** * Dragging output to input is shorthand for "consume what that node * publishes": it points the input at the producer's message name. */ const onConnect = useCallback( (connection: Connection) => { const producer = definitions.find((node) => node.id === connection.source) const consumer = definitions.find((node) => node.id === connection.target) if (!producer || !consumer) return const outSpec = (producer.provides ?? []).find( (spec) => portOf(spec) === connection.sourceHandle, ) const inSpec = (consumer.requires ?? []).find( (spec) => portOf(spec) === connection.targetHandle, ) if (!outSpec?.name || !inSpec) return // Already reading something else: the user may want either message, so // offer the extra port rather than assuming a replacement. if (inSpec.name && inSpec.name !== outSpec.name) { setRebind({ nodeId: consumer.id, nodeLabel: consumer.title || consumer.id, port: portOf(inSpec), from: inSpec.name, to: outSpec.name, dtype: outSpec.dtype, }) return } applyBinding(consumer.id, portOf(inSpec), outSpec.name) }, [definitions, applyBinding], ) /** Give the consumer a second input, bound to the producer's message. */ const addInputPort = useCallback( (nodeId: string, message: string, dtype: MessageSpec["dtype"]) => { commit( definitions.map((node) => node.id === nodeId ? { ...node, requires: [ ...(node.requires ?? []), { name: message, port: "", dtype }, ], } : node, ), ) }, [commit, definitions], ) const unbind = useCallback( (message: string) => { const qualified = qualify(flowName, message) commit( definitions.map((node) => ({ ...node, requires: (node.requires ?? []).map((spec) => qualify(flowName, spec.name ?? "") === qualified ? { ...spec, name: "", port: portOf(spec) } : spec, ), })), ) setInspected(null) }, [commit, definitions, flowName], ) const focusNode = useCallback( (qualifiedId: string) => { const id = qualifiedId.startsWith(`${flowName}.`) ? qualifiedId.slice(flowName.length + 1) : qualifiedId fitView({ nodes: [{ id }], duration: 300, maxZoom: 1.2 }) setSelectedId(id) }, [fitView, flowName], ) const selected = definitions.find((node) => node.id === selectedId) ?? null return ( <> commit(definitions, mergeDragged(canvasNodes, dragged)) } onNodesDelete={(deleted) => deleteNodes(deleted.map((node) => node.id))} onNodeClick={(_event, node) => setSelectedId(node.id)} onPaneClick={() => { setSelectedId(null) setInspected(null) }} onEdgeClick={(event, edge) => { const label = (id: string) => { const node = definitions.find((entry) => entry.id === id) return node?.title || node?.id || id } setInspected({ message: (edge.data as { message: string }).message, from: label(edge.source), to: label(edge.target), x: event.clientX, y: event.clientY, }) }} onConnect={onConnect} nodeTypes={nodeTypes} edgeTypes={edgeTypes} proOptions={{ hideAttribution: true }} fitView fitViewOptions={{ maxZoom: 1, padding: 0.25 }} minZoom={0.25} maxZoom={2} nodeDragThreshold={5} connectionRadius={30} connectOnClick autoPanOnConnect edgesReconnectable={false} deleteKeyCode={["Backspace", "Delete"]} className="h-full w-full" > setPaletteOpen(true)} onRun={() => { flush() runMutation.mutate() }} onFocusNode={focusNode} /> {definitions.length === 0 ? (

This flow is empty

Add a node to get started. Press ⌘K, or use the plus in the bar below.

) : null} { if (selected) sourceMutation.mutate({ nodeId: selected.id, code }) }} onClose={() => { flush() setSelectedId(null) }} onDelete={() => selected && deleteNodes([selected.id])} /> setInspected(null)} onUnbind={unbind} /> runMutation.mutate()} /> !open && setRebind(null)} > How should {rebind?.nodeLabel} read this? Its "{rebind?.port}" input already reads{" "} {rebind?.from}. It can take{" "} {rebind?.to} as well, or instead.
) } function mergeDragged( nodes: FlowCanvasNode[], dragged: FlowCanvasNode[], ): FlowCanvasNode[] { const moved = new Map(dragged.map((node) => [node.id, node.position])) return nodes.map((node) => moved.has(node.id) ? { ...node, position: moved.get(node.id)! } : node, ) } /** Seed xyflow's own node state once; it owns positions while you drag. */ function useUnpositionedNodes(definitions: NodeDef_Input[]) { return useNodesState( definitions.map((node) => ({ id: node.id, type: "flow", position: { x: node.position?.x ?? 0, y: node.position?.y ?? 0 }, data: {}, })), ) } export function FlowEditor({ flowName }: { flowName: string }) { const navigate = useNavigate() const onAuthFailure = useCallback(() => { navigate({ to: "/login" }) }, [navigate]) useFlowSocket(onAuthFailure) useEffect(() => { return () => liveStore.reset() }, []) return ( {/* Remounting per flow keeps canvas state from leaking between them. */} ) }