Event-Driven Architecture

Implement event-driven systems with event sourcing, CQRS, and message queues. This skill covers distributed patterns for scalable, resilient applications.

Purpose

Build loosely coupled, scalable systems:

• Implement event sourcing for audit trails
• Apply CQRS for read/write optimization
• Use message queues for async processing
• Handle distributed transactions with sagas
• Ensure eventual consistency
• Build replay and recovery capabilities

Features

1. Event Sourcing

// Event definitions
interface DomainEvent {
  eventId: string;
  eventType: string;
  aggregateId: string;
  aggregateType: string;
  timestamp: Date;
  version: number;
  data: Record<string, any>;
  metadata: {
    userId?: string;
    correlationId?: string;
    causationId?: string;
  };
}

// Order aggregate events
type OrderEvent =
  | { type: 'OrderCreated'; data: { customerId: string; items: OrderItem[] } }
  | { type: 'OrderItemAdded'; data: { item: OrderItem } }
  | { type: 'OrderItemRemoved'; data: { itemId: string } }
  | { type: 'OrderSubmitted'; data: { submittedAt: Date } }
  | { type: 'PaymentReceived'; data: { paymentId: string; amount: number } }
  | { type: 'OrderShipped'; data: { trackingNumber: string; carrier: string } }
  | { type: 'OrderDelivered'; data: { deliveredAt: Date } }
  | { type: 'OrderCancelled'; data: { reason: string } };

// Event store
class EventStore {
  async append(
    aggregateId: string,
    events: DomainEvent[],
    expectedVersion: number
  ): Promise<void> {
    // Optimistic concurrency check
    const currentVersion = await this.getVersion(aggregateId);

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

    // Append events atomically
    await db.$transaction(async (tx) => {
      for (let i = 0; i < events.length; i++) {
        await tx.event.create({
          data: {
            ...events[i],
            version: expectedVersion + i + 1,
          },
        });
      }
    });

    // Publish to event bus
    for (const event of events) {
      await eventBus.publish(event);
    }
  }

  async getEvents(
    aggregateId: string,
    fromVersion?: number
  ): Promise<DomainEvent[]> {
    return db.event.findMany({
      where: {
        aggregateId,
        version: fromVersion ? { gt: fromVersion } : undefined,
      },
      orderBy: { version: 'asc' },
    });
  }

  async getVersion(aggregateId: string): Promise<number> {
    const lastEvent = await db.event.findFirst({
      where: { aggregateId },
      orderBy: { version: 'desc' },
    });

    return lastEvent?.version ?? 0;
  }
}

// Aggregate with event sourcing
class OrderAggregate {
  private id: string;
  private state: OrderState;
  private version: number = 0;
  private uncommittedEvents: OrderEvent[] = [];

  static async load(eventStore: EventStore, id: string): Promise<OrderAggregate> {
    const aggregate = new OrderAggregate(id);
    const events = await eventStore.getEvents(id);

    for (const event of events) {
      aggregate.apply(event, false);
    }

    return aggregate;
  }

  // Command handlers
  create(customerId: string, items: OrderItem[]): void {
    if (this.state) {
      throw new Error('Order already exists');
    }

    this.applyChange({
      type: 'OrderCreated',
      data: { customerId, items },
    });
  }

  addItem(item: OrderItem): void {
    this.ensureState(['draft']);

    this.applyChange({
      type: 'OrderItemAdded',
      data: { item },
    });
  }

  submit(): void {
    this.ensureState(['draft']);

    if (this.state.items.length === 0) {
      throw new Error('Cannot submit empty order');
    }

    this.applyChange({
      type: 'OrderSubmitted',
      data: { submittedAt: new Date() },
    });
  }

  // Event application
  private apply(event: OrderEvent, isNew: boolean): void {
    switch (event.type) {
      case 'OrderCreated':
        this.state = {
          status: 'draft',
          customerId: event.data.customerId,
          items: event.data.items,
          total: this.calculateTotal(event.data.items),
        };
        break;

      case 'OrderItemAdded':
        this.state.items.push(event.data.item);
        this.state.total = this.calculateTotal(this.state.items);
        break;

      case 'OrderSubmitted':
        this.state.status = 'submitted';
        this.state.submittedAt = event.data.submittedAt;
        break;

      // ... other event handlers
    }

    this.version++;

    if (isNew) {
      this.uncommittedEvents.push(event);
    }
  }

