@castore/core

Castore

Making Event Sourcing easy 😎

Event Sourcing is a data storage paradigm that saves changes in your application state rather than the state itself.

It is powerful as it enables rewinding to a previous state and exploring audit trails for debugging or business/legal purposes. It also integrates very well with event-driven architectures.

However, it is tricky to implement 😅

After years of using it at Kumo, we have grown to love it, but also experienced first-hand the lack of consensus and tooling around it. That's where Castore comes from!

---

Castore is a TypeScript library that makes Event Sourcing easy 😎

---

With Castore, you'll be able to:

• Define your event stores
• Fetch and push new events seamlessly
• Implement and test your commands
• ...and much more!

All that with first-class developer experience and minimal boilerplate ✨

🫀 Core Design

Some important decisions that we've made early on:

💭 Abstractions first

Castore has been designed with flexibility in mind. It gives you abstractions that are meant to be used anywhere: React apps, containers, Lambdas... you name it!

For instance, EventStore classes are stack agnostic: They need an EventStorageAdapter class to interact with actual data. You can code your own EventStorageAdapter (simply implement the interface), but it's much simpler to use an off-the-shelf adapter like DynamoDBEventStorageAdapter.

🙅‍♂️ We do NOT deploy resources

While some packages like DynamoDBEventStorageAdapter require compatible infrastructure, Castore is not responsible for deploying it.

Though that is not something we exclude in the future, we are a small team and decided to focus on DevX first.

⛑ Full type safety

Speaking of DevX, we absolutely love TypeScript! If you do too, you're in the right place: We push type-safety to the limit in everything we do!

If you don't, that's fine 👍 Castore is still available in Node/JS. And you can still profit from some nice JSDocs!

📖 Best practices

The Event Sourcing journey has many hidden pitfalls. We ran into them for you!

Castore is opiniated. It comes with a collection of best practices and documented anti-patterns that we hope will help you out!

Table of content

• Getting Started
  • 📥 Installation
  • 📦 Packages structure
• The Basics
  • 📚 Events
  • 🏷 Event Types
  • 🏗 Aggregates
  • ⚙️ Reducers
  • 🎁 Event Store
  • 💾 Event Storage Adapter
  • ✍️ Command
  • 📨 Message Buses & Queues
  • 📸 Snapshots
  • 📖 Read Models
• Resources
  • 🎯 Test Tools
  • 🔗 Packages List
  • 📖 Common Patterns

Getting Started

📥 Installation

# npm
npm install @castore/core

# yarn
yarn add @castore/core

📦 Packages structure

Castore is not a single package, but a collection of packages revolving around a core package. This is made so every line of code added to your project is opt-in, wether you use tree-shaking or not.

Castore packages are released together. Though different versions may be compatible, you are guaranteed to have working code as long as you use matching versions.

Here is an example of working package.json:

{
  ...
  "dependencies": {
    "@castore/core": "1.3.1",
    "@castore/dynamodb-event-storage-adapter": "1.3.1"
    ...
  },
  "devDependencies": {
    "@castore/test-tools": "1.3.1"
    ...
  }
}

The Basics

📚 Events

Event Sourcing is all about saving changes in your application state. Such changes are represented by events, and needless to say, they are quite important 🙃

Events that concern the same business entity (like a User) are aggregated through a common id called aggregateId (and vice versa, events that have the same aggregateId represent changes of the same business entity). The index of an event in such a serie of events is called its version.

In Castore, stored events (also called event details) always have exactly the following attributes:

• aggregateId (string)
• version (integer ≥ 1)
• timestamp (string): A date in ISO 8601 format
• type (string): A string identifying the business meaning of the event
• payload (?any = never): A payload of any type
• metadata (?any = never): Some metadata of any type

import type { EventDetail } from '@castore/core';

type UserCreatedEventDetail = EventDetail<
  'USER_CREATED',
  { name: string; age: number },
  { invitedBy?: string }
>;

// 👇 Equivalent to:
type UserCreatedEventDetail = {
  aggregateId: string;
  version: number;
  timestamp: string;
  type: 'USER_CREATED';
  payload: { name: string; age: number };
  metadata: { invitedBy?: string };
};

🏷 EventType

Events are generally classified in events types (not to confuse with TS types). Castore lets you declare them via the EventType class:

import { EventType } from '@castore/core';

const userCreatedEventType = new EventType<
  'USER_CREATED',
  { name: string; age: number },
  { invitedBy?: string }
>({ type: 'USER_CREATED' });

