⚡ lambda-pipe

LIVE EXAMPLE

🚀 A type-safe plugin & middleware runtime for AWS Lambda APIs.

Build composable serverless backends with typed context, lifecycle hooks, validation, plugins, middleware, caching, runtime adapters, and background jobs.

Basic Route

// src/routes
import {
  defineRoute,
  ok,
} from 'lambda-pipe'

export default defineRoute({
  method: 'GET',

  path: '/health',

  handler: async () => {
    return ok({
      status: 'ok',
    })
  },
})

What is lambda-pipe?

lambda-pipe is a lightweight execution runtime for serverless APIs.

It provides:

• typed request context
• middleware lifecycle
• plugin system
• validation pipeline
• authentication pipeline
• execution metadata
• cache layer
• runtime adapters
• background job handlers

without forcing framework architecture.

Why lambda-pipe?

Most serverless projects become messy over time:

• auth duplicated everywhere
• validation inconsistent
• middleware order unclear
• request context untyped
• plugins tightly coupled
• runtime logic scattered

lambda-pipe provides:

• deterministic execution
• composable plugins
• centralized lifecycle
• typed runtime context
• adapter-friendly architecture

Features

• ⚡ Serverless-first runtime
• 🔌 Plugin system
• 🧩 Middleware pipeline
• 🧠 Cold start lifecycle hooks
• 🛡 Typed validation
• 🔑 Authentication pipeline
• 🍪 Cookie helpers
• 🧱 Runtime adapters
• 📝 Full TypeScript inference
• 💾 In-memory execution cache
• 🌐 Framework agnostic
• 📦 SQS job runtime
• 🚀 Fastify development server

Metal model

Runtime Adapter
      ↓
normalizeRequest()
      ↓
createRequestContext()
      ↓
Plugins
      ↓
Middleware Pipeline
      ↓
Handler
      ↓
serializeResponse()

Installation

npm install lambda-pipe

Route Definition

import {
  defineRoute,
  ok,
} from 'lambda-pipe'

export default defineRoute({
  method: 'GET',

  path: '/users/:id',

  handler: async (context) => {
    return ok({
      id: context.params.id,
    })
  },
})

Typed Context

import {
  defineRoute,
  ok,
} from 'lambda-pipe'

export default defineRoute<
  {
    query: {
      expand?: string
    }

    params: {
      id: string
    }
  }
>({
  method: 'GET',

  path: '/users/:id',

  handler: async (context) => {
    context.query.expand

    context.params.id

    return ok()
  },
})

Route Groups

const users = defineGroup({
  prefix: '/users',
})

export default users(
  defineRoute({
    method: 'GET',

    path: '/:id',

    handler: async () => {},
  }),
)

Validation

import {
  defineRoute,
  ok,
  ValidationError,
} from 'lambda-pipe'

import type {
  Validator,
} from 'lambda-pipe'

const paramsValidator: Validator<{
  id: string
}> = async (input) => {
  const params = input as any

  if (!params.id) {
    throw new ValidationError('missing id')
  }

  return {
    id: params.id,
  }
}

export default defineRoute({
  method: 'GET',

  path: '/users/:id',

  validate: {
    params: paramsValidator,
  },

  handler: async (context) => {
    return ok(context.params)
  },
})

Middleware

import type {
  Middleware,
} from 'lambda-pipe'

const loggerMiddleware: Middleware = {
  onRequest: async (context, next) => {
    console.log(context.req.requestId)

    await next()
  },

  onResponse: async () => {
    console.log('response sent')
  },

  onError: async (error) => {
    console.error(error)
  },
}

Middleware Lifecycle

| Hook | Description | | --------------- | ---------------------------------- | | onInit | Runs once during cold start | | onRequest | Runs before handler | | onResponse | Runs before response serialization | | afterResponse | Runs after response serialization | | onError | Runs on request failure |

Plugins

import type {
  Plugin,
} from 'lambda-pipe'

const authPlugin: Plugin<{
  auth: {
    can: (scope: string) => boolean
  }
}> = {
  name: 'auth',

  setup: async () => {
    console.log('setup once')
  },

  onColdStart: async () => {
    console.log('cold start')
  },

  extend: () => {
    return {
      auth: {
        can: (scope) => {
          return scope === 'read:user'
        },
      },
    }
  },
}

Plugin Context Injection

import {
  defineRoute,
  ok,
} from 'lambda-pipe'

type User = {
  id: string
}

type PluginContext = {
  auth: {
    can: (scope: string) => boolean
  }
}

export default defineRoute<
  {},
  User,
  PluginContext
>({
  plugins: [authPlugin],

  handler: async (context) => {
    context.auth.can('read:user')

    return ok()
  },
})

