From f8c1cc31712741225f435245014ffbf0778df4ef Mon Sep 17 00:00:00 2001 From: Real Codesiman Date: Wed, 16 Sep 2026 13:37:13 +0700 Subject: [PATCH 1/2] feat(flows): create flow with content and publish in one call POST /v1/flows now accepts an optional `spec` (flow-spec DSL) or raw `nodes`/`edges` graph to seed the draft in the same call, plus a `publish: true` flag to validate and publish it immediately. - normalizeAuthoredGraph (flow-config): lays out positionless nodes with the existing BFS layoutNodes, remaps caller-authored node ids to internal createId() ids (baseNodeSchema requires digit-only ids, so an arbitrary caller id would fail publishFlowSchema), and defaults edge ids/handles to the addHandleEdge convention. - resolveFlowGraphInput: resolves flows.create's spec/nodes content into the graph flowService.createDraft persists, validating via publishFlowSchema only when publish is requested. - flowService.createDraft accepts an optional pre-resolved graph, falling back to today's single default start node. - Docs (docs/flows.md, capabilities API descriptions, MCP SKILL.md) updated to describe the new create-with-content workflow. --- .../create-flow-request-schema.test.ts | 90 +++++++++++ .../__tests__/flows-public-api.test.ts | 116 ++++++++++++- .../src/features/capabilities/api/public.ts | 4 +- apps/builder/src/features/flows/api/public.ts | 48 ++++-- .../flows/lib/compile-spec-to-graph.ts | 27 ++-- .../flows/lib/resolve-flow-graph-input.ts | 78 +++++++++ .../src/features/flows/schema/action.ts | 83 ++++++++++ apps/mcp-server/SKILL.md | 2 +- docs/flows.md | 25 ++- .../business/__tests__/flow.service.test.ts | 41 +++++ packages/business/src/flow/service.ts | 47 ++++-- .../authoring/normalize-graph.test.ts | 126 +++++++++++++++ packages/flow-config/src/authoring/errors.ts | 1 + .../src/authoring/normalize-graph.ts | 153 ++++++++++++++++++ packages/flow-config/src/index.ts | 6 + 15 files changed, 805 insertions(+), 42 deletions(-) create mode 100644 apps/builder/__tests__/create-flow-request-schema.test.ts create mode 100644 apps/builder/src/features/flows/lib/resolve-flow-graph-input.ts create mode 100644 packages/flow-config/__tests__/authoring/normalize-graph.test.ts create mode 100644 packages/flow-config/src/authoring/normalize-graph.ts diff --git a/apps/builder/__tests__/create-flow-request-schema.test.ts b/apps/builder/__tests__/create-flow-request-schema.test.ts new file mode 100644 index 0000000000..2cb404b73c --- /dev/null +++ b/apps/builder/__tests__/create-flow-request-schema.test.ts @@ -0,0 +1,90 @@ +import { describe, expect, test } from "vitest" +import { createFlowRequest } from "@/features/flows/schema/action" + +describe("createFlowRequest", () => { + test("accepts a bare name/folderId body with no content", () => { + const result = createFlowRequest.safeParse({ + name: "New flow", + folderId: null, + }) + + expect(result.success).toBe(true) + }) + + test("accepts a raw nodes/edges body with no positions", () => { + const result = createFlowRequest.safeParse({ + name: "New flow", + folderId: null, + nodes: [{ id: "n1" }, { id: "n2" }], + edges: [{ source: "n1", target: "n2" }], + }) + + expect(result.success).toBe(true) + }) + + test("rejects a body combining spec with nodes", () => { + const result = createFlowRequest.safeParse({ + name: "New flow", + folderId: null, + spec: { + formatVersion: 1, + name: "Spec flow", + steps: [{ type: "send", text: "Hi" }], + }, + nodes: [{ id: "n1" }], + }) + + expect(result.success).toBe(false) + expect(result.success ? [] : result.error.issues[0].path).toEqual(["spec"]) + }) + + test("rejects edges without nodes", () => { + const result = createFlowRequest.safeParse({ + name: "New flow", + folderId: null, + edges: [{ source: "n1", target: "n2" }], + }) + + expect(result.success).toBe(false) + expect(result.success ? [] : result.error.issues[0].path).toEqual(["edges"]) + }) + + test("rejects publish: true with neither spec nor nodes", () => { + const result = createFlowRequest.safeParse({ + name: "New flow", + folderId: null, + publish: true, + }) + + expect(result.success).toBe(false) + expect(result.success ? [] : result.error.issues[0].path).toEqual([ + "publish", + ]) + }) + + test("accepts publish: true alongside spec", () => { + const result = createFlowRequest.safeParse({ + name: "New flow", + folderId: null, + spec: { + formatVersion: 1, + name: "Spec flow", + steps: [{ type: "send", text: "Hi" }], + }, + publish: true, + }) + + expect(result.success).toBe(true) + }) + + test("accepts publish: true alongside nodes", () => { + const result = createFlowRequest.safeParse({ + name: "New flow", + folderId: null, + nodes: [{ id: "n1" }], + publish: true, + }) + + expect(result.success).toBe(true) + }) +}) diff --git a/apps/builder/__tests__/flows-public-api.test.ts b/apps/builder/__tests__/flows-public-api.test.ts index cf2863958d..7aa48fdde2 100644 --- a/apps/builder/__tests__/flows-public-api.test.ts +++ b/apps/builder/__tests__/flows-public-api.test.ts @@ -1,4 +1,7 @@ -import { FlowAuthoringException } from "@chatbotx.io/flow-config" +import { + FlowAuthoringException, + sendMessageNodeDefaultFn, +} from "@chatbotx.io/flow-config" import { beforeEach, describe, expect, test, vi } from "vitest" const SPEC_STEP_PATH_PATTERN = /^steps\[/ @@ -216,6 +219,117 @@ describe("POST /v1/flows", () => { data: { name: "New flow" }, }) }) + + test("spec content: createDraft receives a positioned graph and startNodeId; publish is not called", async () => { + flowService.createDraft.mockResolvedValueOnce({ id: "flow-1" }) + + await procedure.handler?.({ + context: { workspace: { id: "workspace-1" } }, + input: { + name: "New flow", + folderId: null, + spec: { + formatVersion: 1, + name: "Spec flow", + steps: [{ type: "send", text: "Hi" }], + }, + }, + }) + + expect(flowService.createDraft).toHaveBeenCalledTimes(1) + const call = flowService.createDraft.mock.calls[0][0] + expect(call.data).toEqual({ name: "New flow", folderId: null }) + expect(call.graph.nodes).toHaveLength(1) + expect(call.graph.nodes[0].position).toEqual({ x: 100, y: 100 }) + expect(call.graph.startNodeId).toBe(call.graph.nodes[0].id) + expect(flowVersionService.publish).not.toHaveBeenCalled() + }) + + test("raw graph content: createDraft receives normalized positions/edges and startNodeId", async () => { + flowService.createDraft.mockResolvedValueOnce({ id: "flow-1" }) + + await procedure.handler?.({ + context: { workspace: { id: "workspace-1" } }, + input: { + name: "New flow", + folderId: null, + nodes: [ + { id: "n1", type: "sendMessage", data: {} }, + { id: "n2", type: "sendMessage", data: {} }, + ], + edges: [{ source: "n1", target: "n2" }], + }, + }) + + const call = flowService.createDraft.mock.calls[0][0] + const [nodeN1, nodeN2] = call.graph.nodes + // Authored ids ("n1"/"n2") are request-scoped wiring tokens, remapped to + // fresh internal ids server-side — assert the relationship, not the + // literal authored string. + expect(nodeN1.id).not.toBe("n1") + expect(nodeN2.id).not.toBe("n2") + expect(nodeN1.position).toEqual({ x: 100, y: 100 }) + expect(call.graph.edges[0]).toEqual( + expect.objectContaining({ + source: nodeN1.id, + target: nodeN2.id, + sourceHandle: nodeN1.id, + targetHandle: nodeN2.id, + }), + ) + expect(call.graph.startNodeId).toBe(nodeN1.id) + expect(flowVersionService.publish).not.toHaveBeenCalled() + }) + + test("publish: true validates and publishes the graph createDraft returned an id for", async () => { + flowService.createDraft.mockResolvedValueOnce({ id: "flow-9" }) + // Authored id is deliberately non-numeric ("n1") — normalizeAuthoredGraph + // must remap it to an internal createId() id before publishFlowSchema + // validates it, since baseNodeSchema.id requires digits only. + const node = sendMessageNodeDefaultFn({ + nodeProps: { id: "n1" }, + dataProps: {}, + detailProps: {}, + }) + + await procedure.handler?.({ + context: { workspace: { id: "workspace-1" } }, + input: { + name: "New flow", + folderId: null, + nodes: [node], + edges: [], + publish: true, + }, + }) + + const createDraftCall = flowService.createDraft.mock.calls[0][0] + expect(createDraftCall.graph.nodes[0].id).not.toBe("n1") + expect(flowVersionService.publish).toHaveBeenCalledTimes(1) + expect(flowVersionService.publish).toHaveBeenCalledWith({ + workspaceId: "workspace-1", + flowId: "flow-9", + nodes: createDraftCall.graph.nodes, + edges: createDraftCall.graph.edges, + }) + }) + + test("publish: true rejects an invalid node before createDraft is called", async () => { + await expect( + procedure.handler?.({ + context: { workspace: { id: "workspace-1" } }, + input: { + name: "New flow", + folderId: null, + nodes: [{ id: "n1", type: "sendMessage", data: {} }], + edges: [], + publish: true, + }, + }), + ).rejects.toBeInstanceOf(FlowAuthoringException) + + expect(flowService.createDraft).not.toHaveBeenCalled() + }) }) describe("PATCH /v1/flows/{id}", () => { diff --git a/apps/builder/src/features/capabilities/api/public.ts b/apps/builder/src/features/capabilities/api/public.ts index a888675fd7..a026e5d26c 100644 --- a/apps/builder/src/features/capabilities/api/public.ts +++ b/apps/builder/src/features/capabilities/api/public.ts @@ -53,7 +53,7 @@ export const capabilitiesPublicRouter = { path: "/v1/capabilities", summary: "Discover workspace capabilities", description: - "Returns compact (id + name, plus a couple of decisive fields) lists of the workspace entities an agent needs to reference by id — inboxes, WhatsApp templates, custom/bot fields, tags, AI agents, sequences, and flows — plus the flow-spec DSL's step types and valid wait units/channels. Bot fields are reference data only; they cannot be used as a branch condition's `field` (only custom fields and built-in contact fields can). Use `include` (comma-separated) to narrow the response; omit it for the default set an agent needs to build a flow. Call this before `flows.publish`/`flows.updateDraft`/`flows.validate` so names in a flow spec resolve to real ids instead of guesses.", + "Returns compact (id + name, plus a couple of decisive fields) lists of the workspace entities an agent needs to reference by id — inboxes, WhatsApp templates, custom/bot fields, tags, AI agents, sequences, and flows — plus the flow-spec DSL's step types and valid wait units/channels. Bot fields are reference data only; they cannot be used as a branch condition's `field` (only custom fields and built-in contact fields can). Use `include` (comma-separated) to narrow the response; omit it for the default set an agent needs to build a flow. Call this before `flows.create`/`flows.publish`/`flows.updateDraft`/`flows.validate` so names in a flow spec resolve to real ids instead of guesses.", tags: ["Capabilities"], spec: mcpSpec({ visibility: "default", alwaysVisible: true }), }) @@ -76,7 +76,7 @@ export const schemasPublicRouter = { path: "/v1/schemas/flow-spec", summary: "Get JSON Schema for flow-spec DSL", description: - "Returns the JSON Schema for the `spec` object accepted by `flows.publish`'s, `flows.updateDraft`'s, and `flows.validate`'s `{ spec }` input — the authoritative reference for every step type's fields. Use `capabilities.get` first to resolve the names (templates, flows, tags, custom fields) a spec references into real ids.", + "Returns the JSON Schema for the `spec` object accepted by `flows.create`'s, `flows.publish`'s, `flows.updateDraft`'s, and `flows.validate`'s `{ spec }` input — the authoritative reference for every step type's fields. Use `capabilities.get` first to resolve the names (templates, flows, tags, custom fields) a spec references into real ids.", tags: ["Capabilities"], spec: mcpSpec({ visibility: "default" }), }) diff --git a/apps/builder/src/features/flows/api/public.ts b/apps/builder/src/features/flows/api/public.ts index ea7739d5b3..a2e7f5fb1b 100644 --- a/apps/builder/src/features/flows/api/public.ts +++ b/apps/builder/src/features/flows/api/public.ts @@ -23,8 +23,9 @@ import { compileAndValidateSpec, compileSpecToGraph, } from "../lib/compile-spec-to-graph" +import { resolveFlowGraphInput } from "../lib/resolve-flow-graph-input" import { - createFlowSchema, + createFlowRequest, flowSpecRequest, publishFlowRequest, publishFlowSchema, @@ -97,21 +98,38 @@ export const flowsPublicRouter = { path: "/v1/flows", summary: "Create flow", description: - "Starts a draft flow with its default start node. Use `flows.list` to inspect existing flows first, then call `flows.updateDraft` or `flows.publish` to complete it.", + "Creates a flow. With no `spec`/`nodes` it starts a draft with one default start node. Supply `spec` (flow-spec DSL, see `GET /v1/schemas/flow-spec`) or a raw `nodes`/`edges` graph to seed the draft in the same call — node `position`/`measured`, node ids, and edge ids/handles are all generated server-side (a raw node's `id` is only a request-scoped token for wiring `edges`), so only the flow's content needs sending. Add `publish: true` to validate the graph exactly like `flows.publish` and create the flow's first version immediately. Use `flows.list` to inspect existing flows first.", successStatus: 201, tags: ["Flows"], spec: mcpSpec({ visibility: "default" }), }) - .input(createFlowSchema) + .input(createFlowRequest) .output(z.object({ id: z.string() })) .errors(possibleErrorsOnCreatingResource) - .handler( - async ({ context, input }) => - await flowService.createDraft({ - workspaceId: context.workspace.id, - data: input, - }), - ), + .handler(async ({ context, input }) => { + const workspaceId = context.workspace.id + const { name, folderId, spec, nodes, edges, publish } = input + // Compile/validate before any write, so a rejected graph creates no flow row. + const graph = await resolveFlowGraphInput( + { spec, nodes, edges }, + workspaceId, + { validate: publish === true }, + ) + const flow = await flowService.createDraft({ + workspaceId, + data: { name, folderId }, + graph, + }) + if (publish && graph) { + await flowVersionService.publish({ + workspaceId, + flowId: flow.id, + nodes: graph.nodes, + edges: graph.edges, + }) + } + return flow + }), update: workspaceTokenAuthAPI .route({ @@ -211,9 +229,13 @@ export const flowsPublicRouter = { .input(flowSpecRequest) .output(publishFlowSchema) .errors(possibleErrorsOnMutatingResource) - .handler(async ({ context, input }) => - compileAndValidateSpec(input.spec, context.workspace.id), - ), + .handler(async ({ context, input }) => { + const { nodes, edges } = await compileAndValidateSpec( + input.spec, + context.workspace.id, + ) + return { nodes, edges } + }), updateDraft: workspaceTokenAuthAPI .route({ diff --git a/apps/builder/src/features/flows/lib/compile-spec-to-graph.ts b/apps/builder/src/features/flows/lib/compile-spec-to-graph.ts index 20a755cc94..8d742f1ea0 100644 --- a/apps/builder/src/features/flows/lib/compile-spec-to-graph.ts +++ b/apps/builder/src/features/flows/lib/compile-spec-to-graph.ts @@ -27,9 +27,16 @@ async function compileWithContext( export async function compileSpecToGraph( spec: FlowSpec, workspaceId: string, -): Promise<{ nodes: FlowVersionSchema[]; edges: EdgeSchema[] }> { - const { nodes, edges } = await compileWithContext(spec, workspaceId) - return { nodes, edges } +): Promise<{ + nodes: FlowVersionSchema[] + edges: EdgeSchema[] + startNodeId: string +}> { + const { nodes, edges, startNodeId } = await compileWithContext( + spec, + workspaceId, + ) + return { nodes, edges, startNodeId } } /** @@ -70,11 +77,13 @@ function mapPublishIssuePath( export async function compileAndValidateSpec( spec: FlowSpec, workspaceId: string, -): Promise<{ nodes: FlowVersionSchema[]; edges: EdgeSchema[] }> { - const { nodes, edges, specPathByNodeId } = await compileWithContext( - spec, - workspaceId, - ) +): Promise<{ + nodes: FlowVersionSchema[] + edges: EdgeSchema[] + startNodeId: string +}> { + const { nodes, edges, specPathByNodeId, startNodeId } = + await compileWithContext(spec, workspaceId) const result = publishFlowSchema.safeParse({ nodes, edges }) if (!result.success) { @@ -85,5 +94,5 @@ export async function compileAndValidateSpec( ) } - return result.data + return { ...result.data, startNodeId } } diff --git a/apps/builder/src/features/flows/lib/resolve-flow-graph-input.ts b/apps/builder/src/features/flows/lib/resolve-flow-graph-input.ts new file mode 100644 index 0000000000..feb9948749 --- /dev/null +++ b/apps/builder/src/features/flows/lib/resolve-flow-graph-input.ts @@ -0,0 +1,78 @@ +import { + type AuthoredEdgeInput, + type AuthoredNodeInput, + type EdgeSchema, + FlowAuthoringException, + type FlowSpec, + type FlowVersionSchema, + normalizeAuthoredGraph, + zodErrorToFlowAuthoringErrors, +} from "@chatbotx.io/flow-config" +import { publishFlowSchema } from "../schema/action" +import { + compileAndValidateSpec, + compileSpecToGraph, +} from "./compile-spec-to-graph" + +export type FlowGraphInput = { + spec?: FlowSpec + nodes?: AuthoredNodeInput[] + edges?: AuthoredEdgeInput[] +} + +export type ResolvedFlowGraph = { + nodes: FlowVersionSchema[] + edges: EdgeSchema[] + startNodeId: string +} + +/** + * Resolves one of `flows.create`'s two mutually exclusive content shapes + * (`{ spec }` or raw `{ nodes, edges }`) into the graph `flowService.createDraft` + * persists. `undefined` when the request carried no content, so the caller + * keeps today's default-start-node draft behavior. + * + * `options.validate` mirrors the `flows.publish` vs `flows.updateDraft` + * split: `true` runs the same `publishFlowSchema` check `flows.publish` runs + * (used when the caller also passed `publish: true`); `false` skips it, + * since draft nodes are not otherwise schema-validated. + */ +export function resolveFlowGraphInput( + input: FlowGraphInput, + workspaceId: string, + options: { validate: boolean }, +): Promise { + if (input.spec) { + return options.validate + ? compileAndValidateSpec(input.spec, workspaceId) + : compileSpecToGraph(input.spec, workspaceId) + } + + if (!input.nodes) { + return Promise.resolve(undefined) + } + + const graph = normalizeAuthoredGraph(input.nodes, input.edges ?? []) + + if (!options.validate) { + // Draft nodes are unvalidated by design, same as `flows.updateDraft`'s + // raw `{ nodes, edges }` path — `authorableNodeSchema` only requires an + // `id`, so the graph is not `FlowVersionSchema`-shaped until publish time. + return Promise.resolve(graph as unknown as ResolvedFlowGraph) + } + + const result = publishFlowSchema.safeParse({ + nodes: graph.nodes, + edges: graph.edges, + }) + if (!result.success) { + // No spec-path remapper here (unlike `compileAndValidateSpec`): the + // caller wrote these node paths directly, so the raw zod path is already + // the path they need to fix. + throw new FlowAuthoringException( + zodErrorToFlowAuthoringErrors(result.error, "invalidStep"), + ) + } + + return Promise.resolve({ ...result.data, startNodeId: graph.startNodeId }) +} diff --git a/apps/builder/src/features/flows/schema/action.ts b/apps/builder/src/features/flows/schema/action.ts index d55cd372d1..118e778740 100644 --- a/apps/builder/src/features/flows/schema/action.ts +++ b/apps/builder/src/features/flows/schema/action.ts @@ -17,6 +17,89 @@ export const createFlowSchema = z.object({ }) export type CreateFlowSchema = z.infer +const authorableNodeSchema = z.looseObject({ + id: z + .string() + .min(1) + .describe( + "Caller-chosen id, unique within this request, used only to wire `edges` together. The server assigns the node's actual persisted id — this value is not stored verbatim.", + ), +}) + +const authorableEdgeSchema = z.object({ + id: z.optional(z.string()).describe("Edge id. Omit to generate one."), + source: z.string().describe("Id of the node the edge leaves."), + sourceHandle: z + .optional(z.string()) + .describe( + "Handle on the source node. Omit for the node-level Continue handle.", + ), + target: z.string().describe("Id of the node the edge enters."), + targetHandle: z + .optional(z.string()) + .describe("Handle on the target node. Omit for the node's default input."), +}) + +// `createFlowRequest`'s body is always `name`/`folderId` plus at most one +// optional content shape, unlike `publishFlowRequest`/`updateDraftFlowRequest` +// which are pure `{ spec }` vs `{ nodes, edges }` alternatives. A +// `z.union([...])` here would duplicate `name`/`folderId` across branches and +// surface a union mismatch instead of naming the offending key, so this uses +// `.extend()` + `.superRefine()` instead. +export const createFlowRequest = createFlowSchema + .extend({ + spec: z + .optional(flowSpecSchema) + .describe( + "Flow-spec DSL document compiled server-side into the draft's nodes/edges (see `GET /v1/schemas/flow-spec`). Mutually exclusive with `nodes`/`edges`.", + ), + nodes: z + .optional(z.array(authorableNodeSchema).min(1)) + .describe( + "Raw node graph. `position`/`measured` are optional — omitted positions are laid out automatically — and each node's `id` is a request-scoped token for wiring `edges`, remapped to a real persisted id server-side. Mutually exclusive with `spec`.", + ), + edges: z + .optional(z.array(authorableEdgeSchema)) + .describe("Edges between `nodes`. Requires `nodes`."), + publish: z + .optional(z.boolean()) + .describe( + "Validate the supplied graph exactly like `flows.publish` and create the flow's first version immediately. Requires `spec` or `nodes`.", + ), + }) + .superRefine((data, ctx) => { + if ( + data.spec !== undefined && + (data.nodes !== undefined || data.edges !== undefined) + ) { + ctx.addIssue({ + code: "custom", + path: ["spec"], + message: + 'Cannot combine "spec" with "nodes"/"edges" in the same request body.', + }) + } + if (data.edges !== undefined && data.nodes === undefined) { + ctx.addIssue({ + code: "custom", + path: ["edges"], + message: '"edges" requires "nodes".', + }) + } + if ( + data.publish === true && + data.spec === undefined && + data.nodes === undefined + ) { + ctx.addIssue({ + code: "custom", + path: ["publish"], + message: '"publish" requires "spec" or "nodes".', + }) + } + }) +export type CreateFlowRequest = z.infer + export const updateFlowSchema = z.object({ name: z .optional(z.string().trim().min(1).max(255)) diff --git a/apps/mcp-server/SKILL.md b/apps/mcp-server/SKILL.md index f9facd3f5b..7560df3a0b 100644 --- a/apps/mcp-server/SKILL.md +++ b/apps/mcp-server/SKILL.md @@ -237,7 +237,7 @@ Call `capabilities_get` and `token_get` first — both are always visible regard |---|---| | `capabilities_get` | Discover the workspace's inboxes, templates, fields, tags, sequences, and flows — the ids other tools need. | | `token_get` | Get the calling token's workspace id, permission (`read_only`/`full`), and scopes. | -| `schemas_flow_spec` | JSON Schema for the flow-spec DSL `flows_publish`/`flows_update_draft`/`flows_validate` accept. | +| `schemas_flow_spec` | JSON Schema for the flow-spec DSL `flows_create`/`flows_publish`/`flows_update_draft`/`flows_validate` accept. | | Category | Tool | |---|---| diff --git a/docs/flows.md b/docs/flows.md index 7610b2ab1f..b4f7f72a15 100644 --- a/docs/flows.md +++ b/docs/flows.md @@ -23,7 +23,7 @@ versions, and reverting a draft back to the current published content. | Operation | Code path | Behavior | |---|---|---| -| Create | `createFlowAction` | Inserts the flow and a single draft version with a default start node. No published version exists yet. | +| Create | `createFlowAction` / public `flows.create` | Inserts the flow and a single draft version. The builder form and a bare API call seed it with a default start node; the public API can instead seed the draft with `{ spec }` or a raw `{ nodes, edges }` graph in the same call (see [Publish / validate workflow](#publish--validate-workflow)). No published version exists yet unless `publish: true` was sent. | | Edit / autosave | `updateDraftFlowVersionAction` | Overwrites the draft row's `nodes` and `edges` from the canvas. | | Publish | `publishFlowAction` | Copies the current draft content into a new published snapshot, marks it `isLatest`, and updates `currentVersionId`. | | Restore older version | `flowVersionService.restore()` | Marks the chosen version as current published, updates `currentVersionId`, and copies that version's content into the draft row. | @@ -124,17 +124,34 @@ non-terminal step continues into the node-level "Continue" edge by default. ### Publish / validate workflow -1. **`flows.validate`** — compiles a `{ spec }` and validates the result exactly like +1. **`flows.create`** — accepts `name`/`folderId` plus, optionally, exactly one of + `{ spec }` or a raw `{ nodes, edges }` graph to seed the draft in the same call + (a body carrying both, or `edges` without `nodes`, is rejected with a 422 naming + the offending key). For the raw graph shape, node `position`/`measured`, node + ids, and edge `id`/handles are all optional or caller-arbitrary — filled in or + remapped server-side by + [`normalizeAuthoredGraph`](../packages/flow-config/src/authoring/normalize-graph.ts), + which lays out positionless nodes with the same BFS `layoutNodes` the spec + compiler uses, remaps every node's caller-supplied `id` to a fresh internal + `createId()` id (a raw node's `id` is only a request-scoped token for wiring + `edges` — it is never the persisted id, since persisted node ids are numeric + snowflakes and a caller id like `"n1"` would otherwise fail `publishFlowSchema`), + and defaults edge handles to the node-id convention `addHandleEdge` produces + against the remapped ids. Add `publish: true` to validate the resulting graph + exactly like `flows.publish` and create the flow's first version in the same + call; omitting content or `publish` keeps today's default-start-node draft + behavior. +2. **`flows.validate`** — compiles a `{ spec }` and validates the result exactly like `flows.publish` would, without persisting anything. On success it returns the compiled `{ nodes, edges }` graph; on failure a 422 with structured errors (`path`/`code`/`message`/`hint`/`candidates`), each `path` remapped back onto the *spec-relative* location (e.g. `steps[2].buttons[0].then[0]`) rather than a compiled-node path. Fix and retry before publishing. -2. **`flows.publish`** — accepts either the raw `{ nodes, edges }` graph the builder +3. **`flows.publish`** — accepts either the raw `{ nodes, edges }` graph the builder UI sends, or `{ spec }`; a request must send exactly one of the two shapes — a body carrying both is rejected with a 422 rather than silently discarding one of them. Creates an immutable version from the draft and syncs the draft to match. -3. **`flows.updateDraft`** — same either/or `{ nodes, edges }` vs `{ spec }` +4. **`flows.updateDraft`** — same either/or `{ nodes, edges }` vs `{ spec }` acceptance as `flows.publish`, but overwrites the draft in place without publishing; draft nodes are not otherwise validated. diff --git a/packages/business/__tests__/flow.service.test.ts b/packages/business/__tests__/flow.service.test.ts index 20c14b8bb0..7b1ca5aef6 100644 --- a/packages/business/__tests__/flow.service.test.ts +++ b/packages/business/__tests__/flow.service.test.ts @@ -495,4 +495,45 @@ describe("flowService.createDraft", () => { ) expect(result).toEqual({ id: "flow-1" }) }) + + test("uses a pre-resolved graph instead of the default start node when one is supplied", async () => { + mockDbTransaction.mockImplementation(async (callback) => + callback(transaction), + ) + mockInsertValues.mockImplementation(() => + Object.assign(Promise.resolve(undefined), { + returning: mockInsertReturning, + }), + ) + mockCreateId + .mockReturnValueOnce("flow-2") + .mockReturnValueOnce("analytics-2") + .mockReturnValueOnce("version-2") + mockInsertReturning.mockResolvedValue([{ id: "flow-2" }]) + + const result = await flowService.createDraft({ + workspaceId: "ws-1", + data: { name: "New flow" }, + graph: { + nodes: [{ id: "node-1" }], + edges: [{ id: "e1", source: "node-1", target: "node-1" }], + startNodeId: "node-1", + }, + }) + + expect(mockInsert).toHaveBeenNthCalledWith(3, flowVersionModel) + expect(mockInsertValues).toHaveBeenNthCalledWith( + 3, + expect.objectContaining({ + id: "version-2", + workspaceId: "ws-1", + flowId: "flow-2", + nodes: [{ id: "node-1" }], + edges: [{ id: "e1", source: "node-1", target: "node-1" }], + isDraft: true, + startNodeId: "node-1", + }), + ) + expect(result).toEqual({ id: "flow-2" }) + }) }) diff --git a/packages/business/src/flow/service.ts b/packages/business/src/flow/service.ts index 4704c25ff9..c8a44dee1d 100644 --- a/packages/business/src/flow/service.ts +++ b/packages/business/src/flow/service.ts @@ -79,6 +79,26 @@ const resolveManifestIdMap = async ( return { idMap, createdIds } } +/** `flows.create`'s default content when no `spec`/`nodes` is supplied: today's single "Send Message" start node. */ +const defaultDraftGraph = (): { + nodes: FlowVersionModel["nodes"] + edges: FlowVersionModel["edges"] + startNodeId: string +} => { + const defaultNode = sendMessageNodeDefaultFn({ + dataProps: { + name: "Send Message #1", + isStartNode: true, + }, + }) + return { + // biome-ignore lint/suspicious/noExplicitAny: temporary any to bypass circular dependency between flow and flow version + nodes: [defaultNode as any], + edges: [], + startNodeId: defaultNode.id, + } +} + class FlowService extends BaseService { async findBy( input: { workspaceId: string; id: string }, @@ -285,8 +305,12 @@ class FlowService extends BaseService { } /** - * The builder create-flow form's flow: a single new (unpublished) draft - * version seeded with one default "Send Message" start node — unlike + * The builder create-flow form's flow, and the public `flows.create` + * API's default: a single new (unpublished) draft version. With no + * `graph`, it's seeded with one default "Send Message" start node + * (`defaultDraftGraph`); the public API passes a pre-resolved `graph` + * (compiled from `spec`, or normalized from a raw `nodes`/`edges` body) + * to seed the draft with caller-supplied content instead. Unlike * `createPublishedDefault` (template install: draft + published version * pair, external `tx`), this owns its own transaction and audits the * result. @@ -294,6 +318,11 @@ class FlowService extends BaseService { async createDraft(input: { workspaceId: string data: { name: string; folderId?: string | null } + graph?: { + nodes: FlowVersionModel["nodes"] + edges: FlowVersionModel["edges"] + startNodeId: string + } }): Promise<{ id: string }> { const { workspaceId, data } = input @@ -305,12 +334,7 @@ class FlowService extends BaseService { }) } - const defaultNode = sendMessageNodeDefaultFn({ - dataProps: { - name: "Send Message #1", - isStartNode: true, - }, - }) + const graph = input.graph ?? defaultDraftGraph() const flow = await db.transaction(async (tx) => { const flowId = createId() @@ -333,11 +357,10 @@ class FlowService extends BaseService { id: createId(), workspaceId, flowId, - // biome-ignore lint/suspicious/noExplicitAny: temporary any to bypass circular dependency between flow and flow version - nodes: [defaultNode as any], - edges: [], + nodes: graph.nodes, + edges: graph.edges, isDraft: true, - startNodeId: defaultNode.id, + startNodeId: graph.startNodeId, }) return created diff --git a/packages/flow-config/__tests__/authoring/normalize-graph.test.ts b/packages/flow-config/__tests__/authoring/normalize-graph.test.ts new file mode 100644 index 0000000000..c759e579cf --- /dev/null +++ b/packages/flow-config/__tests__/authoring/normalize-graph.test.ts @@ -0,0 +1,126 @@ +import { describe, expect, test } from "vitest" +import { FlowAuthoringException } from "../../src/authoring/errors" +import { normalizeAuthoredGraph } from "../../src/authoring/normalize-graph" +import { DEFAULT_NODE_MEASURED } from "../../src/nodes/base" + +const COLUMN_WIDTH = DEFAULT_NODE_MEASURED.width + 120 + +describe("normalizeAuthoredGraph", () => { + test("lays out a chain with no positions like the compiler would", () => { + const result = normalizeAuthoredGraph( + [{ id: "a" }, { id: "b" }, { id: "c" }], + [ + { source: "a", target: "b" }, + { source: "b", target: "c" }, + ], + ) + + // Node order is preserved, so index still tracks the authored a/b/c order + // even though the returned ids are freshly generated, not "a"/"b"/"c". + const [nodeA, nodeB, nodeC] = result.nodes + + expect(nodeA.position).toEqual({ x: 100, y: 100 }) + expect(nodeB.position).toEqual({ x: 100 + COLUMN_WIDTH, y: 100 }) + expect(nodeC.position).toEqual({ x: 100 + COLUMN_WIDTH * 2, y: 100 }) + for (const node of result.nodes) { + expect(node.measured).toEqual(DEFAULT_NODE_MEASURED) + } + }) + + test("keeps a caller-supplied position instead of laying it out", () => { + const result = normalizeAuthoredGraph( + [{ id: "a", position: { x: 42, y: 7 } }], + [], + ) + + expect(result.nodes[0].position).toEqual({ x: 42, y: 7 }) + }) + + test("remaps every authored node id to a fresh internal id", () => { + const result = normalizeAuthoredGraph([{ id: "a" }, { id: "b" }], []) + + const ids = result.nodes.map((node) => node.id) + expect(ids).toHaveLength(2) + expect(new Set(ids).size).toBe(2) + for (const id of ids) { + expect(id).not.toBe("a") + expect(id).not.toBe("b") + } + }) + + test("generates edge ids and node-id handles, rewriting them to the remapped node ids", () => { + const result = normalizeAuthoredGraph( + [{ id: "a" }, { id: "b" }, { id: "c" }], + [ + { source: "a", target: "b" }, + { + id: "custom-edge", + source: "b", + sourceHandle: "yes", + target: "c", + targetHandle: "in", + }, + ], + ) + const [nodeA, nodeB, nodeC] = result.nodes + + const generated = result.edges.find((edge) => edge.source === nodeA.id) + expect(generated?.id).toBeTruthy() + expect(generated?.sourceHandle).toBe(nodeA.id) + expect(generated?.target).toBe(nodeB.id) + expect(generated?.targetHandle).toBe(nodeB.id) + + // A caller-supplied handle is an arbitrary handle name, not a node id — + // it passes through untouched even though source/target are remapped. + const supplied = result.edges.find((edge) => edge.id === "custom-edge") + expect(supplied).toEqual({ + id: "custom-edge", + source: nodeB.id, + sourceHandle: "yes", + target: nodeC.id, + targetHandle: "in", + }) + }) + + test("start node falls back to the first node when none is flagged", () => { + const result = normalizeAuthoredGraph([{ id: "a" }, { id: "b" }], []) + + expect(result.startNodeId).toBe(result.nodes[0].id) + }) + + test("start node uses the flagged node, and exactly one node stays flagged", () => { + const result = normalizeAuthoredGraph( + [ + { id: "a", data: { isStartNode: true } }, + { id: "b", data: { isStartNode: true } }, + ], + [], + ) + + expect(result.startNodeId).toBe(result.nodes[0].id) + const flaggedNodes = result.nodes.filter( + (node) => node.data?.isStartNode === true, + ) + expect(flaggedNodes).toHaveLength(1) + expect(flaggedNodes[0].id).toBe(result.nodes[0].id) + }) + + test("collects a duplicate node id and an unknown edge target in one exception", () => { + expect.assertions(3) + try { + normalizeAuthoredGraph( + [{ id: "a" }, { id: "a" }], + [{ source: "a", target: "missing" }], + ) + } catch (error) { + expect(error).toBeInstanceOf(FlowAuthoringException) + const authoringError = error as FlowAuthoringException + expect(authoringError.errors.map((issue) => issue.path)).toContain( + "nodes[1].id", + ) + expect(authoringError.errors.map((issue) => issue.path)).toContain( + "edges[0].target", + ) + } + }) +}) diff --git a/packages/flow-config/src/authoring/errors.ts b/packages/flow-config/src/authoring/errors.ts index 0f43a5d29d..5828f61ae2 100644 --- a/packages/flow-config/src/authoring/errors.ts +++ b/packages/flow-config/src/authoring/errors.ts @@ -20,6 +20,7 @@ export type FlowAuthoringErrorCode = | "selfLoopGoto" | "duplicateStepId" | "invalidStep" + | "invalidGraph" | "compileFailed" | "templateNotApproved" diff --git a/packages/flow-config/src/authoring/normalize-graph.ts b/packages/flow-config/src/authoring/normalize-graph.ts new file mode 100644 index 0000000000..f7cf3a8e40 --- /dev/null +++ b/packages/flow-config/src/authoring/normalize-graph.ts @@ -0,0 +1,153 @@ +import { createId } from "@chatbotx.io/utils" +import { DEFAULT_NODE_MEASURED } from "../nodes/base" +import type { EdgeSchema } from "../nodes/index" +import { FlowAuthoringException } from "./errors" +import { type LayoutPosition, layoutNodes } from "./layout" + +export type AuthoredNodeInput = { + id: string + position?: LayoutPosition + measured?: { width: number; height: number } + data?: { isStartNode?: boolean } +} + +export type AuthoredEdgeInput = { + id?: string + source: string + sourceHandle?: string + target: string + targetHandle?: string +} + +export type NormalizedAuthoredGraph = { + nodes: T[] + edges: EdgeSchema[] + startNodeId: string +} + +/** + * Normalizes a caller-authored `{ nodes, edges }` graph into the shape the + * builder canvas and `flowVersionModel` expect: every node's id is remapped + * to a fresh internal `createId()` id (the caller's id is only a + * request-scoped token for wiring `edges`, since node ids elsewhere in this + * codebase are numeric snowflakes — see `zodBigintAsString`), every node + * gets a `position` (a caller-supplied one always wins; otherwise it's laid + * out with the same `layoutNodes` BFS the spec compiler uses) and a + * `measured` footprint, and every edge gets an `id` and handles remapped to + * the new node ids, using the same node-id-as-handle convention + * `addHandleEdge`/`addContinueEdge` use for a "Continue" edge + * (`./compile.ts`). Exactly one node ends up with `data.isStartNode: true`. + * + * Collects every problem before throwing, mirroring `compileFlowSpec`'s + * all-errors-at-once contract so a caller sees every issue in one 422. + */ +export function normalizeAuthoredGraph( + nodes: readonly T[], + edges: readonly AuthoredEdgeInput[], +): NormalizedAuthoredGraph { + const errors: { + path: string + code: "invalidGraph" + message: string + }[] = [] + + if (nodes.length === 0) { + errors.push({ + path: "nodes", + code: "invalidGraph", + message: "A flow needs at least one node.", + }) + } + + const nodeIds = new Set() + for (const [index, node] of nodes.entries()) { + if (nodeIds.has(node.id)) { + errors.push({ + path: `nodes[${index}].id`, + code: "invalidGraph", + message: `Duplicate node id "${node.id}".`, + }) + continue + } + nodeIds.add(node.id) + } + + for (const [index, edge] of edges.entries()) { + if (!nodeIds.has(edge.source)) { + errors.push({ + path: `edges[${index}].source`, + code: "invalidGraph", + message: `Edge source "${edge.source}" does not match any node id.`, + }) + } + if (!nodeIds.has(edge.target)) { + errors.push({ + path: `edges[${index}].target`, + code: "invalidGraph", + message: `Edge target "${edge.target}" does not match any node id.`, + }) + } + } + + if (errors.length > 0) { + throw new FlowAuthoringException(errors) + } + + // A caller-authored id is just a request-scoped token for wiring `edges` + // together; it is never persisted as the node's actual id. Every real + // node id in this codebase is a `createId()` snowflake (`baseNodeSchema.id` + // / `zodBigintAsString` requires digits only), and `publishFlowSchema` + // enforces that pattern — an arbitrary caller string like "n1" would + // otherwise 422 the moment `publish: true` validates the graph. Remapping + // unconditionally (not only on the publish path) keeps the persisted + // shape identical whether or not the caller also asked to publish. + const idByAuthoredId = new Map( + nodes.map((node) => [node.id, createId()] as const), + ) + const internalId = (authoredId: string): string => { + const mapped = idByAuthoredId.get(authoredId) + if (!mapped) { + throw new Error(`Unmapped authored node id "${authoredId}".`) + } + return mapped + } + + const startNodeAuthoredId = + nodes.find((node) => node.data?.isStartNode === true)?.id ?? nodes[0].id + const startNodeId = internalId(startNodeAuthoredId) + + const positions = layoutNodes( + nodes.map((node) => node.id), + edges, + startNodeAuthoredId, + ) + + const resolvedNodes = nodes.map((node) => { + const id = internalId(node.id) + return { + ...node, + id, + position: node.position ?? positions.get(node.id), + measured: node.measured ?? DEFAULT_NODE_MEASURED, + data: { ...node.data, isStartNode: id === startNodeId }, + } + }) as T[] + + const resolvedEdges: EdgeSchema[] = edges.map((edge) => { + const source = internalId(edge.source) + const target = internalId(edge.target) + return { + id: edge.id ?? createId(), + source, + // A caller-supplied handle is an arbitrary handle name (e.g. a + // quick-reply button's handle), not a node id — leave it untouched. + // The default is the node-id-as-handle "Continue" convention, so it + // must track the remapped id, not the authored one. + sourceHandle: edge.sourceHandle ?? source, + target, + targetHandle: edge.targetHandle ?? target, + } + }) + + return { nodes: resolvedNodes, edges: resolvedEdges, startNodeId } +} diff --git a/packages/flow-config/src/index.ts b/packages/flow-config/src/index.ts index 4370537011..b3cbb91de5 100644 --- a/packages/flow-config/src/index.ts +++ b/packages/flow-config/src/index.ts @@ -11,6 +11,12 @@ export { formatZodPathSegment, zodErrorToFlowAuthoringErrors, } from "./authoring/errors" +export { + type AuthoredEdgeInput, + type AuthoredNodeInput, + type NormalizedAuthoredGraph, + normalizeAuthoredGraph, +} from "./authoring/normalize-graph" export { type FlowSpec, type FlowSpecStepType, From 1985cccf11c4a0babe0318cb275e524b39198dd3 Mon Sep 17 00:00:00 2001 From: Real Codesiman Date: Wed, 16 Sep 2026 16:24:22 +0700 Subject: [PATCH 2/2] feat(flows): make flows.create transactional and return persisted node ids - flows.create publish:true now runs FlowService.createPublished, which inserts the flow, its analytics session, draft version, and published version in one transaction via createPublishedDefault, so a downstream failure never leaves an orphaned unpublished flow. - normalizeAuthoredGraph returns a nodeIds map (authored id -> persisted id); flows.create's response surfaces it so callers can reference the nodes they just authored without a round-trip fetch. - Document the nodeIds response and the verbatim (non-remapped) id behavior of flows.updateDraft's raw-graph path in docs/flows.md. --- .../__tests__/flows-public-api.test.ts | 39 ++++++--- apps/builder/src/features/flows/api/public.ts | 42 ++++++---- .../flows/lib/resolve-flow-graph-input.ts | 17 +++- docs/flows.md | 18 +++-- .../business/__tests__/flow.service.test.ts | 80 +++++++++++++++++++ packages/business/src/flow/service.ts | 46 +++++++++++ .../authoring/normalize-graph.test.ts | 8 ++ .../src/authoring/normalize-graph.ts | 23 ++++-- 8 files changed, 229 insertions(+), 44 deletions(-) diff --git a/apps/builder/__tests__/flows-public-api.test.ts b/apps/builder/__tests__/flows-public-api.test.ts index 7aa48fdde2..6c18db914c 100644 --- a/apps/builder/__tests__/flows-public-api.test.ts +++ b/apps/builder/__tests__/flows-public-api.test.ts @@ -56,6 +56,7 @@ const flowService = { list: vi.fn(), findById: vi.fn(), createDraft: vi.fn(), + createPublished: vi.fn(), update: vi.fn(), deleteMany: vi.fn(), duplicate: vi.fn(), @@ -223,7 +224,7 @@ describe("POST /v1/flows", () => { test("spec content: createDraft receives a positioned graph and startNodeId; publish is not called", async () => { flowService.createDraft.mockResolvedValueOnce({ id: "flow-1" }) - await procedure.handler?.({ + const response = await procedure.handler?.({ context: { workspace: { id: "workspace-1" } }, input: { name: "New flow", @@ -243,12 +244,15 @@ describe("POST /v1/flows", () => { expect(call.graph.nodes[0].position).toEqual({ x: 100, y: 100 }) expect(call.graph.startNodeId).toBe(call.graph.nodes[0].id) expect(flowVersionService.publish).not.toHaveBeenCalled() + expect(flowService.createPublished).not.toHaveBeenCalled() + // A `{ spec }` graph has no authored node ids to report. + expect(response.nodeIds).toBeUndefined() }) test("raw graph content: createDraft receives normalized positions/edges and startNodeId", async () => { flowService.createDraft.mockResolvedValueOnce({ id: "flow-1" }) - await procedure.handler?.({ + const response = await procedure.handler?.({ context: { workspace: { id: "workspace-1" } }, input: { name: "New flow", @@ -279,10 +283,14 @@ describe("POST /v1/flows", () => { ) expect(call.graph.startNodeId).toBe(nodeN1.id) expect(flowVersionService.publish).not.toHaveBeenCalled() + expect(flowService.createPublished).not.toHaveBeenCalled() + // The response reports the authored → persisted id mapping so the + // caller can edit the graph it just created without a `flows.get`. + expect(response.nodeIds).toEqual({ n1: nodeN1.id, n2: nodeN2.id }) }) - test("publish: true validates and publishes the graph createDraft returned an id for", async () => { - flowService.createDraft.mockResolvedValueOnce({ id: "flow-9" }) + test("publish: true validates the graph and creates it atomically via flowService.createPublished, not createDraft + publish", async () => { + flowService.createPublished.mockResolvedValueOnce({ id: "flow-9" }) // Authored id is deliberately non-numeric ("n1") — normalizeAuthoredGraph // must remap it to an internal createId() id before publishFlowSchema // validates it, since baseNodeSchema.id requires digits only. @@ -292,7 +300,7 @@ describe("POST /v1/flows", () => { detailProps: {}, }) - await procedure.handler?.({ + const response = await procedure.handler?.({ context: { workspace: { id: "workspace-1" } }, input: { name: "New flow", @@ -303,14 +311,19 @@ describe("POST /v1/flows", () => { }, }) - const createDraftCall = flowService.createDraft.mock.calls[0][0] - expect(createDraftCall.graph.nodes[0].id).not.toBe("n1") - expect(flowVersionService.publish).toHaveBeenCalledTimes(1) - expect(flowVersionService.publish).toHaveBeenCalledWith({ - workspaceId: "workspace-1", - flowId: "flow-9", - nodes: createDraftCall.graph.nodes, - edges: createDraftCall.graph.edges, + expect(flowService.createDraft).not.toHaveBeenCalled() + expect(flowVersionService.publish).not.toHaveBeenCalled() + expect(flowService.createPublished).toHaveBeenCalledTimes(1) + const createPublishedCall = flowService.createPublished.mock.calls[0][0] + expect(createPublishedCall.workspaceId).toBe("workspace-1") + expect(createPublishedCall.data).toEqual({ + name: "New flow", + folderId: null, + }) + expect(createPublishedCall.graph.nodes[0].id).not.toBe("n1") + expect(response).toEqual({ + id: "flow-9", + nodeIds: { n1: createPublishedCall.graph.nodes[0].id }, }) }) diff --git a/apps/builder/src/features/flows/api/public.ts b/apps/builder/src/features/flows/api/public.ts index a2e7f5fb1b..f28b2339ef 100644 --- a/apps/builder/src/features/flows/api/public.ts +++ b/apps/builder/src/features/flows/api/public.ts @@ -98,13 +98,22 @@ export const flowsPublicRouter = { path: "/v1/flows", summary: "Create flow", description: - "Creates a flow. With no `spec`/`nodes` it starts a draft with one default start node. Supply `spec` (flow-spec DSL, see `GET /v1/schemas/flow-spec`) or a raw `nodes`/`edges` graph to seed the draft in the same call — node `position`/`measured`, node ids, and edge ids/handles are all generated server-side (a raw node's `id` is only a request-scoped token for wiring `edges`), so only the flow's content needs sending. Add `publish: true` to validate the graph exactly like `flows.publish` and create the flow's first version immediately. Use `flows.list` to inspect existing flows first.", + "Creates a flow. With no `spec`/`nodes` it starts a draft with one default start node. Supply `spec` (flow-spec DSL, see `GET /v1/schemas/flow-spec`) or a raw `nodes`/`edges` graph to seed the draft in the same call — node `position`/`measured`, node ids, and edge ids/handles are all generated server-side (a raw node's `id` is only a request-scoped token for wiring `edges`; the response's `nodeIds` maps each authored id to its persisted id). Add `publish: true` to validate the graph exactly like `flows.publish` and create the flow's first version immediately. Use `flows.list` to inspect existing flows first.", successStatus: 201, tags: ["Flows"], spec: mcpSpec({ visibility: "default" }), }) .input(createFlowRequest) - .output(z.object({ id: z.string() })) + .output( + z.object({ + id: z.string(), + nodeIds: z + .optional(z.record(z.string(), z.string())) + .describe( + "Authored node id → persisted node id, present only when the request sent raw `nodes` (omitted for `spec` input, which has no authored ids).", + ), + }), + ) .errors(possibleErrorsOnCreatingResource) .handler(async ({ context, input }) => { const workspaceId = context.workspace.id @@ -115,20 +124,19 @@ export const flowsPublicRouter = { workspaceId, { validate: publish === true }, ) - const flow = await flowService.createDraft({ - workspaceId, - data: { name, folderId }, - graph, - }) - if (publish && graph) { - await flowVersionService.publish({ - workspaceId, - flowId: flow.id, - nodes: graph.nodes, - edges: graph.edges, - }) - } - return flow + const flow = + publish && graph + ? await flowService.createPublished({ + workspaceId, + data: { name, folderId }, + graph, + }) + : await flowService.createDraft({ + workspaceId, + data: { name, folderId }, + graph, + }) + return { id: flow.id, nodeIds: graph?.nodeIds } }), update: workspaceTokenAuthAPI @@ -243,7 +251,7 @@ export const flowsPublicRouter = { path: "/v1/flows/{id}/draft", summary: "Update flow draft", description: - "Overwrites the draft version's nodes/edges in place, without publishing. Accepts either the raw `{ nodes, edges }` graph the builder UI sends, or `{ spec }` compiled server-side into that same graph — draft nodes are not otherwise validated (see `flows.validate` to check a spec before writing it).", + "Overwrites the draft version's nodes/edges in place, without publishing. Accepts either the raw `{ nodes, edges }` graph the builder UI sends, or `{ spec }` compiled server-side into that same graph — draft nodes are not otherwise validated (see `flows.validate` to check a spec before writing it). Unlike `flows.create`, a raw node's `id` is persisted verbatim, not remapped, so it must already be a numeric string (the same format `flows.create`'s `nodeIds` response and `flows.get` return).", successStatus: 204, tags: ["Flows"], spec: mcpSpec({ visibility: "default" }), diff --git a/apps/builder/src/features/flows/lib/resolve-flow-graph-input.ts b/apps/builder/src/features/flows/lib/resolve-flow-graph-input.ts index feb9948749..32d60f0b1a 100644 --- a/apps/builder/src/features/flows/lib/resolve-flow-graph-input.ts +++ b/apps/builder/src/features/flows/lib/resolve-flow-graph-input.ts @@ -21,9 +21,16 @@ export type FlowGraphInput = { } export type ResolvedFlowGraph = { - nodes: FlowVersionSchema[] + // Validated (`options.validate: true`) graphs are `FlowVersionSchema`-shaped; + // an unvalidated raw `{ nodes, edges }` draft graph stays `AuthoredNodeInput`-shaped + // (`authorableNodeSchema` only requires an `id`) until publish time. `flowService.createDraft`/ + // `createPublished` only need the loose `FlowVersionModel["nodes"]` shape (`{ id: string, ... }`), + // which both branches satisfy, so no cast is needed to hand either one off. + nodes: FlowVersionSchema[] | AuthoredNodeInput[] edges: EdgeSchema[] startNodeId: string + /** Authored id → persisted id, populated only for the raw `{ nodes, edges }` path (`normalizeAuthoredGraph`'s remap map). `undefined` for `{ spec }` input, which has no authored node ids. */ + nodeIds?: Record } /** @@ -58,7 +65,7 @@ export function resolveFlowGraphInput( // Draft nodes are unvalidated by design, same as `flows.updateDraft`'s // raw `{ nodes, edges }` path — `authorableNodeSchema` only requires an // `id`, so the graph is not `FlowVersionSchema`-shaped until publish time. - return Promise.resolve(graph as unknown as ResolvedFlowGraph) + return Promise.resolve(graph) } const result = publishFlowSchema.safeParse({ @@ -74,5 +81,9 @@ export function resolveFlowGraphInput( ) } - return Promise.resolve({ ...result.data, startNodeId: graph.startNodeId }) + return Promise.resolve({ + ...result.data, + startNodeId: graph.startNodeId, + nodeIds: graph.nodeIds, + }) } diff --git a/docs/flows.md b/docs/flows.md index b4f7f72a15..7b614109e3 100644 --- a/docs/flows.md +++ b/docs/flows.md @@ -137,10 +137,15 @@ non-terminal step continues into the node-level "Continue" edge by default. `edges` — it is never the persisted id, since persisted node ids are numeric snowflakes and a caller id like `"n1"` would otherwise fail `publishFlowSchema`), and defaults edge handles to the node-id convention `addHandleEdge` produces - against the remapped ids. Add `publish: true` to validate the resulting graph - exactly like `flows.publish` and create the flow's first version in the same - call; omitting content or `publish` keeps today's default-start-node draft - behavior. + against the remapped ids. The response's `nodeIds` maps each authored id to + its persisted id (present only for the raw-graph shape; omitted for `{ + spec }` input, which has no authored ids). Add `publish: true` to validate + the resulting graph exactly like `flows.publish` and create the flow's + first version in the same call — the flow, its draft version, and its + published version are inserted in one transaction + (`FlowService.createPublished`/`createPublishedDefault`), so a downstream + failure never leaves an orphaned, unpublished flow behind. Omitting + content or `publish` keeps today's default-start-node draft behavior. 2. **`flows.validate`** — compiles a `{ spec }` and validates the result exactly like `flows.publish` would, without persisting anything. On success it returns the compiled `{ nodes, edges }` graph; on failure a 422 with structured errors @@ -153,7 +158,10 @@ non-terminal step continues into the node-level "Continue" edge by default. them. Creates an immutable version from the draft and syncs the draft to match. 4. **`flows.updateDraft`** — same either/or `{ nodes, edges }` vs `{ spec }` acceptance as `flows.publish`, but overwrites the draft in place without - publishing; draft nodes are not otherwise validated. + publishing; draft nodes are not otherwise validated. Unlike `flows.create`, + a raw node's `id` is persisted **verbatim, not remapped** — it must already + be a numeric string (`zodBigintAsString`), e.g. one returned by + `flows.create`'s `nodeIds` response or a `flows.get` call. The DSL compiler resolves `templateName`/`customFieldName`/`flowName` against the **entire workspace**, not a possibly-truncated capabilities page — see diff --git a/packages/business/__tests__/flow.service.test.ts b/packages/business/__tests__/flow.service.test.ts index 7b1ca5aef6..2f7d42e6ad 100644 --- a/packages/business/__tests__/flow.service.test.ts +++ b/packages/business/__tests__/flow.service.test.ts @@ -12,6 +12,7 @@ const { mockInsert, mockInsertReturning, mockInsertValues, + mockInvalidateList, mockTopLevelFlowFindFirst, mockUpdate, mockUpdateSet, @@ -40,6 +41,7 @@ const { mockInsert, mockInsertReturning, mockInsertValues, + mockInvalidateList: vi.fn(), mockTopLevelFlowFindFirst: vi.fn(), mockUpdate, mockUpdateSet, @@ -126,6 +128,7 @@ vi.mock("../src/errors", () => ({ vi.mock("../src/flow-version", () => ({ flowVersionService: { findDraft: mockFindDraft, + invalidateList: mockInvalidateList, }, })) @@ -537,3 +540,80 @@ describe("flowService.createDraft", () => { expect(result).toEqual({ id: "flow-2" }) }) }) + +describe("flowService.createPublished", () => { + afterEach(() => { + vi.clearAllMocks() + }) + + test("checks the folder exists before opening the transaction", async () => { + mockDbTransaction.mockImplementation(async (callback) => + callback(transaction), + ) + mockInsertValues.mockResolvedValue(undefined) + mockCreateId + .mockReturnValueOnce("flow-1") + .mockReturnValueOnce("draft-1") + .mockReturnValueOnce("published-1") + .mockReturnValueOnce("analytics-1") + + await flowService.createPublished({ + workspaceId: "ws-1", + data: { name: "New flow", folderId: "folder-1" }, + graph: { + nodes: [{ id: "node-1" }] as never, + edges: [] as never, + startNodeId: "node-1", + }, + }) + + expect(mockEnsureExists).toHaveBeenCalledWith({ + id: "folder-1", + workspaceId: "ws-1", + folderType: "flow", + }) + }) + + test("inserts flow + draft + published version in one transaction, invalidates the versions cache, and audits", async () => { + mockDbTransaction.mockImplementation(async (callback) => + callback(transaction), + ) + mockInsertValues.mockResolvedValue(undefined) + mockCreateId + .mockReturnValueOnce("flow-1") + .mockReturnValueOnce("draft-1") + .mockReturnValueOnce("published-1") + .mockReturnValueOnce("analytics-1") + + const result = await flowService.createPublished({ + workspaceId: "ws-1", + data: { name: "New flow" }, + graph: { + nodes: [{ id: "node-1" }] as never, + edges: [] as never, + startNodeId: "node-1", + }, + }) + + expect(result).toEqual({ id: "flow-1" }) + expect(mockDbTransaction).toHaveBeenCalledTimes(1) + expect(mockInsert).toHaveBeenNthCalledWith(3, flowVersionModel) + expect(mockInsertValues).toHaveBeenNthCalledWith(3, [ + expect.objectContaining({ + id: "draft-1", + isDraft: true, + isLatest: false, + }), + expect.objectContaining({ + id: "published-1", + isDraft: false, + isLatest: true, + }), + ]) + expect(mockInvalidateList).toHaveBeenCalledWith("flow-1") + expect(mockAudit).toHaveBeenCalledWith( + "create", + "created a new flow (#flow-1)", + ) + }) +}) diff --git a/packages/business/src/flow/service.ts b/packages/business/src/flow/service.ts index c8a44dee1d..c9ed27ed9c 100644 --- a/packages/business/src/flow/service.ts +++ b/packages/business/src/flow/service.ts @@ -371,6 +371,52 @@ class FlowService extends BaseService { return { id: flow.id } } + /** + * The public `flows.create` API's `publish: true` path: inserts the flow, + * its analytics session, a draft version, and a published version — all in + * one transaction via `createPublishedDefault` — so a downstream failure + * (a DB error, a constraint, the cache invalidation) never leaves behind + * an orphaned, invisible, unpublished flow the caller was never told the + * id of. Runs the same folder guard `createDraft` runs before opening the + * transaction, and invalidates/audits after it commits, matching + * `createPublishedDefault`'s doc-comment contract. + */ + async createPublished(input: { + workspaceId: string + data: { name: string; folderId?: string | null } + graph: { + nodes: FlowVersionModel["nodes"] + edges: FlowVersionModel["edges"] + startNodeId: string + } + }): Promise<{ id: string }> { + const { workspaceId, data, graph } = input + + if (data.folderId) { + await folderService.ensureExists({ + id: data.folderId, + workspaceId, + folderType: "flow", + }) + } + + const { flowId } = await db.transaction((tx) => + this.createPublishedDefault(tx, { + workspaceId, + name: data.name, + folderId: data.folderId, + startNodeId: graph.startNodeId, + nodes: graph.nodes, + edges: graph.edges, + }), + ) + + await flowVersionService.invalidateList(flowId) + await this.audit("create", `created a new flow (#${flowId})`) + + return { id: flowId } + } + /** * Partial update of a flow's name/active/enableInInbox. No-ops (and skips * the audit record) when every field matches the current row, mirroring diff --git a/packages/flow-config/__tests__/authoring/normalize-graph.test.ts b/packages/flow-config/__tests__/authoring/normalize-graph.test.ts index c759e579cf..0e0c599ac9 100644 --- a/packages/flow-config/__tests__/authoring/normalize-graph.test.ts +++ b/packages/flow-config/__tests__/authoring/normalize-graph.test.ts @@ -48,6 +48,14 @@ describe("normalizeAuthoredGraph", () => { } }) + test("returns the authored id -> persisted id map for every node", () => { + const result = normalizeAuthoredGraph([{ id: "a" }, { id: "b" }], []) + + expect(Object.keys(result.nodeIds).sort()).toEqual(["a", "b"]) + expect(result.nodeIds.a).toBe(result.nodes[0].id) + expect(result.nodeIds.b).toBe(result.nodes[1].id) + }) + test("generates edge ids and node-id handles, rewriting them to the remapped node ids", () => { const result = normalizeAuthoredGraph( [{ id: "a" }, { id: "b" }, { id: "c" }], diff --git a/packages/flow-config/src/authoring/normalize-graph.ts b/packages/flow-config/src/authoring/normalize-graph.ts index f7cf3a8e40..4e78e5be23 100644 --- a/packages/flow-config/src/authoring/normalize-graph.ts +++ b/packages/flow-config/src/authoring/normalize-graph.ts @@ -23,6 +23,8 @@ export type NormalizedAuthoredGraph = { nodes: T[] edges: EdgeSchema[] startNodeId: string + /** Authored id (e.g. `"n1"`) → persisted internal id, for callers that need to reference the nodes they just authored without a round-trip fetch. */ + nodeIds: Record } /** @@ -51,6 +53,10 @@ export function normalizeAuthoredGraph( message: string }[] = [] + // The public `flows.create` API pre-enforces `.min(1)` on `nodes` + // (`createFlowRequest`, `apps/builder/src/features/flows/schema/action.ts`), + // so this guard is unreachable from there; it exists for other package + // callers, and `nodes[0].id` below still depends on non-emptiness. if (nodes.length === 0) { errors.push({ path: "nodes", @@ -59,9 +65,9 @@ export function normalizeAuthoredGraph( }) } - const nodeIds = new Set() + const seenNodeIds = new Set() for (const [index, node] of nodes.entries()) { - if (nodeIds.has(node.id)) { + if (seenNodeIds.has(node.id)) { errors.push({ path: `nodes[${index}].id`, code: "invalidGraph", @@ -69,18 +75,18 @@ export function normalizeAuthoredGraph( }) continue } - nodeIds.add(node.id) + seenNodeIds.add(node.id) } for (const [index, edge] of edges.entries()) { - if (!nodeIds.has(edge.source)) { + if (!seenNodeIds.has(edge.source)) { errors.push({ path: `edges[${index}].source`, code: "invalidGraph", message: `Edge source "${edge.source}" does not match any node id.`, }) } - if (!nodeIds.has(edge.target)) { + if (!seenNodeIds.has(edge.target)) { errors.push({ path: `edges[${index}].target`, code: "invalidGraph", @@ -149,5 +155,10 @@ export function normalizeAuthoredGraph( } }) - return { nodes: resolvedNodes, edges: resolvedEdges, startNodeId } + return { + nodes: resolvedNodes, + edges: resolvedEdges, + startNodeId, + nodeIds: Object.fromEntries(idByAuthoredId), + } }