Event Sourcing & CQRS

// --- Domain Events ---
interface BaseEvent {
  id: string;
  aggregateId: string;
  type: string;
  data: unknown;
  metadata: {
    userId: string;
    timestamp: string;
    version: number;
    correlationId: string;
  };
}

interface OrderPlaced extends BaseEvent {
  type: "OrderPlaced";
  data: { customerId: string; items: OrderItem[]; totalAmount: number };
}

interface PaymentReceived extends BaseEvent {
  type: "PaymentReceived";
  data: { paymentId: string; amount: number; method: string };
}

interface OrderShipped extends BaseEvent {
  type: "OrderShipped";
  data: { trackingNumber: string; carrier: string; estimatedDelivery: string };
}

interface OrderCancelled extends BaseEvent {
  type: "OrderCancelled";
  data: { reason: string; refundAmount: number };
}

type OrderEvent = OrderPlaced | PaymentReceived | OrderShipped | OrderCancelled;

interface OrderItem {
  productId: string;
  name: string;
  quantity: number;
  price: number;
}

// --- Event Store (append-only) ---
class EventStore {
  private events: BaseEvent[] = [];
  private subscribers: ((event: BaseEvent) => void)[] = [];

  async append(event: BaseEvent): Promise<void> {
    // In production: INSERT INTO events with optimistic concurrency control
    // Check that the expected version matches
    const existing = this.events.filter(
      (e) => e.aggregateId === event.aggregateId
    );

    if (event.metadata.version !== existing.length + 1) {
      throw new Error(
        `Concurrency conflict: expected version ${existing.length + 1}, got ${event.metadata.version}`
      );
    }

    this.events.push(event);

    // Notify subscribers (projections)
    for (const sub of this.subscribers) {
      sub(event);
    }
  }

  async getEvents(aggregateId: string): Promise<BaseEvent[]> {
    return this.events
      .filter((e) => e.aggregateId === aggregateId)
      .sort((a, b) => a.metadata.version - b.metadata.version);
  }

  async getAllEvents(fromVersion?: number): Promise<BaseEvent[]> {
    if (fromVersion) {
      return this.events.slice(fromVersion);
    }
    return [...this.events];
  }

  subscribe(handler: (event: BaseEvent) => void): void {
    this.subscribers.push(handler);
  }
}

// --- Order Aggregate ---
interface OrderState {
  id: string;
  status: "pending" | "paid" | "shipped" | "cancelled";
  customerId: string;
  items: OrderItem[];
  totalAmount: number;
  paymentId?: string;
  trackingNumber?: string;
  version: number;
}

class OrderAggregate {
  private state: OrderState;

  constructor(id: string) {
    this.state = {
      id,
      status: "pending",
      customerId: "",
      items: [],
      totalAmount: 0,
      version: 0,
    };
  }

  // Rebuild state from events
  static fromEvents(events: BaseEvent[]): OrderAggregate {
    if (events.length === 0) throw new Error("No events found");
    const order = new OrderAggregate(events[0].aggregateId);
    for (const event of events) {
      order.apply(event as OrderEvent);
    }
    return order;
  }

  // Apply an event to update state (no side effects)
  private apply(event: OrderEvent): void {
    switch (event.type) {
      case "OrderPlaced":
        this.state.customerId = event.data.customerId;
        this.state.items = event.data.items;
        this.state.totalAmount = event.data.totalAmount;
        this.state.status = "pending";
        break;
      case "PaymentReceived":
        this.state.paymentId = event.data.paymentId;
        this.state.status = "paid";
        break;
      case "OrderShipped":
        this.state.trackingNumber = event.data.trackingNumber;
        this.state.status = "shipped";
        break;
      case "OrderCancelled":
        this.state.status = "cancelled";
        break;
    }
    this.state.version = event.metadata.version;
  }

  getState(): OrderState {
    return { ...this.state };
  }
}

// --- Command Handlers ---
class OrderCommandHandler {
  constructor(private eventStore: EventStore) {}

  async placeOrder(command: {
    orderId: string;
    customerId: string;
    items: OrderItem[];
    userId: string;
    correlationId: string;
  }): Promise<void> {
    const totalAmount = command.items.reduce(
      (sum, item) => sum + item.price * item.quantity, 0
    );

    // Validate
    if (command.items.length === 0) {
      throw new Error("Order must have at least one item");
    }
    if (totalAmount <= 0) {
      throw new Error("Order total must be positive");
    }

    const event: OrderPlaced = {
      id: crypto.randomUUID(),
      aggregateId: command.orderId,
      type: "OrderPlaced",
      data: {
        customerId: command.customerId,
        items: command.items,
        totalAmount,
      },
      metadata: {
        userId: command.userId,
        timestamp: new Date().toISOString(),
        version: 1,
        correlationId: command.correlationId,
      },
    };

    await this.eventStore.append(event);
  }

