webhook-ingest

Composable webhook ingestion for Express, with idempotency, async error handling, and pluggable queue adapters.

The problem

Every Express project that handles webhooks ends up writing the same things: signature verification, duplicate event detection, try/catch boilerplate, and some mechanism to hand off processing to a background queue. webhook-ingest extracts that pattern into a typed, composable library so you stop copying it between projects.

Packages

Package	Description
webhook-ingest	Core — router, types, in-memory queue, idempotency store
webhook-ingest/bull	Bull adapter (in development)
webhook-ingest/pg-boss	pg-boss adapter (in development)

The core package is self-contained with zero production dependencies beyond Express. Queue adapters are opt-in.

Installation

npm install webhook-ingest

Quick start

import express from "express";
import {
  createWebhookRouter,
  createDispatcher,
  MemoryQueue,
  InMemoryIdempotencyStore,
} from "webhook-ingest";
import { WebhookDefinition } from "webhook-ingest/types";

const stripeWebhook: WebhookDefinition = {
  path: "/stripe",
  middleware: express.raw({ type: "application/json" }),
  verifier: async ({ req }) => {
    // verify signature, parse payload
    const payload = req.body as { type: string; id: string };
    return {
      id: payload.id,
      type: payload.type,
      payload,
    };
  },
};

const queue = new MemoryQueue();
const store = new InMemoryIdempotencyStore();

const router = createWebhookRouter([stripeWebhook], queue, {
  idempotencyStore: store,
});

const dispatch = createDispatcher({
  "payment_intent.succeeded": async (payload, { eventId }) => {
    console.log(`Fulfilling order for event ${eventId}`);
  },
});

const app = express();
app.use("/webhooks", router);

// In production, run this in a separate worker process
app.use(async (_req, _res, next) => {
  await queue.processAll(dispatch);
  next();
});

Architecture

Ingestion (per request)

• def.middleware — runs first; use this for body parsing and signature verification
• def.verifier — returns { id, type, payload } to accept, or throws to reject with a 400
• idempotencyStore.has(id) — skips already-processed events (optional)
• queue.enqueue(job) — hands off to MemoryQueue, Bull, or pg-boss
• 202 Accepted — response is sent immediately; processing is async

Processing (worker)

• queue.processAll(dispatch) — drains the queue, job by job
• createDispatcher — routes each job.type to the registered handler

Each concern is a separate, swappable piece. You can bring your own verifier, your own queue, your own idempotency store, the router just wires them together.

API

createWebhookRouter(definitions, queue, options?)

Creates an Express router from one or more webhook definitions.

Parameter	Type	Description
definitions	WebhookDefinition[]	One entry per webhook endpoint
queue	QueueAdapter	Where verified jobs are enqueued
options.idempotencyStore	IdempotencyStore	Optional — skips duplicate event ids

WebhookDefinition

type WebhookDefinition<T = unknown> = {
  path: string;
  middleware?: RequestHandler | RequestHandler[];
  verifier: (context: WebhookContext) => Promise<VerifiedWebhook<T>>;
  toQueueJob?: (verified: VerifiedWebhook<T>) => QueueJob;
};

verifier is where your signature verification logic lives. Throw to reject the request; return a VerifiedWebhook to accept it.

toQueueJob is optional — use it to reshape the verified payload before it enters the queue. Defaults to { id, type, payload }.

createDispatcher(handlers)

Maps event types to async handler functions.

const dispatch = createDispatcher({
  "payment_intent.succeeded": async (payload, { eventId }) => { ... },
  "customer.subscription.deleted": async (payload, { eventId }) => { ... },
});

Unregistered event types log a warning and are skipped.

QueueAdapter

interface QueueAdapter {
  enqueue(job: QueueJob): Promise<void>;
}

Implement this interface to bring your own queue backend. First-party Bull and pg-boss adapters are in development.

IdempotencyStore

interface IdempotencyStore {
  has(id: string): Promise<boolean>;
  set(id: string): Promise<void>;
}

InMemoryIdempotencyStore is included for development. In production, back this with Redis or a database table.

Using the MemoryQueue

MemoryQueue is suitable for development and testing. For production, use a real queue and run the worker in a separate process.

const queue = new MemoryQueue();

// Enqueues are automatic via the router.
// Process manually in tests:
await queue.processAll(dispatch);

// Inspect queued jobs:
const pending = queue.getJobs();

Production considerations

• Idempotency store — swap InMemoryIdempotencyStore for a Redis or database-backed implementation to survive restarts.
• Queue backend — run webhook-ingest/bull or webhook-ingest/pg-boss with a dedicated worker process separate from your HTTP server.
• Raw body — providers like Stripe require access to the raw request body for signature verification. Use express.raw({ type: 'application/json' }) as your middleware rather than express.json().

To add a queue adapter, implement QueueAdapter from webhook-ingest and publish as webhook-ingest/<adapter-name>.

License

MIT