Authentication

import {
  defineRoute,
  ok,
} from 'lambda-pipe'

export default defineRoute({
  authenticate: async (token) => {
    if (token !== 'valid-token') {
      return null
    }

    return {
      user: {
        id: 'u1',
      },

      scope: ['read:user'],
    }
  },

  handler: async (context) => {
    return ok({
      user: context.req.user,
    })
  },
})

Request Context

| Property | Description | | ------------------------ | ------------------ | | context.req.user | authenticated user | | context.req.scope | auth scopes | | context.req.metadata | auth metadata | | context.req.requestId | request id | | context.req.ip | client ip | | context.req.userAgent | request user agent | | context.req.cookies | parsed cookies | | context.req.receivedAt | request timestamp |

Execution Context

| Property | Description | | -------------------------- | -------------------- | | context.exec.isColdStart | lambda cold start | | context.exec.containerId | runtime container id | | context.exec.cache | in-memory cache |

Cache Example

const cacheKey = `user:${context.params.id}`

const user = await context.exec.cache.getOrSet(
  cacheKey,
  async () => {
    return db.user.findUnique({
      where: {
        id: context.params.id,
      },
    })
  },
  10000,
)

return ok(user)

Response Helpers

import {
  ok,
  json,
  badRequest,
  unauthorized,
  forbidden,
} from 'lambda-pipe'

| Helper | Status | | ---------------- | ------ | | ok() | 200 | | json() | custom | | badRequest() | 400 | | unauthorized() | 401 | | forbidden() | 403 |

Context Response API

context.setHeader('x-request-id', '123')

context.status(200)

context.setCookie('session', 'abc', {
  httpOnly: true,
  secure: true,
  path: '/',
  maxAge: 3600,
})

Error Handling

import {
  HttpError,
} from 'lambda-pipe'

throw new HttpError(
  403,
  'Forbidden',
)

Built-in Errors

import {
  UnauthorizedError,
  ForbiddenError,
  ValidationError,
  NotFoundError,
} from 'lambda-pipe'

AWS Lambda Example

// src/functions/user.ts

import {
  createAWSHandler,
} from 'lambda-pipe/runtime'

import route from '../routes/user'

export const handler =
  createAWSHandler(route)

Production-ready AWS Lambda handler for API Gateway and Lambda Function URLs.

SQS Job Example

import {
  defineJob,
  sqsHandler,
} from 'lambda-pipe/jobs'

const userCreatedJob = defineJob({
  name: 'user-created',

  handler: async (payload: {
    id: string
  }) => {
    console.log(payload.id)
  },
})

export const handler =
  sqsHandler(userCreatedJob)

Utilities

import {
  normalizeHeaders,
  parseCookies,
  extractBearerToken,
  safeJsonParse,
} from 'lambda-pipe'

Route Example

import {
  defineRoute,
  ok,
  HttpError,
} from 'lambda-pipe'

import type {
  Middleware,
  Plugin,
} from 'lambda-pipe'

type User = {
  id: string
}

type PluginContext = {
  auth: {
    can: (
      scope: string,
    ) => boolean
  }
}

/**
 * plugin
 */

const authPlugin: Plugin<
  PluginContext
> = {
  name: 'auth',

  extend: () => ({
    auth: {
      can: (scope: string) =>
        scope === 'read:user',
    },
  }),
}

/**
 * middleware
 */

const authMiddleware: Middleware<
  any,
  User,
  PluginContext
> = {
  onRequest: async (
    context,
    next,
  ) => {
    if (!context.req.user) {
      throw new HttpError(
        401,
        'Unauthorized',
      )
    }

    await next()
  },
}

/**
 * route
 */

export default defineRoute<
  {
    params: {
      id: string
    }
  },
  User,
  PluginContext
>({
  method: 'GET',

  path: '/users/:id',

  plugins: [authPlugin],

  middleware: [
    authMiddleware,
  ],

  authenticate: async (token) => {
    if (token !== 'valid-token') {
      return null
    }

    return {
      user: {
        id: 'u1',
      },

      scope: ['read:user'],
    }
  },

  handler: async (context) => {
    return ok({
      id: context.params.id,

      canRead:
        context.auth.can(
          'read:user',
        ),

      requestId:
        context.req.requestId,
    })
  },
})

Development Runtime

Local development server that simulates AWS Lambda execution using a Fastify-based runtime.

It converts HTTP requests into Lambda-like events and executes routes with full middleware + plugin lifecycle.

import { createDevApp } from 'lambda-pipe/cli'

await createDevApp({
  port: 3000,
  routesDir: 'src/routes',
})

npm run dev

See: DOC

WebSocket Runtime

