Event-Driven Architecture

What is Event-Driven Architecture?

Event-Driven Architecture (EDA) is an architectural pattern where systems communicate through events - immutable records of something that happened in the past. Unlike request-response patterns where components directly call each other, event-driven systems use an event bus or message broker to decouple producers and consumers.

The Fundamental Shift

Traditional architectures use synchronous request-response patterns: Service A calls Service B and waits for a response. Event-driven architectures flip this: Service A publishes an event (“OrderCreated”) and continues. Service B (and potentially C, D, E) subscribe to that event and react asynchronously.

Key Characteristics:

• Events are immutable facts - They represent something that already happened (past tense: “OrderCreated”, not “CreateOrder”)
• Decoupled communication - Producers don’t know who consumes their events
• Asynchronous processing - Consumers process events at their own pace
• Reactive - Systems react to events rather than being proactively called
• Scalable - Add consumers without modifying producers

Simple Analogy

Think of events like news articles:

• Event = News article (“Earthquake happened”)
• Producer = Reporter (writes article)
• Event Bus = Newspaper (distributes articles)
• Consumer = Reader (reads article, reacts)

Multiple readers can read same article. Readers don’t know about each other. Articles are immutable (can’t change history).

Key Principles

1. Events are Immutable - Can’t change past
2. Events are Facts - Something happened
3. Decoupled - Producers don’t know consumers
4. Asynchronous - Non-blocking
5. Reactive - Consumers react to events

---

Event Sourcing

Event sourcing is a pattern where you store all changes to application state as a sequence of events, rather than storing the current state. The current state is derived by replaying all events from the beginning (or from a snapshot).

The Core Idea

Instead of updating a record in a database (overwriting the previous state), you append a new event to an event log. To get the current state, you replay all events. To see the state at any point in time, you replay events up to that point.

Why This Matters:

Traditional databases store the current state - you can see what something is now, but not how it got there. Event sourcing stores the history - you can see every change that ever happened, when it happened, and why it happened (if events include context).

This enables powerful capabilities: audit trails (see who changed what and when), time travel (see state at any historical point), replay (rebuild state from scratch), and debugging (replay events to understand what happened).

Traditional Approach (State-Based)

Problem: Lost history! Can’t see how state changed.

Event Sourcing Approach

Benefits:

• Full history - Every change recorded
• Replay - Rebuild state from events
• Audit trail - See what happened when
• Time travel - See state at any point

Event Sourcing Example

💡 Tip: Click dropdown to switch between languages

Python

Python

"event_sourcing.py

from typing import List, Dict, Any
from datetime import datetime
from dataclasses import dataclass
from enum import Enum

class EventType(Enum):
    USER_CREATED = "user.created"
    BALANCE_ADDED = "balance.added"
    BALANCE_SUBTRACTED = "balance.subtracted"

@dataclass
class Event:
    """Base event class"""
    event_id: str
    event_type: EventType
    aggregate_id: str
    data: Dict[str, Any]
    timestamp: datetime
    version: int

class EventStore:
    """Event store - append-only log"""

    def __init__(self):
        self.events: List[Event] = []

    def append(self, event: Event):
        """Append event (immutable)"""
        self.events.append(event)

    def get_events(self, aggregate_id: str) -> List[Event]:
        """Get all events for aggregate"""
        return [e for e in self.events if e.aggregate_id == aggregate_id]

    def replay(self, aggregate_id: str) -> Dict[str, Any]:
        """Replay events to rebuild state"""
        events = self.get_events(aggregate_id)
        state = {}

        for event in events:
            if event.event_type == EventType.USER_CREATED:
                state['id'] = event.data['user_id']
                state['balance'] = 0
                state['status'] = 'active'
            elif event.event_type == EventType.BALANCE_ADDED:
                state['balance'] = state.get('balance', 0) + event.data['amount']
            elif event.event_type == EventType.BALANCE_SUBTRACTED:
                state['balance'] = state.get('balance', 0) - event.data['amount']

        return state

class UserAggregate:
    """User aggregate with event sourcing"""

    def __init__(self, event_store: EventStore):
        self.event_store = event_store

    def create_user(self, user_id: str, initial_balance: int = 0):
        """Create user (command)"""
        event = Event(
            event_id=f"evt-{datetime.now().timestamp()}",
            event_type=EventType.USER_CREATED,
            aggregate_id=user_id,
            data={'user_id': user_id, 'initial_balance': initial_balance},
            timestamp=datetime.now(),
            version=1
        )
        self.event_store.append(event)

    def add_balance(self, user_id: str, amount: int):
        """Add balance (command)"""
        event = Event(
            event_id=f"evt-{datetime.now().timestamp()}",
            event_type=EventType.BALANCE_ADDED,
            aggregate_id=user_id,
            data={'amount': amount},
            timestamp=datetime.now(),
            version=self._get_next_version(user_id)
        )
        self.event_store.append(event)

    def get_user_state(self, user_id: str) -> Dict[str, Any]:
        """Get current state (query) - replay events"""
        return self.event_store.replay(user_id)

    def _get_next_version(self, aggregate_id: str) -> int:
        """Get next version number"""
        events = self.event_store.get_events(aggregate_id)
        return len(events) + 1

