Event-Driven Architecture Patterns Guide 2025: CQRS, Event Sourcing, Saga Pattern

Event Types
├── Domain Event
│   └── A fact that occurred in the business domain
│       Example: OrderPlaced, PaymentFailed
├── Integration Event
│   └── Events for cross-bounded context communication
│       Example: OrderConfirmedIntegrationEvent
├── Event Notification
│   └── Contains minimal data. Consumer queries if needed
│       Example: OrderStatusChanged (contains only orderId)
└── Event-Carried State Transfer
    └── Contains full state. No additional query needed
        Example: OrderDetails (includes full order info)

// TypeScript - Event Bus Interface
interface EventBus {
  publish<T extends DomainEvent>(event: T): Promise<void>;
  subscribe<T extends DomainEvent>(
    eventType: string,
    handler: EventHandler<T>
  ): void;
  unsubscribe(eventType: string, handler: EventHandler<unknown>): void;
}

interface DomainEvent {
  eventId: string;
  eventType: string;
  aggregateId: string;
  occurredAt: Date;
  version: number;
  metadata: Record<string, string>;
}

interface EventHandler<T extends DomainEvent> {
  handle(event: T): Promise<void>;
}

// Base Domain Event
abstract class BaseDomainEvent implements DomainEvent {
  readonly eventId: string;
  readonly occurredAt: Date;
  readonly version: number;
  abstract readonly eventType: string;
  abstract readonly aggregateId: string;
  metadata: Record<string, string> = {};

  constructor() {
    this.eventId = crypto.randomUUID();
    this.occurredAt = new Date();
    this.version = 1;
  }
}

// Order Placed Event
class OrderPlaced extends BaseDomainEvent {
  readonly eventType = 'order.placed';
  readonly aggregateId: string;

  constructor(
    public readonly orderId: string,
    public readonly customerId: string,
    public readonly items: OrderItem[],
    public readonly totalAmount: Money,
    public readonly shippingAddress: Address
  ) {
    super();
    this.aggregateId = orderId;
  }
}

// Payment Completed Event
class PaymentCompleted extends BaseDomainEvent {
  readonly eventType = 'payment.completed';
  readonly aggregateId: string;

  constructor(
    public readonly paymentId: string,
    public readonly orderId: string,
    public readonly amount: Money,
    public readonly paymentMethod: string
  ) {
    super();
    this.aggregateId = paymentId;
  }
}

// Inventory Reserved Event
class InventoryReserved extends BaseDomainEvent {
  readonly eventType = 'inventory.reserved';
  readonly aggregateId: string;

  constructor(
    public readonly reservationId: string,
    public readonly orderId: string,
    public readonly items: ReservedItem[]
  ) {
    super();
    this.aggregateId = reservationId;
  }
}

// Event Schema (Avro-like)
const OrderPlacedSchema = {
  type: 'record',
  name: 'OrderPlaced',
  namespace: 'com.ecommerce.orders.v1',
  fields: [
    { name: 'eventId', type: 'string' },
    { name: 'eventType', type: 'string' },
    { name: 'aggregateId', type: 'string' },
    { name: 'occurredAt', type: 'string' },  // ISO 8601
    { name: 'version', type: 'int' },
    { name: 'orderId', type: 'string' },
    { name: 'customerId', type: 'string' },
    {
      name: 'items',
      type: {
        type: 'array',
        items: {
          type: 'record',
          name: 'OrderItem',
          fields: [
            { name: 'productId', type: 'string' },
            { name: 'quantity', type: 'int' },
            { name: 'unitPrice', type: 'long' },
          ],
        },
      },
    },
    { name: 'totalAmount', type: 'long' },
    { name: 'currency', type: 'string' },
  ],
};

Traditional Approach (CRUD):
┌─────────────────────────┐
│ orders table            │
│ id: ORD-001             │
│ status: shipped         │ ← Only current state stored
│ total: 50000            │
│ updated_at: 2025-03-24  │
└─────────────────────────┘

