@asaidimu/workflows

A TypeScript workflow engine that compiles visual node‑edge graphs into executable pipelines. Includes a runtime with multiple execution modes, a built‑in HTTP server, and a registry of built‑in node types.

Features

• Graph compiler — Convert a set of WorkflowNode + WorkflowEdge objects into a compiled Workflow with triggers and pipelines.
• Built‑in nodes — Trigger, Code, HTTP Request, Switch, If/Else, For‑Each, While, Try/Catch, Delay, Transformer, Arithmetic, Gemini AI, and more.
• Routing — Nodes can define custom routers to direct execution flow (switch/if/for‑each/while).
• Bounded nodes — Support for nodes with body sub‑graphs (try‑catch, for‑each, while).
• Resource injection — Resource nodes provide handles (DB connections, etc.) to dependent executable nodes via dependency edges.
• Pipeline references — Compose workflows by referencing other compiled pipelines.
• Multiple execution modes:
  • transient — Concurrent runs with configurable concurrency/capacity.
  • serialized — FIFO serialized execution.
  • loop — At most one active run; supports drop/signal/replace on duplicate events.
  • exclusive — Run exclusively, optionally queue one pending run.
• Event‑driven — Workflows are triggered by events emitted on an EventBus.
• State interpolation — Config values support {{ state.path }}, {{ $res.resource }}, and {{ $results.path }} template syntax.
• Built‑in HTTP server — Bun server exposing REST endpoints to compile, register, run, and manage workflows.
• Extensible — Register custom executable, resource, and pause node definitions.

Installation

npm install @asaidimu/workflows
# or
bun add @asaidimu/workflows

Usage

Compile a workflow from nodes and edges

import { compileWorkflow } from "@asaidimu/workflows";

const nodes = [
  {
    id: "trigger-1",
    type: "executable",
    data: { kind: "trigger", config: { initialState: {} } },
    position: { x: 0, y: 0 },
  },
  {
    id: "code-1",
    type: "executable",
    data: { kind: "code", config: { code: "return { greeting: 'hello' };" } },
    position: { x: 200, y: 0 },
  },
];

const edges = [
  { id: "e1", source: "trigger-1", target: "code-1", data: { role: "flow" } },
];

const workflow = await compileWorkflow(nodes, edges);

Run a workflow with the engine

import { WorkflowsEngine } from "@asaidimu/workflows";

const engine = await WorkflowsEngine.create();

const controller = new AbortController();
const result = await engine.run(nodes, edges, controller.signal);

const outcome = engine.getRunOutcome(result!.runId);
console.log(outcome);

Advanced runtime usage

import {
  WorkflowRuntime,
  compileWorkflow,
  createEventBus,
} from "@asaidimu/workflows";
import { StoreRegistry } from "@asaidimu/utils-store";

const bus = createEventBus();
const storeRegistry = new StoreRegistry();
const runtime = new WorkflowRuntime({ bus, storeRegistry });

const workflow = await compileWorkflow(nodes, edges);

await runtime.register(workflow, {
  mode: { type: "transient", concurrency: 5 },
  onPrepare: async (ctx) => {
    /* ... */
  },
  onComplete: async (result) => {
    /* ... */
  },
});

// Emit the trigger event
bus.emit({ name: "__manual__", payload: {} });

Custom node definition

import { defineNode } from "@asaidimu/workflows";

export const myNode = defineNode<{ message: string }>({
  kind: "my-custom",
  label: "My Custom Node",
  description: "Does something custom.",
  configSchema: {
    version: "1.0.0",
    name: "my-custom",
    fields: {
      message: {
        name: "message",
        type: "string",
        default: "hello",
        required: true,
      },
    },
  },
  handles: () => [
    { type: "target", id: "" },
    { type: "source", id: "" },
  ],
  run: async ({ config, state }) => {
    return { result: `${config.message} world` };
  },
});

Resource definition

import { defineResource } from "@asaidimu/workflows";

export const myResource = defineResource({
  kind: "my-db",
  label: "My Database",
  description: "A custom database resource.",
  configSchema: { version: "1.0.0", name: "my-db", fields: {} },
  handles: () => [{ type: "source", id: "db", kind: "resource" }],
  init: async (config, ctx) => {
    // Resolve services from the artifact container
    const createStore = await ctx.use((deps) => deps.require("store-factory"));
    return { client: "connected", createStore };
  },
  cleanup: async (config, handle) => {
    /* disconnect */
  },
});

Resource nodes are compiled into WorkflowServiceDefinition entries with scope: "run" and registered on the run's artifact container. Executable nodes wire them via dependency edges (not flow edges). In the step's run function the resource handle is available through the resources parameter: resources.database.

