@studnicky/pipeline

Generic typed async pipeline for sequential context transforms.

Install

bash

pnpm add @studnicky/pipeline

Requires @studnicky:registry=https://npm.pkg.github.com in .npmrc.

Usage

Build a Pipeline<T> instance, register stages with add(), and run a context through all of them with run(). Each stage receives the context and returns a (possibly transformed) copy. add() returns a removal function:

import type { NumCtxTypeEntity } from './entities/NumCtxTypeEntity.js';

import { Pipeline } from '../src/index.js';

const pipeline = Pipeline.create<NumCtxTypeEntity.Type>();

// Stage 0: multiply by 2
const removeDouble = pipeline.add((ctx) => { return { 'value': ctx.value * 2 }; });

// Stage 1: add 10
pipeline.add((ctx) => { return { 'value': ctx.value + 10 }; });

// Stage 2: multiply by 3
pipeline.add((ctx) => { return { 'value': ctx.value * 3 }; });

console.log(`Stages registered: ${pipeline.stages.length}`);

class PipelineRunDemo {
  // Runs the 3-stage pipeline, removes the doubling stage, then re-runs — returning
  // both results so the caller ends up with a single top-level binding.
  static async run(): Promise<{ 'withDouble': number; 'withoutDouble': number }> {
    // Run: (5 * 2 + 10) * 3 = 60
    const result = await pipeline.run({ 'value': 5 });
    console.log(`Result with 3 stages: ${result.value}`);

    // Remove stage 0 (the doubling stage) using the unsubscribe function
    removeDouble();
    console.log(`Stages after removal: ${pipeline.stages.length}`);

    // Re-run without the double stage: (5 + 10) * 3 = 45
    const resultWithout = await pipeline.run({ 'value': 5 });
    console.log(`Result without double stage: ${resultWithout.value}`);

    return { 'withDouble': result.value, 'withoutDouble': resultWithout.value };
  }
}

const results = await PipelineRunDemo.run();

Try it

The builder demo constructs a Pipeline via Pipeline.builder<RequestCtx>().build() and registers three stages. Watch how each stage receives the transformed context from the previous one — status, headers, and body all accumulate in order.

Pipeline builder

/**
 * builderPipeline — constructs a Pipeline via Pipeline.builder().build(),
 * registers stages with add(), and runs a context through all stages.
 * Run: npx tsx examples/builderPipeline.ts
 */

import assert from 'node:assert/strict';

// #region usage
import { Pipeline } from '../src/index.js';

type RequestCtx = {
  'body': string;
  'headers': Record<string, string>;
  'status': number;
};

// Build a pipeline using the fluent builder
const pipeline = Pipeline.builder<RequestCtx>().build();

console.log('Pipeline built. Stages:', pipeline.stages.length);

// Stage 0: set initial status
pipeline.add((ctx) => { return { ...ctx, 'status': 200 }; });

// Stage 1: attach a request-id header
pipeline.add((ctx) => { return { ...ctx, 'headers': { ...ctx.headers, 'x-request-id': 'abc-123' } }; });

// Stage 2: append a signed footer to the body
pipeline.add((ctx) => { return { ...ctx, 'body': `${ctx.body} [signed]` }; });

console.log('Stages registered:', pipeline.stages.length);

const result = await pipeline.run({ 'body': 'Hello', 'headers': {}, 'status': 0 });
console.log('Status:', result.status);
console.log('Headers:', result.headers);
console.log('Body:', result.body);
// #endregion usage

assert.equal(result.status, 200, 'Status stage applied');
assert.equal(result.headers['x-request-id'], 'abc-123', 'Request-id header injected');
assert.equal(result.body, 'Hello [signed]', 'Body stage applied');

console.log('builderPipeline: all assertions passed');

Output

Press Execute to run this example against the real library.

The hooks demo subclasses Pipeline and overrides all eight protected lifecycle hooks, then runs both a happy path and a failing path. Watch the happy path emit runStart → beforeStage → stageStart → stageSuccess → afterStage for each of three stages, then runComplete. The failing path shows stageError at index 1 followed by runError wrapping the stage failure in a PipelineError.

Pipeline lifecycle hooks

