stroblmeandClaude Fable 5
a1572f91e3
Co-Authored-By: Claude Fable 5 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01H7LwYgJfpkbLCTeiAf8U4A
1222 lines
40 KiB
TypeScript
1222 lines
40 KiB
TypeScript
import {
|
|
Background,
|
|
BackgroundVariant,
|
|
type Connection,
|
|
type Node as FlowCanvasNode,
|
|
ReactFlow,
|
|
ReactFlowProvider,
|
|
useNodesState,
|
|
useReactFlow,
|
|
useStore,
|
|
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,
|
|
type FlowDetail,
|
|
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 { inCodeEditor, useShortcuts } from "@/lib/shortcuts"
|
|
import { cn } from "@/lib/utils"
|
|
import { CanvasTitle } from "./CanvasTitle"
|
|
import { CommandPalette } from "./CommandPalette"
|
|
import { bindingsKey, deriveEdges, portOf, qualify } from "./deriveEdges"
|
|
import { EdgeInspector, type InspectedEdge } from "./EdgeInspector"
|
|
import { EndpointNode } from "./EndpointNode"
|
|
import {
|
|
deriveEndpoints,
|
|
ENDPOINT_TYPE,
|
|
isEndpointNode,
|
|
placementsFor,
|
|
rememberPlacement,
|
|
} from "./endpoints"
|
|
import { FIT_VIEW, FlowDock } from "./FlowDock"
|
|
import { FlowNode, type FlowNodeData } from "./FlowNode"
|
|
import { FlowPanel } from "./FlowPanel"
|
|
import { LiveEdge } from "./LiveEdge"
|
|
import { NodePanel } from "./NodePanel"
|
|
import "./flow.css"
|
|
import { liveStore, useFlowPaused } from "./liveStore"
|
|
import {
|
|
flowKeys,
|
|
flowQueryOptions,
|
|
flowsQueryOptions,
|
|
nodeTypesQueryOptions,
|
|
useAutosave,
|
|
useDiscardDraft,
|
|
usePublish,
|
|
} from "./queries"
|
|
import { useFlowSocket } from "./useFlowSocket"
|
|
|
|
const nodeTypes = { flow: FlowNode, [ENDPOINT_TYPE]: EndpointNode }
|
|
|
|
/** Is this canvas node actually part of the flow document? */
|
|
const isDocumentNode = (node: { id: string }) => !isEndpointNode(node)
|
|
const edgeTypes = { live: LiveEdge }
|
|
|
|
type Rebind = {
|
|
nodeId: string
|
|
nodeLabel: string
|
|
port: string
|
|
from: string
|
|
to: string
|
|
dtype: MessageSpec["dtype"]
|
|
}
|
|
|
|
/** A finished message rename, waiting on an answer about the rest of the flow. */
|
|
type MessageRename = {
|
|
from: string
|
|
to: string
|
|
/** How many other nodes are still bound to the old name. */
|
|
count: number
|
|
}
|
|
|
|
/** Documents to step back and forward through, newest last. */
|
|
type History = {
|
|
past: FlowDef_Input[]
|
|
future: FlowDef_Input[]
|
|
/** When the last entry was recorded, so a burst of typing stays one edit. */
|
|
at: number
|
|
}
|
|
|
|
const HISTORY_LIMIT = 50
|
|
const TYPING_WINDOW = 500
|
|
|
|
/** Nodes copied here, and the code of the ones carrying their own. */
|
|
type NodeClipboard = {
|
|
nodes: NodeDef_Input[]
|
|
sources: Record<string, string>
|
|
}
|
|
|
|
// ponytail: localStorage, so a copy survives a flow switch or a second tab but
|
|
// not a second browser. The system clipboard if that ever matters.
|
|
const CLIPBOARD_KEY = "fluksio.nodeClipboard"
|
|
|
|
/**
|
|
* Remember the document as it was before a change.
|
|
*
|
|
* Fields commit on every keystroke, so consecutive edits that leave the same
|
|
* nodes in place fold into the entry already on the stack. Anything carrying
|
|
* positions — a drag, a new node — is a finished action and starts its own.
|
|
*/
|
|
function record(
|
|
history: History,
|
|
previous: FlowDef_Input,
|
|
next: FlowDef_Input,
|
|
settled: boolean,
|
|
) {
|
|
const before = previous.nodes ?? []
|
|
const after = next.nodes ?? []
|
|
const sameNodes =
|
|
before.length === after.length &&
|
|
before.every((node, index) => node.id === after[index].id)
|
|
const stillTyping =
|
|
!settled && sameNodes && Date.now() - history.at < TYPING_WINDOW
|
|
|
|
if (!stillTyping) {
|
|
history.past.push(previous)
|
|
if (history.past.length > HISTORY_LIMIT) history.past.shift()
|
|
}
|
|
history.at = settled ? 0 : Date.now()
|
|
// A new change is a new branch: what was undone is not coming back.
|
|
history.future = []
|
|
}
|
|
|
|
function toCanvasNodes(definitions: NodeDef_Input[]): FlowCanvasNode[] {
|
|
return definitions.map((node) => ({
|
|
id: node.id,
|
|
type: "flow",
|
|
position: { x: node.position?.x ?? 0, y: node.position?.y ?? 0 },
|
|
data: {},
|
|
}))
|
|
}
|
|
|
|
/**
|
|
* The dot grid lives in flow space, so React Flow shrinks it along with the
|
|
* zoom. Below 1:1 that leaves sub-pixel dots on a fractional grid, which the
|
|
* canvas can only render as a moiré haze — the whole viewport reads as blurry.
|
|
* Taking the zoom back out of both the radius and the spacing gives the same
|
|
* crisp 1.5px dots, 24px apart, whatever the zoom is.
|
|
*/
|
|
function CanvasBackground() {
|
|
const zoom = useStore((state) => state.transform[2])
|
|
// In octaves, so the grid halves rather than drifting as you zoom.
|
|
const step = 2 ** Math.round(Math.log2(1 / zoom))
|
|
return (
|
|
<Background
|
|
variant={BackgroundVariant.Dots}
|
|
gap={24 * step}
|
|
size={1.5 / zoom}
|
|
/>
|
|
)
|
|
}
|
|
|
|
/** 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,
|
|
onReload,
|
|
}: {
|
|
flowName: string
|
|
onReload: () => void
|
|
}) {
|
|
const navigate = useNavigate()
|
|
const queryClient = useQueryClient()
|
|
const { showErrorToast, showSuccessToast } = 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,
|
|
conflict,
|
|
resolveConflict,
|
|
mutation: saving,
|
|
} = useAutosave(flowName)
|
|
const publish = usePublish(flowName)
|
|
const discard = useDiscardDraft(flowName)
|
|
|
|
// The whole working document, so a flow-level edit is an edit like any
|
|
// other — undoable, and on screen before the server has seen it.
|
|
const [flowDoc, setFlowDoc] = useState<FlowDef_Input>(() => detail.definition)
|
|
const definitions = useMemo(() => flowDoc.nodes ?? [], [flowDoc])
|
|
const [canvasNodes, setCanvasNodes, onNodesChange] = useUnpositionedNodes(
|
|
detail.definition.nodes ?? [],
|
|
)
|
|
const [selectedId, setSelectedId] = useState<string | null>(null)
|
|
const [paletteOpen, setPaletteOpen] = useState(false)
|
|
const [inspected, setInspected] = useState<InspectedEdge | null>(null)
|
|
const [rebind, setRebind] = useState<Rebind | null>(null)
|
|
const [renamed, setRenamed] = useState<MessageRename | null>(null)
|
|
const [flowPanelOpen, setFlowPanelOpen] = useState(false)
|
|
// The editor at full size takes the width the canvas chrome does not need.
|
|
const [editorExpanded, setEditorExpanded] = useState(false)
|
|
// The dock hosts the logs, but a failing node opens them too, at its own
|
|
// lines — so which node and whether it is open belong here.
|
|
const [logsOpen, setLogsOpen] = useState(false)
|
|
const [logsNode, setLogsNode] = useState<string | null>(null)
|
|
|
|
const issues = detail.issues ?? []
|
|
const paused = useFlowPaused(flowName)
|
|
|
|
// Keep the latest document in a ref so autosave never captures a stale copy.
|
|
const latest = useRef<FlowDef_Input>(detail.definition)
|
|
const history = useRef<History>({ past: [], future: [], at: 0 })
|
|
|
|
/** Put a document on the canvas and on its way to the server. */
|
|
const applyDoc = useCallback(
|
|
(next: FlowDef_Input) => {
|
|
setFlowDoc(next)
|
|
latest.current = next
|
|
save(next)
|
|
},
|
|
[save],
|
|
)
|
|
|
|
/** Apply a change to the whole document and make it undoable. */
|
|
const commitDoc = useCallback(
|
|
(next: FlowDef_Input, settled = false) => {
|
|
record(history.current, latest.current, next, settled)
|
|
applyDoc(next)
|
|
},
|
|
[applyDoc],
|
|
)
|
|
|
|
/** The same, for the changes that only touch the nodes. */
|
|
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
|
|
})
|
|
commitDoc({ ...latest.current, nodes: placed }, Boolean(positions))
|
|
},
|
|
[canvasNodes, commitDoc],
|
|
)
|
|
|
|
/**
|
|
* Step through the history. Undo and redo save like any other edit, so the
|
|
* autosave debounce collapses a run of them into one write.
|
|
*/
|
|
const step = useCallback(
|
|
(back: boolean) => {
|
|
const { past, future } = history.current
|
|
const remembered = (back ? past : future).pop()
|
|
if (!remembered) return
|
|
;(back ? future : past).push(latest.current)
|
|
history.current.at = 0
|
|
// xyflow owns the positions, so hand it the remembered ones too.
|
|
setCanvasNodes(toCanvasNodes(remembered.nodes ?? []))
|
|
applyDoc(remembered)
|
|
setInspected(null)
|
|
},
|
|
[applyDoc, setCanvasNodes],
|
|
)
|
|
|
|
const typeLabels = useMemo(
|
|
() => new Map((nodeTypeInfo ?? []).map((info) => [info.type, info.title])),
|
|
[nodeTypeInfo],
|
|
)
|
|
|
|
// Which types bring code of their own, so a copy has something to carry.
|
|
const sourceTypes = useMemo(
|
|
() =>
|
|
new Set(
|
|
(nodeTypeInfo ?? [])
|
|
.filter((info) => info.has_source)
|
|
.map((info) => info.type),
|
|
),
|
|
[nodeTypeInfo],
|
|
)
|
|
|
|
// Which types came from an installed connector rather than the engine. They
|
|
// cannot be in the icon map, so they share one.
|
|
const pluginTypes = useMemo(
|
|
() =>
|
|
new Set(
|
|
(nodeTypeInfo ?? [])
|
|
.filter((info) => info.plugin)
|
|
.map((info) => info.type),
|
|
),
|
|
[nodeTypeInfo],
|
|
)
|
|
|
|
const issuesByNode = useMemo(() => {
|
|
const map = new Map<string, string[]>()
|
|
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])
|
|
|
|
const showLogs = useCallback((nodeId: string) => {
|
|
setLogsNode(nodeId)
|
|
setLogsOpen(true)
|
|
}, [])
|
|
|
|
// 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 ?? "",
|
|
isPlugin: pluginTypes.has(definition?.type ?? ""),
|
|
issueText: nodeIssues.join("\n"),
|
|
onShowLogs: showLogs,
|
|
} satisfies FlowNodeData,
|
|
}
|
|
}),
|
|
[
|
|
canvasNodes,
|
|
definitions,
|
|
flowName,
|
|
issuesByNode,
|
|
pluginTypes,
|
|
selectedId,
|
|
showLogs,
|
|
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<string>()
|
|
const consumed = new Set<string>()
|
|
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])
|
|
|
|
// Endpoints are movable but are not the flow's to store, so where they were
|
|
// put lives in the browser rather than in flow.json.
|
|
const [moved, setMoved] = useState<Record<string, { x: number; y: number }>>(
|
|
() => placementsFor(flowName),
|
|
)
|
|
|
|
/** Where clicking an endpoint takes you: the thing it stands for. */
|
|
const openEndpoint = useCallback(
|
|
(id: string) => {
|
|
const [kind, rest] = id.split(":", 2)
|
|
if (kind === "dashboard") {
|
|
navigate({
|
|
to: "/dashboards/$name",
|
|
params: { name: (rest ?? "").split(":")[0] },
|
|
})
|
|
} else if (kind === "flow") {
|
|
navigate({
|
|
to: "/flows/$flowName",
|
|
params: { flowName: (rest ?? "").split(".")[0] },
|
|
})
|
|
}
|
|
},
|
|
[navigate],
|
|
)
|
|
|
|
// Dashboards and other flows wired into this one. They are drawn but never
|
|
// stored: they join at render, after everything that reads or writes
|
|
// canvasNodes, so an autosave, an undo or a delete cannot reach them.
|
|
// biome-ignore lint/correctness/useExhaustiveDependencies: positions change on every drag frame; the key covers the wiring.
|
|
const external = useMemo(
|
|
() =>
|
|
deriveEndpoints(
|
|
detail.endpoints ?? [],
|
|
definitions,
|
|
flowName,
|
|
new Map(canvasNodes.map((node) => [node.id, node.position])),
|
|
moved,
|
|
),
|
|
[detail.endpoints, key, flowName, moved],
|
|
)
|
|
|
|
// Edges follow from the name bindings, so they are derived, never stored.
|
|
// Kept off `external` deliberately: an endpoint's edges depend on which
|
|
// messages it touches, never on where it sits, so dragging one must not
|
|
// rebuild the edge array on every frame.
|
|
// biome-ignore lint/correctness/useExhaustiveDependencies: the key is the dependency; the array identity changes on every drag frame.
|
|
const edges = useMemo(
|
|
() => [...deriveEdges(definitions, flowName), ...external.edges],
|
|
[key, flowName, detail.endpoints],
|
|
)
|
|
|
|
const shownNodes = useMemo(
|
|
() => [...renderedNodes, ...external.nodes],
|
|
[renderedNodes, external],
|
|
)
|
|
|
|
// 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 enableMutation = useMutation({
|
|
mutationFn: (next: boolean) =>
|
|
next
|
|
? FlowsService.startFlow({ name: flowName })
|
|
: FlowsService.stopFlow({ name: flowName }),
|
|
onSuccess: (detail) => {
|
|
queryClient.setQueryData(flowKeys.detail(flowName), detail)
|
|
queryClient.invalidateQueries({ queryKey: flowKeys.all, exact: true })
|
|
},
|
|
onError: () => showErrorToast("The flow could not be started or stopped."),
|
|
})
|
|
|
|
const pauseMutation = useMutation({
|
|
mutationFn: (next: boolean) =>
|
|
next
|
|
? FlowsService.pauseFlow({ name: flowName })
|
|
: FlowsService.resumeFlow({ name: flowName }),
|
|
// The engine answers with a flow_paused event, which is what the dock reads.
|
|
onError: () => showErrorToast("The flow could not be paused."),
|
|
})
|
|
|
|
const stepMutation = useMutation({
|
|
mutationFn: () => FlowsService.stepFlow({ name: flowName }),
|
|
// Says which node ran, or that nothing was held back — both are answers.
|
|
onSuccess: (result) => showSuccessToast(result.message),
|
|
onError: () => showErrorToast("The flow could not be stepped."),
|
|
})
|
|
|
|
const deleteMutation = useMutation({
|
|
mutationFn: () => FlowsService.deleteFlow({ name: flowName }),
|
|
onSuccess: () => {
|
|
queryClient.invalidateQueries({ queryKey: flowKeys.all })
|
|
setFlowPanelOpen(false)
|
|
navigate({ to: "/flows", replace: true })
|
|
},
|
|
onError: () => showErrorToast("The flow could not be deleted."),
|
|
})
|
|
|
|
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, sourceRef?: string) => {
|
|
const id = uniqueNodeId(definitions, sourceRef ?? 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: [],
|
|
// A shared node brings the code; the ports and settings are this
|
|
// flow's own.
|
|
...(sourceRef ? { source_ref: sourceRef } : {}),
|
|
}
|
|
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],
|
|
)
|
|
|
|
/**
|
|
* A message name is shared, not owned: renaming it where it is published
|
|
* strands every node still reading the old one. Offer to bring them along
|
|
* rather than deciding for the user.
|
|
*/
|
|
const proposeRename = useCallback(
|
|
(from: string, to: string) => {
|
|
if (!from || !to) return
|
|
const message = qualify(flowName, from)
|
|
const bound = definitions.filter((node) =>
|
|
[...(node.requires ?? []), ...(node.provides ?? [])].some(
|
|
(spec) => qualify(flowName, spec.name ?? "") === message,
|
|
),
|
|
)
|
|
if (bound.length) setRenamed({ from, to, count: bound.length })
|
|
},
|
|
[definitions, flowName],
|
|
)
|
|
|
|
/** Carry a rename to everything bound to the old name, as one edit. */
|
|
const applyRename = useCallback(
|
|
({ from, to }: MessageRename) => {
|
|
const message = qualify(flowName, from)
|
|
const follow = (specs: MessageSpec[] | undefined) =>
|
|
(specs ?? []).map((spec) =>
|
|
qualify(flowName, spec.name ?? "") === message
|
|
? { ...spec, name: to, port: "" }
|
|
: spec,
|
|
)
|
|
commit(
|
|
definitions.map((node) => ({
|
|
...node,
|
|
requires: follow(node.requires),
|
|
provides: follow(node.provides),
|
|
})),
|
|
)
|
|
setRenamed(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],
|
|
)
|
|
|
|
/** Put the stored draft live. Publishing what is queued means saving first. */
|
|
const publishFlow = useCallback(async () => {
|
|
await flush()
|
|
// Publish what was actually stored: the version only advances once the
|
|
// queued save has landed.
|
|
const current = queryClient.getQueryData<FlowDetail>(
|
|
flowKeys.detail(flowName),
|
|
)
|
|
publish.mutate(current?.definition.version ?? 1)
|
|
}, [flowName, flush, publish.mutate, queryClient])
|
|
|
|
/** Copy the selected nodes, with the code of the ones carrying their own. */
|
|
const copyNodes = useCallback(async () => {
|
|
const ids = new Set(
|
|
canvasNodes.filter((node) => node.selected).map((node) => node.id),
|
|
)
|
|
if (selectedId) ids.add(selectedId)
|
|
const picked = definitions.filter((node) => ids.has(node.id))
|
|
if (!picked.length) return
|
|
|
|
const sources: Record<string, string> = {}
|
|
await Promise.all(
|
|
picked
|
|
// A shared node already points at the library copy; only a private
|
|
// one has code that has to travel with it.
|
|
.filter((node) => !node.source_ref && sourceTypes.has(node.type ?? ""))
|
|
.map(async (node) => {
|
|
const { code } = await FlowsService.readNodeSource({
|
|
name: flowName,
|
|
nodeId: node.id,
|
|
})
|
|
sources[node.id] = code
|
|
}),
|
|
).catch(() => undefined)
|
|
|
|
localStorage.setItem(
|
|
CLIPBOARD_KEY,
|
|
JSON.stringify({ nodes: picked, sources } satisfies NodeClipboard),
|
|
)
|
|
}, [canvasNodes, definitions, flowName, selectedId, sourceTypes])
|
|
|
|
/**
|
|
* Paste them here, renamed around whatever this flow already holds. The
|
|
* document only names a node's code, so the copy is written per node once
|
|
* the definition is on its way.
|
|
*/
|
|
const pasteNodes = useCallback(() => {
|
|
const stored = localStorage.getItem(CLIPBOARD_KEY)
|
|
if (!stored) return
|
|
let clipboard: NodeClipboard
|
|
try {
|
|
clipboard = JSON.parse(stored)
|
|
} catch {
|
|
return
|
|
}
|
|
|
|
let pool = definitions
|
|
const pasted: NodeDef_Input[] = []
|
|
for (const node of clipboard.nodes ?? []) {
|
|
const position = freePosition(pool, {
|
|
// Offset, so a copy of a node in this flow is visibly its own.
|
|
x: (node.position?.x ?? 0) + 48,
|
|
y: (node.position?.y ?? 0) + 48,
|
|
})
|
|
const copy = { ...node, id: uniqueNodeId(pool, node.id), position }
|
|
pool = [...pool, copy]
|
|
pasted.push(copy)
|
|
}
|
|
if (!pasted.length) return
|
|
|
|
const nextCanvas = [
|
|
...canvasNodes,
|
|
...pasted.map(
|
|
(node) =>
|
|
({
|
|
id: node.id,
|
|
type: "flow",
|
|
position: node.position,
|
|
data: {},
|
|
}) as FlowCanvasNode,
|
|
),
|
|
]
|
|
setCanvasNodes(nextCanvas)
|
|
commit(pool, nextCanvas)
|
|
setSelectedId(pasted[pasted.length - 1].id)
|
|
|
|
clipboard.nodes.forEach((node, index) => {
|
|
const code = clipboard.sources?.[node.id]
|
|
if (code !== undefined) {
|
|
sourceMutation.mutate({ nodeId: pasted[index].id, code })
|
|
}
|
|
})
|
|
}, [canvasNodes, commit, definitions, setCanvasNodes, sourceMutation.mutate])
|
|
|
|
useShortcuts(
|
|
{
|
|
"mod+z": () => step(true),
|
|
"mod+shift+z": () => step(false),
|
|
"mod+c": () => void copyNodes(),
|
|
"mod+v": pasteNodes,
|
|
"mod+k": () => setPaletteOpen((open) => !open),
|
|
// Inside the code editor ⌘S applies that code, which the node panel
|
|
// owns; anywhere else on the canvas it puts the flow live.
|
|
"mod+s": (event) => {
|
|
if (!inCodeEditor(event.target)) void publishFlow()
|
|
},
|
|
},
|
|
// Both stay reachable while typing: one is the editor's own save, the
|
|
// other is how you reach anything at all.
|
|
["mod+s", "mod+k"],
|
|
)
|
|
|
|
const selected = definitions.find((node) => node.id === selectedId) ?? null
|
|
// A panel is the view you are working in, so it takes the room — but never
|
|
// the lanes the bars sit in: publishing is most wanted right after editing.
|
|
const panelOpen = Boolean(selected) || flowPanelOpen
|
|
|
|
return (
|
|
<>
|
|
<ReactFlow
|
|
nodes={shownNodes}
|
|
edges={edges}
|
|
onNodesChange={onNodesChange}
|
|
onNodeDrag={(_event, _node, dragged) => {
|
|
// An endpoint's position is ours, not React Flow's, so it only
|
|
// follows the pointer if we move it every frame.
|
|
const endpoints = dragged.filter(isEndpointNode)
|
|
if (!endpoints.length) return
|
|
setMoved((current) => {
|
|
const next = { ...current }
|
|
for (const node of endpoints) next[node.id] = node.position
|
|
return next
|
|
})
|
|
}}
|
|
onNodeDragStop={(_event, _node, dragged) => {
|
|
for (const node of dragged.filter(isEndpointNode)) {
|
|
// Written once at the end; every frame would be a write per pixel.
|
|
rememberPlacement(flowName, node.id, node.position)
|
|
}
|
|
const own = dragged.filter(isDocumentNode)
|
|
if (own.length) commit(definitions, mergeDragged(canvasNodes, own))
|
|
}}
|
|
onNodesDelete={(deleted) =>
|
|
deleteNodes(deleted.filter(isDocumentNode).map((node) => node.id))
|
|
}
|
|
onNodeClick={(_event, node) => {
|
|
// An endpoint is somewhere else's: opening its panel here would
|
|
// offer to edit a node this flow does not contain.
|
|
if (!isDocumentNode(node)) {
|
|
openEndpoint(node.id)
|
|
return
|
|
}
|
|
setFlowPanelOpen(false)
|
|
setSelectedId(node.id)
|
|
}}
|
|
onPaneClick={() => {
|
|
// Clicking the canvas is how you put a panel away, whichever one it
|
|
// is: the graph is what you went back to look at.
|
|
setSelectedId(null)
|
|
setFlowPanelOpen(false)
|
|
setEditorExpanded(false)
|
|
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),
|
|
sourceId: edge.source,
|
|
x: event.clientX,
|
|
y: event.clientY,
|
|
})
|
|
}}
|
|
onConnect={onConnect}
|
|
nodeTypes={nodeTypes}
|
|
edgeTypes={edgeTypes}
|
|
proOptions={{ hideAttribution: true }}
|
|
fitView
|
|
fitViewOptions={FIT_VIEW}
|
|
minZoom={0.25}
|
|
maxZoom={2}
|
|
nodeDragThreshold={5}
|
|
connectionRadius={30}
|
|
connectOnClick
|
|
autoPanOnConnect
|
|
edgesReconnectable={false}
|
|
deleteKeyCode={["Backspace", "Delete"]}
|
|
className="h-full w-full"
|
|
>
|
|
<CanvasBackground />
|
|
</ReactFlow>
|
|
|
|
{/*
|
|
* The bars keep their own lane rather than making way for a panel: they
|
|
* carry Publish, which is what you reach for the moment a node is done.
|
|
* The floating panel is inset out of that lane instead.
|
|
*/}
|
|
<div
|
|
className={cn(
|
|
"pointer-events-none absolute inset-0 transition-[right] duration-200",
|
|
panelOpen && !editorExpanded && "md:right-[27rem]",
|
|
)}
|
|
>
|
|
<CanvasTitle>
|
|
<span className="truncate px-3 py-1.5 text-sm font-medium">
|
|
{flowDoc.title || flowName}
|
|
</span>
|
|
</CanvasTitle>
|
|
|
|
<FlowDock
|
|
flow={flowName}
|
|
issues={issues}
|
|
running={runMutation.isPending}
|
|
enabled={detail.enabled ?? true}
|
|
paused={paused}
|
|
saving={saving.isPending}
|
|
hasDraft={detail.has_draft ?? false}
|
|
publishing={publish.isPending || saving.isPending}
|
|
onEditFlow={() => {
|
|
setSelectedId(null)
|
|
setFlowPanelOpen(true)
|
|
}}
|
|
onPublish={() => void publishFlow()}
|
|
logs={{
|
|
open: logsOpen,
|
|
node: logsNode,
|
|
onOpenChange: (open) => {
|
|
setLogsOpen(open)
|
|
if (!open) setLogsNode(null)
|
|
},
|
|
onClearNode: () => setLogsNode(null),
|
|
}}
|
|
onAddNode={() => setPaletteOpen(true)}
|
|
onRun={async () => {
|
|
// Running executes what is stored, so the queued edit goes first.
|
|
await flush()
|
|
runMutation.mutate()
|
|
}}
|
|
onTogglePause={() => pauseMutation.mutate(!paused)}
|
|
onStep={() => stepMutation.mutate()}
|
|
stepping={stepMutation.isPending}
|
|
onFocusNode={focusNode}
|
|
/>
|
|
</div>
|
|
|
|
{definitions.length === 0 ? (
|
|
<div
|
|
className="pointer-events-none absolute inset-0 flex items-center justify-center"
|
|
data-testid="flow-empty"
|
|
>
|
|
<div className="flex flex-col items-center gap-3 text-center">
|
|
<span className="flex size-14 items-center justify-center rounded-full bg-muted text-muted-foreground">
|
|
<Workflow className="size-6" />
|
|
</span>
|
|
<p className="text-lg font-medium">This flow is empty</p>
|
|
<p className="max-w-xs text-sm text-muted-foreground">
|
|
Add a node to get started. Press ⌘K, or use the plus in the bar
|
|
below.
|
|
</p>
|
|
</div>
|
|
</div>
|
|
) : null}
|
|
|
|
<FlowPanel
|
|
open={flowPanelOpen && !selected}
|
|
definition={flowDoc}
|
|
nodeCount={definitions.length}
|
|
// Confirmed, so it is a finished edit rather than a run of keystrokes.
|
|
onChange={(next) => commitDoc({ ...next, nodes: definitions }, true)}
|
|
onDelete={() => deleteMutation.mutate()}
|
|
enabled={detail.enabled ?? true}
|
|
toggling={enableMutation.isPending}
|
|
onToggleEnabled={(next) => enableMutation.mutate(next)}
|
|
hasDraft={detail.has_draft ?? false}
|
|
discarding={discard.isPending}
|
|
onDiscardDraft={() => {
|
|
discard.mutate(undefined, {
|
|
// The published document replaces what is on the canvas, and the
|
|
// version counter goes back with it.
|
|
onSuccess: () => {
|
|
setFlowPanelOpen(false)
|
|
onReload()
|
|
},
|
|
})
|
|
}}
|
|
onClose={() => setFlowPanelOpen(false)}
|
|
/>
|
|
|
|
<NodePanel
|
|
node={selected}
|
|
flow={flowName}
|
|
nodeTypes={nodeTypeInfo ?? []}
|
|
suggestions={suggestions}
|
|
expanded={editorExpanded}
|
|
onToggleExpand={() => setEditorExpanded((wide) => !wide)}
|
|
onChange={updateNode}
|
|
onRenameMessage={proposeRename}
|
|
onSaveSource={(code) => {
|
|
if (selected) sourceMutation.mutate({ nodeId: selected.id, code })
|
|
}}
|
|
// Sharing rewrites the stored document, so the canvas takes the
|
|
// server's copy rather than keeping its own.
|
|
onShared={onReload}
|
|
onClose={() => {
|
|
flush()
|
|
setEditorExpanded(false)
|
|
setSelectedId(null)
|
|
}}
|
|
onDelete={() => selected && deleteNodes([selected.id])}
|
|
/>
|
|
|
|
<EdgeInspector
|
|
edge={inspected}
|
|
flow={flowName}
|
|
onClose={() => setInspected(null)}
|
|
onUnbind={unbind}
|
|
/>
|
|
|
|
<CommandPalette
|
|
open={paletteOpen}
|
|
onOpenChange={setPaletteOpen}
|
|
nodeTypes={nodeTypeInfo ?? []}
|
|
flows={flows.data}
|
|
onAddNode={addNode}
|
|
onAddSharedNode={(libName) => addNode("python", libName)}
|
|
onRun={() => runMutation.mutate()}
|
|
/>
|
|
|
|
<Dialog
|
|
open={Boolean(rebind)}
|
|
onOpenChange={(open) => !open && setRebind(null)}
|
|
>
|
|
<DialogContent>
|
|
<DialogHeader>
|
|
<DialogTitle>How should {rebind?.nodeLabel} read this?</DialogTitle>
|
|
<DialogDescription>
|
|
Its "{rebind?.port}" input already reads{" "}
|
|
<span className="font-mono">{rebind?.from}</span>. It can take{" "}
|
|
<span className="font-mono">{rebind?.to}</span> as well, or
|
|
instead.
|
|
</DialogDescription>
|
|
</DialogHeader>
|
|
<DialogFooter className="sm:justify-between">
|
|
<Button variant="ghost" onClick={() => setRebind(null)}>
|
|
Cancel
|
|
</Button>
|
|
<div className="flex gap-2">
|
|
<Button
|
|
variant="outline"
|
|
onClick={() => {
|
|
if (rebind) {
|
|
applyBinding(rebind.nodeId, rebind.port, rebind.to)
|
|
}
|
|
setRebind(null)
|
|
}}
|
|
>
|
|
Replace
|
|
</Button>
|
|
<Button
|
|
onClick={() => {
|
|
if (rebind) {
|
|
addInputPort(rebind.nodeId, rebind.to, rebind.dtype)
|
|
}
|
|
setRebind(null)
|
|
}}
|
|
>
|
|
Add as another input
|
|
</Button>
|
|
</div>
|
|
</DialogFooter>
|
|
</DialogContent>
|
|
</Dialog>
|
|
|
|
<Dialog
|
|
open={Boolean(renamed)}
|
|
onOpenChange={(open) => !open && setRenamed(null)}
|
|
>
|
|
<DialogContent data-testid="rename-message">
|
|
<DialogHeader>
|
|
<DialogTitle>Rename this message everywhere?</DialogTitle>
|
|
<DialogDescription>
|
|
{renamed?.count === 1
|
|
? "One other node is"
|
|
: `${renamed?.count} other nodes are`}{" "}
|
|
still bound to <span className="font-mono">{renamed?.from}</span>.
|
|
They keep the old name unless they follow it to{" "}
|
|
<span className="font-mono">{renamed?.to}</span>.
|
|
</DialogDescription>
|
|
</DialogHeader>
|
|
<DialogFooter className="sm:justify-between">
|
|
<Button variant="ghost" onClick={() => setRenamed(null)}>
|
|
Leave them
|
|
</Button>
|
|
<Button onClick={() => renamed && applyRename(renamed)}>
|
|
Rename everywhere
|
|
</Button>
|
|
</DialogFooter>
|
|
</DialogContent>
|
|
</Dialog>
|
|
|
|
{/*
|
|
* Not dismissable: until one version wins, every further save fails, so
|
|
* there is nothing useful to go back to.
|
|
*/}
|
|
<Dialog open={conflict}>
|
|
<DialogContent
|
|
data-testid="save-conflict"
|
|
showCloseButton={false}
|
|
onEscapeKeyDown={(event) => event.preventDefault()}
|
|
onPointerDownOutside={(event) => event.preventDefault()}
|
|
>
|
|
<DialogHeader>
|
|
<DialogTitle>Someone else changed this flow</DialogTitle>
|
|
<DialogDescription>
|
|
Another editor saved <span className="font-mono">{flowName}</span>{" "}
|
|
while you were working on it. Load their version, or keep yours
|
|
and write over theirs.
|
|
</DialogDescription>
|
|
</DialogHeader>
|
|
<DialogFooter className="sm:justify-between">
|
|
<Button
|
|
variant="ghost"
|
|
onClick={() => {
|
|
void resolveConflict("theirs")
|
|
onReload()
|
|
}}
|
|
>
|
|
Load theirs
|
|
</Button>
|
|
<Button onClick={() => void resolveConflict("mine")}>
|
|
Keep mine
|
|
</Button>
|
|
</DialogFooter>
|
|
</DialogContent>
|
|
</Dialog>
|
|
</>
|
|
)
|
|
}
|
|
|
|
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<FlowCanvasNode>(toCanvasNodes(definitions))
|
|
}
|
|
|
|
export function FlowEditor({ flowName }: { flowName: string }) {
|
|
const navigate = useNavigate()
|
|
const onAuthFailure = useCallback(() => {
|
|
navigate({ to: "/login" })
|
|
}, [navigate])
|
|
// Bumped when the canvas has to take the server's document over its own:
|
|
// local edits live in state seeded on mount, so a remount is the reset.
|
|
const [epoch, setEpoch] = useState(0)
|
|
const reload = useCallback(() => setEpoch((n) => n + 1), [])
|
|
|
|
useFlowSocket(onAuthFailure)
|
|
|
|
return (
|
|
<ReactFlowProvider>
|
|
{/*
|
|
* Remounting per flow keeps canvas state from leaking between them, and
|
|
* it is what makes `fitView` run once per flow: xyflow queues the fit on
|
|
* mount and resolves it as soon as the nodes have been measured.
|
|
*/}
|
|
<FlowEditorInner
|
|
key={`${flowName}:${epoch}`}
|
|
flowName={flowName}
|
|
onReload={reload}
|
|
/>
|
|
</ReactFlowProvider>
|
|
)
|
|
}
|