  private applyChange(event: OrderEvent): void {
    this.apply(event, true);
  }

  async save(eventStore: EventStore): Promise<void> {
    const domainEvents = this.uncommittedEvents.map((e, i) => ({
      eventId: uuid(),
      eventType: e.type,
      aggregateId: this.id,
      aggregateType: 'Order',
      timestamp: new Date(),
      version: this.version - this.uncommittedEvents.length + i + 1,
      data: e.data,
      metadata: {},
    }));

    await eventStore.append(
      this.id,
      domainEvents,
      this.version - this.uncommittedEvents.length
    );

    this.uncommittedEvents = [];
  }
}

2. CQRS Pattern

// Command side (writes)
interface Command {
  type: string;
  payload: any;
  metadata: {
    userId: string;
    timestamp: Date;
    correlationId: string;
  };
}

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

  register(commandType: string, handler: CommandHandler): void {
    this.handlers.set(commandType, handler);
  }

  async dispatch(command: Command): Promise<void> {
    const handler = this.handlers.get(command.type);

    if (!handler) {
      throw new Error(`No handler for command: ${command.type}`);
    }

    await handler.handle(command);
  }
}

// Command handler
class CreateOrderHandler implements CommandHandler {
  constructor(
    private eventStore: EventStore,
    private orderRepository: OrderRepository
  ) {}

  async handle(command: CreateOrderCommand): Promise<void> {
    const order = new OrderAggregate(uuid());
    order.create(command.payload.customerId, command.payload.items);
    await order.save(this.eventStore);
  }
}

// Query side (reads)
interface Query {
  type: string;
  params: any;
}

class QueryBus {
  private handlers = new Map<string, QueryHandler>();

  register(queryType: string, handler: QueryHandler): void {
    this.handlers.set(queryType, handler);
  }

  async execute<T>(query: Query): Promise<T> {
    const handler = this.handlers.get(query.type);

    if (!handler) {
      throw new Error(`No handler for query: ${query.type}`);
    }

    return handler.handle(query);
  }
}

// Read model projection
class OrderReadModel {
  async project(event: DomainEvent): Promise<void> {
    switch (event.eventType) {
      case 'OrderCreated':
        await db.orderView.create({
          data: {
            id: event.aggregateId,
            customerId: event.data.customerId,
            status: 'draft',
            itemCount: event.data.items.length,
            total: event.data.total,
            createdAt: event.timestamp,
          },
        });
        break;

      case 'OrderSubmitted':
        await db.orderView.update({
          where: { id: event.aggregateId },
          data: {
            status: 'submitted',
            submittedAt: event.data.submittedAt,
          },
        });
        break;

      case 'OrderShipped':
        await db.orderView.update({
          where: { id: event.aggregateId },
          data: {
            status: 'shipped',
            trackingNumber: event.data.trackingNumber,
          },
        });
        break;
    }
  }

  // Rebuild projection from events
  async rebuild(): Promise<void> {
    // Clear existing read model
    await db.orderView.deleteMany();

    // Replay all events
    const events = await eventStore.getAllEvents();

    for (const event of events) {
      await this.project(event);
    }
  }
}

3. Message Queues with RabbitMQ

import amqp from 'amqplib';

class RabbitMQBroker {
  private connection: amqp.Connection;
  private channel: amqp.Channel;

  async connect(): Promise<void> {
    this.connection = await amqp.connect(process.env.RABBITMQ_URL!);
    this.channel = await this.connection.createChannel();

    // Setup exchanges
    await this.channel.assertExchange('events', 'topic', { durable: true });
    await this.channel.assertExchange('commands', 'direct', { durable: true });
    await this.channel.assertExchange('dlx', 'fanout', { durable: true });
  }

  async publish(exchange: string, routingKey: string, message: any): Promise<void> {
    const content = Buffer.from(JSON.stringify(message));

    this.channel.publish(exchange, routingKey, content, {
      persistent: true,
      contentType: 'application/json',
      messageId: uuid(),
      timestamp: Date.now(),
    });
  }

