Event-Driven Architecture Complete Guide

Event-driven architecture (EDA) enables services to communicate asynchronously through events, creating loose coupling and enabling better scalability.

What Is Event-Driven Architecture?

In EDA, services produce and consume events rather than making direct API calls.

text

Traditional Request/Response:
Service A → [Request] → Service B
Service A ← [Response] ← Service B

Event-Driven:
Service A → [Event] → Event Bus → Service B, C, D

Core Concepts

Events

Events are facts that happened in the system.

javascript

// Event examples
{
  "type": "OrderCreated",
  "id": "evt_123",
  "timestamp": "2024-03-27T10:00:00Z",
  "data": {
    "orderId": "ord_456",
    "userId": "usr_789",
    "total": 99.99,
    "items": [...]
  }
}

Event Bus

The event bus routes events from producers to consumers.

javascript

// Simple in-memory event bus
class EventBus {
  constructor() {
    this.subscribers = new Map();
  }

  subscribe(eventType, callback) {
    if (!this.subscribers.has(eventType)) {
      this.subscribers.set(eventType, []);
    }
    this.subscribers.get(eventType).push(callback);
  }

  publish(event) {
    const callbacks = this.subscribers.get(event.type) || [];
    callbacks.forEach(callback => callback(event));
  }
}

const eventBus = new EventBus();

// Subscribe to events
eventBus.subscribe('OrderCreated', async (event) => {
  await sendConfirmationEmail(event.data);
});

eventBus.subscribe('OrderCreated', async (event) => {
  await updateInventory(event.data);
});

// Publish events
eventBus.publish({
  type: 'OrderCreated',
  data: { orderId: '123', total: 99.99 }
});

Event Patterns

Event Notification

Simple fire-and-forget notifications.

javascript

// Producer
class OrderService {
  async createOrder(orderData) {
    const order = await this.db.create(orderData);

    // Notify others that order was created
    await eventBus.publish({
      type: 'OrderCreated',
      data: order,
    });

    return order;
  }
}

// Consumer
class EmailService {
  constructor() {
    eventBus.subscribe('OrderCreated', this.handleOrderCreated);
  }

  async handleOrderCreated(event) {
    await this.sendEmail(event.data.userEmail, {
      subject: 'Order Confirmation',
      body: `Order ${event.data.id} created!`,
    });
  }
}

Event-Carried State Transfer

Events carry all necessary state.

javascript

{
  "type": "ProductUpdated",
  "data": {
    "id": "prod_123",
    "name": "Product Name",
    "price": 99.99,
    "description": "Full description...",
    "stock": 100,
    "version": 5
  }
}

// Consumers don't need to fetch product details
class SearchService {
  async handleProductUpdated(event) {
    await this.searchIndex.update(event.data);
    // All data included, no additional fetch needed
  }
}

Event Sourcing

Store events as the source of truth.

javascript

class EventStore {
  async saveEvents(aggregateId, events) {
    await this.db.insert({
      aggregateId,
      events,
      version: events.length,
      timestamp: new Date(),
    });
  }

  async getEvents(aggregateId) {
    return await this.db
      .where({ aggregateId })
      .orderBy('version', 'asc');
  }

  async replay(aggregateId) {
    const events = await this.getEvents(aggregateId);

    return events.reduce((state, event) => {
      return this.applyEvent(state, event);
    }, {});
  }
}

// Usage
class Order {
  static create(orderData) {
    return {
      id: generateId(),
      status: 'created',
      events: [{
        type: 'OrderCreated',
        data: orderData,
      }],
    };
  }

  static async fromEvents(eventStore, orderId) {
    const events = await eventStore.getEvents(orderId);
    return events.reduce((order, event) => {
      return Order.apply(order, event);
    }, null);
  }
}

Message Brokers

RabbitMQ

javascript

const amqp = require('amqplib');

async function setupRabbitMQ() {
  const connection = await amqp.connect('amqp://localhost');
  const channel = await connection.createChannel();

  // Declare exchange
  await channel.assertExchange('orders', 'topic', { durable: true });

  // Publish events
  async function publishEvent(routingKey, event) {
    channel.publish(
      'orders',
      routingKey,
      Buffer.from(JSON.stringify(event))
    );
  }

  // Consume events
  async function subscribeEvents(pattern, callback) {
    const queue = await channel.assertQueue('', { exclusive: true });
    await channel.bindQueue(queue.queue, 'orders', pattern);

    channel.consume(queue.queue, (msg) => {
      const event = JSON.parse(msg.content.toString());
      callback(event);
      channel.ack(msg);
    });
  }

  return { publishEvent, subscribeEvents };
}

Kafka

javascript

const { Kafka } = require('kafkajs');

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

const producer = kafka.producer();

async function publishEvent(topic, event) {
  await producer.send({
    topic,
    messages: [{
      key: event.data.id,
      value: JSON.stringify(event),
    }],
  });
}

const consumer = kafka.consumer({ groupId: 'email-service' });

async function subscribeEvents(topic, callback) {
  await consumer.subscribe({ topic, fromBeginning: false });

  await consumer.run({
    eachMessage: async ({ topic, partition, message }) => {
      const event = JSON.parse(message.value.toString());
      await callback(event);
    },
  });
}

AWS SNS/SQS

javascript

const AWS = require('aws-sdk');
const sns = new AWS.SNS();
const sqs = new AWS.SQS();

// Publish to SNS
async function publishEvent(topicArn, event) {
  await sns.publish({
    TopicArn: topicArn,
    Message: JSON.stringify(event),
    MessageAttributes: {
      EventType: {
        DataType: 'String',
        StringValue: event.type,
      },
    },
  }).promise();
}

