Events and workers

@dotpress/events lets your code announce what happened, and lets other code react to it, later, in workers. A route that places an order emits order.placed and answers at once; the confirmation email and the warehouse notification run in a worker, retried on their own when they fail.

The code that emits never knows where the listeners run: in the same process during development, in a container reading Redis, or in a Lambda function reading SQS. Changing the transport changes one line.

npm

yarn

pnpm

bun

deno

npm install @dotpress/events zod

Concepts

	Role	Where
Event (`defineEvent`)	a fact, with a name and the schema of its payload	shared code
Listener (`defineListener`)	a reaction to an event, with a unique name	shared code
Bus (`createEventBus`)	validates the events emitted, sends one message per listener to its transport	the code that emits: API, scripts
Transport	sends the messages to a queue: memory, Redis, SQS, Postgres…	the configuration of the bus
Worker (`createWorker`)	runs the listeners of the messages it receives	a container, a Lambda function, `dotpress-events work`

emit() sends one message per listener: each listener is retried on its own, and can even run on another platform.

Events and listeners

From the Events example:

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]

load imports the code of a listener, the default export of its module, only in the worker: the API that emits only needs the names of the listeners, not their dependencies. Event names use letters, digits, ., _, : and -; a listener name is unique, and renaming it orphans the messages still in queues under the old name. A listener can also be written inline, with handle:

const const userCreated: EventDefinition<z.ZodObject<{
    userId: z.ZodString;
}, z.core.$strip>>
userCreated = defineEvent<z.ZodObject<{
    userId: z.ZodString;
}, z.core.$strip>>(options: {
    name: string;
    schema: z.ZodObject<{
        userId: z.ZodString;
    }, z.core.$strip>;
    transport?: TransportName | undefined;
}): EventDefinition<z.ZodObject<{
    userId: z.ZodString;
}, z.core.$strip>>
Defines an event: a fact that listeners react to, with the schema of its
payload.
export const userCreated = defineEvent({
  name: 'user.created',
  schema: z.object({ userId: z.string() }),
})
defineEvent({
  name: stringUnique name: letters, digits, ., _, : and -.name: 'user.created',
  schema: z.ZodObject<{
    userId: z.ZodString;
}, z.core.$strip>
Schema of the payload. Payloads travel as JSON: describe JSON values
(dates as strings, or coerced on the way in).schema: import zz.function object<{
    userId: z.ZodString;
}>(shape?: {
    userId: z.ZodString;
} | undefined, params?: string | {
    error?: string | z.core.$ZodErrorMap<NonNullable<z.core.$ZodIssueUnrecognizedKeys | z.core.$ZodIssueInvalidType<unknown>>> | undefined;
    message?: string | undefined | undefined;
} | undefined): z.ZodObject<{
    userId: z.ZodString;
}, z.core.$strip>
object({ userId: z.ZodStringuserId: import zz.function string(params?: string | z.core.$ZodStringParams): z.ZodString (+1 overload)string() }),
})

const const syncCrm: ListenerDefinition<z.ZodObject<{
    userId: z.ZodString;
}, z.core.$strip>>
syncCrm = defineListener<z.ZodObject<{
    userId: z.ZodString;
}, z.core.$strip>>(event: EventDefinition<z.ZodObject<{
    userId: z.ZodString;
}, z.core.$strip>>, options: ListenerOptions<{
    userId: string;
}>): ListenerDefinition<z.ZodObject<{
    userId: z.ZodString;
}, z.core.$strip>>
Defines a listener of an event. Its handler runs in a worker, once per
emission, and is retried when it throws.
export const sendWelcomeMail = defineListener(userCreated, {
  name: 'send-welcome-mail',
  load: () => import('./listeners/send-welcome-mail.js'),
})
defineListener(const userCreated: EventDefinition<z.ZodObject<{
    userId: z.ZodString;
}, z.core.$strip>>
userCreated, {
  name: stringUnique name: letters, digits, ., _, : and -. It identifies the
listener in messages: renaming it orphans messages still in queues.name: 'sync-crm',
  handle: ListenerHandler<{
    userId: string;
}>
handle: async ({ payload: {
    userId: string;
}
Payload, validated by the schema of the event.payload, logger: EventsLoggerlogger }) => {
    await const crm: {
    addContact(userId: string): Promise<void>;
}
crm.function addContact(userId: string): Promise<void>addContact(payload: {
    userId: string;
}
Payload, validated by the schema of the event.payload.userId: stringuserId)
    logger: EventsLoggerlogger.EventsLogger.info(data: object, message: string): voidinfo({ userId: stringuserId: payload: {
    userId: string;
}
Payload, validated by the schema of the event.payload.userId: stringuserId }, 'Contact added to the CRM')
  },
})