Note that we only provide TS types for payload and metadata properties. That is because, as stated in the core design, Castore is meant to be as flexible as possible, and that includes the validation library you want to use: The EventType class is not meant to be used directly, but rather extended by other classes which will add run-time validation methods to it 👍

See the following packages for examples:

• JSON Schema Event Type
• Zod Event Type

Constructor:

• type (string): The event type

import { EventType } from '@castore/core';

const userCreatedEventType = new EventType({ type: 'USER_CREATED' });

Properties:

• type (string): The event type

const eventType = userCreatedEventType.type;
// => 'USER_CREATED'

Type Helpers:

• EventTypeDetail: Returns the event detail TS type of an EventType

import type { EventTypeDetail } from '@castore/core';

type UserCreatedEventTypeDetail = EventTypeDetail<typeof userCreatedEventType>;

// 👇 Equivalent to:
type UserCreatedEventTypeDetail = {
  aggregateId: string;
  version: number;
  timestamp: string;
  type: 'USER_CREATED';
  payload: { name: string; age: number };
  metadata: { invitedBy?: string };
};

• EventTypesDetails: Return the events details of a list of EventType

import type { EventTypesDetails } from '@castore/core';

type UserEventTypesDetails = EventTypesDetails<
  [typeof userCreatedEventType, typeof userRemovedEventType]
>;
// => EventTypeDetail<typeof userCreatedEventType>
// | EventTypeDetail<typeof userRemovedEventType>

🏗 Aggregate

Eventhough entities are stored as series of events, we still want to use a stable interface to represent their states at a point in time rather than directly using events. In Castore, it is implemented by a TS type called Aggregate.

☝️ Think of aggregates as "what the data would look like in CRUD"

In Castore, aggregates necessarily contain an aggregateId and version attributes (the version of the latest event). But for the rest, it's up to you 🤷‍♂️

For instance, we can include a name, age and status properties to our UserAggregate:

import type { Aggregate } from '@castore/core';

// Represents a User at a point in time
interface UserAggregate extends Aggregate {
  name: string;
  age: number;
  status: 'CREATED' | 'REMOVED';
}

// 👇 Equivalent to:
interface UserAggregate {
  aggregateId: string;
  version: number;
  name: string;
  age: number;
  status: 'CREATED' | 'REMOVED';
}

⚙️ Reducer

Aggregates are derived from their events by reducing them through a reducer function. It defines how to update the aggregate when a new event is pushed:

import type { Reducer } from '@castore/core';

const usersReducer: Reducer<UserAggregate, UserEventsDetails> = (
  userAggregate,
  newEvent,
) => {
  const { version, aggregateId } = newEvent;

  switch (newEvent.type) {
    case 'USER_CREATED': {
      const { name, age } = newEvent.payload;

      // 👇 Return the next version of the aggregate
      return {
        aggregateId,
        version,
        name,
        age,
        status: 'CREATED',
      };
    }
    case 'USER_REMOVED':
      return { ...userAggregate, version, status: 'REMOVED' };
  }
};

const johnDowAggregate: UserAggregate = johnDowEvents.reduce(usersReducer);

☝️ Aggregates are always computed on the fly, and NOT stored. Changing them does not require any data migration whatsoever.

🎁 EventStore

Once you've defined your event types and how to aggregate them, you can bundle them together in an EventStore class.

Each event store in your application represents a business entity. Think of event stores as "what tables would be in CRUD", except that instead of directly updating data, you just append new events to it!

In Castore, EventStore classes are NOT responsible for actually storing data (this will come with event storage adapters). But rather to provide a boilerplate-free and type-safe interface to perform many actions such as:

• Listing aggregate ids
• Accessing events of an aggregate
• Building an aggregate with the reducer
• Pushing new events etc...

import { EventStore } from '@castore/core';

const userEventStore = new EventStore({
  eventStoreId: 'USERS',
  eventTypes: [
    userCreatedEventType,
    userRemovedEventType,
    ...
  ],
  reducer: usersReducer,
});
// ...and that's it 🥳

☝️ The EventStore class is the heart of Castore, it even gave it its name!

Constructor:

• eventStoreId (string): A string identifying the event store
• eventStoreEvents (EventType[]): The list of event types in the event store
• reduce (EventType[]): A reducer function that can be applied to the store event types
• storageAdapter (?EventStorageAdapter): See EventStorageAdapter

☝️ The return type of the reducer is used to infer the Aggregate type of the EventStore, so it is important to type it explicitely.