lambda-pipe supports AWS API Gateway WebSocket APIs with a simple handler abstraction.

It handles:

• $connect
• $disconnect
• custom route messages

import { wsHandler } from 'lambda-pipe/websocket'

export const handler = wsHandler({
  connect: async () => {
    console.log('client connected')
  },

  message: async (ctx) => {
    return {
      statusCode: 200,
      body: JSON.stringify({ ok: true }),
    }
  },

  disconnect: async () => {
    console.log('client disconnected')
  },
})

See: DOC

Server Runtime

Production-ready Fastify runtime for LambdaPipe.

Designed for:

• bundled deployments
• lazy-loaded route manifests
• serverless adapters
• container deployments
• runtime middleware pipelines
• runtime plugin lifecycle hooks

Unlike the dev runtime, production mode does not scan the filesystem dynamically.

Routes are registered through a static manifest for deterministic builds and deployment-safe startup.

import { bootstrap } from 'lambda-pipe/server'

import routes from './routes.manifest'

await bootstrap({
  port: 3000,

  routes,
})

// routes.manifest.ts

import type {
  RouteLoader,
} from 'lambda-pipe/server'

const routes: RouteLoader[] = [
  () => import('./routes/health'),

  () => import('./routes/user'),
]

export default routes

See: DOC

Universal Route with AWS + Fastify

A lightweight architecture pattern that allows a single route definition to run seamlessly across multiple runtimes (AWS Lambda and Fastify) using adapter-based abstraction.

See: DOC

Fastify Runtime

Use LambdaPipe routes directly inside a Fastify server.

Server Setup

// src/server.ts

import Fastify from 'fastify'

import {
  createFastifyHandler,
} from 'lambda-pipe/runtime'

import userRoute from './routes/user'
import healthRoute from './routes/health'

const app = Fastify({
  logger: true,
})

// ============================================
// register routes
// ============================================

app.route({
  method: userRoute.method,

  url: userRoute.path,

  handler:
    createFastifyHandler(userRoute),
})

app.route({
  method: healthRoute.method,

  url: healthRoute.path,

  handler:
    createFastifyHandler(healthRoute),
})

// ============================================
// start server
// ============================================

await app.listen({
  port: 3000,

  host: '0.0.0.0',
})

console.log(
  '🚀 http://localhost:3000',
)

Route Example

// src/routes/user.ts

import {
  defineRoute,
  ok,
} from 'lambda-pipe'

export default defineRoute({
  method: 'GET',

  path: '/user/:id',

  handler: async (context) => {
    return ok({
      id:
        context.pathParameters.id,

      runtime: 'fastify',
    })
  },
})

Health Route

// src/routes/health.ts

import {
  defineRoute,
  ok,
} from 'lambda-pipe'

export default defineRoute({
  method: 'GET',

  path: '/health',

  handler: async () => {
    return ok({
      ok: true,
    })
  },
})

Run Server

tsx src/server.ts

# curl http://localhost:3000/user/123

Proxy Runtime

Production-ready Fastify proxy runtime for LambdaPipe.

Supports:

• dynamic route loading
• lazy imports
• Fastify proxy server
• serverless-style handlers
• shared route manifest
• deployment-safe routing

Create Proxy Runtime

import { createProxyRuntime } from 'lambda-pipe/runtime'

import routes from './routes.manifest'

const app = await createProxyRuntime({
  routes,
})

await app.listen({
  port: 3000,
})

Route Manifest

// routes.manifest.ts

import type {
  RouteManifestItem,
} from 'lambda-pipe/runtime'

const routes: RouteManifestItem[] = [
  {
    path: '/health',

    load: () =>
      import('./routes/health'),
  },

  {
    path: '/user/:id',

    load: () =>
      import('./routes/user'),
  },
]

export default routes

Example Route

import {
  defineRoute,
  ok,
} from 'lambda-pipe'

export default defineRoute({
  method: 'GET',

  path: '/health',

  handler: async () => {
    return ok({
      status: 'ok',
    })
  },
})

Example Startup

import { createProxyRuntime } from 'lambda-pipe/runtime'

import routes from './routes.manifest'

const app = await createProxyRuntime({
  routes,
})

await app.listen({
  port: 3000,

  host: '0.0.0.0',
})

Philosophy

You control:
- plugins
- middleware
- runtime
- architecture

lambda-pipe controls:
- execution
- lifecycle
- typing
- orchestration

Design Principles

• Explicit execution
• Typed runtime
• Composition over inheritance
• Runtime agnostic
• Serverless-first architecture

When to Use

• Serverless APIs
• Lambda backends
• Internal platforms
• Runtime adapters
• Typed middleware systems
• Plugin-based architecture

License

MIT