Event-Driven Architecture: Building Scalable Systems

Event-driven architecture (EDA) enables loosely coupled, scalable systems. This guide covers core patterns, implementation strategies, and best practices.

Core Concepts#

Events vs Commands vs Queries#

// Event: Something that happened (past tense, immutable)
interface OrderPlaced {
  type: 'OrderPlaced';
  orderId: string;
  customerId: string;
  items: OrderItem[];
  timestamp: Date;
}

// Command: Request to do something (imperative)
interface PlaceOrder {
  type: 'PlaceOrder';
  customerId: string;
  items: OrderItem[];
}

// Query: Request for information
interface GetOrderStatus {
  type: 'GetOrderStatus';
  orderId: string;
}

Event Flow#

┌─────────┐    Command     ┌──────────┐    Event      ┌──────────┐
│ Client  │ ─────────────> │ Service  │ ────────────> │  Queue   │
└─────────┘                └──────────┘               └──────────┘
                                                            │
                           ┌──────────────────────────────────┤
                           │                │                 │
                           ▼                ▼                 ▼
                     ┌──────────┐    ┌──────────┐      ┌──────────┐
                     │ Service  │    │ Service  │      │ Service  │
                     │    A     │    │    B     │      │    C     │
                     └──────────┘    └──────────┘      └──────────┘

Event Sourcing#

Store events as the source of truth:

// Event Store
interface EventStore {
  append(streamId: string, events: DomainEvent[]): Promise<void>;
  read(streamId: string, fromVersion?: number): Promise<DomainEvent[]>;
}

// Domain Events
type OrderEvent =
  | { type: 'OrderCreated'; orderId: string; customerId: string }
  | { type: 'ItemAdded'; orderId: string; item: OrderItem }
  | { type: 'ItemRemoved'; orderId: string; itemId: string }
  | { type: 'OrderSubmitted'; orderId: string; timestamp: Date }
  | { type: 'OrderShipped'; orderId: string; trackingNumber: string };

// Aggregate
class Order {
  private state: OrderState;
  private uncommittedEvents: OrderEvent[] = [];

  constructor(private id: string) {
    this.state = { status: 'draft', items: [] };
  }

  // Rebuild from events
  static fromEvents(id: string, events: OrderEvent[]): Order {
    const order = new Order(id);
    events.forEach(event => order.apply(event));
    return order;
  }