The listener receives the parsed payload, typed by the schema, and:

Field	Content
`id`	the identifier of the message, the same for every attempt
`attempt`	the attempt number, from 1
`emissionId`, `emittedAt`	the emission
`metadata`	the metadata of the emission: the `requestId` with the plugin
`logger`	the logger of the worker
`event`, `listener`	the names

Payloads

Payloads travel as JSON. The bus serializes the payload, then validates it with the schema: an invalid payload throws an EventPayloadError at emit(), not later in a worker. Dates travel as ISO strings: convert them in the schema (z.coerce.date(), .transform()) if listeners need Date objects. Workers validate the payload again, against the current schema: a message emitted by a previous version of your code is checked too.

Pass identifiers, not secrets

Messages are stored by queues, shown in their consoles and kept in dead-letter queues. Send { userId }, never a password, a token or personal data the listener can load itself.

Retries and idempotency

A listener throws to be retried. The transport delivers the message again, then moves it to its dead-letter queue after its retry limit.

Make listeners idempotent. Queues deliver messages at least once: a message can run twice, after a timeout or a duplicate delivery. Use id to ignore duplicates (a unique constraint, an idempotency key sent to the API you call), or make running twice harmless:

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

await const events: EventBusevents.EventBus.emit<EventDefinition<z.ZodObject<{
    invoiceId: z.ZodString;
    paidAt: z.ZodISODateTime;
}, z.core.$strip>>>(event: EventDefinition<z.ZodObject<{
    invoiceId: z.ZodString;
    paidAt: z.ZodISODateTime;
}, z.core.$strip>>, payload: {
    invoiceId: string;
    paidAt: string;
}, options?: EmitOptions): Promise<EmitResult>
Emits an event: validates its payload, then sends one message per
listener to the transport of the listener.emit(
  const invoicePaid: EventDefinition<z.ZodObject<{
    invoiceId: z.ZodString;
    paidAt: z.ZodISODateTime;
}, z.core.$strip>>
invoicePaid,
  { invoiceId: stringinvoiceId: 'F-1', paidAt: stringpaidAt: new var Date: DateConstructor
new () => Date (+3 overloads)
Date().Date.toISOString(): stringReturns a date as a string value in ISO format.toISOString() },
  { EmitOptions.delay?: number | undefinedDelay before delivery, in seconds, within the limit of the transport.delay: 3600, EmitOptions.metadata?: Readonly<Record<string, string>> | undefinedMetadata of the emission, passed to the listeners.metadata: { source: stringsource: 'payments' } } // optional
)

The payload is typed by the schema of the event. delay, in seconds, postpones the listeners, within the limit of the transport. emit() returns the emissionId, and one { id, listener, transport } per listener; an event without listeners is not sent.

Emit after the transaction commits

An event describes something that happened. Emitted inside a database transaction that rolls back, it describes something that did not. Commit first, then emit.

When some messages cannot be sent (the queue is down), emit() throws an EventEmitError, which lists the sent and failed messages. Emitting again starts a new emission, with new message ids: the listeners whose message was already sent run twice. Make listeners idempotent on the data they change (a unique constraint on the order, an idempotency key derived from it), not only on the message id.