# Usage
event_store = EventStore()
user_aggregate = UserAggregate(event_store)

# Create user
user_aggregate.create_user('user-123', initial_balance=0)

# Add balance
user_aggregate.add_balance('user-123', 50)
user_aggregate.add_balance('user-123', 100)
user_aggregate.add_balance('user-123', -30)

# Get current state (replay events)
state = user_aggregate.get_user_state('user-123')
print(f"Current balance: {state['balance']}")  # 120

# Get full history
events = event_store.get_events('user-123')
for event in events:
    print(f"{event.event_type}: {event.data}")

Java

"EventSourcing.java

import java.time.LocalDateTime;
import java.util.*;

class Event {
    private final String eventId;
    private final EventType eventType;
    private final String aggregateId;
    private final Map<String, Object> data;
    private final LocalDateTime timestamp;
    private final int version;

    // Constructor, getters...
}

enum EventType {
    USER_CREATED, BALANCE_ADDED, BALANCE_SUBTRACTED
}

class EventStore {
    private final List<Event> events = new ArrayList<>();

    public void append(Event event) {
        events.add(event);
    }

    public List<Event> getEvents(String aggregateId) {
        return events.stream()
            .filter(e -> e.getAggregateId().equals(aggregateId))
            .collect(Collectors.toList());
    }

    public Map<String, Object> replay(String aggregateId) {
        List<Event> events = getEvents(aggregateId);
        Map<String, Object> state = new HashMap<>();

        for (Event event : events) {
            switch (event.getEventType()) {
                case USER_CREATED:
                    state.put("id", event.getData().get("user_id"));
                    state.put("balance", 0);
                    state.put("status", "active");
                    break;
                case BALANCE_ADDED:
                    state.put("balance",
                        (Integer) state.getOrDefault("balance", 0) +
                        (Integer) event.getData().get("amount"));
                    break;
                case BALANCE_SUBTRACTED:
                    state.put("balance",
                        (Integer) state.getOrDefault("balance", 0) -
                        (Integer) event.getData().get("amount"));
                    break;
            }
        }

        return state;
    }
}

class UserAggregate {
    private final EventStore eventStore;

    public void createUser(String userId, int initialBalance) {
        Event event = new Event(
            UUID.randomUUID().toString(),
            EventType.USER_CREATED,
            userId,
            Map.of("user_id", userId, "initial_balance", initialBalance),
            LocalDateTime.now(),
            1
        );
        eventStore.append(event);
    }

    public Map<String, Object> getUserState(String userId) {
        return eventStore.replay(userId);
    }
}

---

CQRS (Command Query Responsibility Segregation)

CQRS (Command Query Responsibility Segregation) is a pattern that separates the read and write models of your application. Instead of using the same data model for both reading and writing, you have separate models optimized for each operation.

The Core Concept

In traditional applications, you use the same database schema for both:

• Commands (writes) - Create, Update, Delete operations
• Queries (reads) - Get, List, Search operations

CQRS recognizes that these operations have different requirements:

• Commands need consistency, validation, and business logic
• Queries need speed, denormalization, and read optimization

The Separation:

• Write Model - Handles commands, enforces business rules, stores normalized data, optimized for consistency
• Read Model - Handles queries, stores denormalized data, optimized for read performance

The read model is typically updated asynchronously by consuming events from the write model, ensuring eventual consistency.

Benefits:

• Optimize independently - Write for consistency, read for speed
• Scale independently - More read replicas
• Different databases - SQL for writes, NoSQL for reads
• Performance - Read model optimized for queries

CQRS Implementation

💡 Tip: Click dropdown to switch between languages

Python

Python

"cqrs.py

class WriteModel:
    """Write model - handles commands"""

    def __init__(self, event_store):
        self.event_store = event_store

    def create_user(self, user_id: str, email: str):
        """Command: Create user"""
        event = Event(
            event_id=generate_id(),
            event_type=EventType.USER_CREATED,
            aggregate_id=user_id,
            data={'user_id': user_id, 'email': email},
            timestamp=datetime.now(),
            version=1
        )
        self.event_store.append(event)
        # Publish event for read model update
        event_bus.publish(event)