Event Sourcing Approach:
┌─────────────────────────────────────────┐
│ event_store                             │
│ 1. OrderPlaced    (ORD-001, 2025-03-24) │
│ 2. PaymentReceived(ORD-001, 2025-03-24) │
│ 3. ItemPacked     (ORD-001, 2025-03-24) │
│ 4. OrderShipped   (ORD-001, 2025-03-24) │ ← Full history preserved
└─────────────────────────────────────────┘

// Event Store Interface
interface EventStore {
  // Append events (optimistic concurrency control)
  append(
    streamId: string,
    events: DomainEvent[],
    expectedVersion: number
  ): Promise<void>;

  // Read events from stream
  readStream(
    streamId: string,
    fromVersion?: number
  ): Promise<DomainEvent[]>;

  // Subscribe to all events
  subscribe(
    handler: (event: DomainEvent) => Promise<void>
  ): void;
}

// PostgreSQL-based Event Store
class PostgresEventStore implements EventStore {
  async append(
    streamId: string,
    events: DomainEvent[],
    expectedVersion: number
  ): Promise<void> {
    const client = await this.pool.connect();
    try {
      await client.query('BEGIN');

      // Optimistic concurrency check
      const result = await client.query(
        'SELECT MAX(version) as current_version FROM events WHERE stream_id = $1',
        [streamId]
      );
      const currentVersion = result.rows[0]?.current_version ?? 0;

      if (currentVersion !== expectedVersion) {
        throw new ConcurrencyError(
          `Expected version ${expectedVersion}, but current is ${currentVersion}`
        );
      }

      // Store events
      let version = expectedVersion;
      for (const event of events) {
        version++;
        await client.query(
          `INSERT INTO events (stream_id, version, event_type, data, metadata, created_at)
           VALUES ($1, $2, $3, $4, $5, NOW())`,
          [streamId, version, event.eventType, JSON.stringify(event), JSON.stringify(event.metadata)]
        );
      }

      await client.query('COMMIT');
    } catch (error) {
      await client.query('ROLLBACK');
      throw error;
    } finally {
      client.release();
    }
  }

  async readStream(streamId: string, fromVersion: number = 0): Promise<DomainEvent[]> {
    const result = await this.pool.query(
      'SELECT data FROM events WHERE stream_id = $1 AND version > $2 ORDER BY version ASC',
      [streamId, fromVersion]
    );
    return result.rows.map(row => JSON.parse(row.data));
  }
}

// Event Sourced Aggregate base class
abstract class EventSourcedAggregate {
  private uncommittedEvents: DomainEvent[] = [];
  version: number = 0;

  protected apply(event: DomainEvent): void {
    this.when(event);
    this.version++;
    this.uncommittedEvents.push(event);
  }

  loadFromHistory(events: DomainEvent[]): void {
    for (const event of events) {
      this.when(event);
      this.version++;
    }
  }

  getUncommittedEvents(): DomainEvent[] {
    return [...this.uncommittedEvents];
  }

  clearUncommittedEvents(): void {
    this.uncommittedEvents = [];
  }

  protected abstract when(event: DomainEvent): void;
}

// Order Aggregate
class Order extends EventSourcedAggregate {
  private id!: string;
  private customerId!: string;
  private items!: OrderItem[];
  private totalAmount!: Money;
  private status!: OrderStatus;

  // Command: Place Order
  static place(orderId: string, customerId: string, items: OrderItem[], total: Money): Order {
    const order = new Order();
    order.apply(new OrderPlaced(orderId, customerId, items, total, new Address()));
    return order;
  }

  // Command: Confirm Payment
  confirmPayment(paymentId: string): void {
    if (this.status !== OrderStatus.PLACED) {
      throw new InvalidOperationError('Can only confirm payment for placed orders');
    }
    this.apply(new OrderPaymentConfirmed(this.id, paymentId));
  }

