Event-driven architecture (EDA) is a powerful pattern for building scalable, loosely coupled microservices. By designing services that communicate through events rather than direct calls, you can create systems that are more resilient, flexible, and easier to evolve. Let's explore how to implement effective event-driven architectures.
Understanding Event-Driven Architecture
At its core, event-driven architecture is about services producing and consuming events. An event represents a fact that has happened—something that cannot be changed. This fundamental shift from synchronous to asynchronous communication changes how we design and build systems.
typescript
// Synchronous approach (tight coupling)
class OrderService {
async createOrder(orderData: OrderData) {
const order = await this.db.save(orderData);
await this.inventoryService.reserveItems(order.items); // Direct call
await this.paymentService.processPayment(order.total); // Direct call
await this.notificationService.sendEmail(order.email); // Direct call
return order;
}
}
// Event-driven approach (loose coupling)
class OrderService {
constructor(private eventBus: EventBus) {}
async createOrder(orderData: OrderData) {
const order = await this.db.save(orderData);
// Publish event, don't call other services directly
await this.eventBus.publish({
type: 'OrderCreated',
aggregateId: order.id,
payload: {
orderId: order.id,
userId: order.userId,
items: order.items,
total: order.total,
email: order.email
},
timestamp: new Date(),
version: 1
});
return order;
}
}Event Types and Their Uses
Different types of events serve different purposes in your architecture.
Domain Events
Domain events represent something that happened in your domain that's important to other parts of the system.
typescript
interface DomainEvent {
type: string;
aggregateId: string;
payload: any;
timestamp: Date;
version: number;
correlationId?: string;
causationId?: string;
}
// Example domain events
const events = {
OrderCreated: {
type: 'OrderCreated',
aggregateId: 'order-123',
payload: {
orderId: 'order-123',
userId: 'user-456',
items: [
{ productId: 'prod-789', quantity: 2 }
],
total: 99.99
},
timestamp: new Date('2024-03-15T10:30:00Z'),
version: 1
},
PaymentCompleted: {
type: 'PaymentCompleted',
aggregateId: 'payment-999',
payload: {
paymentId: 'payment-999',
orderId: 'order-123',
amount: 99.99,
method: 'credit_card'
},
timestamp: new Date('2024-03-15T10:31:00Z'),
version: 1
}
};Integration Events
Integration events cross service boundaries and enable loose coupling between services.
typescript
class IntegrationEventPublisher {
constructor(
private messageBroker: MessageBroker,
private serializer: EventSerializer
) {}
async publish(event: IntegrationEvent) {
const message = this.serializer.serialize(event);
await this.messageBroker.publish({
topic: event.type,
message,
headers: {
'content-type': 'application/json',
'message-id': this.generateId(),
'correlation-id': event.correlationId,
'timestamp': event.timestamp.toISOString()
}
});
}
}
// Usage
await publisher.publish({
type: 'OrderCompleted',
correlationId: orderId,
payload: {
orderId: 'order-123',
userId: 'user-456',
completedAt: new Date()
},
timestamp: new Date()
});Message Broker Options
Choosing the right message broker is crucial for your event-driven architecture.
RabbitMQ Implementation
typescript
import amqp from 'amqplib';
class RabbitMQEventBus {
private connection: amqp.Connection;
private channel: amqp.Channel;
async connect(url: string) {
this.connection = await amqp.connect(url);
this.channel = await this.connection.createChannel();
}
async publish(event: DomainEvent) {
const exchange = 'domain-events';
await this.channel.assertExchange(exchange, 'topic', { durable: true });
const message = Buffer.from(JSON.stringify(event));
const routingKey = event.type.toLowerCase().replace(/./g, '.');
this.channel.publish(exchange, routingKey, message, {
contentType: 'application/json',
messageId: this.generateId(),
timestamp: event.timestamp.getTime()
});
}
async subscribe(eventPattern: string, handler: EventHandler) {
const exchange = 'domain-events';
const queue = await this.channel.assertQueue('', { exclusive: true });
await this.channel.bindQueue(queue.queue, exchange, eventPattern);
await this.channel.consume(queue.queue, async (msg) => {
try {
const event: DomainEvent = JSON.parse(msg.content.toString());
await handler.handle(event);
this.channel.ack(msg);
} catch (error) {
console.error('Error handling event:', error);
this.channel.nack(msg, false, false);
}
});
}
}Kafka Implementation
typescript
import { Kafka } from 'kafkajs';
class KafkaEventBus {
private kafka: Kafka;
private producer: any;
private consumer: any;
constructor(brokers: string[]) {
this.kafka = new Kafka({
clientId: 'order-service',
brokers
});
}
async publish(event: DomainEvent) {
this.producer = this.kafka.producer();
await this.producer.connect();
await this.producer.send({
topic: event.type,
messages: [{
key: event.aggregateId,
value: JSON.stringify(event),
headers: {
'correlation-id': event.correlationId || '',
'timestamp': event.timestamp.toISOString()
}
}]
});
}
async subscribe(topics: string[], handler: EventHandler) {
this.consumer = this.kafka.consumer({ groupId: 'order-service-group' });
await this.consumer.connect();
await this.consumer.subscribe({ topics });
await this.consumer.run({
eachMessage: async ({ topic, partition, message }) => {
try {
const event: DomainEvent = JSON.parse(message.value.toString());
await handler.handle(event);
} catch (error) {
console.error('Error handling event:', error);
}
}
});
}
}Event Sourcing
Event sourcing stores all changes to an application state as a sequence of events.
typescript
class EventStore {
constructor(private db: Database) {}
async saveEvents(aggregateId: string, events: DomainEvent[], expectedVersion: number) {
await this.db.transaction(async (tx) => {
// Check version for optimistic concurrency
const current = await tx.findOne('aggregates', { aggregateId });
if (current.version !== expectedVersion) {
throw new Error('Concurrency conflict');
}
// Save events
for (const event of events) {
await tx.insert('events', {
aggregateId,
eventType: event.type,
payload: JSON.stringify(event.payload),
timestamp: event.timestamp,
version: expectedVersion + 1
});
}
// Update aggregate version
await tx.update('aggregates',
{ aggregateId },
{ version: expectedVersion + events.length }
);
});
}
async getEvents(aggregateId: string): Promise<DomainEvent[]> {
const records = await this.db.find('events', { aggregateId });
return records.map(record => ({
type: record.eventType,
aggregateId: record.aggregateId,
payload: JSON.parse(record.payload),
timestamp: record.timestamp,
version: record.version
}));
}
}Event Handler Patterns
Different patterns for handling events effectively.
Simple Event Handler
typescript
class InventoryEventHandler {
constructor(private inventoryService: InventoryService) {}
async handle(event: DomainEvent) {
switch (event.type) {
case 'OrderCreated':
await this.handleOrderCreated(event);
break;
case 'PaymentCompleted':
await this.handlePaymentCompleted(event);
break;
case 'OrderCancelled':
await this.handleOrderCancelled(event);
break;
}
}
private async handleOrderCreated(event: DomainEvent) {
const { items } = event.payload;
await this.inventoryService.reserveItems(items);
}
private async handlePaymentCompleted(event: DomainEvent) {
await this.inventoryService.confirmReservation(event.payload.orderId);
}
private async handleOrderCancelled(event: DomainEvent) {
await this.inventoryService.releaseReservation(event.payload.orderId);
}
}Saga Event Handler
typescript
class OrderSagaEventHandler {
constructor(
private sagaRepository: SagaRepository,
private orderService: OrderService,
private paymentService: PaymentService,
private inventoryService: InventoryService
) {}
async handle(event: DomainEvent) {
const saga = await this.sagaRepository.findByCorrelationId(event.correlationId);
if (!saga) {
return; // Not related to any saga
}
switch (event.type) {
case 'PaymentCompleted':
await saga.handlePaymentCompleted(event);
break;
case 'InventoryReserved':
await saga.handleInventoryReserved(event);
break;
case 'PaymentFailed':
await saga.handlePaymentFailed(event);
break;
}
}
}Event Versioning
Events must evolve over time while maintaining backward compatibility.
typescript
interface OrderCreatedEventV1 {
type: 'OrderCreated';
version: 1;
payload: {
orderId: string;
userId: string;
items: Array<{ productId: string; quantity: number }>;
total: number;
};
}
interface OrderCreatedEventV2 {
type: 'OrderCreated';
version: 2;
payload: {
orderId: string;
userId: string;
items: Array<{ productId: string; quantity: number; price: number }>; // Added price
total: number;
currency: string; // Added currency
shippingAddress: Address; // Added address
};
}
// Event upgrader
class EventUpgrader {
upgrade(event: DomainEvent): DomainEvent {
switch (event.type) {
case 'OrderCreated':
if (event.version === 1) {
return this.upgradeOrderCreatedV1ToV2(event as any);
}
break;
}
return event;
}
private upgradeOrderCreatedV1ToV2(v1: OrderCreatedEventV1): OrderCreatedEventV2 {
return {
type: 'OrderCreated',
version: 2,
payload: {
...v1.payload,
items: v1.payload.items.map(item => ({
...item,
price: 0 // Default price for old events
})),
currency: 'USD', // Default currency
shippingAddress: null // Will be null for old events
}
};
}
}Error Handling and Dead Letter Queues
Handle failed events gracefully with dead letter queues.
typescript
class ResilientEventHandler {
constructor(
private deadLetterQueue: DeadLetterQueue,
private retryPolicy: RetryPolicy
) {}
async handleWithRetry(event: DomainEvent, handler: EventHandler) {
let attempt = 0;
const maxAttempts = 3;
while (attempt < maxAttempts) {
try {
await handler.handle(event);
return; // Success
} catch (error) {
attempt++;
if (attempt >= maxAttempts) {
// Send to dead letter queue
await this.deadLetterQueue.add({
event,
error: error.message,
attempts,
failedAt: new Date()
});
return;
}
// Exponential backoff
const delay = Math.pow(2, attempt) * 1000;
await this.sleep(delay);
}
}
}
}
class DeadLetterQueue {
async add(item: DeadLetterItem) {
await this.db.insert('dead_letter_queue', {
eventData: JSON.stringify(item.event),
error: item.error,
attempts: item.attempts,
failedAt: item.failedAt,
processed: false
});
}
async reprocess(handler: EventHandler) {
const items = await this.db.find('dead_letter_queue', {
processed: false
});
for (const item of items) {
try {
const event: DomainEvent = JSON.parse(item.eventData);
await handler.handle(event);
await this.db.update('dead_letter_queue',
{ id: item.id },
{ processed: true, processedAt: new Date() }
);
} catch (error) {
console.error('Failed to reprocess event:', error);
}
}
}
}Testing Event-Driven Systems
Test your event-driven architecture effectively.
typescript
describe('OrderService', () => {
it('should publish OrderCreated event', async () => {
const eventBus = new InMemoryEventBus();
const service = new OrderService(eventBus);
const order = await service.createOrder({
userId: 'user-123',
items: [{ productId: 'prod-456', quantity: 2 }]
});
const events = eventBus.getEvents();
expect(events).toHaveLength(1);
expect(events[0].type).toBe('OrderCreated');
expect(events[0].payload.orderId).toBe(order.id);
});
});
describe('InventoryEventHandler', () => {
it('should reserve inventory when order created', async () => {
const inventoryService = new MockInventoryService();
const handler = new InventoryEventHandler(inventoryService);
await handler.handle({
type: 'OrderCreated',
payload: {
items: [{ productId: 'prod-123', quantity: 2 }]
}
});
expect(inventoryService.reserveCalled).toBe(true);
});
});Conclusion
Event-driven architecture enables building scalable, resilient microservices. Focus on proper event design, reliable messaging, and comprehensive error handling. Remember that events represent facts that have happened—they're immutable and cannot be changed.
Start with a simple event bus, evolve your event schemas carefully, and invest in good observability. The benefits of loose coupling and scalability make event-driven architecture worth the effort for complex distributed systems.