/** observedPipeline — trace every hook in a multi-stage pipeline. Run: npx tsx examples/observedPipeline.ts */

import assert from 'node:assert/strict';

// #region usage
import type { StepCtxTypeEntity } from './entities/StepCtxTypeEntity.js';

import { Pipeline, PipelineError } from '../src/index.js';

class TracingPipeline<T extends StepCtxTypeEntity.Type> extends Pipeline<T> {
  readonly stageStartEvents: { 'ctx': T; 'index': number }[] = [];
  readonly stageSuccessEvents: { 'ctx': T; 'index': number }[] = [];
  readonly stageErrorEvents: { 'error': unknown; 'index': number }[] = [];
  readonly runErrorEvents: { 'error': unknown }[] = [];

  protected override onRunStart(ctx: T): T {
    console.log('[pipeline] runStart');
    return ctx;
  }

  protected override beforeStage(ctx: T, index: number): T {
    console.log(`[pipeline] beforeStage index=${index}`);
    return ctx;
  }

  protected override onStageStart(index: number, ctx: T): void {
    console.log(`[pipeline] stageStart index=${index}`);
    this.stageStartEvents.push({ 'ctx': ctx, 'index': index });
  }

  protected override onStageSuccess(index: number, ctx: T): void {
    console.log(`[pipeline] stageSuccess index=${index}`);
    this.stageSuccessEvents.push({ 'ctx': ctx, 'index': index });
  }

  protected override afterStage(ctx: T, index: number): T {
    console.log(`[pipeline] afterStage index=${index}`);
    return ctx;
  }

  protected override onStageError(index: number, error: unknown): void {
    const msg = error instanceof Error ? error.message : String(error);
    console.log(`[pipeline] stageError index=${index} error=${msg}`);
    this.stageErrorEvents.push({ 'error': error, 'index': index });
  }

  protected override onRunError(error: unknown): void {
    const msg = error instanceof PipelineError ? `PipelineError: ${error.message}` : String(error);
    console.log(`[pipeline] runError error=${msg}`);
    this.runErrorEvents.push({ 'error': error });
  }

  protected override onRunComplete(ctx: T): T {
    console.log('[pipeline] runComplete');
    return ctx;
  }
}

// ── Happy-path run: 3 stages that mutate step/value ───────────────────────────

const successPipeline = TracingPipeline.create<StepCtxTypeEntity.Type>();

successPipeline.add((ctx) => { return { 'step': ctx.step + 1, 'value': `${ctx.value}->alpha` }; });
successPipeline.add((ctx) => { return { 'step': ctx.step + 1, 'value': `${ctx.value}->beta` }; });
successPipeline.add((ctx) => { return { 'step': ctx.step + 1, 'value': `${ctx.value}->gamma` }; });

console.log('\n--- happy path ---');
const successResult = await successPipeline.run({ 'step': 0, 'value': 'start' });
console.log(`result: step=${successResult.step} value=${successResult.value}`);

// ── Failing run: 2 stages where the second throws ────────────────────────────

const failPipeline = TracingPipeline.create<StepCtxTypeEntity.Type>();

failPipeline.add((ctx) => { return { 'step': ctx.step + 1, 'value': `${ctx.value}->alpha` }; });
failPipeline.add((_ctx) => { throw new Error('stage 1 fails'); });

console.log('\n--- failing path ---');
try {
  await failPipeline.run({ 'step': 0, 'value': 'start' });
} catch (err: unknown) {
  const msg = err instanceof PipelineError ? `PipelineError: ${err.message}` : String(err);
  console.log(`caught: ${msg}`);
}
// #endregion usage

// ── Assertions ────────────────────────────────────────────────────────────────

// Success pipeline: 3 stages all started and succeeded
assert.strictEqual(successPipeline.stageStartEvents.length, 3);
assert.strictEqual(successPipeline.stageSuccessEvents.length, 3);
assert.strictEqual(successPipeline.stageErrorEvents.length, 0);
assert.strictEqual(successPipeline.runErrorEvents.length, 0);

assert.strictEqual(successPipeline.stageStartEvents[0]?.index, 0);
assert.strictEqual(successPipeline.stageStartEvents[1]?.index, 1);
assert.strictEqual(successPipeline.stageStartEvents[2]?.index, 2);

