Repositories

Aggregate repositories serve as a bridge between your aggregates and the event store. Their responsibility is to retrieve aggregates from the event store by rehydrating them from a stream of events and persisting new events that represent changes in the aggregate’s state. The repository provides an interface for interacting with the underlying event store without exposing the complexity of event sourcing to the rest of your application.

Defining an Aggregate Repository

An aggregate repository typically involves two operations:

1. Retrieving an aggregate by loading its historical events and rehydrating it.
2. Persisting changes by appending new events to the event stream.

Here is an example of an AccountRepository that handles the Account aggregate:

import { Injectable } from '@nestjs/common';
import { EventStore, EventStream } from '@ocoda/event-sourcing';
import { Account, AccountId } from '../../domain/models';

@Injectable()
export class AccountRepository {
  constructor(private readonly eventStore: EventStore) {}

  async getById(accountId: AccountId): Promise<Account> {
    const eventStream = EventStream.for<Account>(Account, accountId);

    const account = new Account();

    const eventCursor = this.eventStore.getEvents(eventStream, { fromVersion: account.version + 1 });

    await account.loadFromHistory(eventCursor);

    return account;
  }

  async save(account: Account): Promise<void> {
    const events = account.getUncommittedEvents();
    const stream = EventStream.for<Account>(Account, account.id);

    await this.eventStore.appendEvents(stream, events, { expectedVersion: account.committedVersion });
    account.markCommitted(events);
  }
}

Breakdown of the code

• Constructor injection: the EventStore is injected into the repository through the constructor, enabling interaction with the underlying event sourcing persistence layer.
• getById(accountId: AccountId):
  • retrieves an Account aggregate from the event store.
  • the EventStream.for<Account>(Account, accountId) is used to define the event stream specific to the Account aggregate with the accountId.
  • the aggregate is then rehydrated by loading its history using account.loadFromHistory(eventCursor), where the eventCursor fetches events starting from the aggregate’s current version.
• save(account: Account):
  • account.getUncommittedEvents() returns the events the aggregate applied since it was loaded or last saved.
  • this.eventStore.appendEvents(stream, events, { expectedVersion: account.committedVersion }) appends them. committedVersion is the version the stream had when the aggregate was loaded, so the append fails with a version conflict when another writer appended in the meantime.
  • account.markCommitted(events) runs only once the append succeeded: it marks the appended events as committed, so committedVersion catches up. When the append fails, the events stay uncommitted and save() can be called again. Passing events keeps an event that was applied to the aggregate while the append ran uncommitted, for the next save; markCommitted() without arguments would count it as committed without storing it.
  • saving an aggregate without uncommitted events appends nothing.

Registering the Repository

To use the AccountRepository in your application, you need to register it as a provider in your NestJS module:

import { Module } from '@nestjs/common';

@Module({ providers: [AccountRepository] })
export class AccountModule {}

Guidelines for working with Repositories

1. Repositories Should Be Focused on Aggregates

   Each repository should handle only one type of aggregate. In this example, AccountRepository handles only the Account aggregate. This keeps the repository focused and aligned with DDD principles.

2. Event Store as the Source of Truth

   The event store serves as the source of truth for your aggregates. Repositories should always reconstruct aggregates by replaying events from the event store, which ensures that your domain state is always derived from the historical sequence of events.

3. Use Event Streams for Versioning

   Event streams ensure that changes to aggregates are versioned properly. When saving an aggregate, pass the version the stream is expected to have, account.committedVersion, as the expectedVersion of the append. This prevents concurrency issues, ensuring that multiple users don’t overwrite each other’s changes when working with the same aggregate.

   account.version is account.committedVersion plus the number of uncommitted events. 3.x repositories passed account.version, the version after the append, in the 3.x form appendEvents(stream, account.version, events, pool), which is deprecated since 4.0.

   When another writer got there first, appendEvents rejects with an EventStoreVersionConflictException and stores nothing. Match it on its code rather than its message, then reload the aggregate and retry, or report the conflict:

   import { EventSourcingErrorCode, isEventSourcingError } from '@ocoda/event-sourcing';
   
   try {
     await this.accountRepository.save(account);
   } catch (error) {
     if (isEventSourcingError(error, EventSourcingErrorCode.EventStoreVersionConflict)) {
       // error is narrowed to EventStoreVersionConflictException:
       // error.streamId, error.expectedVersion and error.actualVersion describe the conflict
     }
     throw error;
   }

   Every exception class the library exports has a stable code from EventSourcingErrorCode and the fields that describe it; the original error, if any, is its cause (since 4.0).

4. Lazy Loading of Events

   While fetching aggregates, events should be loaded lazily or in a paginated fashion, especially for aggregates with long histories. This example uses eventStore.getEvents(eventStream, { fromVersion: account.version + 1 }) to only fetch events after the aggregate’s current version, optimizing performance and ensuring only relevant events are processed. When aggregates start containing a lot of events and this becomes a performance bottleneck, you can consider using snapshots to optimize the loading process.

5. Mark Events as Committed After the Append

   Save the aggregate after all business logic has been executed, so the events represent the result of valid domain operations, and call markCommitted(events) only once the append succeeded. The deprecated commit() returned the events and cleared them in one call, before the append: when the append failed, they were lost.