Claude Code for Event-Driven Architecture: Event Sourcing, CQRS, and Message Brokers

Event-driven architecture solves coupling problems that direct service calls create — services communicate through events they publish and subscribe to, with no direct dependencies. The complexity moves into event schema management, ordering guarantees, and eventual consistency. Claude Code generates the event sourcing aggregates, CQRS read models, and reliable delivery patterns that make event-driven systems production-worthy.

Event Sourcing

Implement event sourcing for our order management.
Each state change is an event. Current state is derived from event history.
Support replaying events to rebuild state.

// domain/events.ts — immutable event types
export type OrderEvent =
  | { type: 'OrderPlaced'; orderId: string; customerId: string; items: OrderItem[]; timestamp: Date }
  | { type: 'PaymentCaptured'; orderId: string; amountCents: number; paymentId: string; timestamp: Date }
  | { type: 'OrderShipped'; orderId: string; trackingNumber: string; carrier: string; timestamp: Date }
  | { type: 'OrderCancelled'; orderId: string; reason: string; refundAmountCents: number; timestamp: Date };

// domain/Order.ts — aggregate that rebuilds from events
export class Order {
  readonly orderId: string;
  readonly customerId: string;
  status: 'pending' | 'paid' | 'shipped' | 'cancelled' = 'pending';
  items: OrderItem[] = [];
  trackingNumber?: string;
  version = 0;

  // Uncommitted events to be persisted
  readonly uncommittedEvents: OrderEvent[] = [];

  private constructor(orderId: string, customerId: string) {
    this.orderId = orderId;
    this.customerId = customerId;
  }

  // Factory — reconstruct from event history
  static rehydrate(events: OrderEvent[]): Order {
    if (events.length === 0) throw new Error('Cannot rehydrate from empty events');

    const first = events[0];
    if (first.type !== 'OrderPlaced') throw new Error('First event must be OrderPlaced');

    const order = new Order(first.orderId, first.customerId);
    for (const event of events) {
      order.apply(event);
    }
    return order;
  }

  // Business command — validates then records event
  capturePayment(amountCents: number, paymentId: string): void {
    if (this.status !== 'pending') {
      throw new Error(`Cannot capture payment for order in status: ${this.status}`);
    }

    const event: OrderEvent = {
      type: 'PaymentCaptured',
      orderId: this.orderId,
      amountCents,
      paymentId,
      timestamp: new Date(),
    };

    this.apply(event);
    this.uncommittedEvents.push(event);
  }

  ship(trackingNumber: string, carrier: string): void {
    if (this.status !== 'paid') {
      throw new Error(`Cannot ship order in status: ${this.status}`);
    }

    const event: OrderEvent = {
      type: 'OrderShipped',
      orderId: this.orderId,
      trackingNumber,
      carrier,
      timestamp: new Date(),
    };

    this.apply(event);
    this.uncommittedEvents.push(event);
  }

  // Apply event to update in-memory state — deterministic, no side effects
  private apply(event: OrderEvent): void {
    switch (event.type) {
      case 'OrderPlaced':
        this.items = event.items;
        this.status = 'pending';
        break;
      case 'PaymentCaptured':
        this.status = 'paid';
        break;
      case 'OrderShipped':
        this.status = 'shipped';
        this.trackingNumber = event.trackingNumber;
        break;
      case 'OrderCancelled':
        this.status = 'cancelled';
        break;
    }
    this.version++;
  }
}

// infrastructure/OrderRepository.ts
import { EventStore } from './EventStore';

export class OrderRepository {
  constructor(private eventStore: EventStore) {}

  async load(orderId: string): Promise<Order> {
    const events = await this.eventStore.getEvents(`order-${orderId}`);
    return Order.rehydrate(events);
  }

  async save(order: Order): Promise<void> {
    if (order.uncommittedEvents.length === 0) return;

    await this.eventStore.appendEvents(
      `order-${order.orderId}`,
      order.uncommittedEvents,
      order.version - order.uncommittedEvents.length, // Optimistic concurrency
    );

    order.uncommittedEvents.length = 0; // Clear after successful save
  }
}

CQRS Read Models

The order list query is too complex to derive from event history in real time.
Build a read model that subscribes to order events and maintains a
denormalized orders_summary table for fast queries.

// read-models/OrderSummaryProjection.ts
import { EventBus } from '../infrastructure/EventBus';
import { db } from '../infrastructure/db';

export class OrderSummaryProjection {
  constructor(private eventBus: EventBus) {}

  start() {
    this.eventBus.subscribe('OrderPlaced', this.onOrderPlaced.bind(this));
    this.eventBus.subscribe('PaymentCaptured', this.onPaymentCaptured.bind(this));
    this.eventBus.subscribe('OrderShipped', this.onOrderShipped.bind(this));
    this.eventBus.subscribe('OrderCancelled', this.onOrderCancelled.bind(this));
  }

