Event-Driven Architecture: Patterns and Implementation

Event-driven systems communicate through events rather than direct calls, enabling loose coupling.

Core Concepts

// Domain Event
interface DomainEvent {
  eventId: string;
  eventType: string;
  occurredAt: Date;
  aggregateId: string;
}

class OrderPlaced implements DomainEvent {
  readonly eventId = crypto.randomUUID();
  readonly eventType = 'order.placed';
  readonly occurredAt = new Date();

  constructor(
    readonly aggregateId: string,
    readonly customerId: string,
    readonly items: OrderItem[],
    readonly total: number
  ) {}
}

Event Sourcing

Store state as a sequence of events rather than current snapshot.

class OrderAggregate {
  private events: DomainEvent[] = [];
  private state: OrderState = { status: 'pending', items: [] };

  // Apply events to rebuild state
  private apply(event: DomainEvent): void {
    if (event instanceof OrderPlaced) {
      this.state = { status: 'placed', items: event.items };
    } else if (event instanceof OrderShipped) {
      this.state = { ...this.state, status: 'shipped', trackingId: event.trackingId };
    }
  }

  // Command: place order
  place(customerId: string, items: OrderItem[]) {
    const event = new OrderPlaced(this.id, customerId, items, this.calculateTotal(items));
    this.events.push(event);
    this.apply(event);
    return event;
  }

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

CQRS (Command Query Responsibility Segregation)

// Write side (Commands)
class PlaceOrderCommand {
  constructor(
    readonly customerId: string,
    readonly items: OrderItem[]
  ) {}
}

class PlaceOrderHandler {
  constructor(private repo: OrderRepository, private bus: EventBus) {}

  async handle(cmd: PlaceOrderCommand): Promise<string> {
    const order = Order.create(cmd.customerId, cmd.items);
    await this.repo.save(order);
    await this.bus.publish(order.events);
    return order.id;
  }
}

// Read side (Queries) - separate optimized read models
class OrderSummaryQuery {
  constructor(readonly customerId: string) {}
}

class OrderSummaryHandler {
  constructor(private readDb: ReadDatabase) {}

  async handle(query: OrderSummaryQuery): Promise<OrderSummary[]> {
    return this.readDb.query(
      'SELECT * FROM order_summaries WHERE customer_id = $1',
      [query.customerId]
    );
  }
}

Saga Pattern for Distributed Transactions

// Choreography-based saga
class OrderSaga {
  // Each service publishes/subscribes to events
  // No central coordinator

  // OrderService
  async handlePaymentCompleted(event: PaymentCompleted) {
    await this.orderRepo.updateStatus(event.orderId, 'payment_confirmed');
    await this.bus.publish(new OrderReadyToShip(event.orderId));
  }

  // Handle compensation on failure
  async handlePaymentFailed(event: PaymentFailed) {
    await this.orderRepo.cancel(event.orderId);
    await this.bus.publish(new OrderCancelled(event.orderId, 'payment_failed'));
  }
}

Kafka Integration

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

const kafka = new Kafka({ clientId: 'order-service', brokers: ['kafka:9092'] });

// Producer
class EventPublisher {
  private producer: Producer;

  async publish(event: DomainEvent) {
    await this.producer.send({
      topic: event.eventType.replace('.', '-'),
      messages: [{
        key: event.aggregateId,
        value: JSON.stringify(event),
        headers: { 'event-type': event.eventType },
      }],
    });
  }
}

// Consumer
class OrderEventConsumer {
  async start() {
    const consumer = kafka.consumer({ groupId: 'shipping-service' });
    await consumer.subscribe({ topic: 'order-placed', fromBeginning: false });

    await consumer.run({
      eachMessage: async ({ message }) => {
        const event = JSON.parse(message.value!.toString());
        await this.handleOrderPlaced(event);
      },
    });
  }
}

Event-driven architectures excel at decoupling services in microservices environments.