Building an Order and Payment State Machine with TypeScript and KurrentDB Using Event Sourcing and Point-in-Time Reconstruction

-- Ordinary CRUD: only the current state remains
UPDATE orders SET status = 'PAID', updated_at = NOW() WHERE id = '123';

[
  { "type": "OrderCreated",     "orderId": "123", "createdAt": "2026-07-15T09:00:00Z" },
  { "type": "PaymentProcessed", "orderId": "123", "amount": 59000, "processedAt": "2026-07-15T09:01:23Z" },
  { "type": "OrderShipped",     "orderId": "123", "trackingNumber": "KR123456", "shippedAt": "2026-07-15T14:30:00Z" }
]

docker run -d --name kurrentdb \
  -p 2113:2113 \
  -p 2114:2114 \
  --env KURRENTDB_INSECURE=true \
  kurrent/kurrentdb:latest

npm install @kurrent/kurrentdb-client @nestjs/cqrs zod drizzle-orm

// src/order/events/order.events.ts
import { z } from 'zod';
 
export interface EventMetadata {
  correlationId: string; // ID that groups the same business flow together
  causationId:   string; // ID of the event that caused this event
  userId?:       string;
}
 
export const OrderCreatedSchema = z.object({
  orderId:     z.string().uuid(),
  customerId:  z.string().uuid(),
  items: z.array(z.object({
    productId: z.string(),
    quantity:  z.number().int().positive(),
    price:     z.number().positive(),
  })),
  totalAmount: z.number().positive(),
  createdAt:   z.string().datetime(),
});
 
export const PaymentProcessedSchema = z.object({
  orderId:     z.string().uuid(),
  paymentId:   z.string().uuid(),
  amount:      z.number().positive(),
  method:      z.enum(['CARD', 'BANK_TRANSFER', 'VIRTUAL_ACCOUNT']),
  processedAt: z.string().datetime(),
});
 
export const OrderShippedSchema = z.object({
  orderId:        z.string().uuid(),
  trackingNumber: z.string(),
  carrier:        z.string(),
  shippedAt:      z.string().datetime(),
});
 
export type OrderCreated     = z.infer<typeof OrderCreatedSchema>;
export type PaymentProcessed = z.infer<typeof PaymentProcessedSchema>;
export type OrderShipped     = z.infer<typeof OrderShippedSchema>;
 
export type OrderEvent =
  | { type: 'OrderCreated';     data: OrderCreated;     metadata: EventMetadata }
  | { type: 'PaymentProcessed'; data: PaymentProcessed; metadata: EventMetadata }
  | { type: 'OrderShipped';     data: OrderShipped;     metadata: EventMetadata };

// src/order/aggregates/order.aggregate.ts
import { OrderEvent, EventMetadata } from '../events/order.events';
 
export type OrderStatus = 'CREATED' | 'PAYMENT_PROCESSED' | 'SHIPPED' | 'CANCELLED';
 
export interface OrderItem {
  productId: string;
  quantity:  number;
  price:     number;
}
 
export interface OrderState {
  orderId:         string;
  customerId:      string;
  status:          OrderStatus;
  totalAmount:     number;
  items:           OrderItem[];
  paymentId?:      string;
  trackingNumber?: string;
  version:         number;
}
 
export class OrderAggregate {
  private state: Partial<OrderState> = { version: 0 };
  private uncommittedEvents: OrderEvent[] = [];
 
  static reconstitute(events: OrderEvent[]): OrderAggregate {
    const aggregate = new OrderAggregate();
    for (const event of events) aggregate.apply(event);
    return aggregate;
  }
 
  static fromSnapshot(state: OrderState): OrderAggregate {
    const aggregate = new OrderAggregate();
    aggregate.state = { ...state };
    return aggregate;
  }
 
  replayEvents(events: OrderEvent[]): void {
    for (const event of events) this.apply(event);
  }
 
  createOrder(
    orderId:     string,
    customerId:  string,
    items:       OrderItem[],
    totalAmount: number,
    metadata:    EventMetadata,
  ): void {
    if (this.state.orderId) throw new Error('Order already exists');
    this.raise({
      type: 'OrderCreated',
      data: { orderId, customerId, items, totalAmount, createdAt: new Date().toISOString() },
      metadata,
    });
  }
 
  processPayment(
    paymentId: string,
    amount:    number,
    method:    'CARD' | 'BANK_TRANSFER' | 'VIRTUAL_ACCOUNT',
    metadata:  EventMetadata,
  ): void {
    if (this.state.status !== 'CREATED') {
      throw new Error(`Cannot process payment: current status is ${this.state.status}`);
    }
    this.raise({
      type: 'PaymentProcessed',
      data: {
        orderId:     this.state.orderId!,
        paymentId,
        amount,
        method,
        processedAt: new Date().toISOString(),
      },
      metadata,
    });
  }
 
  private raise(event: OrderEvent): void {
    this.apply(event);
    this.uncommittedEvents.push(event);
  }
 
