@setu-ts/events-plugin

In-process domain events. Registers an IEventBus under CAPABILITIES.EVENTS ('events'), backed by InMemoryEventBus.

For events that must cross a process boundary, use @setu-ts/messaging-plugin — or bridge the two with its EventsMessagingBridge.

Installation

import { EventsPlugin } from '@setu-ts/events-plugin';

Usage

import { createApplication } from '@setu-ts/kernel';
import { RuntimePlugin } from '@setu-ts/runtime';
import { defineDomainEvent, EventsPlugin } from '@setu-ts/events-plugin';
import { CAPABILITIES, type IEventBus, type IRuntimeServices } from '@setu-ts/common';

const app = createApplication({
  plugins: [RuntimePlugin(), EventsPlugin()],
});
await app.start({ port: 3000 });

// Bind the event bases to the runtime once, then subclass them.
const runtime = app.services.get<IRuntimeServices>(CAPABILITIES.RUNTIME);
const { DomainEvent } = defineDomainEvent(runtime);

class UserCreated extends DomainEvent<{ userId: string }> {
  override readonly type = 'user.created';
}

async function welcome(userId: string): Promise<void> {
  // Deliver the welcome message through your own transport.
  await Promise.resolve(userId);
}

const bus = app.services.get<IEventBus>(CAPABILITIES.EVENTS);
bus.subscribe<{ userId: string }>('user.created', async (event) => {
  await welcome(event.data.userId);
});

await bus.publish(new UserCreated({ userId: '123' }));
await bus.publishBatch([new UserCreated({ userId: '124' }), new UserCreated({ userId: '125' })]);

Aggregate-local domain events

An aggregate can retain the facts raised during its current operation without inheriting a framework base class or publishing while its persistence policy is unresolved:

import { createDomainEvents, type IDomainEvent } from '@setu-ts/events-plugin';
import { CAPABILITIES, type IEventBus, type IRuntimeServices } from '@setu-ts/common';

class Order {
  readonly events = createDomainEvents();

  place(event: IDomainEvent<{ readonly orderId: string }>): void {
    // Check invariants and mutate order state first.
    this.events.record(event);
  }
}

async function dispatchAfterSave(bus: IEventBus, event: IDomainEvent<unknown>): Promise<void> {
  // Runs after the aggregate is saved. A rejection leaves the event pending, so
  // the loop below removes it only once it has been published.
  await bus.publish(event);
}

function orderPlaced(
  runtime: IRuntimeServices,
  orderId: string,
): IDomainEvent<{ readonly orderId: string }> {
  return {
    type: 'order.placed',
    id: 'event-1',
    occurredOn: new Date(runtime.now()),
    data: { orderId },
  };
}

const bus = app.services.get<IEventBus>(CAPABILITIES.EVENTS);
const order = new Order();
order.place(orderPlaced(runtime, 'order-1'));

// The application owns this boundary: save the aggregate, then dispatch or
// persist the pending facts. Remove or clear only after its chosen policy succeeds.
const pending = order.events.pending();
for (const domainEvent of pending) {
  await dispatchAfterSave(bus, domainEvent);
  order.events.remove(domainEvent);
}

pending() returns an ordered snapshot, so it never exposes the recorder’s mutable collection. remove(event) compares the exact event reference, removes only the first occurrence, and returns false without mutation when that reference is absent. A reconstructed aggregate creates a fresh recorder and therefore begins with no pending facts. This helper never calls IEventBus; automatic publication and durable outbox coordination remain application responsibilities.

Options

errorHandler defaults to logging through the optional logger capability when one is registered, otherwise a no-op. A failing handler never makes publish reject in either dispatch mode.

Diagnostics

EventsPlugin({ diagnostics }) opts into minimized event-dispatch observations for the local diagnostics connector’s GET /v1/event (@setu-ts/diagnostics-plugin). It is an acknowledgement, not a toggle: the option is absent by default, and when supplied enabled must be the literal true — enabled: false is refused at construction.

Only the exact event types listed in events are observed, each aggregated under its approved alias; a type outside the map is neither observed nor counted, and handler function names and identities are never captured. No payload, identifier or thrown value is admitted to the collector. setu devtool enable writes this option for you; this section is for the reader who configures it by hand. Enable only on an approved development dataset.

import { EventsPlugin } from '@setu-ts/events-plugin';

EventsPlugin({
  diagnostics: {
    enabled: true,
    alias: 'orders',
    events: { 'order.placed': 'OrderPlaced', 'order.cancelled': 'OrderCancelled' },
  },
});

Exports

Generated from the package barrel by pnpm run docs:exports; pnpm run check:docs fails when it drifts.

Full API

Every export and option is documented in PUBLIC_API.md.

API reference

EventsPlugin() (@setu-ts/events-plugin)

Provides in-memory event bus for domain events.

Registration

import { EventsPlugin } from '@setu-ts/events-plugin';

app.register(
  EventsPlugin({
    async: true,
    errorHandler: (error, event) => {
      const logger = ctx.services.get('logger');
      logger.error('Event handler failed', { error, eventType: event.type });
    },
  }),
);

Handlers can also be subscribed declaratively, which is what setu generate event-handler wires for you. Each entry is a { type, handler } pair, and the plugin subscribes it through the same subscribeHandler you would call by hand — so the two routes cannot diverge:

