Event Publishers and Subscribers

Whenever the EventStore appends events, the produced EventEnvelopes get published on the EventBus. You can hook into that stream in two ways:

an event subscriber reacts to specific events inside your application, e.g. to update a read model or to send an email.
an event publisher receives every envelope, e.g. to push it to Redis, SNS or Kafka.

Both receive the EventEnvelope, not the event instance. Next to the serialized payload of the event, the envelope carries the name of the event (event) and its metadata: the eventId, the aggregateId, the version of the aggregate, when the event occurredOn, its globalPosition in the pool, and optionally a correlationId, a causationId and headers (see Envelopes).

Event subscribers

Create an event subscriber by decorating a class with @EventSubscriber(), passing the events it’s interested in, and by implementing the IEventSubscriber interface. Its handle method is called for every published envelope of those events.

import { type EventEnvelope, EventSubscriber, type IEventSubscriber } from '@ocoda/event-sourcing';

@EventSubscriber(AccountOpenedEvent)
export class AccountOpenedEventSubscriber implements IEventSubscriber {
  constructor(private readonly accountProjection: AccountProjection) {}

  async handle({ payload, metadata }: EventEnvelope<AccountOpenedEvent>): Promise<void> {
    await this.accountProjection.add(payload.accountId, metadata.occurredOn);
  }
}

A subscriber can listen to multiple events, e.g. @EventSubscriber(AccountOpenedEvent, AccountClosedEvent). The name of the event that an envelope carries is available in envelope.event.

Register the subscriber as a provider in one of your modules, like any other injectable class, and its dependencies get injected. You don’t need to register it anywhere else: subscribers are discovered when the application bootstraps, see Module configuration.

Event publishers

A default publisher takes care of publishing the envelopes inside your application, which is what makes the subscribers above work. To push your envelopes to an external system, create an additional publisher by decorating a class with @EventPublisher() and implementing the IEventPublisher interface. Register it as a provider, and it’s used next to the default publisher, it doesn’t replace it.

When publish returns a promise, the bus awaits it before it passes the publisher the next envelope, and appendEvents resolves once every publisher is done (see delivery semantics). Whatever else publish returns is ignored.

@EventPublisher()
export class CustomEventPublisher implements IEventPublisher {
  // Your own client for the system you publish to, e.g. a Kafka producer or an SNS client
  constructor(private readonly broker: MessageBroker) {}

  async publish(envelope: EventEnvelope): Promise<void> {
    await this.broker.send('account-events', {
      event: envelope.event,
      payload: envelope.payload,
      eventId: envelope.metadata.eventId.value,
      aggregateId: envelope.metadata.aggregateId,
      version: envelope.metadata.version,
      occurredOn: envelope.metadata.occurredOn.toISOString(),
    });
  }
}

A publisher that sends a whole append at once, for example as one batch or one transaction of your broker, implements publishAll(envelopes) as well. The bus then calls publishAll once per append, with the envelopes in commit order, instead of calling publish for each of them:

@EventPublisher()
export class BatchEventPublisher implements IEventPublisher {
  constructor(private readonly broker: MessageBroker) {}

  async publish(envelope: EventEnvelope): Promise<void> {
    await this.publishAll([envelope]);
  }

  async publishAll(envelopes: readonly EventEnvelope[]): Promise<void> {
    await this.broker.sendBatch('account-events', envelopes.map((envelope) => envelope.toJSON()));
  }
}

Delivery semantics

Knowing how events are delivered matters, especially when you publish them to other systems.

