Snapshots

Snapshots are an optimization that is completely optional. However, they become useful when event streams grow large, and reading them becomes slow. Snapshots allow you to store the state of an aggregate at a particular point in time, so you don’t need to replay all events from the beginning of the stream each time the aggregate is loaded.

To start using snapshots using this library you will need to create a SnapshotRepository.

Creating a Snapshot Repository

A SnapshotRepository is responsible for saving and loading snapshots, which are essentially instances of an aggregate at a certain revision. To create a snapshot repository for a specific aggregate you will need to make a snapshot repository class that extends the base SnapshotRepository class and uses the store behind the scenes to manage snapshots.

The base class provides the following methods:

• save(id: Id, aggregate: A, pool?: ISnapshotPool): Promise<void>: Saves a snapshot of the aggregate when one is due, see when snapshots are taken.
• load(id: Id, pool?: ISnapshotPool): Promise<A>: Loads the latest snapshot of the aggregate or returns a blank aggregate.
• loadMany(ids: Id[], pool?: ISnapshotPool): Promise<A[]>: Loads the latest snapshots of multiple aggregates.
• *loadAll(filter?: { aggregateId?: Id; limit?: number; pool?: string }): AsyncGenerator<SnapshotEnvelope<A>[]>: Reads the latest snapshot of every aggregate, in descending binary order of their ids, as an async iterator of batches. aggregateId is an exclusive cursor: pass the id of the last aggregate of a page to read the next page (see paging).

The two methods you will need to implement are serialize and deserialize.

import { type ISnapshot, Snapshot, SnapshotRepository } from '@ocoda/event-sourcing';

@Snapshot(Account, { name: 'account', interval: 5 })
export class AccountSnapshotRepository extends SnapshotRepository<Account> {
  serialize({ id, ownerIds, balance, openedOn, closedOn }: Account): ISnapshot<Account> {
    return {
      id: id.value,
      ownerIds: ownerIds.map(({ value }) => value),
      balance,
      openedOn: openedOn ? openedOn.toISOString() : undefined,
      closedOn: closedOn ? closedOn.toISOString() : undefined,
    };
  }

  deserialize({ id, ownerIds, balance, openedOn, closedOn }: ISnapshot<Account>): Account {
    const account = new Account();
    account.id = AccountId.from(id);
    account.ownerIds = ownerIds.map(AccountOwnerId.from);
    account.balance = balance;
    account.openedOn = openedOn && new Date(openedOn);
    account.closedOn = closedOn && new Date(closedOn);

    return account;
  }
}

Breakdown of the code

• @Snapshot(Account, { name: 'account', interval: 5 }):

  • The @Snapshot() decorator marks the class as a snapshot handler and specifies the aggregate type and the snapshot name.
  • The interval option specifies how often a snapshot should be taken and defaults to 10. In this case, a snapshot is taken each time the aggregate passes a multiple of 5 versions, see below.

• serialize({ id, ownerIds, balance, openedOn, closedOn }: Account):

  • The serialize method is responsible for converting an aggregate instance into a snapshot object. This method is called when saving a snapshot.

• deserialize({ id, ownerIds, balance, openedOn, closedOn }: ISnapshot<Account>): Account:

  • The deserialize method is responsible for converting a snapshot object into an aggregate instance. This method is called when loading a snapshot.

When are snapshots taken?

Calling save() doesn’t store a snapshot every time. The snapshot repository takes a snapshot when:

• the save crosses a multiple of the interval, or
• it’s the first save of an aggregate, which brings it to version 1 (or further, when the aggregate is created with several events at once).

The interval defaults to 10 when you don’t provide one. The snapshot contains the state of the aggregate at the version it has when it’s saved, not at the multiple that was crossed. With interval: 5, this is what happens:

Save of the aggregate	Snapshot	Why
version 0 → 1	yes	first save of the aggregate
version 0 → 3	yes	first save of the aggregate, created with several events
version 1 → 3	no	no multiple of 5 crossed
version 3 → 5	yes	lands on 5
version 5 → 6	no	no multiple of 5 crossed
version 4 → 6	yes	crosses 5, the snapshot is taken at version 6
version 6 → 12	yes	crosses 10, the snapshot is taken at version 12

To know which versions a save covers, the snapshot repository looks at the events that the aggregate’s last markCommitted() marked as committed. Append the events, call markCommitted(), then pass that same instance to save(), as the aggregate repository below does. A save that committed nothing, such as the save of an unchanged aggregate, takes no snapshot (since 4.0). While the aggregate has uncommitted events, save() logs a warning and takes no snapshot (since 4.0): a snapshot ahead of the stored events would make every later save of the stream conflict. When the committed versions aren’t known, because the aggregate was never marked as committed, the repository falls back to taking a snapshot at version 1 and at exact multiples of the interval.

save() never rejects (since 4.0). It runs after the events were appended, and the aggregate still loads from its events without the snapshot, so when the snapshot can’t be serialized or stored the repository logs the failure instead: a version conflict, which means that the stream already has that snapshot or a newer one, as a warning, and any other failure as an error. 3.x rejected, which failed a command whose events were already stored.

Plugging it into your aggregate repository

If you created a snapshot repository and registered it as a provider, you can now plug it into your aggregate repository to optimize the loading process.

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

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

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

    const account = await this.accountSnapshotRepository.load(accountId);

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

    await account.loadFromHistory(events);

    if (account.version < 1) {
      throw new AccountNotFoundException(accountId.value);
    }

    return account;
  }

  async getByIds(accountIds: AccountId[]) {
    const accounts = await this.accountSnapshotRepository.loadMany(accountIds);

    for (const account of accounts) {
      const eventStream = EventStream.for<Account>(Account, account.id);
      const eventCursor = this.eventStore.getEvents(eventStream, { fromVersion: account.version + 1 });
      await account.loadFromHistory(eventCursor);
    }

    return accounts;
  }

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

    // Append the events to the event store
    await this.eventStore.appendEvents(stream, events, { expectedVersion: account.committedVersion });
    account.markCommitted(events);
    // Save a snapshot of the account, if one is due
    await this.accountSnapshotRepository.save(account.id, account);
  }
}

By default the events and snapshots are stored in the default pool. If you separate the data of your tenants, pass the name of the pool to the event store and the snapshot repository, e.g. this.accountSnapshotRepository.load(accountId, 'tenant-1') and this.eventStore.getEvents(eventStream, { pool: 'tenant-1', fromVersion: account.version + 1 }). See Multitenancy for more information.