Events

Events are facts that happened in your domain. Use an event when the code says "this happened" and multiple parts of the app may care: a post was published, a user registered, an invoice was paid, or a comment was added.

Beignet events are typed definitions. Event buses and listeners decide how the fact is delivered.

bun add @beignet/core

Define an event

import { defineEvent } from "@beignet/core/events";
import { z } from "zod";

export const PostPublished = defineEvent("post.published", {
  payload: z.object({
    postId: z.string().uuid(),
    slug: z.string(),
    publishedAt: z.string().datetime(),
  }),
});

The event name is the stable identity. Beignet's producer helpers parse the payload schema before publication. Registered listeners validate payloads that arrive without in-process producer validation state, including values decoded by serialized transports.

Emit events from use cases

Use cases declare which events they may emit with .emits(...). The handler receives an events helper scoped to that declaration:

const publishPost = useCase
  .command("posts.publish")
  .input(PublishPostInput)
  .output(PostOutput)
  .emits([PostPublished])
  .run(async ({ ctx, input, events }) => {
    return ctx.ports.uow.transaction(async (tx) => {
      const published = await tx.posts.publish(input.slug);

      await events.record(tx.events, PostPublished, {
        postId: published.id,
        slug: published.slug,
        publishedAt: published.publishedAt,
      });

      return published;
    });
  });

events.record(...) catches undeclared events at compile time and throws UseCaseEventDeclarationError if an undeclared event is emitted dynamically.

Record events inside transactions

Recording through tx.events keeps events transactional: if the transaction rolls back, recorded events are discarded; if it commits, the Unit of Work validates and publishes them. Beignet's use-case event helpers forward the canonical, transport-stable Standard Schema output. In-process delivery preserves that validation state, so listeners do not repeat schema transforms. See side effects after commit for the rule, Database and transactions for the Unit of Work wiring, and Outbox when event delivery itself must be durable ā€” the outbox keeps the same events.record(tx.events, ...) API but records events as database rows and drains them after commit with retries.

For simple non-transactional workflows, call events.publish(ctx.ports.eventBus, PostPublished, payload). It validates and publishes the canonical output through the event bus.

Beignet requires every event's parsed output to be canonical transport data. Before publishing or recording an event, producer helpers convert the output to strict JSON and validate the decoded JSON again. EventTransportError rejects the operation before the event reaches the bus or outbox when the output is not JSON-safe or changes during that second validation.

Use null, strings, finite numbers, booleans, arrays, and plain objects. Model timestamps as strings or numbers rather than Date objects. Normalizing transforms such as z.string().trim() are valid because their canonical output does not change when parsed again. Transforms that increment a value, toggle a flag, produce a class instance, or reject or further change their canonical output are not transport-stable. Keep event validation deterministic and side-effect-free because Beignet can run it at producer and receiving boundaries.

Define listeners

Listeners react to events. Create the app-bound defineListener builder once in lib/listeners.ts with createListeners<AppContext>() (see app-bound builders), then define listeners in feature files:

import { defineListener } from "@/lib/listeners";
import { PostPublished } from "@/features/posts/domain/events";

export const enqueuePublishedEmail = defineListener(
  "posts.enqueue-published-email",
  {
    event: PostPublished,
    async handle({ payload, ctx }) {
      await ctx.ports.jobs.dispatch(SendPostPublishedEmailJob, payload);
    },
  },
);

Listeners should live with domain or application code, then be collected in server/listeners.ts and registered from server provider wiring.

Register listeners

// server/listeners.ts
import { postListeners } from "@/features/posts/listeners";

export const listeners = [...postListeners] as const;
// server/providers.ts
import { registerListeners } from "@beignet/core/events";
import { listeners } from "@/server/listeners";

const registration = registerListeners(eventBus, listeners, {
  ctx,
  onError(error, listener) {
    ctx.ports.logger.error("Listener failed", {
      error,
      listener: listener.name,
    });
  },
});

await registration.ready;

ready resolves only after every initial transport subscription is active. registerListeners(...) applies one 10-second registration deadline to the complete registry by default; pass readyTimeoutMs to choose another positive timeout. If setup fails or times out, Beignet starts cleanup for every subscription within the remaining deadline. Cleanup that does not settle in time is included in the rejected aggregate as a ListenerRegistrationCleanupTimeoutError, so startup remains bounded without claiming rollback completed. Call await registration.unsubscribe() during normal teardown and apply a host-level shutdown deadline when required.

Generated listener wiring performs this work in a provider start() hook, so server creation does not resolve until the registry is ready. Readiness covers initial registration only. It is not an ongoing transport health check, and it does not make an ephemeral event bus durable or replay messages missed during a later disconnect.