Properties:

• eventStoreId (string)

const userEventStoreId = userEventStore.eventStoreId;
// => 'USERS'

• eventStoreEvents (EventType[])

const userEventStoreEvents = userEventStore.eventStoreEvents;
// => [userCreatedEventType, userRemovedEventType...]

• reduce ((Aggregate, EventType) => Aggregate)

const reducer = userEventStore.reduce;
// => usersReducer

• storageAdapter ?EventStorageAdapter: See EventStorageAdapter

const storageAdapter = userEventStore.storageAdapter;
// => undefined (we did not provide one in this example)

☝️ The storageAdapter is not read-only so you do not have to provide it right away.

Sync Methods:

• getStorageAdapter (() => EventStorageAdapter): Returns the event store event storage adapter if it exists. Throws an UndefinedStorageAdapterError if it doesn't.

import { UndefinedStorageAdapterError } from '@castore/core';

expect(() => userEventStore.getStorageAdapter()).toThrow(
  new UndefinedStorageAdapterError({ eventStoreId: 'USERS' }),
);
// => true

• buildAggregate ((eventDetails: EventDetail[], initialAggregate?: Aggregate) => Aggregate | undefined): Applies the event store reducer to a serie of events.

const johnDowAggregate = userEventStore.buildAggregate(johnDowEvents);

Async Methods:

The following methods interact with the data layer of your event store through its EventStorageAdapter. They will throw an UndefinedStorageAdapterError if you did not provide one.

• getEvents ((aggregateId: string, opt?: OptionsObj = {}) => Promise<ResponseObj>): Retrieves the events of an aggregate, ordered by version. Returns an empty array if no event is found for this aggregateId.

  OptionsObj contains the following attributes:

  • minVersion (?number): To retrieve events above a certain version
  • maxVersion (?number): To retrieve events below a certain version
  • limit (?number): Maximum number of events to retrieve
  • reverse (?boolean = false): To retrieve events in reverse order (does not require to swap minVersion and maxVersion)

  ResponseObj contains the following attributes:

  • events (EventDetail[]): The aggregate events (possibly empty)

const { events: allEvents } = await userEventStore.getEvents(aggregateId);
// => typed as UserEventDetail[] 🙌

// 👇 Retrieve a range of events
const { events: rangedEvents } = await userEventStore.getEvents(aggregateId, {
  minVersion: 2,
  maxVersion: 5,
});

// 👇 Retrieve the last event of the aggregate
const { events: onlyLastEvent } = await userEventStore.getEvents(aggregateId, {
  reverse: true,
  limit: 1,
});

• getAggregate ((aggregateId: string, opt?: OptionsObj = {}) => Promise<ResponseObj>): Retrieves the events of an aggregate and build it.

  OptionsObj contains the following attributes:

  • maxVersion (?number): To retrieve aggregate below a certain version

  ResponseObj contains the following attributes:

  • aggregate (?Aggregate): The aggregate (possibly undefined)
  • events (EventDetail[]): The aggregate events (possibly empty)
  • lastEvent (?EventDetail): The last event (possibly undefined)

const { aggregate: johnDow } = await userEventStore.getAggregate(aggregateId);
// => typed as UserAggregate | undefined 🙌

// 👇 Retrieve an aggregate below a certain version
const { aggregate: aggregateBelowVersion } = await userEventStore.getAggregate(
  aggregateId,
  { maxVersion: 5 },
);

// 👇 Returns the events if you need them
const { aggregate, events } = await userEventStore.getAggregate(aggregateId);

• getExistingAggregate ((aggregateId: string, opt?: OptionsObj = {}) => Promise<ResponseObj>): Same as getAggregate method, but ensures that the aggregate exists. Throws an AggregateNotFoundError if no event is found for this aggregateId.

import { AggregateNotFoundError } from '@castore/core';

expect(async () =>
  userEventStore.getExistingAggregate(unexistingId),
).resolves.toThrow(
  new AggregateNotFoundError({
    eventStoreId: 'USERS',
    aggregateId: unexistingId,
  }),
);
// true

const { aggregate } = await userEventStore.getAggregate(aggregateId);
// => 'aggregate' and 'lastEvent' are always defined 🙌

• pushEvent ((eventDetail: EventDetail) => Promise<void>): Pushes a new event to the event store. Throws an EventAlreadyExistsError if an event already exists for the corresponding aggregateId and version.