class ReadModel:
    """Read model - optimized for queries"""

    def __init__(self):
        self.users = {}  # Denormalized, optimized for reads

    def handle_user_created(self, event: Event):
        """Projection: Update read model"""
        self.users[event.aggregate_id] = {
            'id': event.data['user_id'],
            'email': event.data['email'],
            'created_at': event.timestamp
        }

    def get_user(self, user_id: str) -> Dict:
        """Query: Get user (fast!)"""
        return self.users.get(user_id)

    def list_users(self) -> List[Dict]:
        """Query: List users (fast!)"""
        return list(self.users.values())

# Usage
event_store = EventStore()
write_model = WriteModel(event_store)
read_model = ReadModel()

# Subscribe read model to events
event_bus.subscribe(EventType.USER_CREATED, read_model.handle_user_created)

# Write (command)
write_model.create_user('user-123', '[email protected]')

# Read (query) - fast, optimized
user = read_model.get_user('user-123')
users = read_model.list_users()

Java

"CQRS.java

class WriteModel {
    private final EventStore eventStore;
    private final EventBus eventBus;

    public void createUser(String userId, String email) {
        Event event = new Event(
            UUID.randomUUID().toString(),
            EventType.USER_CREATED,
            userId,
            Map.of("user_id", userId, "email", email),
            LocalDateTime.now(),
            1
        );
        eventStore.append(event);
        eventBus.publish(event);
    }
}

class ReadModel {
    private final Map<String, UserDTO> users = new HashMap<>();

    public void handleUserCreated(Event event) {
        users.put(event.getAggregateId(), new UserDTO(
            event.getData().get("user_id"),
            event.getData().get("email"),
            event.getTimestamp()
        ));
    }

    public UserDTO getUser(String userId) {
        return users.get(userId);
    }

    public List<UserDTO> listUsers() {
        return new ArrayList<>(users.values());
    }
}

TypeScript

cqrs.ts

interface Event {
    eventId: string;
    eventType: EventType;
    aggregateId: string;
    data: Record<string, any>;
    timestamp: Date;
    version: number;
}

enum EventType {
    USER_CREATED = "user.created"
}

class WriteModel {
    // Write model - handles commands
    constructor(
        private eventStore: EventStore,
        private eventBus: EventBus
    ) {}

    createUser(userId: string, email: string): void {
        // Command: Create user
        const event: Event = {
            eventId: `evt-${Date.now()}`,
            eventType: EventType.USER_CREATED,
            aggregateId: userId,
            data: { user_id: userId, email },
            timestamp: new Date(),
            version: 1
        };
        this.eventStore.append(event);
        // Publish event for read model update
        this.eventBus.publish(event);
    }
}

class ReadModel {
    // Read model - optimized for queries
    private users: Map<string, UserDTO> = new Map();

    handleUserCreated(event: Event): void {
        // Projection: Update read model
        this.users.set(event.aggregateId, {
            id: event.data.user_id,
            email: event.data.email,
            createdAt: event.timestamp
        });
    }

    getUser(userId: string): UserDTO | undefined {
        // Query: Get user (fast!)
        return this.users.get(userId);
    }

    listUsers(): UserDTO[] {
        // Query: List users (fast!)
        return Array.from(this.users.values());
    }
}

interface UserDTO {
    id: string;
    email: string;
    createdAt: Date;
}

// Usage
const eventStore = new EventStore();
const eventBus = new EventBus();
const writeModel = new WriteModel(eventStore, eventBus);
const readModel = new ReadModel();

// Subscribe read model to events
eventBus.subscribe(EventType.USER_CREATED, (event) => {
    readModel.handleUserCreated(event);
});

// Write (command)
writeModel.createUser('user-123', '[email protected]');

// Read (query) - fast, optimized
const user = readModel.getUser('user-123');
const users = readModel.listUsers();

C++

cqrs.cpp

#include <string>
#include <map>
#include <vector>
#include <functional>

class WriteModel {
    // Write model - handles commands
private:
    EventStore& eventStore;
    EventBus& eventBus;

public:
    WriteModel(EventStore& eventStore, EventBus& eventBus)
        : eventStore(eventStore), eventBus(eventBus) {}

    void createUser(const std::string& userId, const std::string& email) {
        // Command: Create user
        Event event;
        event.eventId = "evt-" + std::to_string(std::time(nullptr));
        event.eventType = EventType::USER_CREATED;
        event.aggregateId = userId;
        event.data["user_id"] = userId;
        event.data["email"] = email;
        event.timestamp = std::chrono::system_clock::now();
        event.version = 1;

        eventStore.append(event);
        // Publish event for read model update
        eventBus.publish(event);
    }
};

