Vorfall

Type-safe event sourcing with MongoDB and projections

A modern, type-safe event sourcing library for Node.js applications using MongoDB as the event store. Events are stored as CloudEvents, grouped into streams by subject, with inline projections maintained on every append.

Why This Package?

Most event sourcing solutions are either too complex for simple use cases or lack proper TypeScript support. Vorfall bridges this gap by providing:

• Full TypeScript support - End-to-end type safety from events to projections
• MongoDB integration - Leverages MongoDB's document model
• Built-in projections - Create inline read models automatically from events
• CQRS pattern support - Helper functions to handle commands and events with type safety

Requirements

• Node.js >= 20 (the package is ESM-only)
• MongoDB 4.4 or later running as a replica set — appends use multi-document transactions, which are not available on standalone servers, and create a collection inside the transaction on the first append to a new entity

Quick Start

import type { DomainEvent, Subject } from 'vorfall'
import {
  createDomainEvent,
  createEventStore,
  createProjectionDefinition,
  createSubject,
  findOneProjection,
} from 'vorfall'

// Define your domain events
type UserRegistered = DomainEvent<
  'user.registered',
  { userId: string, email: string, name: string },
  undefined,
  Subject<'user'>
>

type UserProfileUpdated = DomainEvent<
  'user.profileUpdated',
  { userId: string, name?: string, email?: string },
  undefined,
  Subject<'user'>
>

type UserEvent = UserRegistered | UserProfileUpdated

interface UserProfile {
  userId: string
  email: string
  name: string
}

// Create a projection for user profiles.
// canHandle is a list of event types; evolve only ever receives those.
// Returning null removes the projection from the stream document; the next
// applicable event starts again from initialState().
const userProfileProjection = createProjectionDefinition({
  name: 'userProfile',
  canHandle: ['user.registered', 'user.profileUpdated'],
  evolve: (state: UserProfile | null, event: UserEvent): UserProfile | null => {
    switch (event.type) {
      case 'user.registered':
        return { ...event.data }
      case 'user.profileUpdated': {
        if (!state)
          return state
        const { userId, ...changes } = event.data
        return { ...state, ...changes }
      }
    }
  },
  initialState: () => null,
})

// Setup event store (synchronous — the connection is established lazily)
const eventStore = createEventStore({
  connectionString: 'mongodb://localhost:27017',
  databaseName: 'myapp',
  projections: [userProfileProjection],
})

// Append events. The stream is derived from the event subject:
// "user/123/registered" belongs to the stream "user/123" in collection "user".
async function registerUser(userId: string, email: string, name: string) {
  const event = createDomainEvent({
    type: 'user.registered' as const,
    subject: createSubject(`user/${userId}/registered`),
    data: { userId, email, name },
  })

  // 'any' deliberately opts out of the concurrency check — see
  // Optimistic Concurrency Control below.
  await eventStore.appendOrCreateStream([event], { expectedVersions: 'any' })
}

// Query the projection maintained for a stream
async function getUserProfile(userId: string) {
  const stream = await findOneProjection(
    eventStore,
    createSubject(`user/${userId}`),
    { projectionName: 'userProfile' },
  )
  return stream?.projections.userProfile ?? null
}

// Or rebuild state directly from the events
async function getUserState(userId: string) {
  const { state, version, streamExists } = await eventStore.aggregateStream<UserProfile | null, UserEvent>(
    createSubject(`user/${userId}`),
    {
      initialState: () => null,
      evolve: (state, event) => userProfileProjection.evolve(state, event),
    },
  )
  return state
}

Optimistic Concurrency Control

Every stream document carries a version (the number of events in the stream). aggregateStream and getEventStreamBySubject return the version seen at read time; pass it back on append to make the write fail if the stream changed in between:

import { ConcurrencyError } from 'vorfall'

const { state, version } = await eventStore.aggregateStream(streamSubject, { evolve, initialState })

try {
  await eventStore.appendOrCreateStream([event], {
    expectedVersions: new Map([[streamSubject, version]]),
  })
}
catch (error) {
  if (error instanceof ConcurrencyError) {
    // Someone else appended first: error.expectedVersion vs. error.actualVersion.
    // Re-read, re-decide, retry — or surface e.g. HTTP 409.
  }
  throw error
}

Expected versions per stream subject can be a number (exact event count, 0 means the stream must not exist yet), 'no-stream' (append must create the stream) or 'any' (no check). On a mismatch the whole append is rolled back, including all other streams in the same call.

Checking is the default: expectedVersions is required, and when it is a map, every stream in the append must be listed — a missing entry throws MissingExpectedVersionError before anything is written. Opting out is always explicit: list a stream with 'any', or pass expectedVersions: 'any' to skip the check for the whole append (e.g. order-insensitive logs or imports where no decision depends on prior state).

handleCommand wires this automatically: the versions observed while aggregating the configured streams are enforced on append, so a concurrent command on the same stream fails with a ConcurrencyError instead of silently interleaving.

The states map handed to the command handler contains one entry per listed stream, with the initial state for streams that do not exist yet. Reading a subject that was not listed throws a StreamNotLoadedError, so a wiring mistake in the caller is not mistaken for a missing stream.

