@beignet/provider-event-bus-memory

Runtime: Beignet requires Node.js 22.12 or newer. Bun is optional.

In-memory EventBusPort adapter for Beignet applications.

Use it for tests, local development, and single-process apps. Distributed systems should adapt a queue, stream, outbox, or message broker behind the same EventBusPort interface.

This provider is not a durable delivery provider. It does not store attempts, compute retry backoff, or dead-letter failed handlers. Use it when losing in-flight events on process shutdown is acceptable, or put a durable outbox or queue-backed event bus behind EventBusPort for production delivery.

Install

bun add @beignet/provider-event-bus-memory @beignet/core
Copy

Direct setup

import { defineEvent } from "@beignet/core/events";
import { createMemoryEventBus } from "@beignet/provider-event-bus-memory";
import { z } from "zod";

// Define your domain events
const UserRegistered = defineEvent("user.registered", {
  payload: z.object({
    userId: z.string(),
    email: z.string().email(),
  }),
});

// Create the event bus
const eventBus = createMemoryEventBus();

// Subscribe to events
const subscription = eventBus.subscribe(UserRegistered, (payload) => {
  console.log(`User registered: ${payload.email}`);
  // Send welcome email, update analytics, etc.
});

await subscription.ready;

// Publish events
await eventBus.publish(UserRegistered, {
  userId: "123",
  email: "user@example.com",
});

// Unsubscribe when done
await subscription.unsubscribe();
Copy

Framework setup

import { createNextServer, createNextServerLoader } from "@beignet/next";
import { createMemoryEventBusProvider } from "@beignet/provider-event-bus-memory";
import { initialPorts } from "@/infra/port-wiring";
import { routes } from "@/server/routes";

export const getServer = createNextServerLoader(() =>
  createNextServer({
    ports: initialPorts,
    providers: [createMemoryEventBusProvider()],
    context: ({ ports }) => ({
      ports,
    }),
    routes,
  }),
);
Copy

Use createMemoryEventBus() directly when you want to manually assign an event bus under ports.

The provider contributes ctx.ports.eventBus, the standard Beignet EventBusPort. It has no provider-specific escape hatch.

This package is optional by provider metadata. beignet doctor --strict treats an installed-but-unregistered memory event bus as an informational hint rather than a required production provider.

Instrumentation

Pass a provider instrumentation target when creating the direct event bus to record published events under the eventBus watcher:

const eventBus = createMemoryEventBus({
  instrumentation: ports,
});
Copy

Provider instrumentation records published event names under the eventBus watcher. Payloads are not recorded.

Using in use cases

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

const OrderPlaced = defineEvent("order.placed", {
  payload: z.object({
    orderId: z.string(),
    total: z.number(),
  }),
});

// Subscribe to events in your application setup
const orderPlacedSubscription = ctx.ports.eventBus.subscribe(
  OrderPlaced,
  async (payload) => {
    // Send order confirmation email
    await ctx.ports.mailer.send({
      to: customer.email,
      subject: "Order Confirmation",
      text: `Your order ${payload.orderId} has been placed!`,
    });
  },
);
await orderPlacedSubscription.ready;

const placeOrder = useCase
  .command("orders.place")
  .input(PlaceOrderInput)
  .output(OrderOutput)
  .emits([OrderPlaced])
  .run(async ({ ctx, input, events }) => {
    return ctx.ports.uow.transaction(async (tx) => {
      const order = await tx.orders.create(input);

      await events.record(tx.events, OrderPlaced, {
        orderId: order.id,
        total: order.total,
      });

      return order;
    });
  });

// Call this from the application's shutdown hook.
export async function stopOrderListeners() {
  await orderPlacedSubscription.unsubscribe();
}
Copy

EventBus port API

publish<E>(event, payload, options?): Promise<void> | void

Publish a domain event with a typed payload. Await the result so payload validation and the configured delivery mode complete before continuing.

await eventBus.publish(UserRegistered, {
  userId: "123",
  email: "user@example.com",
});
Copy

By default, the in-memory bus awaits handlers so local development and tests are deterministic. Handler errors are rethrown unless onHandlerError is provided. Use delivery: "fire-and-forget" when you intentionally want detached in-process delivery. Fire-and-forget delivery reports handler errors through onHandlerError when provided, but it still does not retry or dead-letter failed work.

The optional third { trace } argument is transport metadata used by Beignet listeners. When instrumentation exposes a tracing port, the bus captures the active context automatically. Existing two-argument calls are unchanged.

subscribe<E>(event, handler): EventSubscription

Subscribe to a domain event. The memory adapter is active synchronously, so ready is already resolved. unsubscribe() is idempotent and removes local delivery before its promise resolves.

const subscription = eventBus.subscribe(UserRegistered, (payload, options) => {
  console.log(`New user: ${payload.email}`);
  console.log(options?.trace?.traceparent);
});

await subscription.ready;

// Later, when you want to stop listening:
await subscription.unsubscribe();
Copy

TypeScript support

The event bus provides full type safety:

import type { EventBusPort } from "@beignet/core/ports";
import { definePorts } from "@beignet/core/ports";

// Type-safe ports definition
const initialPorts = definePorts({
  eventBus: createMemoryEventBus() as EventBusPort,
  // ... other ports
});

type AppPorts = typeof initialPorts;
Copy

Testing

The in-memory event bus is perfect for testing:

import { describe, expect, it, mock } from "bun:test";

describe("User Registration", () => {
  it("should publish UserRegistered event", async () => {
    const eventBus = createMemoryEventBus();
    const handler = mock(() => {});
    
    const subscription = eventBus.subscribe(UserRegistered, handler);
    await subscription.ready;
    
    // Perform registration
    await registerUser(ctx, { email: "test@example.com" });
    
    expect(handler).toHaveBeenCalledWith({
      userId: expect.any(String),
      email: "test@example.com",
    });

    await subscription.unsubscribe();
  });
});
Copy

Behavior

• Awaited by default: publish(...) waits for subscribed handlers unless delivery: "fire-and-forget" is configured
• In-process: Events are only delivered within the same process
• Memory-only: No persistence - events are lost if the process crashes
• Order: Handlers are called in the order they were subscribed
• Multiple handlers: Multiple handlers can subscribe to the same event
• Transport-stable payloads: Direct publication, publishEvent(...), and use-case helpers reject non-JSON output and transforms whose canonical JSON changes when validated again. This keeps local behavior aligned with serialized providers.
• Canonical helper delivery: Producer helpers preserve parsed and transport-stable payload state through in-process delivery, so registered listeners do not repeat validation unnecessarily.
• Error handling: Handler errors reject publish(...) unless onHandlerError is configured

In awaited mode, an unhandled failure stops delivery to later handlers. A transport-level retry may then rerun handlers that already succeeded. With onHandlerError, delivery continues but the failed handler is not retried independently. Use separate idempotent jobs or outbox messages when each side effect needs its own durable retry lifecycle.

Local and tests

Use this provider for local development, deterministic use-case tests, and single-process demos. Prefer it over a durable broker in tests unless the test is specifically proving retry, delivery, or worker behavior.

Deployment notes

Use the memory event bus only when in-process, best-effort delivery is acceptable. For multi-process or production workflows that must survive crashes, publish through Beignet outbox/listener workflows or implement EventBusPort over a durable broker.

When to use

Good for:

• Single-process applications
• Development and testing
• Simple event-driven workflows
• Decoupling application components

Not suitable for:

• Distributed systems
• Event persistence requirements
• Guaranteed delivery needs
• Cross-service communication

For production distributed systems, implement EventBusPort with a proper message broker.

License

MIT