class ReadModel {
    // Read model - optimized for queries
private:
    std::map<std::string, UserDTO> users;

public:
    void handleUserCreated(const Event& event) {
        // Projection: Update read model
        users[event.aggregateId] = UserDTO{
            event.data.at("user_id"),
            event.data.at("email"),
            event.timestamp
        };
    }

    UserDTO* getUser(const std::string& userId) {
        // Query: Get user (fast!)
        auto it = users.find(userId);
        return (it != users.end()) ? &it->second : nullptr;
    }

    std::vector<UserDTO> listUsers() {
        // Query: List users (fast!)
        std::vector<UserDTO> result;
        for (const auto& pair : users) {
            result.push_back(pair.second);
        }
        return result;
    }
};

// Usage
EventStore eventStore;
EventBus eventBus;
WriteModel writeModel(eventStore, eventBus);
ReadModel readModel;

// Subscribe read model to events
eventBus.subscribe(EventType::USER_CREATED, [&readModel](const Event& event) {
    readModel.handleUserCreated(event);
});

// Write (command)
writeModel.createUser("user-123", "[email protected]");

// Read (query) - fast, optimized
auto user = readModel.getUser("user-123");
auto users = readModel.listUsers();

C#

CQRS.cs

using System;
using System.Collections.Generic;

public class WriteModel {
    // Write model - handles commands
    private readonly EventStore eventStore;
    private readonly EventBus eventBus;

    public WriteModel(EventStore eventStore, EventBus eventBus) {
        this.eventStore = eventStore;
        this.eventBus = eventBus;
    }

    public void CreateUser(string userId, string email) {
        // Command: Create user
        var eventData = new Event {
            EventId = $"evt-{DateTimeOffset.Now.ToUnixTimeMilliseconds()}",
            EventType = EventType.UserCreated,
            AggregateId = userId,
            Data = new Dictionary<string, object> {
                ["user_id"] = userId,
                ["email"] = email
            },
            Timestamp = DateTime.Now,
            Version = 1
        };
        eventStore.Append(eventData);
        // Publish event for read model update
        eventBus.Publish(eventData);
    }
}

public class ReadModel {
    // Read model - optimized for queries
    private readonly Dictionary<string, UserDTO> users = new Dictionary<string, UserDTO>();

    public void HandleUserCreated(Event eventData) {
        // Projection: Update read model
        users[eventData.AggregateId] = new UserDTO {
            Id = eventData.Data["user_id"].ToString(),
            Email = eventData.Data["email"].ToString(),
            CreatedAt = eventData.Timestamp
        };
    }

    public UserDTO GetUser(string userId) {
        // Query: Get user (fast!)
        return users.GetValueOrDefault(userId);
    }

    public List<UserDTO> ListUsers() {
        // Query: List users (fast!)
        return new List<UserDTO>(users.Values);
    }
}

public class UserDTO {
    public string Id { get; set; }
    public string Email { get; set; }
    public DateTime CreatedAt { get; set; }
}

// Usage
var eventStore = new EventStore();
var eventBus = new EventBus();
var writeModel = new WriteModel(eventStore, eventBus);
var readModel = new ReadModel();

// Subscribe read model to events
eventBus.Subscribe(EventType.UserCreated, readModel.HandleUserCreated);

// Write (command)
writeModel.CreateUser("user-123", "[email protected]");

// Read (query) - fast, optimized
var user = readModel.GetUser("user-123");
var users = readModel.ListUsers();

---

Event Replay

Rebuild state by replaying events.

Use Cases

1. Rebuild Read Model - After crash or migration
2. Debugging - See state at any point
3. Testing - Replay events in test
4. Time Travel - See past state

Snapshot Optimization

Problem: Replaying millions of events is slow!

Solution: Snapshots!

Strategy:

• Create snapshot every N events (e.g., 10,000)
• Replay from latest snapshot
• Much faster!

---

Key Takeaways

Event Sourcing

Store all changes as events. State = replay of events. Enables audit trail, time travel, replay.

CQRS

Separate read and write models. Optimize independently. Scale independently. Better performance.

Event Replay

Replay events to rebuild state. Use snapshots for performance. Enables debugging, testing, time travel.

Events are Facts

Events represent something that happened. Immutable. Past tense. Full history of changes.

---

Next Steps

• Learn Saga Pattern - distributed transaction management
• Master Idempotency - handling duplicate events
• Review Message Queues - events use message queues