Event-Driven Architect

Design robust event-driven architectures for scalable, loosely-coupled microservices.

When to Use

• Designing event-driven microservices
• Implementing event sourcing and CQRS
• Setting up Kafka, RabbitMQ, or similar
• Designing saga patterns for distributed transactions
• Implementing eventual consistency
• Building real-time data pipelines
• Designing publish-subscribe patterns

Key Patterns

Event Sourcing

Store all changes as events, reconstruct state by replaying events.

CQRS (Command Query Responsibility Segregation)

Separate read and write models for better scalability.

Saga Pattern

Manage distributed transactions across microservices.

Event Streaming

Process continuous streams of events in real-time.

Kafka Configuration Example

# Kafka topic configuration
apiVersion: kafka.strimzi.io/v1beta2
kind: KafkaTopic
metadata:
  name: order-events
spec:
  partitions: 10
  replicas: 3
  config:
    retention.ms: 604800000  # 7 days
    compression.type: snappy
    max.message.bytes: 1048576
---
# Event schema (Avro)
{
  "type": "record",
  "name": "OrderCreated",
  "namespace": "com.example.events",
  "fields": [
    {"name": "orderId", "type": "string"},
    {"name": "customerId", "type": "string"},
    {"name": "items", "type": {"type": "array", "items": "OrderItem"}},
    {"name": "totalAmount", "type": "double"},
    {"name": "timestamp", "type": "long"}
  ]
}

Event Sourcing Implementation

// Event store
class OrderEventStore {
  async appendEvent(event: DomainEvent): Promise<void> {
    await this.eventStore.append({
      aggregateId: event.aggregateId,
      eventType: event.constructor.name,
      eventData: JSON.stringify(event),
      timestamp: new Date(),
      version: event.version
    });
    
    // Publish to event bus
    await this.eventBus.publish(event);
  }

  async getEvents(aggregateId: string): Promise<DomainEvent[]> {
    const events = await this.eventStore.find({
      aggregateId,
      orderBy: { version: 'asc' }
    });
    return events.map(e => this.deserialize(e));
  }

  reconstructAggregate(events: DomainEvent[]): Order {
    const order = new Order();
    events.forEach(event => order.apply(event));
    return order;
  }
}

// Domain events
class OrderCreatedEvent {
  constructor(
    public orderId: string,
    public customerId: string,
    public items: OrderItem[],
    public totalAmount: number
  ) {}
}

class OrderConfirmedEvent {
  constructor(public orderId: string) {}
}

// Aggregate
class Order {
  private id: string;
  private status: OrderStatus;
  private items: OrderItem[];

  apply(event: DomainEvent) {
    if (event instanceof OrderCreatedEvent) {
      this.id = event.orderId;
      this.status = 'pending';
      this.items = event.items;
    } else if (event instanceof OrderConfirmedEvent) {
      this.status = 'confirmed';
    }
  }
}

CQRS Pattern

// Command side (write model)
class CreateOrderCommand {
  constructor(
    public customerId: string,
    public items: OrderItem[]
  ) {}
}

class OrderCommandHandler {
  async handle(command: CreateOrderCommand): Promise<string> {
    // Validate
    this.validateCommand(command);
    
    // Create event
    const orderId = uuid();
    const event = new OrderCreatedEvent(
      orderId,
      command.customerId,
      command.items,
      this.calculateTotal(command.items)
    );
    
    // Store event
    await this.eventStore.appendEvent(event);
    
    return orderId;
  }
}

// Query side (read model)
class OrderQueryModel {
  async getOrderById(orderId: string): Promise<OrderDTO> {
    // Read from optimized read database (e.g., MongoDB)
    return await this.orderReadRepo.findById(orderId);
  }

  async getOrdersByCustomer(customerId: string): Promise<OrderDTO[]> {
    return await this.orderReadRepo.find({ customerId });
  }
}

// Projection (updates read model from events)
class OrderProjection {
  @EventHandler(OrderCreatedEvent)
  async onOrderCreated(event: OrderCreatedEvent) {
    await this.orderReadRepo.create({
      id: event.orderId,
      customerId: event.customerId,
      items: event.items,
      totalAmount: event.totalAmount,
      status: 'pending',
      createdAt: new Date()
    });
  }

  @EventHandler(OrderConfirmedEvent)
  async onOrderConfirmed(event: OrderConfirmedEvent) {
    await this.orderReadRepo.update(event.orderId, {
      status: 'confirmed'
    });
  }
}

Saga Pattern (Orchestration)

// Saga orchestrator for order creation
class OrderCreationSaga {
  async execute(orderId: string) {
    try {
      // Step 1: Reserve inventory
      await this.inventoryService.reserve(orderId);
      await this.sagaStore.recordStep(orderId, 'inventory_reserved');

      // Step 2: Process payment
      await this.paymentService.charge(orderId);
      await this.sagaStore.recordStep(orderId, 'payment_processed');

      // Step 3: Confirm order
      await this.orderService.confirm(orderId);
      await this.sagaStore.recordStep(orderId, 'order_confirmed');

      // Success
      await this.sagaStore.complete(orderId);

    } catch (error) {
      // Compensate (rollback)
      await this.compensate(orderId);
      throw new SagaFailedError(orderId, error);
    }
  }

  private async compensate(orderId: string) {
    const steps = await this.sagaStore.getCompletedSteps(orderId);
    
    // Rollback in reverse order
    if (steps.includes('payment_processed')) {
      await this.paymentService.refund(orderId);
    }
    if (steps.includes('inventory_reserved')) {
      await this.inventoryService.release(orderId);
    }
    
    await this.orderService.cancel(orderId);
  }
}

Best Practices

• ✅ Use event versioning from day one
• ✅ Implement idempotent event handlers
• ✅ Design for eventual consistency
• ✅ Use schema registry (Avro, Protobuf)
• ✅ Implement dead letter queues
• ✅ Monitor event lag and throughput
• ✅ Use correlation IDs for tracing
• ✅ Handle duplicate events gracefully

Related Skills

• microservices-orchestrator - Service design
• distributed-tracing-setup - Event tracing
• chaos-engineering-setup - Resilience testing