In-process: the envelopes are published in the process that appended the events. Subscribers in other instances of your application don’t receive them, and neither do consumers outside your application unless you add an event publisher that sends them there.
After the events are persisted: the event store publishes the envelopes once appendEvents has successfully stored the events, through the EventBus the module hands it when it creates the store. When the append fails, for example because of a version conflict, nothing is published, and appendEvents(stream, events, { expectedVersion, publish: false }) stores without publishing. The publishers and subscribers are registered in the module’s onModuleInit, or earlier if a provider of another module appends from its own onModuleInit first, so they receive every event appended from a lifecycle hook on, possibly before their own onModuleInit has run (connect a publisher in an async provider factory, or buffer until it is connected); appending from a provider factory fails with an EventSourcingNotReadyException (see Module configuration). (3.x registered them only when the application had bootstrapped, didn’t publish the events appended before that and logged a warning; 4.0 no longer sets the bus on the store with eventStore.publish = ….)
At-most-once: there’s no outbox, no acknowledgement, no retry and no checkpoint. When the process stops between storing and publishing the events, or when a subscriber or publisher fails, that envelope is not delivered again. A subscriber that wasn’t running when an event was published never receives it afterwards.
Publishers are awaited, in order: each publisher gets the envelopes of an append in commit order, one call at a time: the bus awaits the promise that publish returns before it passes the next envelope. The publishers run concurrently, and appendEvents resolves once every publisher is done, so a slow publisher slows down your commands. Each publisher also gets the appends of a stream one after the other, in the order in which they were stored, even when they run concurrently: while a publisher still handles an append, the next append to the same stream waits for it. That holds back that publisher only, for that stream only, and the commands that wait for it; appends to other streams don’t wait. For one order across all streams, read the events with readAll.
Publisher timeout: a call to publish or publishAll may take at most publishing.publisherTimeout milliseconds, 30 s by default. After that the bus reports a timeout and moves on to the next envelope; set it to 0 to wait as long as a publisher takes. An append waits up to that timeout for each call to each publisher, and for the calls of earlier appends to the same stream that the publisher still handles.
Publishers that publish: a publisher may append events, or call eventBus.publish(), from inside its publish. It gets those envelopes at once, instead of after the call that waits for them. The other publishers and the subscribers get them as usual. Two publishers that each append to the same stream from inside their own call, and await it, wait for each other until their calls time out, or for good with a publisherTimeout of 0. The subscribers are fed after the other publishers, so what a subscriber appends to a stream in reaction to an event reaches the publishers after that event.
Subscribers are not awaited: they are started while the envelopes are being published, in the order in which the envelopes were appended, and run in parallel. appendEvents doesn’t wait for the promises they return, so they can still be running when it resolves. They can overlap and finish in any order, so make them idempotent and don’t depend on the order in which they complete.
Failures are isolated: an error thrown by a subscriber, or a promise it rejects, is logged (logger context EventBus, with the name of the event and the subscriber) and doesn’t affect anything else. The subscriber keeps receiving later events and the other subscribers still get theirs. A publisher that fails or times out is logged as well, it still gets the later envelopes, the other publishers aren’t affected, and appendEvents still resolves, because the events are already stored. Every failure is also reported on deliveryErrors$.

Since a failure is only logged and reported, handle errors that matter inside your subscriber or publisher, for example by retrying or by storing what failed. If you can’t afford to lose an envelope, use the outbox pattern below.

Delivery errors

EventBus.deliveryErrors$ is an observable of every publisher and subscriber failure, one per envelope, for example to count them in your metrics or to keep the envelopes for a retry:

eventBus.deliveryErrors$.subscribe(({ kind, handler, envelope, error }) => {
  // kind: 'publisher' | 'subscriber'; handler: the class name of the publisher or subscriber
  metrics.increment('event_delivery_errors', { kind, handler, event: envelope.event });
});

error is what the publisher or subscriber threw or rejected with. A publisher that ran into its timeout reports a DOMException named TimeoutError, like AbortSignal.timeout() does. A publishAll call that fails reports each of its envelopes.

Waiting for the bus, and shutting down

await eventBus.whenIdle() resolves once no publisher or subscriber is running, which is at once when none is. In tests, use it instead of waiting for a fixed time:

await commandBus.execute(new OpenAccountCommand(ownerId));
await eventBus.whenIdle();
expect(await projection.get(ownerId)).toBeDefined();

whenIdle({ timeout }) rejects with a DOMException named TimeoutError when the bus is still busy after timeout milliseconds; without a timeout, or with 0, it waits as long as it takes. A publisher call that ran into publishing.publisherTimeout no longer counts as running, even when the publisher is still busy with it. Don’t await it inside a publisher or a subscriber: it would wait for itself.

When the application shuts down (app.close(), or a signal with enableShutdownHooks()), the bus waits in beforeApplicationShutdown until the publishers and subscribers that are still running are done, for at most publishing.shutdownTimeout milliseconds (10 s by default, 0 waits as long as it takes; after the timeout it logs a warning). Then, in onApplicationShutdown, it unsubscribes the subscribers and the event and snapshot stores disconnect. NestJS runs onModuleDestroy before that, so a subscriber that is still running can find a provider of yours that has already cleaned up in onModuleDestroy: call await eventBus.whenIdle() before app.close() if yours do.

EventSourcingModule.forRoot({
  events: [AccountOpenedEvent],
  publishing: { publisherTimeout: 5_000, shutdownTimeout: 30_000 },
});

Outbox pattern

The library doesn’t come with an outbox. Publishing happens after the events are stored, and the two aren’t part of one transaction: the process can stop in between, and the envelope is lost for your subscribers and publishers. If you need every event to reach an external system, you can add that guarantee yourself:

A custom event store that writes an outbox. Implement your own event store, or extend an existing one and override persistEvents, which every store implements, so that the events and the outbox rows are inserted in the same database transaction. A separate process reads the unsent rows, sends them to your broker and marks them as sent.
Change data capture (CDC). The events table or collection is append-only, so a CDC tool such as Debezium (or the change streams of MongoDB) can forward every new row to your broker, without your application being involved in the publishing.
Pull instead of push. A consumer can catch up, or rebuild a read model, by reading the stored envelopes with eventStore.readAll({ fromPosition }) and storing the globalPosition of the last envelope it processed as its checkpoint. See Reading all events. On a store whose global order is 'gap-safe', a consumer that resumes after its checkpoint misses no event; one that stops before it saved its checkpoint reads the last events again.

Both the outbox and CDC deliver at-least-once. Consumers can receive an event more than once, so make them idempotent, for example by remembering the eventId of the envelopes they processed.