Start the built-in HTTP server

import "@asaidimu/workflows/server";

Or run it directly:

bun run ./node_modules/@asaidimu/workflows/server.ts

Available endpoints:

| Method | Path | Description | | ------ | ------------------- | ------------------------------------- | | GET | /registry | List registered node types | | POST | /compile | Compile nodes+edges into a workflow | | POST | /run | Compile, register, and run a workflow | | POST | /register | Register a pre-compiled workflow | | POST | /deregister | Remove a registered workflow | | POST | /events | Emit a trigger event | | GET | /runs | List all runs | | GET | /runs/:id/outcome | Get run outcome | | POST | /runs/:id/abort | Abort a running workflow |

Built-in Nodes

| Kind | Node | Description | | ------------- | ---------------- | ---------------------------------------------- | | trigger | Trigger | Starts the workflow with initial state | | code | JavaScript Code | Executes custom JS on workflow state | | http | HTTP Request | Makes HTTP requests to external APIs | | switch | Switch | Routes execution based on a condition | | if | If/Else | Conditional branching | | for-each | For Each | Iterates over an array in state | | while | While | Loops while a condition is true | | try-catch | Try/Catch | Error handling with catch body | | delay | Delay | Pauses execution for a duration | | transformer | Transformer | Transforms state values | | arithmetic | Arithmetic | Performs math operations on state | | database | Database Service | Provides a database connection (resource node) | | query | Database Query | Executes queries against a database collection | | gemini | Gemini AI | Calls the Gemini AI API |

Execution Modes

| Mode | Behavior | | ------------ | ---------------------------------------------------------------------------- | | transient | Concurrent runs up to concurrency limit, queued up to capacity | | serialized | FIFO ordered, one at a time, queued up to capacity | | loop | Single active run; drop/signal/replace on new events | | exclusive | Single run, no queue; or queue exactly one pending (reject/queue_single) |

Dependency edges

Dependency edges (role: "dependency") connect resource nodes to executable nodes without affecting flow execution order. The compiler collects all dependency edges for each executable node and makes the resolved resource handles available in the step's run function via the resources parameter.

// Edge connecting a database resource to a query node
{ id: "d1", source: "db-1", target: "query-1", data: { role: "dependency" } }

Service injection

Resource nodes are compiled into WorkflowServiceDefinition objects with scope: "run". The runtime registers these on the run's artifact container via extendContextContainer(). When a resource's init() function uses ctx.use((deps) => deps.require("service-id")), it resolves services from the container hierarchy (global services first, then workflow services).

This allows resources to depend on services registered at the runtime level without hardcoding imports:

// Runtime-level service registration
const runtime = new WorkflowRuntime({
  bus,
  storeRegistry,
  services: [{ id: "store-factory", factory: () => createEphemeralStore }],
});

// Resource init resolves from the container
init: async (config, ctx) => {
  const createStore = await ctx.use((deps) => deps.require("store-factory"));
  return DatabaseConnection(config, createStore);
},

API

compileWorkflow(nodes, edges, pipelineRegistry?)

Compiles visual graph nodes and edges into a Workflow object.

defineNode(def) / defineResource(def)

Register custom executable or resource node definitions with schema validation, config coercion, and lifecycle hooks.

WorkflowRuntime

Event-driven runtime that manages workflow registration, execution contexts, and run lifecycle.

WorkflowsEngine

High-level facade that wraps WorkflowRuntime for quick start (create(), run(), compile(), register(), emitEvent()).

Project Structure

src/workflows/
├── index.ts           # Public exports
├── types.ts           # Core type definitions
├── schema.ts          # defineNode / defineResource
├── registry.ts        # Built-in node registry
├── compiler.ts        # Graph → pipeline compilation
├── engine.ts          # WorkflowsEngine facade
├── server.ts          # Bun HTTP server
├── generated-nodes.ts # Auto-generated node registries
├── compiler.test.ts   # Compiler tests
├── runtime.test.ts    # Runtime tests
├── database-workflow.test.ts  # Integration test (DB + conditional)
├── utils/
│   └── state.ts       # State interpolation helpers
└── nodes/             # Built-in node implementations
    ├── trigger/
    ├── code/
    ├── http/
    ├── switch/
    ├── if/
    ├── for-each/
    ├── while/
    ├── try-catch/
    ├── delay/
    ├── transformer/
    ├── arithmetic/
    ├── gemini/
    ├── database-service/  # Resource node — provides DB connections
    └── query/             # Executable node — queries a DB collection

License

MIT