assert.strictEqual(successPipeline.stageSuccessEvents[0]?.ctx.value, 'start->alpha');
assert.strictEqual(successPipeline.stageSuccessEvents[1]?.ctx.value, 'start->alpha->beta');
assert.strictEqual(successPipeline.stageSuccessEvents[2]?.ctx.value, 'start->alpha->beta->gamma');

// Fail pipeline: stage 0 succeeded, stage 1 errored, run errored
assert.strictEqual(failPipeline.stageStartEvents.length, 2);
assert.strictEqual(failPipeline.stageSuccessEvents.length, 1);
assert.strictEqual(failPipeline.stageSuccessEvents[0]?.index, 0);
assert.strictEqual(failPipeline.stageErrorEvents.length, 1);
assert.strictEqual(failPipeline.stageErrorEvents[0]?.index, 1);
assert.ok(failPipeline.stageErrorEvents[0]?.error instanceof Error);
assert.strictEqual(failPipeline.runErrorEvents.length, 1);
assert.ok(failPipeline.runErrorEvents[0]?.error instanceof PipelineError);

console.log('observedPipeline: all assertions passed');

Output

Press Execute to run this example against the real library.

Subpath exports

Subpath	Contents
`@studnicky/pipeline`	`Pipeline`, `PipelineBuilder`, `PipelineError`
`@studnicky/pipeline/interfaces`	`PipelineInterface`
`@studnicky/pipeline/pipeline`	`Pipeline`, `PipelineBuilder` (direct subpath)
`@studnicky/pipeline/types`	`PipelineFnType`

Extending

Pipeline exposes four protected hooks (onRunStart, beforeStage, afterStage, and onRunComplete) that subclasses can override to inject timing, logging, or context mutation without coupling the core pipeline to any external dependency:

import { Pipeline } from '../src/index.js';

type RequestCtxType = {
  'elapsed'?: number;
  'headers': Record<string, string>;
  'url': string;
};

class TimedPipeline extends Pipeline<RequestCtxType> {
  private startTime = 0;

  protected override onRunStart(ctx: RequestCtxType): RequestCtxType {
    this.startTime = Date.now();
    return ctx;
  }

  protected override onRunComplete(ctx: RequestCtxType): RequestCtxType {
    return { ...ctx, 'elapsed': Date.now() - this.startTime };
  }
}

const pipeline = TimedPipeline.create<RequestCtxType>();

// Stage: attach an Authorization header
pipeline.add((ctx) => { return {
  ...ctx,
  'headers': { ...ctx.headers, 'Authorization': 'Bearer token-abc' }
}; });

const result = await pipeline.run({ 'headers': {}, 'url': '/api/data' });

console.log(`url:           ${result.url}`);
console.log(`Authorization: ${result.headers.Authorization}`);
console.log(`elapsed:       ${result.elapsed}ms`);

The stages getter exposes a readonly view of all registered transforms, useful for inspection or tooling.

Observability hooks

Pipeline exposes eight protected hooks for every stage of execution. The four intercept hooks (onRunStart, beforeStage, afterStage, onRunComplete) return T, stay in-band, and can transform the context or fail the run. The four observer hooks are void, fire at every stage boundary and error path, and are kept observational so they do not replace the stage result or canonical stage error.

Hook	When it fires	Args
`onRunStart(ctx)`	Before the first stage; return value becomes the initial ctx	`ctx: T`
`beforeStage(ctx, index)`	Before each stage fn; return value is passed to the stage fn	`ctx: T`, `index: number`
`onStageStart(index, ctx)`	After `beforeStage`, before the stage fn — void observer	`index: number`, `ctx: T`
`onStageSuccess(index, ctx)`	After the stage fn succeeds, before `afterStage` — void observer	`index: number`, `ctx: T`
`afterStage(ctx, index)`	After each stage fn; return value becomes ctx for the next stage	`ctx: T`, `index: number`
`onStageError(index, error)`	When a stage fn throws, before the error is wrapped — void observer	`index: number`, `error: unknown`
`onRunError(error)`	When a stage error propagates out of `run()`, after `onStageError` — void observer	`error: unknown`
`onRunComplete(ctx)`	After all stages complete; return value is the resolved result	`ctx: T`

