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, and optionally a correlationId and causationId.

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.

@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(),
		});
	}
}

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. When the append fails, for example because of a version conflict, nothing is published. The event store starts publishing when the application has bootstrapped. Events appended before that are stored, but not published, and a warning is logged once (since 3.0.1).
• 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.
• Not awaited: publishers and subscribers are started while the events are being published, in the order in which the envelopes were appended. appendEvents doesn’t wait for the promises that asynchronous subscribers or publishers 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 (since 3.0.1): 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 failing publisher is logged as well, the remaining publishers and envelopes are still published, and appendEvents still resolves, because the events are already stored.

Since a failure is only logged, 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.

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 appendEvents, 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 and the streams of DynamoDB) 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.getAllEnvelopes() and keeping track of the last envelope it processed. Bear in mind that this is a way to read the events again and not a delivery guarantee.

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.