  // Command: Cancel Order
  cancel(reason: string): void {
    if (this.status === OrderStatus.CANCELLED) {
      throw new InvalidOperationError('Order is already cancelled');
    }
    this.apply(new OrderCancelled(this.id, reason));
  }

  // Apply events to state
  protected when(event: DomainEvent): void {
    if (event instanceof OrderPlaced) {
      this.id = event.orderId;
      this.customerId = event.customerId;
      this.items = event.items;
      this.totalAmount = event.totalAmount;
      this.status = OrderStatus.PLACED;
    } else if (event instanceof OrderPaymentConfirmed) {
      this.status = OrderStatus.CONFIRMED;
    } else if (event instanceof OrderCancelled) {
      this.status = OrderStatus.CANCELLED;
    }
  }
}

class SnapshotStore {
  async saveSnapshot(
    streamId: string,
    snapshot: AggregateSnapshot
  ): Promise<void> {
    await this.pool.query(
      `INSERT INTO snapshots (stream_id, version, data, created_at)
       VALUES ($1, $2, $3, NOW())
       ON CONFLICT (stream_id) DO UPDATE SET version = $2, data = $3, created_at = NOW()`,
      [streamId, snapshot.version, JSON.stringify(snapshot.state)]
    );
  }

  async getLatestSnapshot(streamId: string): Promise<AggregateSnapshot | null> {
    const result = await this.pool.query(
      'SELECT version, data FROM snapshots WHERE stream_id = $1',
      [streamId]
    );
    if (result.rows.length === 0) return null;
    return {
      version: result.rows[0].version,
      state: JSON.parse(result.rows[0].data),
    };
  }
}

// Aggregate loading with snapshots
class OrderRepository {
  async getById(orderId: string): Promise<Order> {
    const order = new Order();

    // 1. Load snapshot
    const snapshot = await this.snapshotStore.getLatestSnapshot(orderId);
    let fromVersion = 0;

    if (snapshot) {
      order.restoreFromSnapshot(snapshot.state);
      fromVersion = snapshot.version;
    }

    // 2. Load only events after snapshot
    const events = await this.eventStore.readStream(orderId, fromVersion);
    order.loadFromHistory(events);

    // 3. Create snapshot after certain number of events
    if (events.length > 50) {
      await this.snapshotStore.saveSnapshot(orderId, {
        version: order.version,
        state: order.toSnapshot(),
      });
    }

    return order;
  }
}

// Projection that builds read model from events
class OrderProjection {
  constructor(private readDb: Pool) {}

  async handle(event: DomainEvent): Promise<void> {
    if (event instanceof OrderPlaced) {
      await this.readDb.query(
        `INSERT INTO order_read_model (id, customer_id, status, total_amount, created_at, version)
         VALUES ($1, $2, 'PLACED', $3, $4, $5)`,
        [event.orderId, event.customerId, event.totalAmount.amount, event.occurredAt, event.version]
      );

      for (const item of event.items) {
        await this.readDb.query(
          `INSERT INTO order_items_read_model (order_id, product_id, quantity, unit_price)
           VALUES ($1, $2, $3, $4)`,
          [event.orderId, item.productId, item.quantity, item.unitPrice]
        );
      }
    } else if (event instanceof OrderPaymentConfirmed) {
      await this.readDb.query(
        `UPDATE order_read_model SET status = 'CONFIRMED', version = $2 WHERE id = $1`,
        [event.orderId, event.version]
      );
    } else if (event instanceof OrderCancelled) {
      await this.readDb.query(
        `UPDATE order_read_model SET status = 'CANCELLED', version = $2 WHERE id = $1`,
        [event.orderId, event.version]
      );
    }
  }
}

