Pipeline

This guide walks through the most common pipeline patterns: registering nodes with actions, using ordered forks for branching, embedding sub-pipelines, and composing pipelines with domain flows. For complete signatures and field tables, see the Definition Functions, Registration Helpers, and Pipeline Instance reference pages.

A Pipeline models a finite, single-run workflow. Execution starts at the initial node, evaluates ordered forks at each step, and runs until it reaches a terminal node or finds no matching fork.

Defining a Pipeline

Use definePipeline() to create a pipeline. The setup function registers nodes using the node() helper and must return { initial } naming the first node to execute.

import { definePipeline, node, action, fork, target, terminal } from '@tde.io/plexis';

type PaymentContext = { cardValid: boolean; amount: number };

const payment = definePipeline<PaymentContext>('payment', () => {
  node('validate', () => {
    action(async (ctx) => {
      const ok = ctx.cardValid && ctx.amount > 0;
      return { validated: ok };
    });
    fork('card-ok', target('charge'), (ctx) => ctx.cardValid);
    fork('card-invalid', target('decline'), (ctx) => !ctx.cardValid);
  });

  node('charge', terminal());
  node('decline', terminal());

  return { initial: 'validate' };
});

Nodes

Register nodes with node(id, fn | terminal(fn?)). The setup function calls scoped helpers:

Helper	Description
action(fn)	Runs when the node executes; return value merges into context
fork(label, target, cond?)	Register an ordered forward transition

Pass terminal() or terminal(fn) as the second argument to register a terminal node with an optional final action.

Actions and Context Patches

An action receives the current context and a PipelineActionInput (node ID, pipeline ID, optional input value, trace ID). It returns a patch merged as { ...prev, ...patch }.

node('enrich', () => {
  action(async (ctx, { input }) => ({
    enrichedAt: Date.now(),
    source: input?.source ?? 'unknown',
  }));
});

Returning undefined, null, or void leaves the context unchanged.

Forks

Forks define ordered forward transitions. The first fork whose condition returns true is taken; remaining forks are not evaluated (first-match-wins).

fork(label, target, condition?)

See fork for the complete parameter reference.

A fork with no condition always matches, making it a catch-all:

node('route', () => {
  fork('high-value', target('high-value'), (ctx) => ctx.amount > 1000);
  fork('standard', target('standard'));  // default path
});

Terminal Nodes

Mark a node as terminal with terminal() or terminal(fn). No forks are evaluated after a terminal node — the pipeline stops and PipelineRunResult is returned.

node('done', terminal());

// With a final action:
node('done', terminal(async (ctx) => ({ completedAt: Date.now() })));

Running a Pipeline

Call pipeline.run(context, input?) to execute it. The pipeline runs to completion unless an error policy stops it.

const result = await payment.run({ cardValid: true, amount: 50 });

console.log(result.status);    // 'completed'
console.log(result.finalNode); // 'charge'
console.log(result.context);   // updated context

See PipelineRunResult for the complete field reference.

Sub-Pipelines

A fork target can be another Pipeline instance. When the fork is taken, the sub-pipeline runs with the current context and its output context flows forward.

const enrichment = definePipeline('enrichment', () => {
  node('fetch-geo', () => {
    action(async () => ({ geo: 'US' }));
    fork('next', target('done'));
  });
  node('done', terminal());
  return { initial: 'fetch-geo' };
});

const main = definePipeline('main', () => {
  node('start', () => {
    fork('enrich', target(enrichment));
  });
  node('finish', terminal());
  return { initial: 'start' };
});

Attaching a Pipeline to a Domain Flow

Pipelines compose with domains: pass a pipeline to a flow’s pipeline option inside on(...). The pipeline runs after the flow action and before the state transition.

import { defineDomain, when, on } from '@tde.io/plexis';

when('pending', () => {
  on('submit', () => { pipeline(payment); return target('processing'); });
});

Execution order on follow('submit'):

1. Guard check
2. exit hook for pending
3. Flow action
4. payment pipeline runs
5. State transitions to processing
6. enter hook for processing
7. State entry pipeline for processing