import type { EventHandlerRegistration } from '@setu-ts/events-plugin';

const handlers: readonly EventHandlerRegistration[] = [
  { type: 'user-created', handler: new UserCreatedEventHandler() },
  // Or a factory that builds the handler from the service registry:
  { type: 'order-placed', handler: createOrderPlacedEventHandler },
];

app.register(EventsPlugin({ handlers }));

The handler accepts an instance or a RegistryFactory ((services: IServiceRegistry) => handler). A factory is called at the onInit phase — the first phase at which the registry holds every capability — so it can resolve any capability (the broker, the queue, the logger) and build the handler with it; the result subscribes through the same subscribeHandler the instance arm uses, so the two arms cannot diverge. A factory that throws rejects start(), naming the option and the entry. Instance handlers keep their register() timing byte-identically. The imperative subscribeHandler remains available for a handler wired from another plugin’s own register.

Defining Events

import { DomainEvent } from '@setu-ts/events-plugin';

class UserCreatedEvent extends DomainEvent<{ userId: string; email: string }> {
  readonly type = 'UserCreated';
}

class OrderPlacedEvent extends DomainEvent<{ orderId: string; total: number }> {
  readonly type = 'OrderPlaced';
}

Publishing Events

app.router.post('/users', async (ctx) => {
  const eventBus = ctx.services.get<IEventBus>('events');
  const user = await createUser(ctx.request.body);

  await eventBus.publish(
    new UserCreatedEvent({
      userId: user.id,
      email: user.email,
    }),
  );

  return ctx.response.status(201).json(user);
});

Recording aggregate-local domain events

Use createDomainEvents() as a field owned by an aggregate. It records local facts only: it does not resolve a capability, call IEventBus, save an aggregate, or provide an outbox guarantee.

import { createDomainEvents, type IDomainEvent } from '@setu-ts/events-plugin';

class Order {
  readonly events = createDomainEvents();

  place(event: IDomainEvent<{ readonly orderId: string }>): void {
    // Invariants and state mutation happen before the fact is recorded.
    this.events.record(event);
  }
}

const order = new Order();
// Save order first. The application then chooses to dispatch or persist each fact.
for (const event of order.events.pending()) {
  await dispatchAfterSave(event);
  order.events.remove(event);
}

pending() is an insertion-ordered, read-only snapshot rather than the mutable backing collection. remove(event) uses reference equality (===), removes the first matching occurrence, and returns false without changing the recorder if the reference is absent. clear() removes every pending fact. A reconstructed aggregate creates a new recorder and begins with no pending facts.

interface IDomainEvents {
  record<T>(event: IDomainEvent<T>): void;
  pending(): readonly IDomainEvent[];
  remove(event: IDomainEvent): boolean;
  clear(): void;
}

function createDomainEvents(): IDomainEvents;

Subscribing to Events

// In a plugin
app.register({
  name: 'user-event-handlers',
  version: '1.0.0',
  dependencies: ['events'],
  register(ctx) {
    const eventBus = ctx.services.get<IEventBus>('events');

    eventBus.subscribe<UserCreatedEvent>('UserCreated', async (event) => {
      const mailer = ctx.services.get<IMailer>('mail');
      await mailer.send({
        to: event.data.email,
        subject: 'Welcome!',
        body: 'Thank you for joining.',
      });
    });
  },
});

Event Interface

interface IEventBus {
  publish<T>(event: IDomainEvent<T>): Promise<void>;
  publishBatch(events: IDomainEvent[]): Promise<void>; // non-generic: batches are heterogeneous
  subscribe<T>(type: string, handler: EventHandler<T>): Unsubscribe; // returns an Unsubscribe fn
}

EventsPlugin Exports

• EventsPlugin — Plugin factory that configures and registers the in-memory event bus with the given dispatch options (async, errorHandler) and subscribes any handlers supplied.
• EventHandlerRegistration — One { type, handler } pair for the handlers option. Exported because that option is typed as an array of it, so without it a caller could not declare its own array in a variable.
• InMemoryEventBus — In-memory publish/subscribe event bus implementing IEventBus.
• DomainEvent — Base class for domain events, generated by defineDomainEvent.
• IntegrationEvent — Semantic subclass of DomainEvent for integration events (no additional fields).
• defineDomainEvent — Factory that binds DomainEvent and IntegrationEvent to a runtime, returning event IDs and timestamps from the runtime’s uuid and now services.
• IDomainEvents — Aggregate-owned local recorder contract. Its snapshots are ordered and isolated from mutable recorder state; it never dispatches events itself.
• createDomainEvents — Factory for a new empty IDomainEvents recorder, intended for an aggregate field rather than a framework base class.
• IEventHandler — Class-based event handler interface with a handle(event) method.
• subscribeHandler — Function that adapts an IEventHandler instance to the EventHandler signature and subscribes it to the bus; returns an Unsubscribe function.
• EventsDiagnosticsOptions — The plugin’s opt-in event-dispatch observation option (diagnostics), typed as { enabled: true, alias: string, events: Record<string, string> }. The collector behind the bus observes ONLY the exact event types the events map lists, replacing each with its approved alias; payloads, event ids, aggregate ids and unapproved types are never retained. See “Event dispatch observations (M98j)” in the diagnostics section.

Re-exports from @setu-ts/common: IEventBus, IDomainEvent, EventHandler, Unsubscribe.