  async receivePayment(command: {
    orderId: string;
    paymentId: string;
    amount: number;
    method: string;
    userId: string;
    correlationId: string;
  }): Promise<void> {
    // Load current state from events
    const events = await this.eventStore.getEvents(command.orderId);
    const order = OrderAggregate.fromEvents(events);
    const state = order.getState();

    // Business rules
    if (state.status !== "pending") {
      throw new Error(`Cannot pay for order in status: ${state.status}`);
    }
    if (command.amount !== state.totalAmount) {
      throw new Error(
        `Payment amount ${command.amount} doesn't match order total ${state.totalAmount}`
      );
    }

    const event: PaymentReceived = {
      id: crypto.randomUUID(),
      aggregateId: command.orderId,
      type: "PaymentReceived",
      data: {
        paymentId: command.paymentId,
        amount: command.amount,
        method: command.method,
      },
      metadata: {
        userId: command.userId,
        timestamp: new Date().toISOString(),
        version: state.version + 1,
        correlationId: command.correlationId,
      },
    };

    await this.eventStore.append(event);
  }

  async shipOrder(command: {
    orderId: string;
    trackingNumber: string;
    carrier: string;
    estimatedDelivery: string;
    userId: string;
    correlationId: string;
  }): Promise<void> {
    const events = await this.eventStore.getEvents(command.orderId);
    const order = OrderAggregate.fromEvents(events);
    const state = order.getState();

    if (state.status !== "paid") {
      throw new Error(`Cannot ship order in status: ${state.status}`);
    }

    const event: OrderShipped = {
      id: crypto.randomUUID(),
      aggregateId: command.orderId,
      type: "OrderShipped",
      data: {
        trackingNumber: command.trackingNumber,
        carrier: command.carrier,
        estimatedDelivery: command.estimatedDelivery,
      },
      metadata: {
        userId: command.userId,
        timestamp: new Date().toISOString(),
        version: state.version + 1,
        correlationId: command.correlationId,
      },
    };

    await this.eventStore.append(event);
  }
}

// --- Read Model Projection ---
interface OrderView {
  id: string;
  customerId: string;
  status: string;
  totalAmount: number;
  itemCount: number;
  trackingNumber?: string;
  lastUpdated: string;
}

class OrderProjection {
  private orders = new Map<string, OrderView>();

  constructor(eventStore: EventStore) {
    // Subscribe to real-time events
    eventStore.subscribe((event) => this.handleEvent(event as OrderEvent));
  }

  private handleEvent(event: OrderEvent): void {
    switch (event.type) {
      case "OrderPlaced":
        this.orders.set(event.aggregateId, {
          id: event.aggregateId,
          customerId: event.data.customerId,
          status: "pending",
          totalAmount: event.data.totalAmount,
          itemCount: event.data.items.length,
          lastUpdated: event.metadata.timestamp,
        });
        break;
      case "PaymentReceived": {
        const order = this.orders.get(event.aggregateId);
        if (order) {
          order.status = "paid";
          order.lastUpdated = event.metadata.timestamp;
        }
        break;
      }
      case "OrderShipped": {
        const order = this.orders.get(event.aggregateId);
        if (order) {
          order.status = "shipped";
          order.trackingNumber = event.data.trackingNumber;
          order.lastUpdated = event.metadata.timestamp;
        }
        break;
      }
      case "OrderCancelled": {
        const order = this.orders.get(event.aggregateId);
        if (order) {
          order.status = "cancelled";
          order.lastUpdated = event.metadata.timestamp;
        }
        break;
      }
    }
  }

  // Rebuild projection from all events (for recovery or new projections)
  async rebuild(eventStore: EventStore): Promise<void> {
    this.orders.clear();
    const allEvents = await eventStore.getAllEvents();
    for (const event of allEvents) {
      this.handleEvent(event as OrderEvent);
    }
  }

  // Query methods
  getOrder(id: string): OrderView | undefined {
    return this.orders.get(id);
  }

  getOrdersByCustomer(customerId: string): OrderView[] {
    return Array.from(this.orders.values()).filter(
      (o) => o.customerId === customerId
    );
  }

  getOrdersByStatus(status: string): OrderView[] {
    return Array.from(this.orders.values()).filter(
      (o) => o.status === status
    );
  }
}

// --- Demo ---
async function main() {
  const store = new EventStore();
  const commands = new OrderCommandHandler(store);
  const projection = new OrderProjection(store);

  const orderId = "order-001";
  const corrId = crypto.randomUUID();

  // Place order
  await commands.placeOrder({
    orderId,
    customerId: "cust-1",
    items: [
      { productId: "p1", name: "Widget", quantity: 2, price: 29.99 },
      { productId: "p2", name: "Gadget", quantity: 1, price: 49.99 },
    ],
    userId: "admin",
    correlationId: corrId,
  });

  // Receive payment
  await commands.receivePayment({
    orderId,
    paymentId: "pay-001",
    amount: 109.97,
    method: "credit_card",
    userId: "admin",
    correlationId: corrId,
  });

  // Ship order
  await commands.shipOrder({
    orderId,
    trackingNumber: "1Z999AA10123456784",
    carrier: "UPS",
    estimatedDelivery: "2025-01-15",
    userId: "admin",
    correlationId: corrId,
  });

  // Query projection
  console.log("Order view:", projection.getOrder(orderId));

  // View full event history
  const events = await store.getEvents(orderId);
  console.log("Event history:", events.map((e) => e.type));
}

main().catch(console.error);