// Subscribe SQS to SNS
async function subscribeQueue(queueUrl, topicArn) {
  await sns.subscribe({
    Protocol: 'sqs',
    TopicArn: topicArn,
    Endpoint: queueUrl,
  }).promise();
}

// Consume from SQS
async function processMessages(queueUrl, callback) {
  const response = await sqs.receiveMessage({
    QueueUrl: queueUrl,
    MaxNumberOfMessages: 10,
  }).promise();

  for (const message of response.Messages || []) {
    const event = JSON.parse(message.Body);
    await callback(event);

    await sqs.deleteMessage({
      QueueUrl: queueUrl,
      ReceiptHandle: message.ReceiptHandle,
    }).promise();
  }
}

Event Design

Event Naming

javascript

// Good: Past tense, clear
{
  "type": "OrderCreated",
  "type": "PaymentProcessed",
  "type": "UserRegistered"
}

// Bad: Ambiguous
{
  "type": "Order",
  "type": "ProcessPayment",
  "type": "RegisterUser"
}

Event Structure

javascript

// Standard event structure
{
  "id": "evt_123456",
  "type": "OrderCreated",
  "source": "/orders",
  "timestamp": "2024-03-27T10:00:00Z",
  "correlationId": "corr_789",
  "causationId": "cause_456",
  "data": {
    "orderId": "ord_123",
    "userId": "usr_456",
    "total": 99.99,
  },
  "datacontenttype": "application/json"
}

Event Processing

At-Least-Once Delivery

javascript

class EventProcessor {
  async processEvent(event) {
    try {
      await this.handleEvent(event);
      await this.markAsProcessed(event.id);
    } catch (error) {
      // Retry or move to dead letter queue
      await this.handleFailure(event, error);
    }
  }

  async markAsProcessed(eventId) {
    await this.db.insert({
      eventId,
      processedAt: new Date(),
      status: 'processed',
    });
  }

  async handleFailure(event, error) {
    await this.db.insert({
      event,
      error: error.message,
      failedAt: new Date(),
      retryCount: 0,
    });
  }
}

Idempotency

javascript

class IdempotentHandler {
  async handle(event) {
    // Check if already processed
    const processed = await this.db.find({
      eventId: event.id,
    });

    if (processed) {
      console.log('Event already processed, skipping');
      return;
    }

    // Process event
    await this.process(event);

    // Mark as processed
    await this.db.insert({
      eventId: event.id,
      processedAt: new Date(),
    });
  }
}

Sagas

Choreography

javascript

// Order Service
class OrderService {
  async createOrder(orderData) {
    const order = await this.db.create(orderData);

    await eventBus.publish({
      type: 'OrderCreated',
      data: order,
    });

    return order;
  }

  async handlePaymentFailed(event) {
    await this.db.update(event.orderId, {
      status: 'payment_failed',
    });

    await eventBus.publish({
      type: 'OrderCancelled',
      data: { orderId: event.orderId },
    });
  }
}

// Payment Service
class PaymentService {
  constructor() {
    eventBus.subscribe('OrderCreated', this.processPayment);
  }

  async processPayment(event) {
    try {
      await this.charge(event.data);
      await eventBus.publish({
        type: 'PaymentSucceeded',
        data: event.data,
      });
    } catch (error) {
      await eventBus.publish({
        type: 'PaymentFailed',
        data: event.data,
      });
    }
  }
}

Orchestration

javascript

class OrderSaga {
  async execute(orderData) {
    const saga = await this.createSaga(orderData);

    try {
      // Step 1: Create order
      const order = await this.createOrder(saga, orderData);

      // Step 2: Process payment
      const payment = await this.processPayment(saga, order);

      // Step 3: Reserve inventory
      await this.reserveInventory(saga, order);

      // Step 4: Confirm order
      await this.confirmOrder(saga, order);

      await this.completeSaga(saga);
    } catch (error) {
      await this.compensate(saga, error);
    }
  }

  async compensate(saga, error) {
    const executedSteps = saga.executedSteps;

    // Compensate in reverse order
    for (const step of executedSteps.reverse()) {
      await this.compensateStep(step);
    }

    await this.failSaga(saga, error);
  }
}

Testing

Unit Testing

javascript

describe('OrderService', () => {
  it('should publish OrderCreated event', async () => {
    const mockEventBus = {
      publish: jest.fn(),
    };

    const service = new OrderService(mockEventBus);
    await service.createOrder({ userId: '123', total: 99.99 });

    expect(mockEventBus.publish).toHaveBeenCalledWith({
      type: 'OrderCreated',
      data: expect.objectContaining({ userId: '123' }),
    });
  });
});

Best Practices

DOs

Use events for communication between services
Make events immutable
Design events for multiple consumers
Handle failures gracefully
Implement idempotency
Monitor event processing
Version your events

DON'Ts

Use events for synchronous operations
Create tight coupling through events
Ignore event ordering when it matters
Skip error handling
Make events too large
Forget about dead letter queues
Break backward compatibility

Conclusion

Event-driven architecture enables loose coupling and better scalability. Use events for notifications, updates, and integrations. Combine with CQRS for complex systems. Always handle failures and implement proper monitoring. The key is designing clear events and reliable processing.

Event-Driven Architecture Complete Guide

What Is Event-Driven Architecture?

Core Concepts

Events

Event Bus

Event Patterns

Event Notification

Event-Carried State Transfer

Event Sourcing

Message Brokers

RabbitMQ

Kafka

AWS SNS/SQS

Event Design

Event Naming

Event Structure

Event Processing

At-Least-Once Delivery

Idempotency

Sagas

Choreography

Orchestration

Testing

Unit Testing

Best Practices

DOs

DON'Ts

Conclusion

Related essays

Design Tokens: Building Scalable Design Systems

OAuth 2.0: Complete Implementation Guide

SAML: Security Assertion Markup Language Complete Guide