await userEventStore.pushEvent({
  aggregateId,
  version: lastVersion + 1,
  timestamp: new Date().toISOString(),
  type: 'USER_CREATED', // <= event type is correctly typed 🙌
  payload, // <= payload is typed according to the provided event type 🙌
  metadata, // <= same goes for metadata 🙌
});

• listAggregateIds ((opt?: OptionsObj = {}) => Promise<ResponseObj>): Retrieves the list of aggregateId of an event store, ordered by timestamp of their first event. Returns an empty array if no aggregate is found.

  OptionsObj contains the following attributes:

  • limit (?number): Maximum number of aggregate ids to retrieve
  • pageToken (?string): To retrieve a paginated result of aggregate ids

  ResponseObj contains the following attributes:

  • aggregateIds (string[]): The list of aggregate ids
  • nextPageToken (?string): A token for the next page of aggregate ids if one exists

const accAggregateIds: string = [];
const { aggregateIds: firstPage, nextPageToken } =
  await userEventStore.listAggregateIds({ limit: 20 });

accAggregateIds.push(...firstPage);

if (nextPageToken) {
  const { aggregateIds: secondPage } = await userEventStore.listAggregateIds({
    // 👇 Previous limit of 20 is passed through the page token
    pageToken: nextPageToken,
  });
  accAggregateIds.push(...secondPage);
}

Type Helpers:

• EventStoreId: Returns the EventStore id

import type { EventStoreId } from '@castore/core';

type UserEventStoreId = EventStoreId<typeof userEventStore>;
// => 'USERS'

• EventStoreEventsTypes: Returns the EventStore list of events types

import type { EventStoreEventsTypes } from '@castore/core';

type UserEventsTypes = EventStoreEventsTypes<typeof userEventStore>;
// => [typeof userCreatedEventType, typeof userRemovedEventType...]

• EventStoreEventsDetails: Returns the union of all the EventStore possible events details

import type { EventStoreEventsDetails } from '@castore/core';

type UserEventsDetails = EventStoreEventsDetails<typeof userEventStore>;
// => EventTypeDetail<typeof userCreatedEventType>
// | EventTypeDetail<typeof userRemovedEventType>
// | ...

• EventStoreReducer: Returns the EventStore reducer

import type { EventStoreReducer } from '@castore/core';

type UserReducer = EventStoreReducer<typeof userEventStore>;
// => Reducer<UserAggregate, UserEventsDetails>

• EventStoreAggregate: Returns the EventStore aggregate

import type { EventStoreAggregate } from '@castore/core';

type UserReducer = EventStoreAggregate<typeof userEventStore>;
// => UserAggregate

💾 EventStorageAdapter

For the moment, we didn't provide any actual way to store our events data. This is the responsibility of the EventStorageAdapter class.

import { EventStore } from '@castore/core';

const userEventStore = new EventStore({
  eventStoreId: 'USERS',
  eventTypes: userEventTypes,
  reducer: usersReducer,
  // 👇 Provide it in the constructor
  storageAdapter: mySuperStorageAdapter,
});

// 👇 ...or set/switch it in context later
userEventStore.storageAdapter = mySuperStorageAdapter;

You can choose to build an event storage adapter that suits your usage. However, we highly recommend using an off-the-shelf adapter:

• DynamoDB Event Storage Adapter
• In-Memory Event Storage Adapter

If the storage solution that you use is missing, feel free to create/upvote an issue, or contribute!

✍️ Command

Modifying the state of your application (i.e. pushing new events to your event stores) is done by executing commands. They typically consist in:

• Fetching the required aggregates (if not the first event of a new aggregate)
• Validating that the modification is acceptable
• Pushing new events with incremented versions

import { Command } from '@castore/core';

// 👇 You can provide several eventStores if needed
type RequiredEventStores = [typeof userEventStore];
type CommandInput = { name: string; age: number };
type CommandOutput = { userId: string };

const createUserCommand = new Command<
  RequiredEventStores,
  // 👇 The type need to be provided twice for technical reasons
  RequiredEventStores,
  CommandInput,
  CommandOutput
>({
  commandId: 'CREATE_USER',
  requiredEventStores: [userEventStore],
  // 👇 Code to execute
  handler: async (commandInput, [userEventStore]) => {
    const { name, age } = commandInput;
    const userId = generateUuid();

    await userEventStore.pushEvent({
      aggregateId: userId,
      version: 1,
      type: 'USER_CREATED',
      timestamp: new Date().toISOString(),
      payload: { name, age },
    });

    return { userId };
  },
});