Installation

# Using npm
npm install vorfall

# Using pnpm
pnpm add vorfall

# Using yarn
yarn add vorfall

Core Concepts

• Events - Immutable facts about what happened in your domain, stored as CloudEvents (createDomainEvent)
• Subjects - Hierarchical identifiers in the form entity/id[/...] (e.g. user/123, user/123/registered). The first segment is the MongoDB collection, the first two segments identify the stream (createSubject, createStreamSubject)
• Event Store - One document per stream, holding its events and projection states (createEventStore, appendOrCreateStream, getEventStreamBySubject, aggregateStream)
• Projections - Read models folded from events and persisted on the stream document on every append (createProjectionDefinition, findOneProjection, findMultipleProjections, countProjections)
• Commands - Operations that may produce events (createCommand, handleCommand)

Storage Layout

Each stream is one document in a collection named after its entity, keyed by the stream subject: the document _id of the stream user/123 is the string 'user/123', in the user collection. The document holds the events, the version, timestamps and the projection states. There is no other identifier and no secondary index; the _id index is what makes a subject unique, and a concurrent create of the same subject fails with a ConcurrencyError.

getCollectionBySubject and getCollectionByEntity hand out the stored shape (StoredEventStream), so a direct query filters on _id. toDocument and fromDocument translate between that shape and EventStream, which carries the subject under streamSubject and never an _id.

Aggregates

defineAggregate derives everything one aggregate needs from a single evolve function: the stream subject, the streams entry for handleCommand and the projection definition for createEventStore. An aggregate has no state until its first event and none again after evolve returns null; both the command-side fold and the projection start from null, so the two never disagree.

import { defineAggregate } from 'vorfall'

const user = defineAggregate({
  name: 'user',
  evolve: (state: UserProfile | null, event: UserEvent): UserProfile | null => { /* ... */ },
})

user.subject('123') // 'user/123'
user.stream('123') // { evolve, initialState, streamSubject: 'user/123' } for handleCommand
user.projection // for createEventStore({ projections: [user.projection] })

The aggregate's projection is selected by entity, not by event type: it folds every event of every stream under user/. evolve is therefore the only place that lists the aggregate's event types. An exhaustive switch in it is the completeness check at compile time; at runtime, an event whose type evolve has no case for fails the append with an UnhandledProjectionEventError instead of leaving the projection silently stale. A projection selected by canHandle remains the right tool when it deliberately observes a subset of events, possibly across entities:

const registrations = createProjectionDefinition({
  name: 'registrations',
  canHandle: ['user.registered'],
  evolve: (state: { count: number } | null) => ({ count: (state?.count ?? 0) + 1 }),
  initialState: () => ({ count: 0 }),
})

CQRS Pattern Support

handleCommand aggregates the current state of one or more streams, runs your command handler to produce new events, and appends them:

import { createCommand, handleCommand } from 'vorfall'

const command = createCommand({
  type: 'registerUser',
  data: { userId: '123', email: 'alice@example.com', name: 'Alice' },
})

await handleCommand({
  eventStore,
  command,
  streams: [
    {
      streamSubject: createSubject('user/123'),
      initialState: () => null,
      evolve: (state: UserProfile | null, event: UserEvent) => state, // fold events into state
    },
  ],
  commandHandlerFunction: ({ command, states }) => {
    // Decide based on current state, return the event(s) to append
    return createDomainEvent({
      type: 'user.registered' as const,
      subject: createSubject(`user/${command.data.userId}/registered`),
      data: command.data,
    })
  },
})

A handler that finds the command already satisfied by the current state returns an empty array. handleCommand then appends nothing and resolves with totalEventsAppended: 0, so an idempotent command needs no sentinel error:

function changeEmail({ command, states }) {
  const subject = createSubject(`user/${command.data.userId}`)
  const user = states.get(subject)
  if (user?.email === command.data.email) {
    return [] // nothing changed, nothing to record
  }
  return createDomainEvent({ type: 'user.emailChanged' as const, subject, data: { email: command.data.email } })
}

Development Setup

To contribute to this project:

# Clone the repository
git clone https://github.com/lennartzellmer/vorfall.git
cd vorfall

# Install dependencies
pnpm install

# Run tests
pnpm test

# Run linting
pnpm lint

# Build packages
pnpm build

Running Tests

The project uses Vitest for testing with MongoDB Memory Server for integration tests:

# Run all tests
pnpm test

# Run tests in watch mode
pnpm test:watch

# Run tests for the eventstore package only
pnpm --filter vorfall test

Contributing

We welcome contributions!

1. Fork the repository
2. Create your feature branch (git checkout -b feature/amazing-feature)
3. Commit your changes (git commit -m 'Add some amazing feature')
4. Push to the branch (git push origin feature/amazing-feature)
5. Open a Pull Request

Publishing

Releases are managed with Changesets. Every user-facing change ships with a changeset; pnpm release versions the packages, and the "Publish Package to npmjs" GitHub Actions workflow publishes to npm on a GitHub release (or via manual dispatch).

License

This project is licensed under the MIT License - see the LICENSE file for details.

---

Built with ❤️ using TypeScript, MongoDB, and CloudEvents.