Workflow

Build custom workflow nodes (IWorkflowNode), register them on the module, trigger workflows from services, and expose controller APIs as discoverable @WorkflowEntry steps.

Overview

The workflow engine runs graphs of nodes. Plugins contribute nodes and can start workflows when domain events happen. Import workflow types from @quan-erp/shared-backend-core.

Custom node — implement IWorkflowNode, register under @Module({ workflowNodes })
Trigger — isTriggerable: true + WorkflowService.findByInitId / executeWorkflow from controllers/services
API entry — @WorkflowEntry on a controller method so the editor can call that HTTP API as a step
Frontend — register matching node UI via workflowNodes() from @quan-erp/base-frontend (see Frontend docs)

Node data model

Props — configuration from the workflow editor UI
Args — values injected when the node runs (previous output or executeWorkflow args)
ReturnType — value this node produces for downstream nodes
type — always `${metadata.name}/${NODE_ID}` for plugin nodes

Props & return schemas

Declare JSON schemas with PropsType and infer TypeScript types via SchemaToType. Schemas drive both process() typing and the frontend editor.

TSXschemas

import {
  PropsType,
  SchemaToType,
} from "@quan-erp/shared-backend-core";

const inputSchema = {
  type: "number",
} as const satisfies PropsType;
type Props = SchemaToType<typeof inputSchema>;

const returnSchema = {
  type: "object",
  properties: {
    bookingId: { type: "number" },
    bookingSource: { type: "string" },
  },
} as const satisfies PropsType;
type ReturnType = SchemaToType<typeof returnSchema>;

export type Args = {
  bookingId: number;
  bookingSource: string;
};

Execution lifecycle

01
process()
Executor calls process(context, uid, props, args). Return NodeResult.single(...) or another NodeResultType.
02
nextToExecute()
After process, executor asks for the next node ids. Return null to follow the UI next edges; return string[] to override routing (if / loop nodes).
03
Routing
null → use the node’s next array from the canvas. string[] → run those node ids instead.

ExecutionContext

Each run gets an ExecutionContext shared by all nodes. Resolve UI-mapped props before use; optionally sub-execute another node.

context.pipeData.resolve(props) — expand ${{node_….output}} expressions; always resolve before reading props
context.execute(nodeId, args) — run another node immediately and await its NodeResultType (T[]) (e.g. AI agent → tool)
toolResult[0] — raw payload from a sub-execution (NodeResultType is an array)
context.findWorkflowNodeById(uid) — inspect current node edges for custom nextToExecute

TSXresolve props

import type {
  ExecutionContext,
  NodeResultType,
  PromisableReturn,
} from "@quan-erp/shared-backend-core";
import { NodeResult } from "@quan-erp/shared-backend-core";

function process(
  context: ExecutionContext,
  uid: string,
  props: { expression: string },
  args: unknown,
): PromisableReturn<NodeResultType<boolean>> {
  const resolvedProps = context.pipeData.resolve(props);
  const evaluateResult = eval(resolvedProps.expression);
  return NodeResult.single(evaluateResult as boolean);
}

TSXsub-execute

import type { ExecutionContext } from "@quan-erp/shared-backend-core";

async function callTool(
  context: ExecutionContext,
  toolNodeId: string,
  toolArgs: Record<string, unknown>,
) {
  const toolResult = await context.execute(toolNodeId, toolArgs);
  return toolResult[0];
}

Custom routing (nextToExecute)

Override the default edge walk for conditionals and loops. Return null to keep canvas edges. nextToExecute receives the NodeResultType array from process (use args[0]).

TSXif-style routing

import type {
  ExecutionContext,
  NodeResultType,
  PromisableReturn,
} from "@quan-erp/shared-backend-core";

function nextToExecute(
  context: ExecutionContext,
  uid: string,
  args: NodeResultType<boolean>,
): PromisableReturn<string[]> | PromisableReturn<null> {
  const current = context.findWorkflowNodeById(uid);
  const ifTrue = current?.next?.[0];
  const ifFalse = current?.next?.[1];
  if (args[0] && ifTrue?.id) return [ifTrue.id];
  if (!args[0] && ifFalse?.id) return [ifFalse.id];
  return null;
}

Full trigger node

Trigger nodes set isTriggerable: true and usually pass Args through as the return value so the rest of the graph can use them.

TSXOnBookingWorkflowNode

