Event Observer

The initEventObserver function starts a background event observation loop that continuously listens for new events from EventSourcingDB. Observers automatically reconnect with exponential backoff on failure, making them ideal for building read-side projections and reactive event handlers.

For full details on observing events, including resuming after connection loss and observing from the last event of a given type, see the Observing Events section in the EventSourcingDB documentation.

An in Depth Example

This guide also has an in depth example of a working application built with EventFabric. Combining DDD, CQRS and Event Sourcing.

Check out the In Depth Example page to learn how everything is connected and works out in a real-world application.

Basic Usage

Event observers are typically configured as part of the client setup:

import { setupEventSourcingDBClient } from "@nimbus-cqrs/eventsourcingdb";
import type { Event } from "eventsourcingdb";

await setupEventSourcingDBClient({
    url: new URL(process.env.ESDB_URL ?? ""),
    apiToken: process.env.ESDB_API_TOKEN ?? "",
    eventObservers: [
        {
            subject: "/",
            recursive: true,
            eventHandler: (event: Event) => {
                console.log("Received event:", event);
            },
        },
    ],
});

You can also start an observer independently after the client has been initialized:

import { initEventObserver } from "@nimbus-cqrs/eventsourcingdb";
import type { Event } from "eventsourcingdb";

initEventObserver({
    subject: "/users",
    recursive: true,
    eventHandler: async (event: Event) => {
        console.log("Received event:", event);
    },
});

EventObserver Options

Option	Type	Default	Description
subject	string	(required)	The subject to observe events for
recursive	boolean	false	Whether to observe events recursively for all sub-subjects
lowerBound	Bound	undefined	The starting position for observation
fromLatestEvent	ObserveFromLatestEvent	undefined	Start observation from a specific latest event
eventHandler	(event: Event) => void	(required)	Handler function called for each observed event
retryOptions	RetryOptions	see below	Options for retry behavior on connection failure

Bound

The lowerBound option defines where observation starts:

{
    id: "last-processed-event-id",
    type: "exclusive", // or "inclusive"
}

Property	Type	Description
id	string	The event ID to start from
type	"inclusive"	"exclusive"

ObserveFromLatestEvent

The fromLatestEvent option starts observation from the latest event matching specific criteria:

{
    subject: "/users",
    type: "at.overlap.nimbus.user-invited",
    ifEventIsMissing: "read-everything", // or "wait-for-event"
}

Property	Type	Description
subject	string	The subject to find the latest event for
type	string	The event type to match
ifEventIsMissing	"read-everything"	"wait-for-event"

Retry Options

Option	Type	Default	Description
maxRetries	number	3	Maximum number of retry attempts before giving up
initialRetryDelayMs	number	3000	Initial delay in milliseconds before the first retry

The observer uses exponential backoff with jitter for retries:

• Base delay doubles with each attempt: initialDelayMs * 2^attempt
• Random jitter of 0-30% is added to avoid thundering-herd effects
• After maxRetries consecutive failures, a critical error is logged and the observer stops

Building Projections

A common use case for event observers is building read-side projections. The observer processes events and updates an in-memory or persistent view:

import { Event, getLogger } from "@eventfabric-cqrs/core";
import { eventSourcingDBEventToNimbusEvent } from "@nimbus-cqrs/eventsourcingdb";
import { Event as EventSourcingDBEvent } from "eventsourcingdb";

const USER_INVITED_EVENT_TYPE = "at.overlap.nimbus.user-invited";
const USER_INVITATION_ACCEPTED_EVENT_TYPE =
    "at.overlap.nimbus.user-invitation-accepted";

const usersStore = new Map();

export const projectViews = (eventSourcingDBEvent: EventSourcingDBEvent) => {
    const event =
        eventSourcingDBEventToNimbusEvent<Event>(eventSourcingDBEvent);

    switch (event.type) {
        case USER_INVITED_EVENT_TYPE: {
            usersStore.set(event.data.id, {
                id: event.data.id,
                revision: event.id,
                email: event.data.email,
                firstName: event.data.firstName,
                lastName: event.data.lastName,
                invitedAt: event.data.invitedAt,
                acceptedAt: null,
            });
            break;
        }
        case USER_INVITATION_ACCEPTED_EVENT_TYPE: {
            const id = event.subject.split("/")[2];
            const currentRow = usersStore.get(id);

            usersStore.set(id, {
                ...currentRow,
                revision: event.id,
                acceptedAt: event.data.acceptedAt,
            });
            break;
        }
        default: {
            getLogger().warn({
                category: "ProjectViews",
                message: `Unknown event type ${event.type}`,
            });
            break;
        }
    }
};

Then register this projection handler as an event observer:

await setupEventSourcingDBClient({
    url: new URL(process.env.ESDB_URL ?? ""),
    apiToken: process.env.ESDB_API_TOKEN ?? "",
    eventObservers: [
        {
            subject: "/",
            recursive: true,
            eventHandler: projectViews,
        },
    ],
});

Automatic Position Tracking

The observer automatically tracks its position in the event stream. After each successfully handled event, the lowerBound is updated so that reconnections resume from the last processed event rather than replaying the entire stream.

OpenTelemetry Tracing

Each observed event is processed within an OpenTelemetry span named eventsourcingdb.observeEvent. If the event carries a traceparent (injected by writeEvents), the span is linked to the original writer’s trace, enabling end-to-end distributed tracing across the write and read sides.