> For AI agents: the complete documentation index is available at https://dotpress.dev/llms.txt, the full documentation bundle is available at https://dotpress.dev/llms-full.txt.

# 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.


```sh [npm]
npm install @dotpress/events zod
```

```sh [yarn]
yarn add @dotpress/events zod
```

```sh [pnpm]
pnpm add @dotpress/events zod
```

```sh [bun]
bun add @dotpress/events zod
```

```sh [deno]
deno add npm:@dotpress/events npm: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](https://dotpress.dev/examples/events.md):

```ts file="<root>/../examples/events/src/events.ts"
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`:

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

declare const crm: { addContact(userId: string): Promise<void> }
// ---cut---
const userCreated = defineEvent({
  name: 'user.created',
  schema: z.object({ userId: z.string() }),
})

const syncCrm = defineListener(userCreated, {
  name: 'sync-crm',
  handle: async ({ payload, logger }) => {
    await crm.addContact(payload.userId)
    logger.info({ userId: payload.userId }, '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.

:::warning 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:

```ts file="<root>/../examples/events/src/listeners/notify-warehouse.ts"
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

```ts twoslash
import { createEventBus, createMemoryTransport, defineEvent } from '@dotpress/events'
import { z } from 'zod'

const invoicePaid = defineEvent({
  name: 'invoice.paid',
  schema: z.object({ invoiceId: z.string(), paidAt: z.iso.datetime() }),
})
const events = createEventBus({ transport: createMemoryTransport(), listeners: [] })
// ---cut---
await events.emit(
  invoicePaid,
  { invoiceId: 'F-1', paidAt: new Date().toISOString() },
  { delay: 3600, metadata: { source: '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.

:::warning 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

```ts
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.

```ts file="<root>/../examples/events/src/bus.ts"
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](https://dotpress.dev/deployment/event-workers.md).

## Workers

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

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

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

```ts file="<root>/../examples/events/src/worker.ts"
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](https://dotpress.dev/deployment/event-workers.md).

## 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](https://dotpress.dev/reference/events.md).

## Testing

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

```ts
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`:

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

See [Testing helpers](https://dotpress.dev/services/testing.md).