Events

An orders API with @dotpress/events, on Fastify. Placing an order emits an order.placed event; its two listeners run later, in a worker, through a Redis queue (BullMQ):

the API answers without waiting for the listeners: it only queues one message per listener;
send-order-confirmation logs the confirmation, and notify-warehouse calls a warehouse service over HTTP, with an idempotency key;
a listener that throws is retried (3 attempts, exponential backoff), then kept in the failed jobs of the queue;
without Redis, an in-memory queue runs the listeners in the API process: nothing to install to try it.

Source: examples/events.

Run it

npx degit dotpressjs/dotpress/examples/events my-api
cd my-api
npm install
npm run dev

curl -X POST localhost:3000/orders -H 'content-type: application/json' \
  -d '{"items":[{"sku":"BOOK-1","quantity":2}]}'

The logs of the API show the listeners running.

With Redis

docker compose up -d # Redis
REDIS_URL=redis://localhost:6379 npm run dev # terminal 1: the API
REDIS_URL=redis://localhost:6379 npm run worker # terminal 2: the worker

Stop the worker, place orders, then start it again: it handles the orders that waited in the queue. npm run worker runs dotpress-events work src/worker.ts --watch.

The event and its listeners

Shared by the API and the worker. load imports the code of a listener in the worker only:

import { defineEvent, defineListener } from '@dotpress/events'
import { z } from 'zod'

// Shared by the API, which emits, and by the worker, which runs the
// listeners.

/** A fact: an order was placed. Payloads travel as JSON. */
export const orderPlaced = defineEvent({
  name: 'order.placed',
  schema: z.object({
    orderId: z.uuid(),
    items: z.array(z.object({ sku: z.string(), quantity: z.number().int() })),
  }),
})

// Each listener gets its own message: it is retried on its own. `load`
// keeps the code of the listeners out of the API, which only needs their
// names.
export const sendOrderConfirmation = defineListener(orderPlaced, {
  name: 'send-order-confirmation',
  load: () => import('./listeners/send-order-confirmation.ts'),
})

export const notifyWarehouse = defineListener(orderPlaced, {
  name: 'notify-warehouse',
  load: () => import('./listeners/notify-warehouse.ts'),
})

export const listeners = [sendOrderConfirmation, notifyWarehouse]

A listener can run twice: the message id is the idempotency key sent to the warehouse service.

import type { EventPayload, ListenerContext } from '@dotpress/events'
import type { orderPlaced } from '../events.ts'

/**
 * Sends the order to the warehouse service (`WAREHOUSE_URL`). When the
 * service fails, the listener throws: the message is retried, then kept in
 * the failed jobs of the queue.
 */
export default async ({
  payload,
  id,
  logger,
}: ListenerContext<EventPayload<typeof orderPlaced>>) => {
  const url = process.env.WAREHOUSE_URL
  if (!url) {
    logger.info(
      { orderId: payload.orderId },
      'WAREHOUSE_URL is not set: the warehouse is not notified'
    )
    return
  }

  const res = await fetch(url, {
    method: 'POST',
    headers: {
      'content-type': 'application/json',
      // A message can be delivered twice (retry after a timeout): the id
      // is the same for every attempt, so the service ignores duplicates.
      'idempotency-key': id,
    },
    body: JSON.stringify(payload),
    signal: AbortSignal.timeout(10_000),
  })
  if (!res.ok) throw new Error(`The warehouse answered ${res.status}`)

  logger.info({ orderId: payload.orderId }, 'Warehouse notified')
}

Emitting

The route stores the order, then emits: an event describes something that happened.

import { createRouter, notFoundError } from '@dotpress/core'
import { z } from 'zod'
import { orderPlaced } from './events.ts'
import type { OrderStore } from './orders.ts'

const Order = z
  .object({
    id: z.uuid(),
    items: z.array(z.object({ sku: z.string(), quantity: z.number().int() })),
    createdAt: z.iso.datetime(),
  })
  .meta({ id: 'Order' })

const ORDER_NOT_FOUND = 'Order not found'

