import type { Edge, Node as FlowCanvasNode } from "@xyflow/react" import type { Endpoint, FlowDef_Input, NodeDef_Input } from "@/client" import type { FlowEdgeData } from "./deriveEdges" import { displayName, portOf, qualify } from "./deriveEdges" /** * Canvas elements for things wired into a flow that are not nodes in it. * * A dashboard control setting one of its messages, a tile showing one, or a * node in another flow across a dotted name. Without these the canvas shows a * value changing with nothing to account for it — and pulses whichever node * happens to be drawn as a producer, which did nothing. * * They are never part of the document: this returns display-only nodes that * are appended after everything which reads or writes `canvasNodes`, so an * autosave, an undo or a delete cannot reach them. */ /** Marks a canvas node as one of these, for anything that has to skip them. */ export const ENDPOINT_TYPE = "endpoint" /** The id prefixes an endpoint may carry, none of which is a node id. */ const PREFIXES = ["dashboard:", "flow:", "input:", "output:"] export type EndpointNodeData = { label: string kind: string detail: string /** Messages it publishes into this flow, and ones it reads out of it. */ provides: string[] requires: string[] [key: string]: unknown } export function isEndpointNode(node: { id: string }): boolean { return PREFIXES.some((prefix) => node.id.startsWith(prefix)) } /** * The flow's own boundary, as endpoints. * * An input is a message written from outside — a control, a run, the API — and * an output is what a batch run reports. They come from the working document * rather than from the server, so declaring one shows up on the canvas before * it has been published. * * An input a dashboard or another flow already accounts for is left out: the * label exists so a value has a visible origin, and it already has one. */ function boundaryOf( doc: FlowDef_Input, flow: string, accounted: Set, ): Endpoint[] { const found: Endpoint[] = [] for (const declared of doc.inputs ?? []) { const message = qualify(flow, declared.spec?.name ?? "") if (!message || accounted.has(message)) continue found.push({ kind: "input", id: `input:${message}`, label: displayName(flow, message), detail: declared.spec?.dtype ?? "float", provides: [message], requires: [], }) } // Only a batch flow has a result; a live one never finishes to have one. if (doc.mode === "batch") { for (const name of doc.outputs ?? []) { const message = qualify(flow, name) if (!message) continue found.push({ kind: "output", id: `output:${message}`, label: displayName(flow, message), detail: "result", provides: [], requires: [message], }) } } return found } /** * Build the endpoints and wire them to the nodes they touch. * * They carry no position: the caller lays them out together with the flow's * own nodes (see `layout.ts`), so an endpoint that publishes lands upstream of * what it feeds and one that reads lands downstream of what feeds it, by the * same rule that orders everything else. */ export function deriveEndpoints( endpoints: Endpoint[], definitions: NodeDef_Input[], flow: string, doc: FlowDef_Input, ): { nodes: FlowCanvasNode[]; edges: Edge[] } { const accounted = new Set(endpoints.flatMap((one) => one.provides ?? [])) const all = [...boundaryOf(doc, flow, accounted), ...endpoints] if (all.length === 0) return { nodes: [], edges: [] } // Which node consumes or produces each message, so a label can sit beside it. const consumers = new Map() const producers = new Map< string, { node: string; port: string; stream?: boolean }[] >() for (const node of definitions) { for (const spec of node.requires ?? []) { const message = qualify(flow, spec.name ?? "") if (!message) continue consumers.set(message, [ ...(consumers.get(message) ?? []), { node: node.id, port: portOf(spec) }, ]) } for (const spec of node.provides ?? []) { const message = qualify(flow, spec.name ?? "") if (!message) continue producers.set(message, [ ...(producers.get(message) ?? []), { node: node.id, port: portOf(spec), stream: spec.stream }, ]) } } const nodes: FlowCanvasNode[] = [] const edges: Edge[] = [] for (const endpoint of all) { const produces = endpoint.provides ?? [] const reads = endpoint.requires ?? [] nodes.push({ id: endpoint.id, type: ENDPOINT_TYPE, // Filled in by the layout, along with the flow's own nodes. position: { x: 0, y: 0 }, selectable: true, deletable: false, data: { label: endpoint.label, kind: endpoint.kind, detail: endpoint.detail ?? "", provides: produces, requires: reads, } satisfies EndpointNodeData, }) // An endpoint that publishes feeds every node consuming that message. for (const message of produces) { for (const consumer of consumers.get(message) ?? []) { edges.push({ id: `${endpoint.id}->${consumer.node}:${consumer.port}`, source: endpoint.id, sourceHandle: message, target: consumer.node, targetHandle: consumer.port, type: "live", data: { message, flow, producerId: endpoint.id, } satisfies FlowEdgeData, }) } } // One that reads is fed by every node producing it. for (const message of reads) { for (const producer of producers.get(message) ?? []) { edges.push({ id: `${producer.node}:${producer.port}->${endpoint.id}`, source: producer.node, sourceHandle: producer.port, target: endpoint.id, targetHandle: message, type: "live", data: { message, flow, producerId: `${flow}.${producer.node}`, stream: producer.stream, } satisfies FlowEdgeData, }) } } } return { nodes, edges } }