import type { StepCtxTypeEntity } from './entities/StepCtxTypeEntity.js';

import { Pipeline, PipelineError } from '../src/index.js';

class TracingPipeline<T extends StepCtxTypeEntity.Type> extends Pipeline<T> {
  readonly stageStartEvents: { 'ctx': T; 'index': number }[] = [];
  readonly stageSuccessEvents: { 'ctx': T; 'index': number }[] = [];
  readonly stageErrorEvents: { 'error': unknown; 'index': number }[] = [];
  readonly runErrorEvents: { 'error': unknown }[] = [];

  protected override onRunStart(ctx: T): T {
    console.log('[pipeline] runStart');
    return ctx;
  }

  protected override beforeStage(ctx: T, index: number): T {
    console.log(`[pipeline] beforeStage index=${index}`);
    return ctx;
  }

  protected override onStageStart(index: number, ctx: T): void {
    console.log(`[pipeline] stageStart index=${index}`);
    this.stageStartEvents.push({ 'ctx': ctx, 'index': index });
  }

  protected override onStageSuccess(index: number, ctx: T): void {
    console.log(`[pipeline] stageSuccess index=${index}`);
    this.stageSuccessEvents.push({ 'ctx': ctx, 'index': index });
  }

  protected override afterStage(ctx: T, index: number): T {
    console.log(`[pipeline] afterStage index=${index}`);
    return ctx;
  }

  protected override onStageError(index: number, error: unknown): void {
    const msg = error instanceof Error ? error.message : String(error);
    console.log(`[pipeline] stageError index=${index} error=${msg}`);
    this.stageErrorEvents.push({ 'error': error, 'index': index });
  }

  protected override onRunError(error: unknown): void {
    const msg = error instanceof PipelineError ? `PipelineError: ${error.message}` : String(error);
    console.log(`[pipeline] runError error=${msg}`);
    this.runErrorEvents.push({ 'error': error });
  }

  protected override onRunComplete(ctx: T): T {
    console.log('[pipeline] runComplete');
    return ctx;
  }
}

// ── Happy-path run: 3 stages that mutate step/value ───────────────────────────

const successPipeline = TracingPipeline.create<StepCtxTypeEntity.Type>();

successPipeline.add((ctx) => { return { 'step': ctx.step + 1, 'value': `${ctx.value}->alpha` }; });
successPipeline.add((ctx) => { return { 'step': ctx.step + 1, 'value': `${ctx.value}->beta` }; });
successPipeline.add((ctx) => { return { 'step': ctx.step + 1, 'value': `${ctx.value}->gamma` }; });

console.log('\n--- happy path ---');
const successResult = await successPipeline.run({ 'step': 0, 'value': 'start' });
console.log(`result: step=${successResult.step} value=${successResult.value}`);

// ── Failing run: 2 stages where the second throws ────────────────────────────

const failPipeline = TracingPipeline.create<StepCtxTypeEntity.Type>();

failPipeline.add((ctx) => { return { 'step': ctx.step + 1, 'value': `${ctx.value}->alpha` }; });
failPipeline.add((_ctx) => { throw new Error('stage 1 fails'); });

console.log('\n--- failing path ---');
try {
  await failPipeline.run({ 'step': 0, 'value': 'start' });
} catch (err: unknown) {
  const msg = err instanceof PipelineError ? `PipelineError: ${err.message}` : String(err);
  console.log(`caught: ${msg}`);
}

The base class never calls any logger or metrics library. Observer hooks are no-ops by default; intercept hooks are the behavioral seams.

The four observer hooks run through a composed HookInvoker (see @studnicky/errors). Pass hookTimeoutMs — via Pipeline.create<T>({ hookTimeoutMs }) or Pipeline.builder<T>().hookTimeoutMs(value) — to bound how long an async observer hook may run before it fails through onHookError with a HookTimeoutError cause. Left unset, a hook may take arbitrarily long, matching prior behavior.

Source on GitHub

@studnicky/pipeline ​

Install ​

Usage ​

Try it ​

Subpath exports ​

Extending ​

Observability hooks ​