┌─────────────────────────────────────────────────┐
│                   Client                        │
│         ┌──────────┐  ┌──────────┐              │
│         │  Write   │  │  Read    │              │
│         │  Request │  │  Request │              │
│         └────┬─────┘  └─────┬────┘              │
└──────────────┼──────────────┼───────────────────┘
               │              │
    ┌──────────▼──────┐  ┌────▼──────────┐
    │  Command Side   │  │  Query Side   │
    │                 │  │               │
    │  Command Handler│  │  Query Handler│
    │       │         │  │       │       │
    │  ┌────▼──────┐  │  │  ┌────▼────┐  │
    │  │ Domain    │  │  │  │ Read DB │  │
    │  │ Model     │  │  │  │ (Denorm)│  │
    │  └────┬──────┘  │  │  └─────────┘  │
    │  ┌────▼──────┐  │  │       ▲       │
    │  │ Write DB  │  │  │       │       │
    │  └────┬──────┘  │  │  Projection   │
    └───────┼─────────┘  └───────┼───────┘
            │                    │
            └─── Events ─────────┘

// Command definition
class PlaceOrderCommand {
  constructor(
    public readonly customerId: string,
    public readonly items: OrderItemDto[],
    public readonly shippingAddress: AddressDto
  ) {}
}

// Command Handler
class PlaceOrderHandler {
  constructor(
    private orderRepo: OrderRepository,
    private eventBus: EventBus
  ) {}

  async handle(command: PlaceOrderCommand): Promise<string> {
    // Business logic
    const orderId = generateOrderId();
    const items = command.items.map(i => new OrderItem(i.productId, i.quantity, i.unitPrice));
    const total = items.reduce((sum, i) => sum + i.quantity * i.unitPrice, 0);

    // Create aggregate and apply events
    const order = Order.place(orderId, command.customerId, items, new Money(total, 'USD'));

    // Save to event store
    await this.orderRepo.save(order);

    // Publish events
    for (const event of order.getUncommittedEvents()) {
      await this.eventBus.publish(event);
    }

    return orderId;
  }
}

// Command Bus
class CommandBus {
  private handlers = new Map<string, CommandHandler<unknown>>();

  register<T extends Command>(commandType: string, handler: CommandHandler<T>): void {
    this.handlers.set(commandType, handler);
  }

  async dispatch<T extends Command>(command: T): Promise<unknown> {
    const handler = this.handlers.get(command.constructor.name);
    if (!handler) {
      throw new Error(`No handler for ${command.constructor.name}`);
    }
    return handler.handle(command);
  }
}

// Query definition
class GetOrderDetailsQuery {
  constructor(public readonly orderId: string) {}
}

class ListCustomerOrdersQuery {
  constructor(
    public readonly customerId: string,
    public readonly page: number = 1,
    public readonly pageSize: number = 20
  ) {}
}

// Query Handler
class GetOrderDetailsHandler {
  constructor(private readDb: Pool) {}

  async handle(query: GetOrderDetailsQuery): Promise<OrderDetailsDto> {
    const result = await this.readDb.query(
      `SELECT o.*, json_agg(i.*) as items
       FROM order_read_model o
       JOIN order_items_read_model i ON o.id = i.order_id
       WHERE o.id = $1
       GROUP BY o.id`,
      [query.orderId]
    );

    if (result.rows.length === 0) {
      throw new NotFoundError(`Order ${query.orderId} not found`);
    }

    return this.mapToDto(result.rows[0]);
  }
}

// Eventual Consistency
// Read model updates after event processing, so slight delays may occur
class OrderQueryService {
  async getOrderWithConsistencyCheck(
    orderId: string,
    expectedVersion?: number
  ): Promise<OrderDetailsDto> {
    const order = await this.getOrderDetails(orderId);

    // Check if specific version is required
    if (expectedVersion && order.version < expectedVersion) {
      // Projection hasn't caught up yet
      // Wait briefly and retry, or query directly from Command side
      await this.waitForProjection(orderId, expectedVersion);
      return this.getOrderDetails(orderId);
    }

    return order;
  }
}