import {
  ExecutionContext,
  IWorkflowNode,
  NodeResult,
  NodeResultType,
  PromisableReturn,
  PropsType,
  SchemaToType,
  WorkflowNodeDefination,
} from "@quan-erp/shared-backend-core";
import metadata from "../../../module.metadata.json" with { type: "json" };

const inputSchema = {
  type: "number",
} as const satisfies PropsType;
type Props = SchemaToType<typeof inputSchema>;

const returnSchema = {
  type: "object",
  properties: {
    bookingId: { type: "number" },
    bookingSource: { type: "string" },
  },
} as const satisfies PropsType;
type ReturnType = SchemaToType<typeof returnSchema>;

export type Args = {
  bookingId: number;
  bookingSource: string;
};

export class OnBookingWorkflowNode
  implements IWorkflowNode<Props, ReturnType, Args>
{
  static NODE_ID = "bookings";

  process(
    context: ExecutionContext,
    id: string,
    props: Props,
    args: Args,
  ): PromisableReturn<NodeResultType<ReturnType>> {
    return NodeResult.single(args);
  }

  nextToExecute(
    context: ExecutionContext,
    uid: string,
    args: NodeResultType<ReturnType>,
  ): PromisableReturn<string[]> | PromisableReturn<null> {
    return null;
  }

  getDefination(): WorkflowNodeDefination {
    return {
      type: `${metadata.name}/${OnBookingWorkflowNode.NODE_ID}`,
      group: ["any"],
      displayName: "On Hotel Booked",
      description: "Triggers when a hotel is booked",
      props: inputSchema,
      isTriggerable: true,
      returnType: {
        type: "array",
        items: returnSchema,
      },
    };
  }
}

Register on the module

Add the node class to workflowNodes so the backend mounts it and exposes the definition to the editor.

TSXmodule

import { Module } from "@quan-erp/shared-backend-core";
import metadata from "../../module.metadata.json" with { type: "json" };
import { OnBookingWorkflowNode } from "./workflow/on-booking.workflow.js";

@Module({
  name: metadata.name,
  providers: [],
  controllers: [],
  entities: [],
  workflowNodes: [OnBookingWorkflowNode],
})
export class MyPluginModule {}

Trigger workflows programmatically

For isTriggerable nodes, inject WorkflowService from the builtin plugin, find active workflows whose init id is `${metadata.name}/${NODE_ID}`, and execute each with Args.

TSXexecuteWorkflow

import {
  ContainerRegistryManager,
  Controller,
  Inject,
  Post,
  ResponseDto,
  WorkflowService,
} from "@quan-erp/shared-backend-core";
import metadata from "../../module.metadata.json" with { type: "json" };
import { OnBookingWorkflowNode } from "../workflow/on-booking.workflow.js";

@Controller("/bookings")
export class BookingController {
  @Inject(WorkflowService, ContainerRegistryManager.BUILTIN_PLUGIN)
  workflowService: WorkflowService;

  @Post("/")
  async createBooking() {
    // … create booking …
    const workflows = await this.workflowService.findByInitId(
      `${metadata.name}/${OnBookingWorkflowNode.NODE_ID}`,
      { active: true },
    );

    await Promise.all(
      workflows.map((w) =>
        this.workflowService.executeWorkflow({
          workflow: w,
          args: {
            bookingId: 1,
            bookingSource: "web",
          },
        }),
      ),
    );

    return ResponseDto.okWithEmpty();
  }
}

@WorkflowEntry (HTTP as a step)

Mark a controller method so the workflow engine can discover and invoke that API as a node. Provide name, description, and a returnType JSON schema matching the response shape.

TSX@WorkflowEntry

import {
  Controller,
  Get,
  WorkflowEntry,
} from "@quan-erp/shared-backend-core";

@Controller("/branch")
export class BranchController {
  @Get("/")
  @WorkflowEntry({
    name: "get-branch",
    description: "get all branches",
    returnType: {
      type: "object",
      properties: {
        payload: {
          type: "array",
          items: {
            type: "object",
            properties: {
              id: { type: "number" },
              name: { type: "string" },
            },
          },
        },
      },
    },
  })
  async list() {
    /* … */
  }
}

Checklist

Schemas: PropsType + SchemaToType for props and returnType
Implement process, nextToExecute, getDefination on IWorkflowNode
type: `${metadata.name}/${NODE_ID}` — never omit the plugin prefix
Register in @Module({ workflowNodes })
Triggerable nodes: fire via WorkflowService on the domain event
Always context.pipeData.resolve(props) before using editor props
Return values with NodeResult.single(...) (or the matching NodeResult helper)