Chapter 74: Event-Driven Architecture & CQRS
Event-Driven Architecture
Event-Driven Architecture (EDA) treats events as first-class citizens in system design. An event represents a significant state change or occurrence in the system, immutable and timestamped.
Core Concepts
Events as First-Class Citizens: Events are the primary means of communication between components. They capture what happened, not what should happen next.
// Event structure
{
"eventId": "evt_12345",
"eventType": "OrderPlaced",
"timestamp": "2026-05-12T10:30:00Z",
"aggregateId": "order_789",
"version": 1,
"data": {
"orderId": "order_789",
"customerId": "cust_456",
"items": [...],
"totalAmount": 99.99
}
}Producers and Consumers
Producers: Components that generate events when state changes occur. They publish events without knowing who will consume them (loose coupling).
Consumers: Components that subscribe to events and react accordingly. Multiple consumers can react to the same event independently.
Producer] -->|OrderPlaced| B[Event Bus] B --> C[Inventory Service
Consumer] B --> D[Notification Service
Consumer] B --> E[Analytics Service
Consumer] B --> F[Shipping Service
Consumer]
Event Bus
The event bus (message broker) acts as the backbone of EDA, routing events from producers to consumers. Common implementations:
- Apache Kafka: High-throughput, distributed log with replay capability
- RabbitMQ: Feature-rich message broker with routing flexibility
- AWS EventBridge: Serverless event bus with schema registry
- Azure Event Hubs: Streaming platform for millions of events/sec
Answer: Implement idempotency keys in events, use transactional outbox pattern for atomic event publishing, maintain consumer offsets with processed event IDs, use event bus features (Kafka transactions, RabbitMQ publisher confirms), and design consumers to be naturally idempotent.
Event Sourcing
Event Sourcing stores all changes to application state as a sequence of events rather than storing only the current state. The current state is derived by replaying events from the beginning.
Store Events, Not State
Traditional systems store current state in a database. Event-sourced systems store the history of state changes:
// Traditional approach - only current state
UPDATE accounts SET balance = 500 WHERE id = 123;
// Event sourcing - append events
INSERT INTO events (aggregate_id, event_type, data) VALUES
('account_123', 'AccountCreated', '{"initialBalance": 1000}'),
('account_123', 'MoneyWithdrawn', '{"amount": 300}'),
('account_123', 'MoneyWithdrawn', '{"amount": 200}');Rebuild State by Replay
Current state is computed by loading all events for an aggregate and applying them sequentially:
class Account {
private string id;
private decimal balance;
private List<Event> changes = new();
public void Apply(Event evt) {
switch (evt) {
case AccountCreated e:
this.id = e.AccountId;
this.balance = e.InitialBalance;
break;
case MoneyWithdrawn e:
this.balance -= e.Amount;
break;
case MoneyDeposited e:
this.balance += e.Amount;
break;
}
}
public static Account Load(string accountId, IEventStore store) {
var events = store.GetEvents(accountId);
var account = new Account();
foreach (var evt in events) {
account.Apply(evt);
}
return account;
}
}Event Store
The event store is an append-only log of events, typically indexed by aggregate ID. Popular implementations:
- EventStoreDB: Purpose-built database for event sourcing
- Kafka: Distributed log with compaction support
- PostgreSQL: JSONB columns with proper indexing
- DynamoDB: With composite keys (aggregateId + version)
Answer: Pros: Complete audit trail, temporal queries (state at any point in time), debugging by replay, natural fit for event-driven systems. Cons: Complexity in querying current state, event schema evolution challenges, storage growth (mitigated by snapshots), eventual consistency for projections, learning curve for teams.
CQRS (Command Query Responsibility Segregation)
CQRS separates read and write operations into different models. Commands modify state, queries read state. Each side can be optimized independently.
Separate Read and Write Models
Write Model (Command Side): Enforces business rules, validates commands, generates events, optimized for consistency and integrity.
Read Model (Query Side): Denormalized views optimized for queries, eventually consistent with write model, can have multiple read models for different use cases.
SQL] D --> F[Read Model 2
Elasticsearch] D --> G[Read Model 3
Redis Cache] end C -->|Events| H[Event Bus] H -->|Update| E H -->|Update| F H -->|Update| G
Implementation Example
// Command side - write operations
public class OrderCommandHandler {
public async Task Handle(PlaceOrderCommand cmd) {
var order = new Order(cmd.OrderId, cmd.CustomerId);
foreach (var item in cmd.Items) {
order.AddItem(item);
}
// Validate business rules
if (order.Total > customer.CreditLimit) {
throw new BusinessRuleViolation("Exceeds credit limit");
}
// Save events
await eventStore.Save(order.Id, order.GetUncommittedEvents());
}
}
// Query side - read operations
public class OrderQueryHandler {
public async Task<OrderView> GetOrder(string orderId) {
// Read from optimized read model
return await readDatabase.Orders
.Include(o => o.Items)
.Include(o => o.Customer)
.FirstOrDefaultAsync(o => o.Id == orderId);
}
public async Task<List<OrderSummary>> GetCustomerOrders(string customerId) {
// Different read model, different schema
return await readDatabase.OrderSummaries
.Where(o => o.CustomerId == customerId)
.OrderByDescending(o => o.OrderDate)
.ToListAsync();
}
}Benefits of CQRS
- Independent Scaling: Scale reads and writes separately
- Optimized Queries: Denormalized read models tailored to UI needs
- Multiple Read Models: Different views for different consumers
- Technology Diversity: Use PostgreSQL for writes, Elasticsearch for search
- Security: Different authorization for commands vs. queries
Answer: Avoid CQRS for simple CRUD applications where reads and writes have similar complexity, when eventual consistency is unacceptable, for small teams unfamiliar with the pattern, or when the added complexity doesn't provide clear benefits. CQRS shines in complex domains with different read/write patterns.
Projections
Projections build read models from events. They subscribe to events and update denormalized views optimized for queries.
Building Read Models from Events
public class OrderProjection {
private readonly IReadDatabase readDb;
public async Task Handle(OrderPlacedEvent evt) {
await readDb.Orders.InsertAsync(new OrderView {
OrderId = evt.OrderId,
CustomerId = evt.CustomerId,
Status = "Pending",
Total = evt.Total,
PlacedAt = evt.Timestamp
});
}
public async Task Handle(OrderShippedEvent evt) {
await readDb.Orders.UpdateAsync(
evt.OrderId,
o => o.Status = "Shipped", o.ShippedAt = evt.Timestamp
);
}
public async Task Handle(PaymentReceivedEvent evt) {
await readDb.Orders.UpdateAsync(
evt.OrderId,
o => o.PaymentStatus = "Paid", o.PaidAt = evt.Timestamp
);
}
}
// Multiple projections from same events
public class AnalyticsProjection {
public async Task Handle(OrderPlacedEvent evt) {
await analyticsDb.Execute(
"UPDATE daily_revenue SET amount = amount + @total WHERE date = @date",
new { total = evt.Total, date = evt.Timestamp.Date }
);
}
}Eventual Consistency
Projections are updated asynchronously, creating eventual consistency between write and read models. Strategies to manage this:
- Version Tracking: Include event version in read models, detect staleness
- Polling: UI polls until expected version appears
- Optimistic UI: Show expected state immediately, reconcile later
- Bounded Staleness: Alert if projection falls too far behind
// Client waits for projection to catch up
public async Task<OrderView> GetOrderWhenReady(string orderId, int expectedVersion) {
const int maxAttempts = 10;
for (int i = 0; i < maxAttempts; i++) {
var order = await queryHandler.GetOrder(orderId);
if (order != null && order.Version >= expectedVersion) {
return order;
}
await Task.Delay(100); // Exponential backoff in production
}
throw new TimeoutException("Projection not ready");
}Rebuilding Projections
One powerful advantage: projections can be rebuilt from scratch by replaying events. Useful for:
- Fixing bugs in projection logic
- Adding new read models
- Schema changes
- Recovering from corruption
Answer: Use dead-letter queues for failed events, implement retry with exponential backoff, track projection position (offset/checkpoint), enable alerting on lag, design projections to be idempotent, use poison message detection (skip after N failures), and maintain multiple projection instances for redundancy.
Event Store Design
The event store is the single source of truth in event-sourced systems. Design considerations are critical for performance and reliability.
Append-Only Storage
Events are never updated or deleted (logical deletion via new events). This ensures:
- Complete audit trail
- No lost history
- Temporal queries possible
- Simplified concurrency (only appends)
CREATE TABLE events (
event_id BIGSERIAL PRIMARY KEY,
aggregate_id VARCHAR(100) NOT NULL,
aggregate_type VARCHAR(100) NOT NULL,
event_type VARCHAR(100) NOT NULL,
event_version INT NOT NULL,
event_data JSONB NOT NULL,
metadata JSONB,
timestamp TIMESTAMPTZ DEFAULT NOW(),
UNIQUE(aggregate_id, event_version)
);
CREATE INDEX idx_events_aggregate ON events(aggregate_id, event_version);
CREATE INDEX idx_events_type ON events(event_type);
CREATE INDEX idx_events_timestamp ON events(timestamp);Snapshots for Performance
As event streams grow, replaying thousands of events becomes slow. Snapshots store aggregate state at a point in time:
CREATE TABLE snapshots (
aggregate_id VARCHAR(100) PRIMARY KEY,
aggregate_type VARCHAR(100) NOT NULL,
version INT NOT NULL,
state JSONB NOT NULL,
timestamp TIMESTAMPTZ DEFAULT NOW()
);
// Load with snapshot
public static Account Load(string accountId, IEventStore store) {
var snapshot = store.GetSnapshot(accountId);
var account = snapshot != null
? Account.FromSnapshot(snapshot)
: new Account();
var events = store.GetEvents(accountId, fromVersion: snapshot?.Version ?? 0);
foreach (var evt in events) {
account.Apply(evt);
}
return account;
}
// Snapshot every N events
if (account.Version % 100 == 0) {
await store.SaveSnapshot(account.Id, account.Version, account.GetState());
}Ordering Guarantees
Event ordering is critical for consistency:
- Per-Aggregate Ordering: Events for same aggregate must be strictly ordered
- Global Ordering: Optional, useful for replaying entire system
- Optimistic Concurrency: Use version numbers to detect conflicts
public async Task Save(string aggregateId, List<Event> events, int expectedVersion) {
using var transaction = await db.BeginTransactionAsync();
// Check current version
var currentVersion = await db.Events
.Where(e => e.AggregateId == aggregateId)
.MaxAsync(e => (int?)e.EventVersion) ?? 0;
if (currentVersion != expectedVersion) {
throw new ConcurrencyException(
$"Expected version {expectedVersion} but found {currentVersion}"
);
}
// Append events
int version = expectedVersion;
foreach (var evt in events) {
version++;
await db.Events.InsertAsync(new EventRecord {
AggregateId = aggregateId,
EventVersion = version,
EventType = evt.GetType().Name,
EventData = JsonSerializer.Serialize(evt),
Timestamp = DateTime.UtcNow
});
}
await transaction.CommitAsync();
}Answer: Use event upcasters to transform old event versions to new ones during replay, store schema version in each event, maintain event upgrade functions, use weak schema (JSON) for flexibility, test upgrades on production event streams before deployment, and consider backward-compatible changes when possible.
Advantages
Complete Audit Trail
Every change is recorded with timestamp, user, and context. Perfect for regulated industries:
// Query: "Who changed this customer's credit limit and when?"
SELECT event_type, event_data->>'oldLimit', event_data->>'newLimit',
metadata->>'userId', timestamp
FROM events
WHERE aggregate_id = 'customer_123'
AND event_type = 'CreditLimitChanged'
ORDER BY timestamp;Temporal Queries
Answer questions about past state: "What was the inventory level last Tuesday at 3pm?"
public async Task<Account> GetAccountStateAt(string accountId, DateTime pointInTime) {
var events = await store.GetEvents(
accountId,
toTimestamp: pointInTime
);
var account = new Account();
foreach (var evt in events) {
account.Apply(evt);
}
return account;
}Replay for Debugging
Reproduce bugs by replaying production events in test environment:
- Extract events for failing aggregate
- Replay in local environment
- Step through with debugger
- Fix bug and verify with same events
Business Intelligence
Events provide rich data for analytics:
- Customer behavior analysis (event patterns)
- Conversion funnels (event sequences)
- A/B testing (compare event streams)
- Predictive modeling (event trends)
Answer: Event sourcing helps with audit requirements (who accessed what, when), but conflicts with right to deletion. Solutions: use crypto-shredding (encrypt personal data, delete key), store PII separately with references in events, use pseudonymization, implement event redaction (replace with "PersonalDataDeleted" event), or use hybrid approach (event sourcing for transactions, separate store for personal data).
Challenges
Eventual Consistency
Read models lag behind write model. Users may not see changes immediately. Mitigation strategies:
- Set user expectations (UI messaging)
- Use optimistic UI updates
- Implement version tracking and polling
- Minimize projection lag (fast processors)
Schema Evolution
Events are immutable, but business requirements change. Strategies:
- Upcasting: Transform old events to new schema on read
- Weak Schema: Use JSON with optional fields
- Multiple Event Versions: Handle v1, v2, v3 in code
- Event Migration: Create new events from old (rare, expensive)
// Upcaster example
public class EventUpgrader {
public Event Upgrade(EventRecord record) {
return record.EventType switch {
"OrderPlaced" when record.Version == 1 =>
UpgradeOrderPlacedV1ToV2(record),
"OrderPlaced" when record.Version == 2 =>
DeserializeOrderPlacedV2(record),
_ => throw new UnknownEventException(record.EventType)
};
}
private OrderPlacedEventV2 UpgradeOrderPlacedV1ToV2(EventRecord record) {
var v1 = JsonSerializer.Deserialize<OrderPlacedEventV1>(record.EventData);
return new OrderPlacedEventV2 {
OrderId = v1.OrderId,
CustomerId = v1.CustomerId,
Items = v1.Items,
ShippingAddress = v1.Address, // Field renamed
PaymentMethod = "Unknown" // New required field, default value
};
}
}Event Versioning
Managing multiple event versions over time:
// Embed version in event type
public abstract class Event {
public int SchemaVersion { get; set; }
}
public class OrderPlacedEvent_V1 : Event {
public OrderPlacedEvent_V1() { SchemaVersion = 1; }
public string OrderId { get; set; }
public string Address { get; set; }
}
public class OrderPlacedEvent_V2 : Event {
public OrderPlacedEvent_V2() { SchemaVersion = 2; }
public string OrderId { get; set; }
public Address ShippingAddress { get; set; } // Structured
public Address BillingAddress { get; set; } // New field
}Storage Growth
Event stores grow indefinitely. Management strategies:
- Snapshots: Reduce need to read old events
- Archiving: Move old events to cold storage
- Event Compaction: Collapse event sequences (carefully)
- Retention Policies: Delete events after N years (compliance permitting)
Query Complexity
Answering queries requires replaying events or maintaining read models. Not suitable for ad-hoc queries across aggregates.
Answer: Don't make external calls during event replay (not idempotent). Options: (1) Process managers that emit commands based on events, making API calls before generating new events. (2) Sagas that coordinate multi-step processes. (3) Event handlers that react to events and call external APIs, storing results as new events. (4) Store "intent events" then have separate processors make calls.
When to Use
Ideal Use Cases
- Complex Domains: DDD aggregates, rich business logic, domain events
- Audit Requirements: Financial systems, healthcare, government
- Event-Heavy Systems: IoT, trading platforms, gaming
- Temporal Analysis: Need to query historical state
- Multiple Read Models: Different consumers need different views
- Collaboration: Multiple users modifying same data
- Debugging Complex Workflows: Replay production issues
When to Avoid
- Simple CRUD: Event sourcing is overkill for basic apps
- Strong Consistency Required: Can't tolerate eventual consistency
- Ad-Hoc Queries: Need flexible querying across all data
- Team Unfamiliarity: Steep learning curve, need training
- Stateless Services: No domain logic, just data pass-through
Hybrid Approaches
Not all-or-nothing. Consider:
- Event sourcing for core domain, CRUD for supporting domains
- CQRS without event sourcing (separate read/write databases)
- Event-driven architecture without event sourcing (events for integration only)
Real-World Examples
- Banking: Account transactions as events (audit trail, temporal queries)
- E-commerce: Order lifecycle events (integration, analytics)
- Healthcare: Patient record changes (compliance, history)
- Gaming: Player actions as events (replay, anti-cheat)
- Logistics: Package tracking events (real-time updates, history)
Answer: Aggregates: Order, Payment, Shipment. Events: OrderPlaced, ItemAdded, PaymentReceived, OrderShipped, OrderDelivered. Event store: PostgreSQL with events table, snapshots every 50 events. Projections: OrderView (current state), OrderHistory (timeline), SalesAnalytics (metrics). Handle distributed transactions with Saga pattern. Scale reads with multiple read models (SQL for orders, Elasticsearch for search, Redis for caching). Use Kafka for event bus with partitioning by order ID.
Production Event Sourcing Patterns
Event sourcing sounds elegant in theory. Here's what you actually encounter in production.
Event Schema Evolution
Events are immutable and live forever. But business requirements change. How do you evolve schemas without breaking replay?
// V1: Original event (2022)
{
"eventId": "evt_001",
"type": "OrderPlaced",
"timestamp": "2022-05-12T10:00:00Z",
"data": {
"orderId": "order_789",
"amount": 100
}
}
// V2: Add new field (2023) - BACKWARD COMPATIBLE
{
"eventId": "evt_002",
"type": "OrderPlaced",
"timestamp": "2023-05-12T10:00:00Z",
"data": {
"orderId": "order_790",
"amount": 150,
"currency": "USD" // NEW FIELD: Old consumers ignore it
}
}
// V3: Renamed field (2024) - BREAKING CHANGE
{
"eventId": "evt_003",
"type": "OrderPlaced",
"timestamp": "2024-05-12T10:00:00Z",
"data": {
"orderId": "order_791",
"total_amount": 200, // RENAMED from "amount"
"currency": "USD"
}
}Handling Breaking Changes: Upcasters
When you rename or restructure fields, use upcasters to transform old events to new schema at read time.
public class OrderPlacedUpcaster {
public OrderPlacedV3 Upcast(EventRecord record) {
return record.SchemaVersion switch {
// V1 → V3: Add missing fields with defaults
1 => new OrderPlacedV3 {
OrderId = record.Data["orderId"],
TotalAmount = record.Data["amount"],
Currency = "USD" // Default for old events
},
// V2 → V3: Rename "amount" to "total_amount"
2 => new OrderPlacedV3 {
OrderId = record.Data["orderId"],
TotalAmount = record.Data["amount"], // Rename
Currency = record.Data["currency"]
},
// V3: Already current
3 => JsonSerializer.Deserialize<OrderPlacedV3>(record.Data)
};
}
}Schema Evolution Strategies
- Backward compatible: Add optional fields (safest, no migration)
- Upcasters: Transform old events at read time (transparent to consumers)
- Event migration: Rewrite old events to new schema (rare, expensive, breaks audit trail)
- Multi-version handlers: Code handles V1, V2, V3 simultaneously (complex but flexible)
Snapshot Frequency Calculation
How often should you snapshot? Math it out.
GIVEN:
- Replaying 10,000 events takes 5 seconds
- Your read latency requirement: <100ms
- Each event replay takes: 5 sec / 10,000 = 0.5ms per event
CALCULATION:
- To stay under 100ms, replay max 200 events
- 100ms / 0.5ms per event = 200 events
CONCLUSION:
- Snapshot every 200 events
- Load latest snapshot + replay events since snapshot
- Read latency: snapshot load (10ms) + 200 events × 0.5ms = 110ms ≈ 100ms targetTrade-off: More frequent snapshots = faster reads but more storage. Less frequent = slower reads but less storage overhead.
Event Store Sizing
Event stores grow forever. Here's how to estimate storage.
E-COMMERCE EXAMPLE:
Daily volume: 1M events/day
Average event size: 500 bytes (JSON payload + metadata)
Retention: 3 years (required for compliance)
RAW STORAGE:
- Per day: 1M events × 500 bytes = 500 MB/day
- Per year: 500 MB × 365 days = 182.5 GB/year
- 3 years: 182.5 GB × 3 = 547.5 GB ≈ 548 GB
WITH COMPRESSION (Snappy 3x):
- Compressed: 548 GB / 3 = 183 GB
WITH SNAPSHOTS:
- Snapshot every 200 events, 1KB per snapshot
- Total aggregates: 10M (e.g., 10M orders)
- Snapshots: 10M × 1KB = 10 GB
- Total: 183 GB (events) + 10 GB (snapshots) = 193 GB
CONCLUSION:
- 3 years of event history = ~200 GB
- Cost (S3): 200 GB × $0.023/GB = $4.60/month
- Cost (DynamoDB): Much higher, use S3 for archivalProjection Rebuilding
One of event sourcing's killer features: rebuild ANY view of your data by replaying events.
When to Rebuild Projections
- Bug fix: Projection had wrong logic, replay with corrected code
- New view: Add a new read model (e.g., analytics dashboard)
- Schema change: Denormalized view needs new fields
- Corruption: Database crashed, restore from events
Replay Performance
At 1M events/sec replay speed:
- 10M events: 10 seconds
- 100M events: 100 seconds (1.7 minutes)
- 1B events: 1,000 seconds (17 minutes)
Parallel Replay
Partition events by aggregate ID, replay in parallel:
- 1B events across 100 workers = 170 seconds (2.8 minutes)
- Requires partitionable projections (no cross-aggregate joins)
// Rebuilding a projection
public async Task RebuildOrderSummaryProjection() {
// 1. Create new empty projection table
await db.Execute("CREATE TABLE order_summary_v2 (...)");
// 2. Replay all OrderPlaced and OrderShipped events
var eventStream = eventStore.GetAllEvents(
eventTypes: ["OrderPlaced", "OrderShipped", "OrderDelivered"]
);
long processedCount = 0;
await foreach (var evt in eventStream) {
await orderSummaryProjection.Handle(evt);
processedCount++;
if (processedCount % 10000 == 0) {
Console.WriteLine($"Processed {processedCount} events...");
}
}
// 3. Atomic swap: replace old projection with new
await db.Execute("DROP TABLE order_summary_old");
await db.Execute("ALTER TABLE order_summary RENAME TO order_summary_old");
await db.Execute("ALTER TABLE order_summary_v2 RENAME TO order_summary");
Console.WriteLine($"Rebuild complete: {processedCount} events replayed");
}Zero-Downtime Projection Updates
How to update a projection's logic without taking the system down:
- Deploy new projection handler (v2) in parallel with old (v1)
- v2 starts replaying events from beginning to catch up
- v1 continues serving reads from current projection
- When v2 catches up (same event position as v1), switch reads to v2
- Decommission v1 after validation period
Lag monitoring: Track event position (offset) of each projection. Alert if v2 lags >1 hour behind v1.
Summary
Event-Driven Architecture, Event Sourcing, and CQRS are powerful patterns for building scalable, auditable systems. Key takeaways:
- Events are immutable facts capturing state changes
- Event sourcing stores events, not state, enabling temporal queries and replay
- CQRS separates read and write models for independent optimization
- Projections build eventual-consistent read models from events
- Trade complexity for auditability, scalability, and flexibility
- Not suitable for simple CRUD or strong consistency requirements
- Snapshots and proper indexing are critical for performance
- Event versioning and schema evolution require careful planning
- Schema evolution: Use upcasters to transform old events at read time. Backward-compatible changes (add optional fields) are safest.
- Snapshot frequency: Snapshot every N events where N = target_latency / time_per_event. Balance read performance vs storage cost.
- Projection rebuilding: Replay 1B events in 17 minutes (single-threaded) or 3 minutes (parallelized). Enables bug fixes and new views.
- Event store sizing: 1M events/day × 500 bytes × 3 years = ~200 GB with compression. Use S3 for archival, DynamoDB/PostgreSQL for hot data.
These patterns excel in complex domains with audit requirements, event-heavy workloads, and diverse read patterns. They require team buy-in and careful design, but deliver significant benefits when applied appropriately.