export const ordersRouter = (store: OrderStore) => {
  const router = createRouter()
  const orders = router.createGroup('/orders')

  orders.defineRoute({
    method: 'post',
    path: '/',
    status: 201,
    schema: (z) => ({
      body: z.object({
        items: z
          .array(
            z.object({
              sku: z.string().trim().min(1).max(64),
              quantity: z.number().int().min(1).max(1000),
            })
          )
          .min(1),
      }),
      response: Order,
    }),
    meta: { summary: 'Place an order', tags: ['orders'] },
    // ctx.events comes from eventsPlugin.
    handler: async ({ req, events }) => {
      // Store first, then emit: an event describes something that
      // happened. With a database, emit after the transaction commits.
      const order = await store.create(req.body.items)
      // Returns once the messages are queued: the listeners run later, in
      // the worker, and the client does not wait for them.
      await events.emit(orderPlaced, { orderId: order.id, items: order.items })
      return order
    },
  })

  orders.defineRoute({
    path: '/:id',
    schema: () => ({ response: Order }),
    meta: {
      summary: 'Get an order',
      tags: ['orders'],
      responses: { 404: ORDER_NOT_FOUND },
    },
    handler: async ({ req }) =>
      (await store.get(req.params.id)) ?? notFoundError(ORDER_NOT_FOUND),
  })

  return router
}

The transport comes from the configuration: BullMQ with REDIS_URL, an in-memory queue without it.

import {
  createEventBus,
  createMemoryTransport,
  type EventBusOptions,
  type Transport,
} from '@dotpress/events'
import { createBullmqTransport } from '@dotpress/events/bullmq'
import { listeners } from './events.ts'

type Env = Record<string, string | undefined>

/** Name of the BullMQ queue, shared by the API and the worker. */
export const queueName = (env: Env = process.env) =>
  env.EVENTS_QUEUE ?? 'events'

/** BullMQ connection options from a Redis URL (`redis://` or `rediss://`). */
export const redisConnection = (redisUrl: string) => {
  const url = new URL(redisUrl)
  return {
    host: url.hostname,
    port: Number(url.port || 6379),
    ...(url.username && { username: decodeURIComponent(url.username) }),
    ...(url.password && { password: decodeURIComponent(url.password) }),
    ...(url.pathname.length > 1 && { db: Number(url.pathname.slice(1)) }),
    ...(url.protocol === 'rediss:' && { tls: {} }),
    // Required by the BullMQ workers.
    maxRetriesPerRequest: null,
  }
}

/** A transport, and how to close it on shutdown. */
export type ClosableTransport = Transport & { close(): Promise<void> }

/**
 * With `REDIS_URL`, the events go to a BullMQ queue, handled by the worker
 * (`npm run worker`). Without it, an in-memory queue runs the listeners in
 * this process: development without Redis.
 */
export const createTransport = (env: Env = process.env): ClosableTransport => {
  if (env.REDIS_URL) {
    return createBullmqTransport({
      name: queueName(env),
      connection: redisConnection(env.REDIS_URL),
    })
  }
  const memory = createMemoryTransport()
  // Waits for the listeners in progress.
  return Object.assign(memory, { close: () => memory.drain() })
}

/** The event bus of the API. */
export const createBus = (
  transport: Transport,
  options: Pick<EventBusOptions, 'logger' | 'onFailure'> = {}
) => createEventBus({ transport, listeners, ...options })

The worker

import { createConsoleLogger } from '@dotpress/core'
import { createWorker } from '@dotpress/events'
import { createBullmqSource } from '@dotpress/events/bullmq'
import { queueName, redisConnection } from './bus.ts'
import { listeners } from './events.ts'

// Entry of `dotpress-events work src/worker.ts`: reads the BullMQ queue and
// runs the listeners, until Ctrl+C or SIGTERM (it then finishes the jobs in
// progress). In production, run it in its own container.
const redisUrl = process.env.REDIS_URL
if (!redisUrl) {
  throw new Error(
    'REDIS_URL is required: without it, the API runs the listeners itself.'
  )
}

export const logger = createConsoleLogger({ bindings: { process: 'worker' } })

export default {
  worker: createWorker({
    listeners,
    logger,
    // The attempts of the BullMQ jobs (3 by default): the last failure is
    // reported as final, for alerts.
    maxAttempts: 3,
    onFailure: (error, { envelope, attempt, final }) => {
      if (final) {
        logger.error(
          { error, listener: envelope?.listener, id: envelope?.id, attempt },
          'Event listener gave up: the job stays in the failed jobs'
        )
      }
    },
  }),
  source: createBullmqSource({
    name: queueName(),
    connection: redisConnection(redisUrl),
    concurrency: 5,
  }),
  logger,
}

In production, run the API and the worker as two processes, with dotpress-events work src/worker.ts (without --watch). On SIGTERM, the worker finishes the jobs in progress before exiting.

Learn more

Events and workers
Deployment › Event workers: SQS, pg-boss, Cloudflare Queues, Pub/Sub, Service Bus
Testing helpers

#Events

#Run it

#With Redis

#The event and its listeners

#Emitting

#The worker