Melvin Strobl
05dccdf88d
Flows can now be taken off the engine and put back. Stopped state lives in a runtime.json beside the flow, not in the flow document: the canvas autosaves that document, so a stopped flow would otherwise start itself again on the next edit. A stopped flow gets no subscriptions, schedules or webhooks, its nodes are skipped by the scheduler, and running it answers 409. Pausing holds a flow's nodes while its values keep arriving, so the canvas still shows what is coming in. Node code is user code and print is how it says things, so stdout is teed through a contextvar sink active only during a node execution — one event per execution, capped, so a chatty node cannot outrun the stream. A node that fails sends its traceback the same way, trimmed to the author's own frames. The dock gains a logs panel and a pause control; the dashboard replaces its placeholder with what is running, stopped or failing; the edge inspector can send the last message again. Single-stepping is deferred and noted: the scheduler keeps no progress between calls, so a step button would re-run the same node rather than advance.
993 lines
31 KiB
TypeScript
993 lines
31 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 { AnimatePresence } from "motion/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 { CommandPalette } from "./CommandPalette"
|
|
import { bindingsKey, deriveEdges, portOf, qualify } from "./deriveEdges"
|
|
import { EdgeInspector, type InspectedEdge } from "./EdgeInspector"
|
|
import { FIT_VIEW, FlowDock } from "./FlowDock"
|
|
import { FlowNode, type FlowNodeData } from "./FlowNode"
|
|
import { FlowPanel } from "./FlowPanel"
|
|
import { FlowTabs } from "./FlowTabs"
|
|
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 }
|
|
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: NodeDef_Input[][]
|
|
future: NodeDef_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
|
|
|
|
/**
|
|
* 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: NodeDef_Input[],
|
|
next: NodeDef_Input[],
|
|
settled: boolean,
|
|
) {
|
|
const sameNodes =
|
|
previous.length === next.length &&
|
|
previous.every((node, index) => node.id === next[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 = []
|
|
}
|
|
|
|
/** Monaco and every text field keep their own undo, so leave theirs alone. */
|
|
function isTextEntry(target: EventTarget | null): boolean {
|
|
return Boolean(
|
|
(target as Element | null)?.closest?.(
|
|
"input, textarea, [contenteditable='true'], .monaco-editor",
|
|
),
|
|
)
|
|
}
|
|
|
|
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 } = 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)
|
|
|
|
const [definitions, setDefinitions] = useState<NodeDef_Input[]>(
|
|
() => detail.definition.nodes ?? [],
|
|
)
|
|
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 covers the canvas, so its chrome steps aside.
|
|
const [editorExpanded, setEditorExpanded] = useState(false)
|
|
|
|
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 set of nodes on the canvas and on their way to the server. */
|
|
const apply = 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],
|
|
)
|
|
|
|
/** Apply a change and make it undoable. */
|
|
const commit = useCallback(
|
|
(nodes: NodeDef_Input[], positions?: FlowCanvasNode[]) => {
|
|
record(
|
|
history.current,
|
|
latest.current.nodes ?? [],
|
|
nodes,
|
|
Boolean(positions),
|
|
)
|
|
apply(nodes, positions)
|
|
},
|
|
[apply],
|
|
)
|
|
|
|
/**
|
|
* 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.nodes ?? [])
|
|
history.current.at = 0
|
|
// xyflow owns the positions, so hand it the remembered ones too.
|
|
const canvas = toCanvasNodes(remembered)
|
|
setCanvasNodes(canvas)
|
|
apply(remembered, canvas)
|
|
setInspected(null)
|
|
},
|
|
[apply, setCanvasNodes],
|
|
)
|
|
|
|
useEffect(() => {
|
|
const onKeyDown = (event: KeyboardEvent) => {
|
|
// Shift makes it a capital Z, so compare on the letter alone.
|
|
const undoKey =
|
|
(event.metaKey || event.ctrlKey) && event.key.toLowerCase() === "z"
|
|
if (!undoKey || isTextEntry(event.target)) return
|
|
event.preventDefault()
|
|
step(!event.shiftKey)
|
|
}
|
|
window.addEventListener("keydown", onKeyDown)
|
|
return () => window.removeEventListener("keydown", onKeyDown)
|
|
}, [step])
|
|
|
|
const typeLabels = useMemo(
|
|
() => new Map((nodeTypeInfo ?? []).map((info) => [info.type, info.title])),
|
|
[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])
|
|
|
|
// 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<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])
|
|
|
|
// 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 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 renameMutation = useMutation({
|
|
mutationFn: (newName: string) =>
|
|
FlowsService.renameFlow({
|
|
name: flowName,
|
|
requestBody: { new_name: newName },
|
|
}),
|
|
onSuccess: (detail) => {
|
|
queryClient.invalidateQueries({ queryKey: flowKeys.all })
|
|
setFlowPanelOpen(false)
|
|
navigate({
|
|
to: "/flows/$flowName",
|
|
params: { flowName: detail.definition.name },
|
|
replace: true,
|
|
})
|
|
},
|
|
onError: () =>
|
|
showErrorToast("That name is taken, or is not a valid flow name."),
|
|
})
|
|
|
|
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) => {
|
|
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],
|
|
)
|
|
|
|
/**
|
|
* 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],
|
|
)
|
|
|
|
const selected = definitions.find((node) => node.id === selectedId) ?? null
|
|
// A panel is the view you are working in: the bars would only compete with
|
|
// it, so they step aside until it closes. On a phone the panel covers them
|
|
// anyway, and its own close button is the way back.
|
|
const panelOpen = Boolean(selected) || flowPanelOpen
|
|
|
|
return (
|
|
<>
|
|
<ReactFlow
|
|
nodes={renderedNodes}
|
|
edges={edges}
|
|
onNodesChange={onNodesChange}
|
|
onNodeDragStop={(_event, _node, dragged) =>
|
|
commit(definitions, mergeDragged(canvasNodes, dragged))
|
|
}
|
|
onNodesDelete={(deleted) => deleteNodes(deleted.map((node) => node.id))}
|
|
onNodeClick={(_event, node) => {
|
|
setFlowPanelOpen(false)
|
|
setSelectedId(node.id)
|
|
}}
|
|
onPaneClick={() => {
|
|
setSelectedId(null)
|
|
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>
|
|
|
|
<AnimatePresence>
|
|
{panelOpen ? null : (
|
|
<FlowTabs
|
|
key="flow-tabs"
|
|
flows={flows.data}
|
|
active={flowName}
|
|
saving={saving.isPending}
|
|
hasDraft={detail.has_draft ?? false}
|
|
publishing={publish.isPending || saving.isPending}
|
|
onPublish={async () => {
|
|
// Publish what was actually stored: the version only advances
|
|
// once the queued save has landed.
|
|
await flush()
|
|
const current = queryClient.getQueryData<FlowDetail>(
|
|
flowKeys.detail(flowName),
|
|
)
|
|
publish.mutate(current?.definition.version ?? 1)
|
|
}}
|
|
onEditFlow={() => {
|
|
setSelectedId(null)
|
|
setFlowPanelOpen(true)
|
|
}}
|
|
/>
|
|
)}
|
|
|
|
{panelOpen ? null : (
|
|
<FlowDock
|
|
key="flow-dock"
|
|
flow={flowName}
|
|
issues={issues}
|
|
running={runMutation.isPending}
|
|
enabled={detail.enabled ?? true}
|
|
paused={paused}
|
|
onAddNode={() => setPaletteOpen(true)}
|
|
onRun={async () => {
|
|
// Running executes what is stored, so the queued edit goes first.
|
|
await flush()
|
|
runMutation.mutate()
|
|
}}
|
|
onTogglePause={() => pauseMutation.mutate(!paused)}
|
|
onFocusNode={focusNode}
|
|
/>
|
|
)}
|
|
</AnimatePresence>
|
|
|
|
{definitions.length === 0 ? (
|
|
<div className="pointer-events-none absolute inset-0 flex items-center justify-center">
|
|
<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={{ ...detail.definition, nodes: definitions }}
|
|
nodeCount={definitions.length}
|
|
renaming={renameMutation.isPending}
|
|
onChange={(next) => {
|
|
flush()
|
|
save({ ...next, nodes: definitions })
|
|
}}
|
|
onRename={async (newName) => {
|
|
await flush()
|
|
renameMutation.mutate(newName)
|
|
}}
|
|
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 })
|
|
}}
|
|
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}
|
|
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)
|
|
|
|
useEffect(() => {
|
|
return () => liveStore.reset()
|
|
}, [])
|
|
|
|
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>
|
|
)
|
|
}
|