cc-event-sourcing Specification

interface Event<TPayload = unknown, TMetadata extends EventMetadata = EventMetadata> {
  /** Globally unique event identifier (UUIDv7 recommended for sortability) */
  id: string;

  /** Event type in domain.action format (e.g., "user.created", "order.shipped") */
  type: string;

  /** ISO 8601 timestamp when event occurred */
  timestamp: string;

  /** Event-specific data */
  payload: TPayload;

  /** Event metadata */
  metadata: TMetadata;
}

interface EventMetadata {
  /** Aggregate ID this event belongs to */
  aggregateId: string;

  /** Aggregate type (e.g., "User", "Order") */
  aggregateType: string;

  /** Sequence number within the aggregate (starts at 1) */
  version: number;

  /** ID of the command/action that caused this event (for correlation) */
  correlationId?: string;

  /** ID linking related events across aggregates (for distributed tracing) */
  causationId?: string;

  /** User or system that triggered the event */
  actor?: string;

  /** Additional context (tags, source system, etc.) */
  context?: Record<string, unknown>;
}

<domain>.<entity>.<action>

import { uuidv7 } from 'uuidv7';
const eventId = uuidv7();

abstract class Aggregate<TState, TEvent extends Event> {
  /** Unique identifier for this aggregate instance */
  abstract readonly id: string;

  /** Type name for this aggregate */
  abstract readonly type: string;

  /** Current state derived from events */
  protected state: TState;

  /** Current version (number of applied events) */
  private _version: number = 0;

  /** Uncommitted events pending persistence */
  private _uncommittedEvents: TEvent[] = [];

  /** Apply an event to update state (pure function) */
  protected abstract apply(state: TState, event: TEvent): TState;

  /** Get initial state for new aggregates */
  protected abstract getInitialState(): TState;
}

// Example: User aggregate handling a ChangeEmail command
class UserAggregate extends Aggregate<UserState, UserEvent> {
  changeEmail(newEmail: string): void {
    // Validate business rules
    if (!isValidEmail(newEmail)) {
      throw new DomainError('Invalid email format');
    }
    if (this.state.email === newEmail) {
      return; // No change needed
    }

    // Emit event (does not persist yet)
    this.emit({
      type: 'user.email.changed',
      payload: {
        oldEmail: this.state.email,
        newEmail
      }
    });
  }
}

protected apply(state: UserState, event: UserEvent): UserState {
  switch (event.type) {
    case 'user.created':
      return {
        id: event.payload.id,
        email: event.payload.email,
        name: event.payload.name,
        createdAt: event.timestamp,
      };

    case 'user.email.changed':
      return {
        ...state,
        email: event.payload.newEmail,
      };

    default:
      return state;
  }
}

interface EventStore<TEvent extends Event = Event> {
  /**
   * Append events to a stream
   * @throws OptimisticConcurrencyError if expectedVersion doesn't match
   */
  append(
    streamId: string,
    events: TEvent[],
    expectedVersion: number
  ): Promise<void>;

  /**
   * Read all events from a stream
   */
  read(
    streamId: string,
    options?: ReadOptions
  ): AsyncIterable<TEvent>;

  /**
   * Read events across all streams (for projections)
   */
  readAll(options?: ReadAllOptions): AsyncIterable<TEvent>;

  /**
   * Subscribe to new events
   */
  subscribe(
    handler: (event: TEvent) => Promise<void>,
    options?: SubscribeOptions
  ): Subscription;

  /**
   * Get current version of a stream
   */
  getVersion(streamId: string): Promise<number>;
}

<aggregateType>-<aggregateId>

// expectedVersion: -1 = stream must not exist (new aggregate)
// expectedVersion: 0+ = stream must be at this version
await store.append('User-123', [event], expectedVersion);

interface ReadOptions {
  /** Start reading from this version (inclusive) */
  fromVersion?: number;

  /** Stop reading at this version (inclusive) */
  toVersion?: number;

  /** Maximum number of events to read */
  limit?: number;

  /** Read events in reverse order */
  reverse?: boolean;
}

function replay<TState, TEvent extends Event>(
  events: Iterable<TEvent>,
  apply: (state: TState, event: TEvent) => TState,
  initialState: TState
): TState {
  let state = initialState;
  for (const event of events) {
    state = apply(state, event);
  }
  return state;
}

