MiddlewarePipeline

Package: @tagna/udiot v1.0.0-rc.1 Source: src/middleware/pipeline.ts Import:

import { MiddlewarePipeline } from '@tagna/udiot';

Constructor

new MiddlewarePipeline<TContext extends BaseContext = BaseContext>(
  options?: PipelineOptions
)

Creates a new pipeline. The generic parameter TContext locks the pipeline to a specific context type. All middleware added via .use() must accept that same context type.

Default option values:

{
  trackPerformance: true,
  timeout: 0,        // 0 means no timeout
  autoOptimize: false,
  debug: false
}

PipelineOptions

ErrorRecoveryOptions

interface ErrorRecoveryOptions {
  strategy: 'halt' | 'continue' | 'retry' | 'fallback';
  maxRetries?: number;    // Used by 'retry' strategy. Default: 3
  retryDelay?: number;    // Milliseconds between retries. Default: 1000
  fallbackHandler?: MiddlewareFunction; // Used by 'fallback' strategy
}

Recovery applies once, in the middleware that raised the error. When that recovery fails, the error passes through the outer middleware without being retried or counted again. Calling next() twice in one middleware throws.

Methods

.use(middleware)

use(middleware: Middleware<TContext> | MiddlewareFunction<TContext>): this

Adds middleware to the pipeline. Accepts either a bare function or a Middleware object with metadata.

When a bare function is passed, it is wrapped in a Middleware object with the name anonymous_<index> and enabled: true.

Returns this for chaining.

Middleware<TContext> object shape:

interface Middleware<TContext extends BaseContext = BaseContext> {
  name: string;              // Unique identifier (used in PipelineResult.middlewareExecuted)
  handler: MiddlewareFunction<TContext>; // The async function
  priority?: number;         // Lower number = executes earlier. Default: 0
  enabled?: boolean;         // When false, skipped during execution. Default: true
  idempotent?: boolean;      // Safe to re-run after it threw; required for 'retry'. Default: false
  config?: Record<string, any>; // Arbitrary configuration
  metadata?: MiddlewareMetadata; // Category, description, dependencies, performance
}

.catch(handler)

catch(handler: ErrorHandler<TContext>): this

Adds a pipeline-level error handler. Multiple handlers may be added; all are called in registration order when the pipeline fails. Returns this for chaining.

.execute(context)

async execute(context: TContext): Promise<PipelineResult>

Executes all enabled middleware in priority order (ascending priority value, then registration order for equal priorities). Returns a PipelineResult whether the pipeline succeeds or fails — it never throws.

PipelineResult

interface PipelineResult<T = any> {
  success: boolean;                // true if no unhandled error occurred
  result?: T;                      // ctx.result at the end of execution
  error?: Error;                   // Set when success is false
  executionTime: number;           // Total wall-clock milliseconds
  middlewareExecuted: string[];    // Names of middleware that completed successfully
}

middlewareExecuted lists only middleware that completed without throwing. If a middleware throws and the strategy is 'halt', it is not included. If the strategy is 'continue', a failed middleware is not included even though execution continued.

Additional Methods

Priority Ordering

Middleware with a lower priority number executes earlier. Middleware with the same priority runs in registration order.

pipeline
  .use({ name: 'cors',     handler: corsHandler,     priority: 10 })
  .use({ name: 'auth',     handler: authHandler,     priority: 20 })
  .use({ name: 'business', handler: businessHandler, priority: 30 });
// execution order: cors → auth → business

The built-in middleware factories do not set a priority. Set priorities explicitly when mixing built-ins with custom middleware that must run in a specific order relative to them.

Performance Tracking

When trackPerformance: true (the default), the pipeline tracks per-middleware metrics internally. Access them via getStats():

const stats = pipeline.getStats();
console.log(stats.slowest);
// [{ name: 'dbQuery', averageTime: 42, callCount: 1200 }, ...]
console.log(stats.mostErrors);
// [{ name: 'externalApi', errorCount: 7, callCount: 200 }, ...]

Metrics are maintained as a running average; they persist across multiple execute() calls on the same pipeline instance.

Examples

Basic pipeline with three middleware

import { MiddlewarePipeline, createHttpContext } from '@tagna/udiot';
import type { HttpContext } from '@tagna/udiot';

const pipeline = new MiddlewarePipeline<HttpContext>();

pipeline
  .use(async (ctx, next) => {
    ctx.metadata.set('startedAt', Date.now());
    await next();
  })
  .use(async (ctx, next) => {
    if (!ctx.headers['x-api-key']) {
      ctx.result = { error: 'Missing API key' };
      return; // short-circuit
    }
    await next();
  })
  .use(async (ctx, next) => {
    ctx.result = { message: 'Hello', user: ctx.user };
    await next();
  });

const ctx = createHttpContext(req, res);
const result = await pipeline.execute(ctx);

if (result.success) {
  res.json(result.result);
} else {
  res.status(500).json({ error: result.error?.message });
}

Pipeline with timeout and error recovery

import {
  MiddlewarePipeline,
  loggingMiddleware,
  authMiddleware
} from '@tagna/udiot';
import type { HttpContext } from '@tagna/udiot';

const pipeline = new MiddlewarePipeline<HttpContext>({
  timeout: 5000, // abort entire pipeline after 5 seconds
  errorRecovery: {
    strategy: 'retry',
    maxRetries: 3,
    retryDelay: 500
  },
  errorHandler: async (error, ctx) => {
    console.error('Pipeline failed:', error.message);
    ctx.result = { error: 'Service unavailable' };
  },
  debug: true
});

pipeline
  .use(loggingMiddleware())
  .use(authMiddleware({ strategy: 'jwt', secret: process.env.JWT_SECRET! }))
  .use({
    name: 'fetch-quote',
    idempotent: true, // a read: safe for the 'retry' strategy to run again
    handler: async (ctx, next) => {
      ctx.result = await fetchFromUnreliableService();
      await next();
    },
  });

Named middleware with metadata and chained .use()

import { MiddlewarePipeline } from '@tagna/udiot';
import type { HttpContext, Middleware } from '@tagna/udiot';

const timingMiddleware: Middleware<HttpContext> = {
  name: 'timing',
  priority: 5,
  handler: async (ctx, next) => {
    const start = Date.now();
    await next();
    ctx.metadata.set('responseTime', Date.now() - start);
  },
  metadata: {
    category: 'observability',
    description: 'Records end-to-end response time'
  }
};

const pipeline = new MiddlewarePipeline<HttpContext>({ trackPerformance: true })
  .use(timingMiddleware)
  .use({ name: 'handler', priority: 50, handler: myHandlerFn });

Common Pitfalls

Forgetting to call next()

If your middleware never calls next(), all downstream middleware are skipped. This is intentional for short-circuit scenarios (auth guards, cache hits) but is a bug when it happens unintentionally. Check every branch in conditional logic.

Mutating context before next() when downstream needs the original value

Middleware that transforms ctx.body before calling next() gives downstream middleware the transformed body. If you need the original for error reporting, save a reference before mutating:

const originalBody = ctx.body;
ctx.body = transform(ctx.body);
try {
  await next();
} catch (err) {
  ctx.metadata.set('originalBody', originalBody);
  throw err;
}

Using timeout with retry

The timeout option applies to the entire pipeline execution. If retry is also configured with retryDelay, repeated retries can exceed the timeout and cause a timeout error instead of the expected retry result. Set timeout generously or disable it when using retry logic.