CQRS & Event Sourcing: Separating Reads from Writes

SELECT user_id, COUNT(*) as orders, SUM(amount) as total
FROM orders
WHERE created_at > NOW() - INTERVAL 30 DAYS
GROUP BY user_id
ORDER BY total DESC

Write database: Stores actual order data
Read database: Same database (locked during query)
Problem: Reads block writes

Write database: Orders table (optimized for inserts)
  ├─ Fast writes (100ms per order)
  └─ Emits: OrderCreated event

Event stream: Kafka (durable)
Read database: Materialized view (pre-computed)
  ├─ user_id, order_count, total_amount (pre-aggregated)
  ├─ Updated asynchronously (1 second behind)
  └─ Can't be locked (separate database)
Result:
- Write: Still 100ms (unaffected)
- Read: Now instant (pre-computed, < 10ms)
- Dashboard: Runs every 100ms without any impact

Write side:
User → Command → Domain Model → Update DB
  └─ Emit event

Read side:
  ↑ Event triggers update of read database
User → Query → Read DB (optimized for queries)
Example:
- Write DB: Orders (normalized)
- Event: OrderCreated
- Read DB: DailyStats (denormalized aggregates)

Don't store current state.
Store all events.
Derive state from replaying events.

Events:
- AccountCreated
- MoneyDeposited
- MoneyWithdrawn
- FeeCharged
Current state = Replay all events from start
Benefits:
- Audit trail (know exactly what happened)
- Can "time travel" (what was state on date X?)
- Can rebuild read model (replay all events to new schema)

Build a CQRS + Event Sourcing system for [USE CASE]:

## Use Case
What are you building?
Example: Financial account system (deposits, withdrawals, transfers)
Read/write ratio:
- Reads per second: [X]
- Writes per second: [Y]
Example: 1000 reads/sec, 100 writes/sec (10:1 ratio)
If ratio < 5:1 → Don't use CQRS (single database is fine)
If ratio > 5:1 → CQRS might help
## Write Model (Command Side)
What commands (writes) does the system support?
1. CreateAccount
   - Input: account_holder, initial_balance
   - Validation: Name not empty, balance >= 0
   - Side effect: Emit AccountCreated event
2. Deposit
   - Input: account_id, amount
   - Validation: amount > 0, account exists
   - Side effect: Emit MoneyDeposited event
3. Withdraw
   - Input: account_id, amount
   - Validation: amount > 0, account exists, balance >= amount
   - Side effect: Emit MoneyWithdrawn event or WithdrawalFailed event
4. Transfer
   - Input: from_account_id, to_account_id, amount
   - Validation: Both accounts exist, from has balance, amount > 0
   - Side effect: Emit MoneyTransferred event
## Event Catalog
Define all events:
1. AccountCreated
   - Fields: account_id, account_holder, initial_balance, created_at
   - Triggering command: CreateAccount
   - Version: 1
2. MoneyDeposited
   - Fields: account_id, amount, deposited_at, transaction_id
   - Triggering command: Deposit
   - Version: 1
3. MoneyWithdrawn
   - Fields: account_id, amount, withdrawn_at, transaction_id
   - Triggering command: Withdraw
   - Version: 1
4. WithdrawalFailed
   - Fields: account_id, amount, reason, attempted_at, transaction_id
   - Triggering command: Withdraw (but validation failed)
   - Version: 1
5. MoneyTransferred
   - Fields: from_account, to_account, amount, transferred_at, transaction_id
   - Triggering command: Transfer
   - Version: 1
6. MoneyTransferFailed
   - Fields: from_account, to_account, amount, reason, attempted_at
   - Version: 1
7. FeeCharged
   - Fields: account_id, fee_amount, reason, charged_at
   - Triggering command: (System/scheduled job)
   - Version: 1