  private async onOrderPlaced(event: Extract<OrderEvent, { type: 'OrderPlaced' }>) {
    const totalCents = event.items.reduce((sum, item) => sum + item.priceCents * item.quantity, 0);

    await db('orders_summary').insert({
      order_id: event.orderId,
      customer_id: event.customerId,
      status: 'pending',
      item_count: event.items.length,
      total_cents: totalCents,
      placed_at: event.timestamp,
    });
  }

  private async onPaymentCaptured(event: Extract<OrderEvent, { type: 'PaymentCaptured' }>) {
    await db('orders_summary')
      .where('order_id', event.orderId)
      .update({ status: 'paid', paid_at: event.timestamp });
  }

  private async onOrderShipped(event: Extract<OrderEvent, { type: 'OrderShipped' }>) {
    await db('orders_summary')
      .where('order_id', event.orderId)
      .update({
        status: 'shipped',
        tracking_number: event.trackingNumber,
        shipped_at: event.timestamp,
      });
  }

  private async onOrderCancelled(event: Extract<OrderEvent, { type: 'OrderCancelled' }>) {
    await db('orders_summary')
      .where('order_id', event.orderId)
      .update({ status: 'cancelled', cancelled_at: event.timestamp });
  }
}

// The read model enables fast queries without touching event history:
// SELECT * FROM orders_summary WHERE customer_id = ? ORDER BY placed_at DESC LIMIT 20;

Outbox Pattern for Reliable Event Publishing

When we save an order, we need to publish an event to Kafka.
But if Kafka is down, we lose the event. Fix this.

// The outbox pattern: save events to DB in same transaction as business data
// A separate process reads and publishes them to Kafka

// In the order service transaction:
async function placeOrder(customerId: string, items: OrderItem[]) {
  await db.transaction(async (trx) => {
    // Business data
    const orderId = randomUUID();
    await trx('orders').insert({ id: orderId, customer_id: customerId, status: 'pending' });

    // Event in outbox — same transaction, guaranteed atomicity
    await trx('outbox_events').insert({
      id: randomUUID(),
      aggregate_type: 'Order',
      aggregate_id: orderId,
      event_type: 'OrderPlaced',
      payload: JSON.stringify({ orderId, customerId, items }),
      created_at: new Date(),
      published_at: null, // Will be set when Kafka publish succeeds
    });
  });
  // If transaction commits, the outbox event is guaranteed to exist
  // If Kafka publish fails, the outbox relay will retry
}

// Outbox relay — separate process, runs every second
async function relayOutboxEvents() {
  const unpublished = await db('outbox_events')
    .whereNull('published_at')
    .orderBy('created_at', 'asc')
    .limit(100)
    .forUpdate()             // Lock selected rows to prevent duplicate processing
    .skipLocked();           // Skip rows locked by other relay instances

  for (const event of unpublished) {
    try {
      await kafka.send({
        topic: `${event.aggregate_type.toLowerCase()}.events`,
        messages: [{ key: event.aggregate_id, value: event.payload }],
      });

      await db('outbox_events').where('id', event.id).update({ published_at: new Date() });
    } catch (error) {
      console.error(`Failed to publish event ${event.id}:`, error);
      // Will retry on next cycle
    }
  }
}

Saga Orchestration

We have a multi-step order process: reserve inventory → charge payment → ship.
If payment fails, release the inventory reservation.
Use a saga to coordinate this.

// sagas/PlaceOrderSaga.ts
export class PlaceOrderSaga {
  private state: 'started' | 'inventory-reserved' | 'payment-captured' | 'shipped' | 'compensating' | 'failed' = 'started';

  constructor(
    private readonly orderId: string,
    private readonly inventoryService: InventoryService,
    private readonly paymentService: PaymentService,
    private readonly shippingService: ShippingService,
  ) {}

  async execute(): Promise<void> {
    let reservationId: string | undefined;

    try {
      // Step 1: Reserve inventory
      reservationId = await this.inventoryService.reserve(this.orderId);
      this.state = 'inventory-reserved';

      // Step 2: Capture payment
      await this.paymentService.capture(this.orderId);
      this.state = 'payment-captured';

      // Step 3: Create shipment
      await this.shippingService.createShipment(this.orderId);
      this.state = 'shipped';

    } catch (error) {
      this.state = 'compensating';

      // Compensate in reverse order
      if (this.state === 'payment-captured') {
        await this.paymentService.refund(this.orderId).catch(e =>
          console.error('Refund failed — manual intervention required:', e)
        );
      }

      if (reservationId) {
        await this.inventoryService.release(reservationId).catch(e =>
          console.error('Release reservation failed:', e)
        );
      }

      this.state = 'failed';
      throw error;
    }
  }
}

For the Kafka setup and message broker configuration used for event publishing, see the Kafka guide. For the microservices architecture that event-driven patterns decompose, see the microservices guide. The Claude Skills 360 bundle includes event-driven architecture skill sets for event sourcing, CQRS, and saga patterns. Start with the free tier to try event-driven scaffolding.