In a route

import { eventsPlugin } from '@dotpress/events/plugin'

const app = await createApp({ adapter, plugins: [eventsPlugin(events)] })

defineRoute({
  method: 'post',
  path: '/orders',
  handler: async ({ req, events }) => {
    const order = await orders.place(req.body)
    await events.emit(orderPlaced, { orderId: order.id, items: order.items })
    return order
  },
})

ctx.events logs with the request logger, and adds the requestId to the metadata of the events: the logs of the listeners link back to the request. Using ctx.events without the plugin throws an explicit error. Outside a route, events.withContext({ logger, metadata }) does the same.

Transports

Transport	Import	Use
`createSyncTransport()`	`@dotpress/events`	runs the listeners inside `emit()`. Development, tests, apps without a queue
`createMemoryTransport()`	`@dotpress/events`	an in-process queue with retries. Development and tests: messages are lost when the process stops
`createBullmqTransport()`	`@dotpress/events/bullmq`	Redis, through BullMQ
`createSqsTransport()`	`@dotpress/events/aws-sqs`	Amazon SQS, with Lambda workers
`createPgBossTransport()`	`@dotpress/events/pg-boss`	Postgres, through pg-boss: no other infrastructure than your database
`createCloudflareQueueTransport()`	`@dotpress/events/cloudflare-queues`	Cloudflare Queues
`createPubSubTransport()`	`@dotpress/events/google-pubsub`	Google Cloud Pub/Sub
`createServiceBusTransport()`	`@dotpress/events/azure-service-bus`	Azure Service Bus

The example picks its transport from the configuration: BullMQ with Redis, an in-memory queue in the API process 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 })

Several transports can be used together: name them in the bus with transports: { default, … }, and route an event, or a single listener, with its transport option. Declare the names in the EventTransports interface to type-check them; the bus checks at startup that every listener has its transport. The setup of each platform (queues, dead letters, Lambda, containers) is described in Event workers.

Workers

A worker runs the listeners of the messages it receives. During development, run it next to your API with dotpress-events work:

npx dotpress-events work src/worker.ts --watch

The module exports the worker, the source of the messages, and optionally a logger:

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,
}

Ctrl+C or SIGTERM stops the worker after the messages in progress; a second signal, a second later or more, exits at once. In production, the same command is the entry point of a container; on AWS Lambda, the SQS handler replaces it. See Event workers.

Failures

Failure	What happens
The listener throws	the message is retried, then moved to the dead-letter queue after the retry limit
Invalid payload, unknown listener	retrying would not help: the message goes to the dead-letter queue at once when the transport can, or fails until the retry limit

onFailure(error, { envelope, attempt, final, reason }), an option of the worker, is called for every failure: use it for alerts. Set maxAttempts to the retry limit of the queue, so that the last attempt is reported as final. An unknown-listener failure usually means a deployment mistake: a listener routed to a queue whose workers do not run it.

The options, the failure reasons, the logs and the errors are listed in the Events reference.

Testing

The memory transport runs the listeners in the test, and drain() waits for them:

const transport = createMemoryTransport({ retryDelay: () => 0 })
const events = createEventBus({ transport, listeners })

await placeOrder(events, order)
await transport.drain()
expect(transport.sent).toMatchObject([{ event: 'order.placed', listener: 'notify-warehouse' }])

To test the code that emits without running the listeners, use createFakeEventBus() and expectEventEmitted() of @dotpress/testing:

const events = createFakeEventBus()
await placeOrder(events, order)
expectEventEmitted(events, orderPlaced, { orderId: order.id })

See Testing helpers.

#Events and workers

#Concepts

#Events and listeners

#Payloads

#Retries and idempotency

#Emitting

#In a route

#Transports

#Workers

#Failures

#Testing