  async subscribe(
    queue: string,
    exchange: string,
    routingKey: string,
    handler: (message: any) => Promise<void>
  ): Promise<void> {
    // Setup queue with dead letter exchange
    await this.channel.assertQueue(queue, {
      durable: true,
      deadLetterExchange: 'dlx',
      deadLetterRoutingKey: `${queue}.dlq`,
    });

    await this.channel.bindQueue(queue, exchange, routingKey);

    // Consume messages
    await this.channel.consume(queue, async (msg) => {
      if (!msg) return;

      try {
        const content = JSON.parse(msg.content.toString());
        await handler(content);
        this.channel.ack(msg);
      } catch (error) {
        console.error('Message processing failed:', error);

        // Retry or dead-letter
        const retryCount = (msg.properties.headers?.['x-retry-count'] || 0) + 1;

        if (retryCount < 3) {
          // Retry with exponential backoff
          setTimeout(() => {
            this.channel.publish(exchange, routingKey, msg.content, {
              ...msg.properties,
              headers: {
                ...msg.properties.headers,
                'x-retry-count': retryCount,
              },
            });
            this.channel.ack(msg);
          }, Math.pow(2, retryCount) * 1000);
        } else {
          // Send to dead letter queue
          this.channel.reject(msg, false);
        }
      }
    });
  }
}

// Event publishing
class EventPublisher {
  constructor(private broker: RabbitMQBroker) {}

  async publish(event: DomainEvent): Promise<void> {
    const routingKey = `${event.aggregateType}.${event.eventType}`;
    await this.broker.publish('events', routingKey, event);
  }
}

// Event consumer
class OrderEventConsumer {
  constructor(
    private broker: RabbitMQBroker,
    private readModel: OrderReadModel
  ) {}

  async start(): Promise<void> {
    await this.broker.subscribe(
      'order-projector',
      'events',
      'Order.*',
      async (event) => {
        await this.readModel.project(event);
      }
    );
  }
}

4. Saga Pattern for Distributed Transactions

// Saga orchestrator
interface SagaStep {
  name: string;
  execute: (context: SagaContext) => Promise<void>;
  compensate: (context: SagaContext) => Promise<void>;
}

class SagaOrchestrator {
  private steps: SagaStep[] = [];
  private executedSteps: SagaStep[] = [];

  addStep(step: SagaStep): this {
    this.steps.push(step);
    return this;
  }

  async execute(context: SagaContext): Promise<void> {
    try {
      for (const step of this.steps) {
        console.log(`Executing step: ${step.name}`);
        await step.execute(context);
        this.executedSteps.push(step);
      }
    } catch (error) {
      console.error('Saga failed, compensating...', error);
      await this.compensate(context);
      throw error;
    }
  }

  private async compensate(context: SagaContext): Promise<void> {
    // Execute compensation in reverse order
    for (const step of this.executedSteps.reverse()) {
      try {
        console.log(`Compensating step: ${step.name}`);
        await step.compensate(context);
      } catch (error) {
        console.error(`Compensation failed for ${step.name}:`, error);
        // Log for manual intervention
        await this.logCompensationFailure(step, context, error);
      }
    }
  }
}

// Order saga example
const createOrderSaga = new SagaOrchestrator()
  .addStep({
    name: 'Reserve Inventory',
    execute: async (ctx) => {
      const reservation = await inventoryService.reserve(ctx.items);
      ctx.reservationId = reservation.id;
    },
    compensate: async (ctx) => {
      if (ctx.reservationId) {
        await inventoryService.releaseReservation(ctx.reservationId);
      }
    },
  })
  .addStep({
    name: 'Process Payment',
    execute: async (ctx) => {
      const payment = await paymentService.charge(ctx.customerId, ctx.total);
      ctx.paymentId = payment.id;
    },
    compensate: async (ctx) => {
      if (ctx.paymentId) {
        await paymentService.refund(ctx.paymentId);
      }
    },
  })
  .addStep({
    name: 'Create Order',
    execute: async (ctx) => {
      const order = await orderService.create({
        customerId: ctx.customerId,
        items: ctx.items,
        paymentId: ctx.paymentId,
        reservationId: ctx.reservationId,
      });
      ctx.orderId = order.id;
    },
    compensate: async (ctx) => {
      if (ctx.orderId) {
        await orderService.cancel(ctx.orderId);
      }
    },
  })
  .addStep({
    name: 'Send Confirmation',
    execute: async (ctx) => {
      await notificationService.sendOrderConfirmation(ctx.orderId);
    },
    compensate: async (ctx) => {
      // No compensation needed for notifications
    },
  });

