Use the event bus

The event bus is in-process, backed by a durable outbox. It exists to keep modules decoupled at the code level — iam doesn't import notifications; instead it raises user.created and a listener in notifications reacts. Since #fix-02, publish() persists every event to the event_outbox table before dispatching, so failed handlers are retried instead of silently dropped.

Implementation

apps/server/src/infrastructure/events/event-bus.ts (interface + a lightweight in-memory implementation kept for tests) and apps/server/src/infrastructure/events/outbox-event-bus.ts (the production bus wired in the container):

export interface IEventBus {
  publish(events: ReadonlyArray<DomainEvent>): Promise<void>;
  subscribe(eventName: string, handler: EventHandler): void;
}

export class InMemoryEventBus implements IEventBus { /* ... */ }

OutboxEventBus.publish() inserts every event as a pending event_outbox row, then dispatches inline. A handler that throws does not fail the publishing aggregate — but it is no longer silently swallowed either: the error goes to Sentry via captureError, the row stays pending, and an in-process poller retries it with exponential backoff (30s doubling, capped at 15m). After 8 attempts the row is dead-lettered (status = failed, lastError populated) — query event_outbox to inspect. Delivery is at-least-once per event: a retry re-runs every handler of that event, so handlers must be idempotent (the billing read-model listeners are upserts). Only a failed outbox insert rejects publish() — in the billing webhook path that becomes a 5xx so the provider redelivers.

1. Raise an event from an aggregate

Inside the aggregate's domain method:

this.addDomainEvent({
  name: 'user.created',
  aggregateId: this.id,
  occurredAt: new Date(),
  payload: { email: this.email.value, locale: this.locale },
});

Events accumulate on the aggregate but are not yet published.

2. Publish in the use case

After persisting the aggregate:

async execute(input: RegisterInput): Promise<Result<User, DomainError>> {
  const userResult = User.create(input);
  if (userResult.isErr()) return userResult;

  await this.users.save(userResult.value);
  await this.bus.publish(userResult.value.pullEvents());

  return userResult;
}

pullEvents() returns the accumulated events and clears them off the aggregate, so re-saving never re-publishes.

3. Subscribe a listener

In the receiving module's bootstrap (typically called from bootstrap/container.ts or a dedicated register-listeners.ts):

// apps/server/src/modules/notifications/infrastructure/listeners.ts
export const registerNotificationListeners = (deps: {
  bus: IEventBus;
  jobs: JobScheduler;
}) => {
  deps.bus.subscribe('user.created', async (event) => {
    await deps.jobs.enqueue('emails', {
      kind: 'welcome',
      to: { email: event.payload.email },
      recipientName: event.payload.recipientName,
      appName: 'UseDeploy',
      locale: event.payload.locale,
    });
  });
};

The listener does the minimum synchronous work — typically: enqueue a BullMQ job. Heavy work (sending the email, calling a third party) belongs in the worker.

4. Type your event names

Add the event name to a shared union if you want the type system to enforce that you only subscribe to existing events:

type ApplicationEvent =
  | { name: 'user.created'; payload: { /* ... */ } }
  | { name: 'subscription.activated'; payload: { /* ... */ } };

When to use events vs a direct call

The bus is not a full message queue, but it is durable: events are persisted to event_outbox before dispatch, and a crashed process resumes pending rows on the next poll (30s interval; both the API process and the worker poll, and an optimistic claim keeps concurrent pollers safe). The remaining gap is a crash exactly between the aggregate save() and publish() — that window is microseconds wide and deliberately not closed (it would require threading a unit-of-work through every repository). Heavy work still belongs in BullMQ jobs; for cross-instance fan-out, layer Redis pub/sub on top. One deliberate exception: the invitation accept email does not ride the bus at all — the plaintext token must never be persisted (the outbox stores payloads), so InviteMemberUseCase sends it directly via an injected SendInvitationEmail callback.

Testing

Inject InMemoryEventBus from the production code or build a recording double:

const recorded: DomainEvent[] = [];
const bus: IEventBus = {
  publish: async (events) => { recorded.push(...events); },
  subscribe: () => {},
};

Assert against recorded after the use case runs. If the use case forgets to publish, the test fails.