Note that we only provide TS types for commandInput and commandOutput properties. That is because, as stated in the core design, Castore is meant to be as flexible as possible, and that includes the validation library you want to use: The Command class is not meant to be used directly, but rather extended by other classes which will add run-time validation methods to it 👍

See the following packages for examples:

• JSON Schema Event Type

Constructor:

• commandId (string): A string identifying the command
• handler ((input: Input, requiredEventsStores: EventStore[]) => Promise<Output>): The code to execute
• requiredEventStores (EventStore[]): A tuple of EventStores that are required by the command for read/write purposes. In TS, you should use the tuple util to preserve tuple ordering in the handler (tuple doesn't mute its input, it simply returns them)

import { Command, tuple } from '@castore/core';

const doSomethingCommand = new Command({
  commandId: 'DO_SOMETHING',
  requiredEventStores: tuple(eventStore1, eventStore2),
  handler: async (commandInput, [eventStore1, eventStore2]) => {
    // ...do something here
  },
});

Properties:

• commandId (string): The command id

const commandId = doSomethingCommand.commandId;
// => 'DO_SOMETHING'

• requiredEventStores (EventStore[]): The required event stores

const requiredEventStores = doSomethingCommand.requiredEventStores;
// => [eventStore1, eventStore2]

• handler ((input: Input, requiredEventsStores: EventStore[]) => Promise<Output>): Function to invoke the command

const output = await doSomethingCommand.handler(input, [
  eventStore1,
  eventStore2,
]);

A few notes on commands handlers:

• Commands handlers should NOT use read models when validating that a modification is acceptable. Read models are like cache: They are not the source of truth, and may not represent the freshest state.

• Fetching and pushing events non-simultaneously exposes your application to race conditions. To counter that, commands are designed to be retried when an EventAlreadyExistsError is triggered (which is part of the EventStorageAdapter interface).

• When writing on several event stores at once, it is important to make sure that all events are written or none, i.e. use transactions: This ensures that the application is not in a corrupt state. Transactions accross event stores cannot be easily abstracted, so check you adapter library on how to achieve this. For instance, the DynamoDBEventStorageAdapter exposes a pushEventsTransaction util.

📨 Message Buses & Queues

As mentioned in the introduction, Event Sourcing integrates very well with event-driven architectures. After having successfully run a command, it can be very useful to push the freshly written events in a Message Bus or a Message Queue system.

There are two kind of messages:

• Notification messages which only carry the events details
• Stateful messages which also carry the corresponding aggregates

Message buses and queues are not implemented in Castore yet, but we have big plans for them, so stay tuned 🙂

📸 Snapshots

As events pile up in your event stores, the performances and costs of your commands can become an issue.

One solution is to periodially persist snapshots of your aggregates (e.g. through a message bus subscription), and only fetch them plus the subsequent events instead of all the events.

Snapshots are not implemented in Castore yet, but we have big plans for them, so stay tuned 🙂

📖 Read Models

Even with snapshots, using the event store for querying needs (like displaying data in a web page) would be slow and inefficient, if not impossible depending on the access pattern.

In Event Sourcing, it is common to use a special type of message bus subscription called projections, responsible for maintaining data specifically designed for querying needs, called read models.

Read models allow for faster read operations and re-indexing. Keep in mind that they are eventually consistent by design, which can be annoying in some use cases (like opening a resource page directly after its creation).

Read models are not implemented in Castore yet, but we have big plans for them, so stay tuned 🙂

Resources

🎯 Test Tools

Castore comes with a handy Test Tool package that facilitates the writing of unit tests: It allows mocking event stores, populating them with an initial state and resetting them to it in a boilerplate-free and type-safe way.

🔗 Packages List

🏷 Event Types

• JSON Schema Event Type: DRY EventType definition using JSON Schemas and json-schema-to-ts
• Zod Event Type: DRY EventType definition using zod

💾 Event Storage Adapters

• DynamoDB Event Storage Adapter: Implementation of the EventStorageAdapter interface based on DynamoDB.
• In-Memory Event Storage Adapter: Implementation of the EventStorageAdapter interface using a local Node/JS object. To be used in manual or unit tests.

📨 Commands

• JSON Schema Command: DRY Command definition using JSON Schemas and json-schema-to-ts

📖 Common Patterns

• Simulating a future/past aggregate state: ...coming soon
• Snapshotting: ...coming soon
• Projecting on read models: ...coming soon
• Replaying events: ...coming soon