// Execute saga
async function handleCreateOrder(command: CreateOrderCommand): Promise<void> {
  const context: SagaContext = {
    customerId: command.customerId,
    items: command.items,
    total: calculateTotal(command.items),
  };

  await createOrderSaga.execute(context);
}

5. Kafka Streaming

import { Kafka, Producer, Consumer, EachMessagePayload } from 'kafkajs';

class KafkaService {
  private kafka: Kafka;
  private producer: Producer;
  private consumers: Map<string, Consumer> = new Map();

  constructor() {
    this.kafka = new Kafka({
      clientId: process.env.SERVICE_NAME,
      brokers: (process.env.KAFKA_BROKERS || '').split(','),
    });
  }

  async connect(): Promise<void> {
    this.producer = this.kafka.producer({
      idempotent: true,
      maxInFlightRequests: 5,
    });

    await this.producer.connect();
  }

  async publish(topic: string, messages: KafkaMessage[]): Promise<void> {
    await this.producer.send({
      topic,
      messages: messages.map(m => ({
        key: m.key,
        value: JSON.stringify(m.value),
        headers: m.headers,
        partition: m.partition,
      })),
    });
  }

  async subscribe(
    groupId: string,
    topics: string[],
    handler: (payload: EachMessagePayload) => Promise<void>
  ): Promise<void> {
    const consumer = this.kafka.consumer({
      groupId,
      sessionTimeout: 30000,
      heartbeatInterval: 3000,
    });

    await consumer.connect();
    await consumer.subscribe({ topics, fromBeginning: false });

    await consumer.run({
      eachMessage: async (payload) => {
        try {
          await handler(payload);
        } catch (error) {
          console.error('Message processing failed:', error);
          // Implement retry/DLQ logic
        }
      },
    });

    this.consumers.set(groupId, consumer);
  }

  async disconnect(): Promise<void> {
    await this.producer.disconnect();
    for (const consumer of this.consumers.values()) {
      await consumer.disconnect();
    }
  }
}

// Stream processing
class OrderStreamProcessor {
  constructor(private kafka: KafkaService) {}

  async start(): Promise<void> {
    await this.kafka.subscribe(
      'order-processor',
      ['order-events'],
      async ({ topic, partition, message }) => {
        const event = JSON.parse(message.value?.toString() || '{}');

        switch (event.type) {
          case 'OrderCreated':
            await this.handleOrderCreated(event);
            break;
          case 'OrderCompleted':
            await this.handleOrderCompleted(event);
            break;
        }
      }
    );
  }

  private async handleOrderCreated(event: any): Promise<void> {
    // Update analytics
    await analyticsService.recordOrder(event.data);

    // Trigger downstream processes
    await this.kafka.publish('inventory-commands', [{
      key: event.aggregateId,
      value: {
        type: 'ReserveInventory',
        orderId: event.aggregateId,
        items: event.data.items,
      },
    }]);
  }
}

Use Cases

1. Order Processing System

// Complete order workflow
async function processOrder(orderId: string): Promise<void> {
  const saga = new SagaOrchestrator()
    .addStep(reserveInventoryStep)
    .addStep(processPaymentStep)
    .addStep(createShipmentStep)
    .addStep(sendNotificationStep);

  await saga.execute({ orderId });
}

2. Real-time Analytics

// Stream aggregation
const orderTotalsStream = kafka.subscribe(
  'analytics-aggregator',
  ['order-events'],
  async (event) => {
    await updateDailySales(event.data.total);
    await updateProductMetrics(event.data.items);
  }
);

Best Practices

Do's

• Design events as facts - Immutable, past-tense naming
• Implement idempotent handlers - Handle duplicates gracefully
• Plan for event versioning - Schema evolution
• Use dead letter queues - Handle failures
• Monitor queue depths - Alert on backlogs
• Test with chaos - Simulate failures

Don'ts

• Don't couple services through shared databases
• Don't ignore message ordering requirements
• Don't skip compensation logic
• Don't forget about exactly-once semantics
• Don't over-engineer for simple use cases
• Don't ignore backpressure

Related Skills

• redis - Pub/sub and caching
• real-time-systems - WebSocket integration
• backend-development - Service architecture

Reference Resources

• Event Sourcing Pattern
• CQRS Pattern
• RabbitMQ Documentation
• Apache Kafka Documentation