beignet make listener updates server/listeners.ts and provider wiring. beignet doctor flags listener and event registration drift; see CLI for the generator and doctor details.

Failure and ordering semantics

Listener delivery is not an independently durable fan-out. On an awaited, sequential event bus, listeners run in registration order. Without registerListeners(..., { onError }), a listener failure propagates to the bus and may prevent later listeners from running. If the transport retries the event, listeners that already succeeded may run again.

Providing onError reports and consumes that listener failure, allowing a sequential bus to continue, but Beignet does not then retry the failed listener independently. Choose this policy deliberately. When every side effect needs its own retry and dead-letter lifecycle, have listeners dispatch separate idempotent jobs or record separate outbox messages instead of relying on event fan-out for delivery isolation.

Event bus adapters

The starter ships no event bus. beignet make event and beignet make resource --events add the eventBus: EventBusPort port and register the memory provider when the app does not have one yet, and skip the wiring when the ports file already mentions eventBus. The in-memory bus suits local development, tests, and single-process apps:

// server/providers.ts
import { createMemoryEventBusProvider } from "@beignet/provider-event-bus-memory";

export const providers = [createMemoryEventBusProvider()] as const;

Tests and app-owned wiring can also create the bus directly with createMemoryEventBus() from the same package.

When implementing an app-owned EventBusPort, call prepareEventPayloadForTransport(...) before publication. Deliver its payload to in-process subscribers, encode its transportValue in serialized adapters, and forward the complete publishOptions object to in-process subscribers. Serialized adapters should transport documented metadata such as the trace carrier and let registerListeners(...) validate decoded payloads in the receiving process.

For multi-process best-effort delivery, use the Redis Pub/Sub provider:

bun add @beignet/provider-event-bus-redis ioredis
// server/providers.ts
import { createRedisEventBusProvider } from "@beignet/provider-event-bus-redis";

export const providers = [createRedisEventBusProvider()] as const;

Set REDIS_EVENT_BUS_URL=redis://localhost:6379, or let the CLI apply the full preset:

bun beignet provider add event-bus-redis

Redis Pub/Sub gives cross-instance delivery while subscribers are online. It does not persist messages, replay missed events, acknowledge handlers, retry failed handlers, or dead-letter failures. Production apps that need durable event delivery should use Outbox, jobs, or an app-owned durable transport behind the same event bus port; feature code keeps publishing through ctx.ports.eventBus either way. The provider opens separate publisher and subscriber connections in every process where it is installed. ioredis reconnects and resubscribes after a transient disconnect, but events published during that gap are lost. Host subscriptions in long-lived web or worker processes rather than ephemeral request runtimes. The env-backed provider accepts a standalone or managed single-endpoint Redis URL; Sentinel and Cluster users should inject app-owned ioredis clients through createRedisEventBus(...) and validate failover against their exact topology. The direct adapter exposes stop() for shutdown: it detaches Beignet's message listener, cancels subscription retries, and awaits channel cleanup without closing the app-owned publisher or subscriber. Failed cleanup rejects and is reported through instrumentation; stop() retries channels whose earlier unsubscribe failed before reporting any remaining cleanup failure. Cleanup has a 10-second deadline by default, configurable with cleanupTimeoutMs or REDIS_EVENT_BUS_CLEANUP_TIMEOUT_MS, so a stuck Redis command cannot hold server rollback open forever. Await stop() before closing or reusing those clients, and force-disconnect caller-owned clients if cleanup rejects. When ports.tracing is installed, the Redis envelope carries Beignet's versioned trace carrier and registerListeners(...) continues the producer trace in the subscriber process. Legacy envelopes and malformed trace metadata still deliver the event payload. Redis subscribe and unsubscribe command failures are reported through the provider's onSubscriberError callback and recorded as eventBus.subscription.failed instrumentation, so a failed listener registration is visible in enabled devtools or app telemetry. A failed subscription retries with capped backoff while at least one local handler remains registered; its ready promise stays pending until Redis acknowledges the subscription or the caller cancels it. Removing the final handler cancels the retry. Publish failures reject the caller and record eventBus.publish.failed; the provider never reports a failed Redis command as successful delivery. Connection errors from env-backed publisher and subscriber clients are recorded as eventBus.connection.failed when the watcher is enabled. Without enabled instrumentation, ioredis's fallback logging remains active. The canonical generated listener provider also reports handler failures once from registerListeners(...).onError. Direct listener registration remains app-owned; do not report again inside a listener handler when the registration boundary already owns the incident. See Runtime recipes for the process topology and readiness implications of best-effort cross-process events.

Where events fit