Choreography:
Services collaborate autonomously through events

Order Service ──OrderPlaced──> Payment Service
                               ──PaymentCompleted──> Inventory Service
                                                     ──InventoryReserved──> Shipping Service

Pros: Low coupling, no single point of failure
Cons: Hard to track complex workflows, possible cyclic dependencies

─────────────────────────────────────────────────

Orchestration:
A central orchestrator controls the workflow

                   ┌──────────────────┐
                   │  Order Saga      │
                   │  Orchestrator    │
                   └───────┬──────────┘
                           │
          ┌────────────────┼────────────────┐
          │                │                │
   ┌──────▼──────┐  ┌─────▼───────┐  ┌─────▼───────┐
   │  Payment    │  │  Inventory  │  │  Shipping   │
   │  Service    │  │  Service    │  │  Service    │
   └─────────────┘  └─────────────┘  └─────────────┘

Pros: Easy to track flow, testable
Cons: Orchestrator is single point of failure

// Saga State
enum SagaState {
  STARTED = 'STARTED',
  PAYMENT_PENDING = 'PAYMENT_PENDING',
  INVENTORY_PENDING = 'INVENTORY_PENDING',
  SHIPPING_PENDING = 'SHIPPING_PENDING',
  COMPLETED = 'COMPLETED',
  COMPENSATING = 'COMPENSATING',
  COMPENSATED = 'COMPENSATED',
  FAILED = 'FAILED',
}

// Order Processing Saga
class OrderProcessingSaga {
  private state: SagaState = SagaState.STARTED;
  private orderId!: string;
  private paymentId?: string;
  private reservationId?: string;

  constructor(
    private sagaStore: SagaStore,
    private commandBus: CommandBus
  ) {}

  // Start: when order is placed
  async start(event: OrderPlaced): Promise<void> {
    this.orderId = event.orderId;
    this.state = SagaState.PAYMENT_PENDING;
    await this.sagaStore.save(this);

    // Step 1: Request payment
    await this.commandBus.dispatch(new ProcessPaymentCommand(
      event.orderId, event.totalAmount
    ));
  }

  async handlePaymentCompleted(event: PaymentCompleted): Promise<void> {
    this.paymentId = event.paymentId;
    this.state = SagaState.INVENTORY_PENDING;
    await this.sagaStore.save(this);

    // Step 2: Reserve inventory
    await this.commandBus.dispatch(new ReserveInventoryCommand(
      this.orderId, event.paymentId
    ));
  }

  async handlePaymentFailed(event: PaymentFailed): Promise<void> {
    this.state = SagaState.COMPENSATING;
    await this.sagaStore.save(this);

    // Compensate: Cancel order
    await this.commandBus.dispatch(new CancelOrderCommand(
      this.orderId, 'Payment failed'
    ));

    this.state = SagaState.COMPENSATED;
    await this.sagaStore.save(this);
  }

  async handleInventoryReserved(event: InventoryReserved): Promise<void> {
    this.reservationId = event.reservationId;
    this.state = SagaState.SHIPPING_PENDING;
    await this.sagaStore.save(this);

    // Step 3: Request shipping
    await this.commandBus.dispatch(new CreateShipmentCommand(
      this.orderId, this.reservationId
    ));
  }

  async handleInventoryFailed(event: InventoryReservationFailed): Promise<void> {
    this.state = SagaState.COMPENSATING;
    await this.sagaStore.save(this);

    // Compensate: Refund payment then cancel order
    await this.commandBus.dispatch(new RefundPaymentCommand(
      this.orderId, this.paymentId!
    ));
    await this.commandBus.dispatch(new CancelOrderCommand(
      this.orderId, 'Inventory unavailable'
    ));

    this.state = SagaState.COMPENSATED;
    await this.sagaStore.save(this);
  }

