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