Packages

@xtaskjs/event-source

Event-sourced aggregates, optimistic stream persistence, stored-event subscribers, and optional queue publication.

npm install @xtaskjs/event-source reflect-metadata Package path: packages/event-source

Overview

What this package owns in the runtime

Event source adds aggregate rehydration, stream persistence, optimistic concurrency checks, and stored-event publication to xtaskjs. It keeps event application inside aggregate roots while exposing repositories, stores, publishers, and subscribers through the same DI lifecycle used by the rest of the framework.

What it provides

EventSourcedAggregate() and ApplyEvent() let aggregates raise and replay domain events with explicit appliers.
EventSourceRepository is registered in the container and supports create(), load(), loadOrCreate(), and save() with optimistic stream appends.
InMemoryEventStore works by default, while createTypeOrmEventStore() adds durable event-stream persistence.
createQueueEventPublisher() and EventSourceSubscriber() connect stored events to queues, projections, and side-effect handlers without replacing CQRS buses.

How it fits

Configured through @EventSource(...) or configureEventSource() before CreateApplication() so store and publisher bindings are ready during bootstrap.
Pairs naturally with @xtaskjs/typeorm for durable event storage and with @xtaskjs/queues for broker-backed event publication.
Works beside @xtaskjs/cqrs when commands load aggregates through repositories and downstream read models react to stored events.

Usage Chart

How strongly this package shapes the runtime

Bootstrap

4/5

Dependency Injection

4/5

Persistence

5/5

Messaging

4/5

Domain Modeling

5/5

Package Flow

How this package moves through xtaskjs runtime phases

Before startup

Configure the package with @EventSource(...) or configureEventSource(), then decorate aggregates and subscribers so metadata exists before the container is scanned.

During CreateApplication()

The lifecycle manager resolves the configured event store and publisher, registers aggregate repositories and tokens in the container, and wires stored-event subscribers from DI-managed services.

During app.close()

Configured stores and publishers are destroyed automatically so in-memory, TypeORM-backed, and queue-backed event infrastructure shuts down with the application.

API Surface

Representative exports from the upstream package

Configuration and publishers

EventSource
configureEventSource
createQueueEventPublisher
QueueEventPublisher
createTypeOrmEventStore
TypeOrmEventStore

Aggregates and repositories

EventSourcedAggregate
ApplyEvent
EventSourcedAggregateRoot
EventSourceRepository
InjectEventSourceRepository

Subscribers and lifecycle

EventSourceSubscriber
StoredEventSubscriber
InjectEventStore
InjectEventSourceBus
InjectEventPublisher
InjectEventSourceLifecycleManager
EventSourceLifecycleManager
initializeEventSourceIntegration
shutdownEventSourceIntegration
getEventSourceLifecycleManager
InMemoryEventStore
EventSourceBus

Usage

Typical adoption flow

1. Configure storage and publication

Start with the default in-memory store for tests and local workflows, or plug in createTypeOrmEventStore() and createQueueEventPublisher() when streams must be durable and publish outside the process.

2. Model aggregates around raised events

Decorate aggregate roots with EventSourcedAggregate(), raise domain events from command methods, and use ApplyEvent() handlers to mutate state during live execution and historical rehydration.

3. Save through repositories and react with subscribers

Inject EventSourceRepository into services or command handlers, save aggregates to append stored events, and attach EventSourceSubscriber() listeners for projections, integrations, or side effects.

Example

Reference snippet

Aggregate registration with durable event storage

import { Service } from "@xtaskjs/core";
import {
  ApplyEvent,
  EventSource,
  EventSourcedAggregate,
  EventSourcedAggregateRoot,
  EventSourceRepository,
  InjectEventSourceRepository,
  createTypeOrmEventStore,
} from "@xtaskjs/event-source";

@EventSource({
  store: createTypeOrmEventStore({ dataSourceName: "write-db", tableName: "event_store" }),
})
class EventSourceConfiguration {}

class UserRegisteredEvent {
  constructor(public readonly id: string, public readonly email: string) {}
}

@EventSourcedAggregate({ stream: "users" })
class UserAggregate extends EventSourcedAggregateRoot {
  public email?: string;

  register(id: string, email: string) {
    this.assignStreamId(id);
    this.raiseEvent(new UserRegisteredEvent(id, email));
  }

  @ApplyEvent(UserRegisteredEvent)
  onRegistered(event: UserRegisteredEvent) {
    this.email = event.email;
  }
}

@Service()
export class UserRegistrationService {
  constructor(
    @InjectEventSourceRepository(UserAggregate)
    private readonly users: EventSourceRepository<UserAggregate>
  ) {}

  async register(id: string, email: string) {
    const user = this.users.create(id);
    user.register(id, email);
    await this.users.save(user);
  }
}

Samples

Official samples to inspect next

Reference samples: 21-event_source_rabbitmq_app and 22-event_source_cqrs_app