  async handleShipmentCreated(event: ShipmentCreated): Promise<void> {
    this.state = SagaState.COMPLETED;
    await this.sagaStore.save(this);

    // Confirm order
    await this.commandBus.dispatch(new ConfirmOrderCommand(this.orderId));
  }
}

Problem Scenario:
1. Save order to DB ✓
2. Publish event to Kafka ✗  ← If this fails, data inconsistency!

Outbox Pattern:
1. Within DB transaction:
   - Save order ✓
   - Save event to outbox table ✓
2. Separate process reads from outbox and publishes to Kafka

CREATE TABLE outbox (
  id UUID PRIMARY KEY DEFAULT gen_random_uuid(),
  aggregate_type VARCHAR(255) NOT NULL,
  aggregate_id VARCHAR(255) NOT NULL,
  event_type VARCHAR(255) NOT NULL,
  payload JSONB NOT NULL,
  created_at TIMESTAMP NOT NULL DEFAULT NOW(),
  published_at TIMESTAMP,
  retry_count INT DEFAULT 0,
  last_error TEXT
);

CREATE INDEX idx_outbox_unpublished ON outbox (created_at)
  WHERE published_at IS NULL;

class OutboxService {
  async saveWithOutbox(
    order: Order,
    events: DomainEvent[],
    client: PoolClient
  ): Promise<void> {
    // Save order and events in same transaction
    await client.query(
      'INSERT INTO orders (id, customer_id, status, total) VALUES ($1, $2, $3, $4)',
      [order.id, order.customerId, order.status, order.total]
    );

    for (const event of events) {
      await client.query(
        `INSERT INTO outbox (aggregate_type, aggregate_id, event_type, payload)
         VALUES ($1, $2, $3, $4)`,
        ['Order', order.id, event.eventType, JSON.stringify(event)]
      );
    }
  }
}

// Outbox Relay: periodically reads outbox and publishes
class OutboxRelay {
  async processOutbox(): Promise<void> {
    const client = await this.pool.connect();
    try {
      // Read unpublished events
      const result = await client.query(
        `SELECT * FROM outbox
         WHERE published_at IS NULL AND retry_count < 5
         ORDER BY created_at ASC
         LIMIT 100
         FOR UPDATE SKIP LOCKED`
      );

      for (const row of result.rows) {
        try {
          await this.kafka.produce(
            `events.${row.aggregate_type.toLowerCase()}`,
            row.aggregate_id,
            row.payload
          );

          // Mark as published
          await client.query(
            'UPDATE outbox SET published_at = NOW() WHERE id = $1',
            [row.id]
          );
        } catch (error) {
          // Increment failure count
          await client.query(
            'UPDATE outbox SET retry_count = retry_count + 1, last_error = $1 WHERE id = $2',
            [error.message, row.id]
          );
        }
      }
    } finally {
      client.release();
    }
  }
}

# Debezium Connector Configuration
{
  "name": "outbox-connector",
  "config": {
    "connector.class": "io.debezium.connector.postgresql.PostgresConnector",
    "database.hostname": "postgres",
    "database.port": "5432",
    "database.user": "debezium",
    "database.password": "secret",
    "database.dbname": "orderdb",
    "table.include.list": "public.outbox",
    "transforms": "outbox",
    "transforms.outbox.type": "io.debezium.transforms.outbox.EventRouter",
    "transforms.outbox.table.field.event.id": "id",
    "transforms.outbox.table.field.event.key": "aggregate_id",
    "transforms.outbox.table.field.event.type": "event_type",
    "transforms.outbox.table.field.event.payload": "payload",
    "transforms.outbox.route.topic.replacement": "events.orders"
  }
}

class IdempotencyStore {
  async isProcessed(idempotencyKey: string): Promise<boolean> {
    const result = await this.pool.query(
      'SELECT 1 FROM processed_events WHERE idempotency_key = $1',
      [idempotencyKey]
    );
    return result.rows.length > 0;
  }