  addItem(item: OrderItem) {
    if (this.state.status !== 'draft') {
      throw new Error('Cannot modify submitted order');
    }

    const event: OrderEvent = {
      type: 'ItemAdded',
      orderId: this.id,
      item,
    };

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

  private apply(event: OrderEvent) {
    switch (event.type) {
      case 'ItemAdded':
        this.state.items.push(event.item);
        break;
      case 'ItemRemoved':
        this.state.items = this.state.items.filter(i => i.id !== event.itemId);
        break;
      case 'OrderSubmitted':
        this.state.status = 'submitted';
        break;
    }
  }

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

CQRS (Command Query Responsibility Segregation)#

Separate read and write models:

// Write Model (Commands)
class OrderCommandHandler {
  constructor(
    private eventStore: EventStore,
    private eventBus: EventBus
  ) {}

  async handle(command: PlaceOrder): Promise<void> {
    const order = new Order(generateId());

    command.items.forEach(item => order.addItem(item));
    order.submit();

    const events = order.getUncommittedEvents();

    await this.eventStore.append(order.id, events);
    await this.eventBus.publish(events);
  }
}

// Read Model (Queries)
class OrderReadModel {
  private orders: Map<string, OrderSummary> = new Map();

  // Update from events
  async handleEvent(event: OrderEvent) {
    switch (event.type) {
      case 'OrderCreated':
        this.orders.set(event.orderId, {
          id: event.orderId,
          customerId: event.customerId,
          items: [],
          status: 'draft',
        });
        break;
      case 'ItemAdded':
        const order = this.orders.get(event.orderId);
        order?.items.push(event.item);
        break;
      case 'OrderShipped':
        const shipped = this.orders.get(event.orderId);
        if (shipped) shipped.status = 'shipped';
        break;
    }
  }

  // Query
  getOrder(orderId: string): OrderSummary | undefined {
    return this.orders.get(orderId);
  }

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

Message Brokers#

Implementing with Redis Streams#

import Redis from 'ioredis';

class RedisEventBus {
  private redis: Redis;
  private subscriber: Redis;

  constructor(url: string) {
    this.redis = new Redis(url);
    this.subscriber = new Redis(url);
  }

  async publish(stream: string, event: DomainEvent): Promise<void> {
    await this.redis.xadd(
      stream,
      '*',
      'type', event.type,
      'data', JSON.stringify(event),
      'timestamp', Date.now().toString()
    );
  }

  async subscribe(
    stream: string,
    group: string,
    consumer: string,
    handler: (event: DomainEvent) => Promise<void>
  ): Promise<void> {
    // Create consumer group if not exists
    try {
      await this.redis.xgroup('CREATE', stream, group, '0', 'MKSTREAM');
    } catch (e) {
      // Group already exists
    }

    while (true) {
      const results = await this.subscriber.xreadgroup(
        'GROUP', group, consumer,
        'BLOCK', 5000,
        'COUNT', 10,
        'STREAMS', stream, '>'
      );

      if (results) {
        for (const [, messages] of results) {
          for (const [id, fields] of messages) {
            const event = JSON.parse(fields[3]); // data field
            await handler(event);
            await this.subscriber.xack(stream, group, id);
          }
        }
      }
    }
  }
}

Implementing with RabbitMQ#

import amqp from 'amqplib';

class RabbitMQEventBus {
  private channel: amqp.Channel;

  async connect(url: string): Promise<void> {
    const connection = await amqp.connect(url);
    this.channel = await connection.createChannel();
  }

  async publish(exchange: string, event: DomainEvent): Promise<void> {
    await this.channel.assertExchange(exchange, 'topic', { durable: true });

    this.channel.publish(
      exchange,
      event.type,
      Buffer.from(JSON.stringify(event)),
      { persistent: true }
    );
  }

  async subscribe(
    exchange: string,
    queue: string,
    pattern: string,
    handler: (event: DomainEvent) => Promise<void>
  ): Promise<void> {
    await this.channel.assertExchange(exchange, 'topic', { durable: true });
    await this.channel.assertQueue(queue, { durable: true });
    await this.channel.bindQueue(queue, exchange, pattern);

    this.channel.consume(queue, async (msg) => {
      if (msg) {
        const event = JSON.parse(msg.content.toString());
        try {
          await handler(event);
          this.channel.ack(msg);
        } catch (error) {
          this.channel.nack(msg, false, true);
        }
      }
    });
  }
}

Saga Pattern#

Coordinate distributed transactions:

interface SagaStep<T> {
  execute(context: T): Promise<void>;
  compensate(context: T): Promise<void>;
}

class OrderSaga {
  private steps: SagaStep<OrderContext>[] = [
    {
      async execute(ctx) {
        ctx.paymentId = await paymentService.charge(ctx.amount);
      },
      async compensate(ctx) {
        await paymentService.refund(ctx.paymentId);
      },
    },
    {
      async execute(ctx) {
        ctx.inventoryReservation = await inventoryService.reserve(ctx.items);
      },
      async compensate(ctx) {
        await inventoryService.release(ctx.inventoryReservation);
      },
    },
    {
      async execute(ctx) {
        ctx.shipmentId = await shippingService.createShipment(ctx.orderId);
      },
      async compensate(ctx) {
        await shippingService.cancelShipment(ctx.shipmentId);
      },
    },
  ];

  async execute(context: OrderContext): Promise<void> {
    const completedSteps: number[] = [];

    try {
      for (let i = 0; i < this.steps.length; i++) {
        await this.steps[i].execute(context);
        completedSteps.push(i);
      }
    } catch (error) {
      // Compensate in reverse order
      for (const stepIndex of completedSteps.reverse()) {
        try {
          await this.steps[stepIndex].compensate(context);
        } catch (compensateError) {
          console.error(`Failed to compensate step ${stepIndex}`, compensateError);
        }
      }
      throw error;
    }
  }
}

Best Practices#

Idempotent handlers: Events may be delivered multiple times
Event versioning: Plan for schema evolution
Dead letter queues: Handle poison messages
Monitoring: Track event processing latency and failures
Event ordering: Use partitioning for ordered processing

Conclusion#

Event-driven architecture enables scalable, loosely coupled systems. Start with simple pub/sub patterns and evolve to event sourcing and CQRS as complexity grows.

Share this article