## Event Store
Store all events:
```sql
CREATE TABLE event_store (
  event_id UUID PRIMARY KEY,
  event_type VARCHAR(100),  -- AccountCreated, MoneyDeposited, etc
  aggregate_id UUID,         -- account_id (partition key)
  version INT,               -- Which version of the event
  payload JSON,              -- Full event data
  timestamp TIMESTAMP,
  created_at TIMESTAMP,

  -- Ensure ordering per aggregate
  UNIQUE(aggregate_id, version)
);
-- Index for querying events
CREATE INDEX idx_aggregate ON event_store(aggregate_id, version);

Query: "What's the current balance of account 123?"

Read model table:
├─ account_id (indexed)
├─ current_balance
├─ last_updated_at
├─ transaction_count
Updated by: MoneyDeposited, MoneyWithdrawn events
Latency: 1 second behind (OK for balance)

Query: "Show me last 100 transactions for account 123"

Read model table:
├─ account_id (indexed)
├─ transaction_id
├─ type (deposit, withdrawal, transfer)
├─ amount
├─ timestamp (indexed for sorting)
├─ description
Updated by: All transaction events
Latency: Instant (materialized view)

Query: "Show me daily summary for account 123 on date 2025-01-01"

Read model table:
├─ account_id
├─ date (indexed)
├─ opening_balance
├─ deposits_count, deposits_total
├─ withdrawals_count, withdrawals_total
├─ fees_count, fees_total
├─ closing_balance
Updated by: All transaction events (batch updated daily)
Latency: 1 day behind (acceptable)

Event emitted
  ↓
Event stored in event store (write DB)
  ↓
Event published to event stream (Kafka/RabbitMQ)
  ↓
Read model updater consumes event
  ↓
Update read model (separate database)
  ↓
Read model is now eventually consistent
  (Usually < 1 second lag)

// When command executes
function deposit(accountId, amount) {
  // 1. Update write model (event store)
  const event = {
    event_id: uuid(),
    event_type: 'MoneyDeposited',
    aggregate_id: accountId,
    version: 5,
    payload: { account_id: accountId, amount, deposited_at: now() }
  };

  eventStore.insert(event);

  // 2. Publish to event stream
  eventBus.publish('MoneyDeposited', event);

  // 3. Return immediately (don't wait for read model update)
  return { success: true, event_id: event.id };
}

// Separate process: Read model updater
eventBus.subscribe('MoneyDeposited', async (event) => {
  const { account_id, amount } = event.payload;

  // Update read model
  db.query(
    'UPDATE account_balance SET current_balance = current_balance + ?, last_updated = NOW() WHERE account_id = ?',
    [amount, account_id]
  );
});

Want to change read model schema?
No problem:

1. Iterate through event store (from start)
2. Replay all events into new schema
3. Validate against old schema
4. Switch to new schema
Example:
Old read model: DailyStats
New read model: HourlyStats
Iterate events:
- Replay all MoneyDeposited, MoneyWithdrawn events
- Compute hourly aggregates
- New schema ready in minutes (not weeks)

---
## Real Example: Bank Account (Full CQRS)
**Scenario:** User checks balance, deposits $100, transfers $50
**Write Model (Event Store):**

**Read Model: Account Balance (denormalized)**

**Read Model: Transaction History**

**Key insight:**
- Write model: Pure event store (immutable log)
- Read models: Derived from events (updatable, denormalized)
- Can rebuild read models by replaying events
- Can time-travel: "What was balance on 10:05?". Answer: $500

---
## Your Next Step: Audit Your System
For your system:
1. **Read/write ratio:** [Reads per sec] / [Writes per sec]
   If < 5:1 → Don't use CQRS
   If > 5:1 → Consider CQRS
2. **Read complexity:** [How complex is your heaviest query?]
   If simple (< 100ms) → Don't need CQRS
   If complex (> 1 second) → CQRS helps
3. **Consistency requirement:** [Do you need instant consistency?]
   If yes → Traditional database
   If no (1-5 sec lag OK) → CQRS works
4. **Scale target:** [How many queries/second?]
   If < 1000 → Traditional database fine
   If > 10,000 → CQRS needed
Based on answers:
- If low ratio + simple queries → Stick with traditional
- If high ratio + complex queries → Implement CQRS
---