@studnicky/pipeline
Generic typed async pipeline for sequential context transforms.
Install
pnpm add @studnicky/pipelineRequires @studnicky:registry=https://npm.pkg.github.com in .npmrc.
Usage
Construct a Pipeline<T> instance with a fixed array of stages, and run a context through all of them with run(). Each stage receives the context and returns a (possibly transformed) copy. The stage list is fixed at construction — a different composition is a different Pipeline.create() call with a different array:
import { NumberContextTypeEntity } from './entities/NumberContextTypeEntity.js';
class NumberStages {
static double(context: NumberContextTypeEntity.Type): NumberContextTypeEntity.Type { return { 'value': context.value * 2 }; }
static addTen(context: NumberContextTypeEntity.Type): NumberContextTypeEntity.Type { return { 'value': context.value + 10 }; }
static timesThree(context: NumberContextTypeEntity.Type): NumberContextTypeEntity.Type { return { 'value': context.value * 3 }; }
}
// Three-stage pipeline: double, then add ten, then multiply by three
const threeStagePipeline = Pipeline.create<NumberContextTypeEntity.Type>([
NumberStages.double, NumberStages.addTen, NumberStages.timesThree
]);
// Two-stage pipeline: add ten, then multiply by three — a different fixed
// composition constructed from a different stage array
const twoStagePipeline = Pipeline.create<NumberContextTypeEntity.Type>([NumberStages.addTen, NumberStages.timesThree]);
console.log(`Three-stage pipeline stages: ${threeStagePipeline.stages.length}`);
console.log(`Two-stage pipeline stages: ${twoStagePipeline.stages.length}`);
class PipelineRunDemo {
// Runs both fixed pipelines against the same input, returning both
// results so the caller ends up with a single top-level binding.
static async run(): Promise<{ 'withDouble': number; 'withoutDouble': number }> {
// (5 * 2 + 10) * 3 = 60
const result = await threeStagePipeline.run(NumberContextTypeEntity.create({ 'value': 5 }));
console.log(`Result with 3 stages: ${result.value}`);
// (5 + 10) * 3 = 45
const resultWithoutDouble = await twoStagePipeline.run(NumberContextTypeEntity.create({ 'value': 5 }));
console.log(`Result without double stage: ${resultWithoutDouble.value}`);
return { 'withDouble': result.value, 'withoutDouble': resultWithoutDouble.value };
}
}
const results = await PipelineRunDemo.run();Try it
The basic demo constructs a Pipeline directly with Pipeline.create<RequestCtx>([...stages]). Each stage receives the transformed context from the previous one.
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.
Public API
Import Pipeline and PipelineError from @studnicky/pipeline; import schema namespaces from @studnicky/pipeline/entities and type contracts from @studnicky/pipeline/interfaces.
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 { HookRequestContextEntity } from './entities/HookRequestContextEntity.js';
class TimedPipeline extends Pipeline<HookRequestContextEntity.Type> {
private startTime = 0;
protected override onRunStart(context: HookRequestContextEntity.Type): HookRequestContextEntity.Type {
this.startTime = Date.now();
return context;
}
protected override onRunComplete(context: HookRequestContextEntity.Type): HookRequestContextEntity.Type {
return { ...context, 'elapsed': Date.now() - this.startTime };
}
}
const pipeline = TimedPipeline.create<HookRequestContextEntity.Type>([
// Stage: attach an Authorization header
(context) => { return {
...context,
'headers': { ...context.headers, 'Authorization': 'Bearer token-abc' }
}; }
]);
const result = await pipeline.run(HookRequestContextEntity.create({ 'headers': {}, 'url': '/api/data' }));
console.log(`url: ${result.url}`);
console.log(`Authorization: ${result.headers.Authorization}`);
console.log(`elapsed: ${result.elapsed}ms`);The stages getter returns a readonly snapshot of all constructed transforms, useful for inspection or tooling.
Observability hooks
Pipeline exposes eight protected hooks for every stage of execution. The four transform 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 { PipelineOptionsEntity } from '../src/entities/index.js';
import type { PipelineFunctionInterface } from '../src/interfaces/index.js';
import { Pipeline, PipelineError } from '../src/index.js';
import { StepContextTypeEntity } from './entities/StepContextTypeEntity.js';
class TracingPipeline<T extends StepContextTypeEntity.Type> extends Pipeline<T> {
public constructor(
stages: readonly PipelineFunctionInterface<T>[],
options?: Readonly<PipelineOptionsEntity.Type>
) {
super(stages, options);
}
readonly stageStartEvents: { 'context': T; 'index': number }[] = [];
readonly stageSuccessEvents: { 'context': T; 'index': number }[] = [];
readonly stageErrorEvents: { 'error': Error; 'index': number }[] = [];
readonly runErrorEvents: { 'error': Error }[] = [];
protected override onRunStart(context: T): T {
console.log('[pipeline] runStart');
return context;
}
protected override beforeStage(context: T, index: number): T {
console.log(`[pipeline] beforeStage index=${index}`);
return context;
}
protected override onStageStart(index: number, context: T): void {
console.log(`[pipeline] stageStart index=${index}`);
this.stageStartEvents.push({ 'context': context, 'index': index });
}
protected override onStageSuccess(index: number, context: T): void {
console.log(`[pipeline] stageSuccess index=${index}`);
this.stageSuccessEvents.push({ 'context': context, 'index': index });
}
protected override afterStage(context: T, index: number): T {
console.log(`[pipeline] afterStage index=${index}`);
return context;
}
protected override onStageError(index: number, error: Error): void {
console.log(`[pipeline] stageError index=${index} error=${error.message}`);
this.stageErrorEvents.push({ 'error': error, 'index': index });
}
protected override onRunError(error: Error): void {
const message = error instanceof PipelineError ? `PipelineError: ${error.message}` : error.message;
console.log(`[pipeline] runError error=${message}`);
this.runErrorEvents.push({ 'error': error });
}
protected override onRunComplete(context: T): T {
console.log('[pipeline] runComplete');
return context;
}
}
// ── Happy-path run: 3 stages that mutate step/value ───────────────────────────
const successPipeline = new TracingPipeline<StepContextTypeEntity.Type>([
(context) => { return { 'step': context.step + 1, 'value': `${context.value}->alpha` }; },
(context) => { return { 'step': context.step + 1, 'value': `${context.value}->beta` }; },
(context) => { return { 'step': context.step + 1, 'value': `${context.value}->gamma` }; }
]);
console.log('\n--- happy path ---');
const successResult = await successPipeline.run(StepContextTypeEntity.create({ 'step': 0, 'value': 'start' }));
console.log(`result: step=${successResult.step} value=${successResult.value}`);
// ── Failing run: 2 stages where the second throws ────────────────────────────
const failPipeline = new TracingPipeline<StepContextTypeEntity.Type>([
(context) => { return { 'step': context.step + 1, 'value': `${context.value}->alpha` }; },
(_context) => { throw RuntimeError.create('stage 1 fails'); }
]);
console.log('\n--- failing path ---');
try {
await failPipeline.run(StepContextTypeEntity.create({ 'step': 0, 'value': 'start' }));
} catch (error: unknown) {
const message = error instanceof PipelineError ? `PipelineError: ${error.message}` : String(error);
console.log(`caught: ${message}`);
}The base class never calls any logger or metrics library. Observer hooks are no-ops by default; transform hooks are the behavioral seams.
The four observer hooks run through a composed HookInvoker (see @studnicky/errors). A throwing observer surfaces as HookInvocationError. Pass hookTimeoutMs to Pipeline.create<T>([...stages], { hookTimeoutMs }) to bound an asynchronous observer; exceeding the bound surfaces through HookTimeoutError. Without hookTimeoutMs, hook invocation is unbounded.
Entities
@studnicky/pipeline/entities exports the pipeline option schema namespace.
import { PipelineOptionsEntity } from '@studnicky/pipeline/entities';Interfaces
@studnicky/pipeline/interfaces exports pipeline stage and runner contracts.
import type { PipelineFunctionInterface } from '@studnicky/pipeline/interfaces';Exports
| Symbol | Purpose | Import path |
|---|---|---|
Pipeline | Runs typed transformation stages in sequence. | @studnicky/pipeline |
PipelineError | Represents pipeline execution failures. | @studnicky/pipeline |