  async markProcessed(
    idempotencyKey: string,
    result: unknown,
    client: PoolClient
  ): Promise<void> {
    await client.query(
      `INSERT INTO processed_events (idempotency_key, result, processed_at)
       VALUES ($1, $2, NOW())
       ON CONFLICT (idempotency_key) DO NOTHING`,
      [idempotencyKey, JSON.stringify(result)]
    );
  }
}

// Idempotent Event Handler
class IdempotentEventHandler<T extends DomainEvent> {
  constructor(
    private idempotencyStore: IdempotencyStore,
    private innerHandler: EventHandler<T>
  ) {}

  async handle(event: T): Promise<void> {
    const key = `${event.eventType}:${event.eventId}`;

    // Check if event already processed
    if (await this.idempotencyStore.isProcessed(key)) {
      console.log(`Event ${key} already processed, skipping`);
      return;
    }

    const client = await this.pool.connect();
    try {
      await client.query('BEGIN');

      // Process event
      await this.innerHandler.handle(event);

      // Mark as processed
      await this.idempotencyStore.markProcessed(key, null, client);

      await client.query('COMMIT');
    } catch (error) {
      await client.query('ROLLBACK');
      throw error;
    } finally {
      client.release();
    }
  }
}

Strong Consistency:
Writer ──Write──> DB ──Read──> Reader
                  └── Always sees latest ──┘

Eventual Consistency:
Writer ──Write──> Write DB ──Event──> Read DB ──Read──> Reader
                                      └── May see stale data temporarily ──┘

// Strategy 1: Optimistic UI Update
// After write, update UI immediately without waiting for read model
class OrderController {
  async placeOrder(req: Request, res: Response): Promise<void> {
    const orderId = await this.commandBus.dispatch(
      new PlaceOrderCommand(req.body)
    );

    // Return orderId immediately
    // Client uses this to poll or subscribe for updates
    res.json({
      orderId,
      status: 'PROCESSING',
      message: 'Order is being processed'
    });
  }
}

// Strategy 2: Version-based consistency check
class OrderApiController {
  async getOrder(req: Request, res: Response): Promise<void> {
    const minVersion = req.headers['x-min-version']
      ? parseInt(req.headers['x-min-version'] as string)
      : undefined;

    const order = await this.queryService.getOrderWithConsistencyCheck(
      req.params.orderId,
      minVersion
    );

    res.json(order);
  }
}

// Strategy 3: Subscription-based updates
class OrderEventsWebSocket {
  handleConnection(client: WebSocket, orderId: string): void {
    this.eventBus.subscribe('order.*', async (event: DomainEvent) => {
      if (event.aggregateId === orderId) {
        client.send(JSON.stringify({
          type: event.eventType,
          data: event,
        }));
      }
    });
  }
}

// Schema evolution strategy
interface OrderPlacedV1 {
  orderId: string;
  customerId: string;
  totalAmount: number;
}

interface OrderPlacedV2 extends OrderPlacedV1 {
  currency: string;        // New field (default: 'USD')
  shippingMethod: string;  // New field (default: 'STANDARD')
}

// Event upcaster: converts old versions to new versions
class OrderPlacedUpcaster {
  upcast(event: OrderPlacedV1): OrderPlacedV2 {
    return {
      ...event,
      currency: 'USD',
      shippingMethod: 'STANDARD',
    };
  }
}

import { Kafka, Partitioners } from 'kafkajs';

const kafka = new Kafka({
  clientId: 'order-service',
  brokers: ['kafka-1:9092', 'kafka-2:9092', 'kafka-3:9092'],
  retry: { retries: 5 },
});

const producer = kafka.producer({
  createPartitioner: Partitioners.DefaultPartitioner,
  idempotent: true,
  maxInFlightRequests: 5,
  transactionalId: 'order-producer-1',
});