  private apply(event: OrderEvent): void {
    const nextVersion = (this.state.version ?? 0) + 1;
    switch (event.type) {
      case 'OrderCreated':
        this.state = {
          ...this.state,
          orderId:     event.data.orderId,
          customerId:  event.data.customerId,
          totalAmount: event.data.totalAmount,
          items:       event.data.items,
          status:      'CREATED',
          version:     nextVersion,
        };
        break;
      case 'PaymentProcessed':
        this.state = {
          ...this.state,
          status:    'PAYMENT_PROCESSED',
          paymentId: event.data.paymentId,
          version:   nextVersion,
        };
        break;
      case 'OrderShipped':
        this.state = {
          ...this.state,
          status:         'SHIPPED',
          trackingNumber: event.data.trackingNumber,
          version:        nextVersion,
        };
        break;
    }
  }
 
  getState(): Readonly<Partial<OrderState>> { return this.state; }
  getUncommittedEvents(): OrderEvent[]       { return [...this.uncommittedEvents]; }
  clearUncommittedEvents(): void             { this.uncommittedEvents = []; }
}

// src/order/repositories/event.repository.ts
import { Injectable } from '@nestjs/common';
import {
  KurrentDBClient,
  jsonEvent,
  FORWARDS,
  START,
  NO_STREAM,
  StreamNotFoundError,
} from '@kurrent/kurrentdb-client';
import { OrderEvent } from '../events/order.events';
 
export interface StoredEvent {
  event:      OrderEvent;
  recordedAt: Date;   // Timestamp recorded by the KurrentDB server (event.created)
  revision:   bigint;
}
 
@Injectable()
export class OrderEventRepository {
  constructor(private readonly client: KurrentDBClient) {}
 
  async append(
    orderId:          string,
    events:           OrderEvent[],
    expectedRevision: typeof NO_STREAM | bigint,
  ): Promise<void> {
    const streamName = `order-${orderId}`;
    const toWrite = events.map(e =>
      jsonEvent({ type: e.type, data: e.data, metadata: e.metadata })
    );
    await this.client.appendToStream(streamName, toWrite, { expectedRevision });
  }
 
  async loadEvents(orderId: string): Promise<StoredEvent[]> {
    return this.readFrom(orderId, START);
  }
 
  async loadEventsAfterVersion(orderId: string, afterVersion: number): Promise<StoredEvent[]> {
    return this.readFrom(orderId, BigInt(afterVersion + 1));
  }
 
  private async readFrom(
    orderId:      string,
    fromRevision: typeof START | bigint,
  ): Promise<StoredEvent[]> {
    const streamName = `order-${orderId}`;
    const results: StoredEvent[] = [];
 
    try {
      const stream = this.client.readStream(streamName, {
        direction: FORWARDS,
        fromRevision,
      });
 
      for await (const { event } of stream) {
        if (!event) continue;
        results.push({
          event: {
            type:     event.type as OrderEvent['type'],
            data:     event.data as OrderEvent['data'],
            metadata: event.metadata as OrderEvent['metadata'],
          },
          recordedAt: event.created,
          revision:   event.revision,
        });
      }
    } catch (err) {
      if (err instanceof StreamNotFoundError) return [];
      throw err;
    }
 
    return results;
  }
}

// src/order/queries/get-order-at-time.query.ts
import { IQueryHandler, QueryHandler } from '@nestjs/cqrs';
import { OrderEventRepository } from '../repositories/event.repository';
import { OrderAggregate } from '../aggregates/order.aggregate';
 
export class GetOrderAtTimeQuery {
  constructor(
    public readonly orderId:   string,
    public readonly timestamp: Date,
  ) {}
}
 
@QueryHandler(GetOrderAtTimeQuery)
export class GetOrderAtTimeQueryHandler
  implements IQueryHandler<GetOrderAtTimeQuery>
{
  constructor(private readonly repo: OrderEventRepository) {}
 
  async execute(query: GetOrderAtTimeQuery) {
    const stored = await this.repo.loadEvents(query.orderId);
 
    const events = stored
      .filter(({ recordedAt }) => recordedAt <= query.timestamp)
      .map(({ event }) => event);
 
    return OrderAggregate.reconstitute(events).getState();
  }
}

// Example usage: querying the order state at 2:32 PM yesterday
const state = await queryBus.execute(
  new GetOrderAtTimeQuery(
    'order-uuid-here',
    new Date('2026-07-14T14:32:00Z'),
  ),
);
// → { status: 'PAYMENT_PROCESSED', paymentId: 'pay-xxx', totalAmount: 59000, ... }

// src/order/projections/order-summary.projection.ts
import { Injectable, OnModuleInit } from '@nestjs/common';
import { KurrentDBClient, START } from '@kurrent/kurrentdb-client';
import { eq } from 'drizzle-orm';
import { DrizzleService } from '../db/drizzle.service';
import { orderSummaries, projectionCheckpoints } from '../db/schema';
import { OrderCreated, PaymentProcessed, OrderShipped } from '../events/order.events';
 
const PROJECTION_ID = 'order-summary';
 
