pipeline

Overview#

Orchestrating a sequence of steps — validation, enrichment, side effects — is extremely common but leads to deeply nested, hard-to-test code. The Pipeline implements the Chain of Responsibility pattern for async, typed workflows: each handler receives a typed context, optionally enriches it, and returns a Result. The pipeline routes success forward and short-circuits on failure — or collects all errors in a single pass, depending on the mode.

Each call to .pipe() returns a new immutable pipeline, so you can share a base pipeline and extend it without side effects. The result includes the final context, any collected errors, and an executedSteps list useful for debugging.

Building a Pipeline#

Create a Pipeline typed with your context shape, then .pipe() handler instances in order. Use .pipeIf(predicate, handler) for conditional steps that only run when the predicate returns true. Call .run(initialContext) to execute.

const pipeline = new Pipeline<PaymentContext>()
  .pipe(new ValidateCardHandler())
  .pipe(new CheckFraudHandler())
  .pipeIf(ctx => ctx.amount > 100, new RequireManagerApprovalHandler())
  .pipe(new ChargeStripeHandler());

const result = await pipeline.run({ userId, amount, card });

if (!result.ok) {
  // result.executedSteps shows which handlers ran before the failure
  logger.debug('Pipeline failed', { steps: result.executedSteps, error: result.error });
}

Writing Handlers#

Extend PipelineHandler<TContext> and implement handle(ctx). Return ok(enrichedCtx) to continue the chain, or fail(error) to abort. The enriched context — with new fields set — flows directly into the next handler.

export class CheckFraudHandler extends PipelineHandler<PaymentContext> {
  async handle(ctx: PaymentContext): Promise<Result<PaymentContext, FraudError>> {
    const score = await this.fraud.score({ userId: ctx.userId, amount: ctx.amount });
    if (score > 80) return fail({ type: 'fraud_detected', score });
    return ok({ ...ctx, fraudScore: score }); // next handler sees fraudScore
  }
}

Execution Modes#

Observability hooks

Attach hooks to instrument every step without modifying handler code. Use them for logging, metrics, or tracing.

NestJS Integration#

Use definePipeline() to create a typed token, register factories in PipelineModule.forRoot(), and inject pipelines into any service — all with full DI support for handler dependencies.

import { PipelineModule, definePipeline } from '@backendkit-labs/pipeline/nestjs';

export const PAYMENT_PIPELINE = definePipeline<PaymentContext, PaymentError>();

@Module({
  imports: [
    PipelineModule.forRoot([
      {
        token: PAYMENT_PIPELINE,
        factory: (fraud: FraudService) => new Pipeline<PaymentContext>()
          .pipe(new ValidateCardHandler())
          .pipe(new CheckFraudHandler(fraud))
          .pipe(new ChargeStripeHandler()),
        inject: [FraudService],
      },
    ]),
  ],
})
export class PaymentModule {}

@Injectable()
export class PaymentService {
  constructor(
    @InjectPipeline(PAYMENT_PIPELINE)
    private readonly pipeline: Pipeline<PaymentContext, PaymentError>,
  ) {}
}

Examples#

From basic to production-grade — copy and adapt.

import { Pipeline } from '@backendkit-labs/pipeline';

interface PaymentContext {
  userId: string; amount: number; currency: string; card: CardDto;
  fraudScore?: number;  // set by CheckFraudHandler
  chargeId?: string;    // set by ChargeStripeHandler
}

const pipeline = new Pipeline<PaymentContext>()
  .pipe(new ValidateCardHandler())
  .pipe(new CheckFraudHandler())
  .pipe(new ChargeStripeHandler())
  .pipe(new SendReceiptHandler());

const result = await pipeline.run({ userId, amount, currency: 'usd', card });

if (result.ok) {
  return { chargeId: result.value.chargeId };
} else {
  throw mapPipelineError(result.error);
}

⚖️ vs. Alternatives#

✅ Supported  ·  ❌ Not supported  ·  ⚠️ Partial / workaround needed. Download counts are approximate weekly npm averages.