async function publishOrderEvent(event: OrderPlaced): Promise<void> {
  await producer.send({
    topic: 'events.orders',
    messages: [
      {
        key: event.orderId,
        value: JSON.stringify(event),
        headers: {
          'event-type': event.eventType,
          'event-id': event.eventId,
          'correlation-id': event.metadata.correlationId || '',
        },
      },
    ],
  });
}

const consumer = kafka.consumer({
  groupId: 'payment-service-group',
  sessionTimeout: 30000,
  heartbeatInterval: 3000,
});

async function startConsumer(): Promise<void> {
  await consumer.connect();
  await consumer.subscribe({
    topics: ['events.orders'],
    fromBeginning: false,
  });

  await consumer.run({
    eachMessage: async ({ topic, partition, message }) => {
      const event = JSON.parse(message.value!.toString());
      const eventType = message.headers?.['event-type']?.toString();

      console.log(`Received: ${eventType} from ${topic}[${partition}]`);

      try {
        switch (eventType) {
          case 'order.placed':
            await paymentHandler.onOrderPlaced(event);
            break;
          case 'order.cancelled':
            await paymentHandler.onOrderCancelled(event);
            break;
          default:
            console.log(`Unknown event type: ${eventType}`);
        }
      } catch (error) {
        console.error(`Failed to process event: ${error}`);
        // DLQ handling or retry
      }
    },
  });
}

class DeadLetterQueueHandler {
  private dlqProducer: Producer;

  async handleFailedMessage(
    originalTopic: string,
    message: KafkaMessage,
    error: Error
  ): Promise<void> {
    await this.dlqProducer.send({
      topic: `${originalTopic}.dlq`,
      messages: [
        {
          key: message.key,
          value: message.value,
          headers: {
            ...message.headers,
            'dlq-reason': error.message,
            'dlq-timestamp': new Date().toISOString(),
            'original-topic': originalTopic,
            'retry-count': String(
              parseInt(message.headers?.['retry-count']?.toString() || '0') + 1
            ),
          },
        },
      ],
    });
  }
}

Exchange Types:
├── Direct Exchange
│   └── Routes by exact routing key match
│       Use case: Specific service targeting
├── Topic Exchange
│   └── Routes by routing key pattern (*.order.#)
│       Use case: Flexible event routing
├── Fanout Exchange
│   └── Broadcasts to all bound queues
│       Use case: Event notification to all consumers
└── Headers Exchange
    └── Routes by message header matching
        Use case: Complex routing logic

import amqp from 'amqplib';

class RabbitMQEventBus {
  private connection!: amqp.Connection;
  private channel!: amqp.Channel;

  async connect(): Promise<void> {
    this.connection = await amqp.connect('amqp://localhost');
    this.channel = await this.connection.createChannel();

    // Declare exchange
    await this.channel.assertExchange('domain-events', 'topic', {
      durable: true,
    });
  }

  async publish(event: DomainEvent): Promise<void> {
    const routingKey = `events.${event.eventType}`;
    this.channel.publish(
      'domain-events',
      routingKey,
      Buffer.from(JSON.stringify(event)),
      {
        persistent: true,
        messageId: event.eventId,
        contentType: 'application/json',
        headers: {
          'event-type': event.eventType,
          'aggregate-id': event.aggregateId,
        },
      }
    );
  }

  async subscribe(pattern: string, handler: (event: DomainEvent) => Promise<void>): Promise<void> {
    const queue = await this.channel.assertQueue('', { exclusive: true });
    await this.channel.bindQueue(queue.queue, 'domain-events', pattern);

    await this.channel.consume(queue.queue, async (msg) => {
      if (!msg) return;
      try {
        const event = JSON.parse(msg.content.toString());
        await handler(event);
        this.channel.ack(msg);
      } catch (error) {
        // Reject and requeue or send to DLQ
        this.channel.nack(msg, false, false);
      }
    });
  }
}