#Events

The definitions, options, errors and contracts of @dotpress/events. For a guide, see Events and workers; for each queue, Event workers.

#defineEvent()

defineEvent({ name: 'invoice.paid', schema: z.object({ invoiceId: z.string() }), transport: 'edge' })

#defineListener()

defineListener(invoicePaid, { name: 'send-receipt', handle: async (context) => { … } })
defineListener(invoicePaid, { name: 'send-receipt', load: () => import('./send-receipt.ts') })

The handler receives:

A handler throws to be retried.

#createEventBus()

The bus checks at startup that every listener has its transport. Names other than default are declared for type checking:

declare module '@dotpress/events' {
  interface EventTransports {
    edge: true
  }
}

#emit()

const { emissionId, messages } = await events.emit(event, payload, { delay, metadata })

messages lists one { id, listener, transport } per listener. An event without listeners is not sent. Each call is a new emission, with new message ids: emitting again after an EventEmitError runs again the listeners whose message was sent.

events.withContext({ logger, metadata }) returns a bus that logs with this logger and adds this metadata: ctx.events is one.

#Transports in the process

The memory transport has sent (the envelopes sent), drain() (waits for every message, retries included) and clear().

createFakeEventBus() checks each emission like a real bus (event, payload, delay, metadata) without running listeners, and records it in emitted; clear() empties it. See Testing helpers.

#createWorker()

info holds the envelope (undefined when the message is not one), the attempt, final (a dead message, or the last attempt), and the reason:

worker.handle(message, { attempt }) returns success, retry or dead, and never throws: the entry points of the transports (Lambda handler, queue() handler…) are built on it.

startWorker(worker, source, { logger, signal }) reads a source until stopped, and returns { stop() }, which finishes the messages in progress.

#eventsPlugin()

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

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

The plugin, named events, provides ctx.events: the bus, with the request logger, and the requestId in the metadata. Using ctx.events without the plugin throws an explicit error.

#Logs

Payloads are never logged.

#Errors

The error classes extend EventsError.

#Custom transports

A transport has a kind and a send(envelopes, { delay }) method; a source has a kind and a consume(worker, { signal, logger }) method. Envelopes are plain JSON objects:

{ v: 1, id, emissionId, event, listener, payload, emittedAt, metadata }