@Injectable()
export class OrderSummaryProjection implements OnModuleInit {
  constructor(
    private readonly client: KurrentDBClient,
    private readonly db:     DrizzleService,
  ) {}
 
  onModuleInit() {
    this.startProjection().catch(console.error);
  }
 
  private async startProjection() {
    const savedPosition = await this.loadCheckpoint();
 
    // Stream name prefix filter to exclude system events — see current client docs for exact filter API
    const subscription = this.client.subscribeToAll({
      fromPosition: savedPosition ?? START,
      filter: { streamName: { prefix: ['order-'] } },
    });
 
    for await (const resolved of subscription) {
      const { event, commitPosition } = resolved;
      if (!event) continue;
 
      switch (event.type) {
        case 'OrderCreated': {
          const data = event.data as OrderCreated;
          await this.db
            .insert(orderSummaries)
            .values({
              orderId:     data.orderId,
              customerId:  data.customerId,
              status:      'CREATED',
              totalAmount: data.totalAmount,
              createdAt:   data.createdAt,
            })
            .onConflictDoNothing(); // Skip already-processed events
          break;
        }
        case 'PaymentProcessed': {
          const data = event.data as PaymentProcessed;
          await this.db
            .update(orderSummaries)
            .set({ status: 'PAYMENT_PROCESSED', paymentId: data.paymentId })
            .where(eq(orderSummaries.orderId, data.orderId));
          break;
        }
        case 'OrderShipped': {
          const data = event.data as OrderShipped;
          await this.db
            .update(orderSummaries)
            .set({ status: 'SHIPPED', trackingNumber: data.trackingNumber })
            .where(eq(orderSummaries.orderId, data.orderId));
          break;
        }
      }
 
      if (commitPosition !== undefined) {
        await this.saveCheckpoint(commitPosition);
      }
    }
  }
 
  private async loadCheckpoint(): Promise<{ commit: bigint; prepare: bigint } | undefined> {
    const row = await this.db.query.projectionCheckpoints.findFirst({
      where: eq(projectionCheckpoints.projectionId, PROJECTION_ID),
    });
    if (!row) return undefined;
    return { commit: BigInt(row.commitPosition), prepare: BigInt(row.preparePosition) };
  }
 
  private async saveCheckpoint(position: { commit: bigint; prepare: bigint }): Promise<void> {
    await this.db
      .insert(projectionCheckpoints)
      .values({
        projectionId:    PROJECTION_ID,
        commitPosition:  position.commit.toString(),
        preparePosition: position.prepare.toString(),
      })
      .onConflictDoUpdate({
        target: projectionCheckpoints.projectionId,
        set: {
          commitPosition:  position.commit.toString(),
          preparePosition: position.prepare.toString(),
        },
      });
  }
}

// src/order/repositories/snapshot.repository.ts
import { Injectable } from '@nestjs/common';
import { OrderState } from '../aggregates/order.aggregate';
 
export interface Snapshot {
  streamName: string;
  version:    number;
  state:      string; // JSON.stringify(OrderState)
  savedAt:    string;
}
 
@Injectable()
export class SnapshotRepository {
  // Replace with a persistent store such as PostgreSQL or Redis in production
  private readonly store = new Map<string, Snapshot>();
 
  async save(snapshot: Snapshot): Promise<void> {
    this.store.set(snapshot.streamName, snapshot);
  }
 
  async findLatest(streamName: string): Promise<Snapshot | null> {
    return this.store.get(streamName) ?? null;
  }
}

// src/order/services/aggregate-loader.service.ts
import { Injectable } from '@nestjs/common';
import { OrderAggregate, OrderState } from '../aggregates/order.aggregate';
import { OrderEventRepository } from '../repositories/event.repository';
import { SnapshotRepository } from '../repositories/snapshot.repository';
 
const SNAPSHOT_THRESHOLD = 100;
 
@Injectable()
export class AggregateLoaderService {
  constructor(
    private readonly eventRepo:    OrderEventRepository,
    private readonly snapshotRepo: SnapshotRepository,
  ) {}
 
  async load(orderId: string): Promise<OrderAggregate> {
    const streamName = `order-${orderId}`;
    const snapshot = await this.snapshotRepo.findLatest(streamName);
 
    if (snapshot) {
      const state = JSON.parse(snapshot.state) as OrderState;
      const aggregate = OrderAggregate.fromSnapshot(state);
      const remaining = await this.eventRepo.loadEventsAfterVersion(orderId, snapshot.version);
      aggregate.replayEvents(remaining.map(s => s.event));
      return aggregate;
    }
 
    const stored = await this.eventRepo.loadEvents(orderId);
    const aggregate = OrderAggregate.reconstitute(stored.map(s => s.event));
 
    const currentVersion = aggregate.getState().version ?? 0;
    if (currentVersion > 0 && currentVersion % SNAPSHOT_THRESHOLD === 0) {
      await this.snapshotRepo.save({
        streamName,
        version: currentVersion,
        state:   JSON.stringify(aggregate.getState()),
        savedAt: new Date().toISOString(),
      });
    }
 
    return aggregate;
  }
}