async function replayAsync<TState, TEvent extends Event>(
  events: AsyncIterable<TEvent>,
  apply: (state: TState, event: TEvent) => TState,
  initialState: TState
): Promise<TState> {
  let state = initialState;
  for await (const event of events) {
    state = apply(state, event);
  }
  return state;
}

Events: [E1] → [E2] → [E3] → [SNAPSHOT @ v3] → [E4] → [E5]

Full replay:   E1 → E2 → E3 → E4 → E5  (5 events)
With snapshot: SNAPSHOT → E4 → E5     (2 events)

interface Snapshot<TState> {
  /** Aggregate ID */
  aggregateId: string;

  /** Aggregate type */
  aggregateType: string;

  /** Version this snapshot represents */
  version: number;

  /** Serialized state */
  state: TState;

  /** When snapshot was created */
  createdAt: string;

  /** Schema version for migration support */
  schemaVersion: number;
}

interface SnapshotPolicy {
  /** Create snapshot every N events */
  everyNEvents?: number;

  /** Create snapshot if replay would take > N ms */
  maxReplayTimeMs?: number;

  /** Create snapshot if > N events since last snapshot */
  eventThreshold?: number;

  /** Create snapshot on specific event types */
  triggerEvents?: string[];
}

// Example: Snapshot every 100 events or on 'order.completed'
const policy: SnapshotPolicy = {
  everyNEvents: 100,
  triggerEvents: ['order.completed']
};

interface SnapshotStore<TState> {
  /** Save a snapshot */
  save(snapshot: Snapshot<TState>): Promise<void>;

  /** Load latest snapshot for an aggregate */
  load(
    aggregateType: string,
    aggregateId: string
  ): Promise<Snapshot<TState> | null>;

  /** Delete snapshots older than a version */
  prune(
    aggregateType: string,
    aggregateId: string,
    keepAfterVersion: number
  ): Promise<number>;
}

interface TimeTravel<TState> {
  /** Get state at a specific timestamp */
  stateAt(timestamp: Date): Promise<TState>;

  /** Get state at a specific version */
  stateAtVersion(version: number): Promise<TState>;

  /** Get state changes between two points */
  diff(from: Date, to: Date): Promise<StateDiff<TState>>;

  /** List all events in a time range */
  eventsInRange(from: Date, to: Date): AsyncIterable<Event>;
}

class TimeTravelDebugger<TState, TEvent extends Event> {
  constructor(
    private store: EventStore<TEvent>,
    private apply: ApplyFunction<TState, TEvent>,
    private initialState: TState,
    private snapshotStore?: SnapshotStore<TState>
  ) {}

  async stateAt(streamId: string, timestamp: Date): Promise<TState> {
    // 1. Find nearest snapshot before timestamp
    // 2. Replay events from snapshot to timestamp
    // 3. Return reconstructed state
  }

  async eventsBefore(streamId: string, timestamp: Date): Promise<TEvent[]> {
    const events: TEvent[] = [];
    for await (const event of this.store.read(streamId)) {
      if (new Date(event.timestamp) > timestamp) break;
      events.push(event);
    }
    return events;
  }
}

// What happened to this aggregate?
const events = await store.read('User-123');

// Who changed what when?
for await (const event of events) {
  console.log(`${event.timestamp}: ${event.metadata.actor} - ${event.type}`);
}

// Find all events by a specific user
const userActions = store.readAll({
  filter: { 'metadata.actor': '[email]' }
});

// Upcaster example
function upcast(event: Event): Event {
  if (event.type === 'user.created' && !event.payload.role) {
    return {
      ...event,
      payload: { ...event.payload, role: 'member' }
    };
  }
  return event;
}

// Given-When-Then style for aggregate testing
describe('UserAggregate', () => {
  it('should emit email changed event', () => {
    // Given: An existing user
    const aggregate = replay<UserAggregate>([
      createUserEvent({ id: '123', email: '[email]' })
    ]);

    // When: Email is changed
    aggregate.changeEmail('[email]');

    // Then: EmailChanged event is emitted
    expect(aggregate.uncommittedEvents).toContainEqual(
      expect.objectContaining({
        type: 'user.email.changed',
        payload: { oldEmail: '[email]', newEmail: '[email]' }
      })
    );
  });
});