Event Sourcing and CQRS: Building Scalable Event-Driven Systems

Traditional CRUD applications have a dirty secret: they delete history. When you update a user’s email, the old email vanishes forever. When you change an order’s status, there’s no record of when or why it changed.

Event sourcing flips this model on its head. Instead of storing current state, you store every change as an immutable event. Want to know what happened? Replay the events. Need to debug a production issue? Look at the exact sequence of events that led to it. Building a new feature? Create a new view of the same data without changing a thing.

This guide covers event sourcing and CQRS (Command Query Responsibility Segregation) — two patterns that work together to build systems that scale, audit, and evolve gracefully. For related architecture patterns, see my guide on building RESTful APIs with Go and Chi.

The Problem with CRUD

Consider a typical e-commerce order:

-- Traditional approach: overwrite current state
UPDATE orders SET status = 'shipped' WHERE id = 123;

This single line destroys information. You can’t answer:

• When was the order shipped?
• Who marked it as shipped?
• What was the previous status?
• How long did fulfillment take?

You might add a status_history table, but now you’re building event logging on top of CRUD — why not start with events?

Event Sourcing Fundamentals

Event sourcing stores every state change as an event:

// Instead of storing current state
interface Order {
  id: string;
  status: 'pending' | 'paid' | 'shipped' | 'delivered';
  total: number;
}

// Store events that led to this state
interface OrderEvent {
  type: 'OrderCreated' | 'PaymentReceived' | 'OrderShipped' | 'OrderDelivered';
  orderId: string;
  timestamp: Date;
  data: Record<string, any>;
  metadata: {
    userId: string;
    correlationId: string;
  };
}

Event Store Design

The event store is an append-only log:

package eventsourcing

import (
    "encoding/json"
    "time"
)

type Event struct {
    ID          string          `json:"id"`
    Type        string          `json:"type"`
    AggregateID string          `json:"aggregateId"`
    Version     int             `json:"version"`
    Timestamp   time.Time       `json:"timestamp"`
    Data        json.RawMessage `json:"data"`
    Metadata    Metadata        `json:"metadata"`
}

type Metadata struct {
    UserID       string `json:"userId"`
    CorrelationID string `json:"correlationId"`
    CausationID   string `json:"causationId"`
}

type EventStore interface {
    Append(aggregateID string, events []Event, expectedVersion int) error
    GetEvents(aggregateID string, afterVersion int) ([]Event, error)
    Subscribe(handler func(Event)) error
}

Rebuilding State from Events

Current state is derived by replaying events:

type Order struct {
    ID      string
    Status  string
    Total   float64
    Version int
}

func (o *Order) Apply(event Event) error {
    switch event.Type {
    case "OrderCreated":
        var data struct {
            Items []Item `json:"items"`
        }
        json.Unmarshal(event.Data, &data)
        o.ID = event.AggregateID
        o.Status = "pending"
        o.Total = calculateTotal(data.Items)
        
    case "PaymentReceived":
        o.Status = "paid"
        
    case "OrderShipped":
        var data struct {
            TrackingNumber string `json:"trackingNumber"`
            Carrier        string `json:"carrier"`
        }
        json.Unmarshal(event.Data, &data)
        o.Status = "shipped"
        
    case "OrderDelivered":
        o.Status = "delivered"
    }
    
    o.Version = event.Version
    return nil
}

func RebuildOrder(events []Event) *Order {
    order := &Order{}
    for _, event := range events {
        order.Apply(event)
    }
    return order
}

CQRS: Separate Reads from Writes

Event sourcing pairs naturally with CQRS — splitting your system into:

• Command side: Handles writes, validates business rules, emits events
• Query side: Handles reads, builds optimized projections from events

┌─────────────────┐     Events      ┌─────────────────┐
│   Command Side   │ ───────────────▶│   Query Side    │
│                 │                 │                 │
│  - Validation   │                 │  - Read Models  │
│  - Business     │                 │  - Projections  │
│    Logic        │                 │  - Queries      │
│  - Event Store  │                 │                 │
└─────────────────┘                 └─────────────────┘
        ▲                                   │
        │                                   │
    Commands                            Queries
        │                                   │
        └────────────── API ────────────────┘

Command Handler

type CommandHandler struct {
    store EventStore
}

type CreateOrderCommand struct {
    OrderID string
    UserID  string
    Items   []Item
}

func (h *CommandHandler) HandleCreateOrder(cmd CreateOrderCommand) error {
    // Load aggregate (replay events)
    events, _ := h.store.GetEvents(cmd.OrderID, 0)
    order := RebuildOrder(events)
    
    // Check business rules
    if order.Version > 0 {
        return errors.New("order already exists")
    }
    
    if len(cmd.Items) == 0 {
        return errors.New("order must have items")
    }
    
    // Create event
    event := Event{
        ID:          uuid.New().String(),
        Type:        "OrderCreated",
        AggregateID: cmd.OrderID,
        Version:     1,
        Timestamp:   time.Now(),
        Data:        mustMarshal(map[string]any{"items": cmd.Items}),
        Metadata: Metadata{
            UserID:        cmd.UserID,
            CorrelationID: uuid.New().String(),
        },
    }
    
    return h.store.Append(cmd.OrderID, []Event{event}, 0)
}

Projection Builder

type OrderSummaryProjection struct {
    db *sql.DB
}

type OrderSummary struct {
    OrderID     string    `json:"orderId"`
    Status      string    `json:"status"`
    Total       float64   `json:"total"`
    CustomerID  string    `json:"customerId"`
    CreatedAt   time.Time `json:"createdAt"`
    ShippedAt   *time.Time `json:"shippedAt,omitempty"`
}

func (p *OrderSummaryProjection) Handle(event Event) error {
    switch event.Type {
    case "OrderCreated":
        var data struct {
            Items []Item `json:"items"`
        }
        json.Unmarshal(event.Data, &data)
        
        _, err := p.db.Exec(`
            INSERT INTO order_summaries (order_id, status, total, created_at)
            VALUES ($1, 'pending', $2, $3)
        `, event.AggregateID, calculateTotal(data.Items), event.Timestamp)
        return err
        
    case "PaymentReceived":
        _, err := p.db.Exec(`
            UPDATE order_summaries SET status = 'paid' WHERE order_id = $1
        `, event.AggregateID)
        return err
        
    case "OrderShipped":
        _, err := p.db.Exec(`
            UPDATE order_summaries 
            SET status = 'shipped', shipped_at = $2 
            WHERE order_id = $1
        `, event.AggregateID, event.Timestamp)
        return err
    }
    return nil
}

When to Use Event Sourcing

Event sourcing shines in specific scenarios:

Use Case	Why Event Sourcing Helps
Financial systems	Perfect audit trail, regulatory compliance
Collaborative apps	Conflict resolution via events
Analytics	Rich historical data for insights
Debugging	Replay exact sequence of events
Feature development	New projections without schema changes

For observability in event-driven systems, see my guide on distributed tracing to trace events across services.

When to Avoid It

Event sourcing adds complexity. Skip it if:

• Your domain has no business logic (simple CRUD)
• You don’t need audit trails or history
• Your team lacks experience with the pattern
• You need real-time queries on rapidly changing data (use caching instead)

For simpler data storage needs, consider SQLite for production which offers an excellent balance of simplicity and capability.

Event Versioning Strategies

Events are immutable, but your domain evolves. How do you handle changes?

1. Multiple Versions

Support old and new event formats:

type OrderCreatedV1 struct {
    OrderID string `json:"orderId"`
    Items   []Item `json:"items"`
}

type OrderCreatedV2 struct {
    OrderID    string `json:"orderId"`
    Items      []Item `json:"items"`
    Source     string `json:"source"` // web, mobile, api
    PromoCode  string `json:"promoCode,omitempty"`
}

func (o *Order) Apply(event Event) error {
    if event.Type == "OrderCreated" {
        // Try V2 first, fall back to V1
        var v2 OrderCreatedV2
        if err := json.Unmarshal(event.Data, &v2); err == nil && v2.Source != "" {
            return o.applyV2(v2)
        }
        
        var v1 OrderCreatedV1
        if err := json.Unmarshal(event.Data, &v1); err != nil {
            return err
        }
        return o.applyV1(v1)
    }
    // ...
}

2. Upcasting

Transform old events when reading:

type Upcaster func(Event) (Event, error)

var upcasters = map[string]map[int]Upcaster{
    "OrderCreated": {
        1: func(e Event) (Event, error) {
            var v1 OrderCreatedV1
            json.Unmarshal(e.Data, &v1)
            
            v2 := OrderCreatedV2{
                OrderID: v1.OrderID,
                Items:   v1.Items,
                Source:  "unknown", // Default value
            }
            
            e.Data, _ = json.Marshal(v2)
            return e, nil
        },
    },
}

func Upcast(event Event) Event {
    if versionUpcasters, ok := upcasters[event.Type]; ok {
        if upcaster, ok := versionUpcasters[event.Version]; ok {
            upcasted, _ := upcaster(event)
            return upcasted
        }
    }
    return event
}

Snapshotting for Performance

Replaying thousands of events is slow. Snapshots capture state at a point:

type Snapshot struct {
    AggregateID string          `json:"aggregateId"`
    Version     int             `json:"version"`
    State       json.RawMessage `json:"state"`
    Timestamp   time.Time       `json:"timestamp"`
}

func (s *SnapshotStore) Load(aggregateID string) (*Order, int, error) {
    snapshot, err := s.getSnapshot(aggregateID)
    if err != nil {
        // No snapshot, replay all events
        events, _ := s.eventStore.GetEvents(aggregateID, 0)
        return RebuildOrder(events), len(events), nil
    }
    
    // Start from snapshot
    var order Order
    json.Unmarshal(snapshot.State, &order)
    
    // Replay events after snapshot
    events, _ := s.eventStore.GetEvents(aggregateID, snapshot.Version)
    for _, event := range events {
        order.Apply(event)
    }
    
    return &order, order.Version, nil
}

Event Sourcing in Microservices

Events become the integration layer between services:

┌──────────────┐    OrderCreated    ┌──────────────┐
│ Order Service│ ──────────────────▶│Inventory Svc │
└──────────────┘                    └──────────────┘
       │                                   │
       │ OrderShipped                      │ StockReserved
       ▼                                   ▼
┌──────────────┐                    ┌──────────────┐
│Notification  │                    │ Analytics Svc│
│   Service    │                    └──────────────┘
└──────────────┘

Each service builds its own projections:

// Analytics service builds its own model
type OrderAnalyticsProjection struct{}

func (p *OrderAnalyticsProjection) Handle(event Event) error {
    switch event.Type {
    case "OrderCreated":
        // Track order creation metrics
        metrics.OrdersCreated.Inc()
        metrics.OrderValue.Observe(extractTotal(event))
    case "OrderShipped":
        metrics.FulfillmentTime.Observe(
            time.Since(extractCreatedAt(event)),
        )
    }
    return nil
}

Practical Implementation Tips

1. Keep Events Small

// Bad: Including too much data
type OrderCreated struct {
    Order    Order    `json:"order"`
    Customer Customer `json:"customer"` // Already exists elsewhere
    Items    []Item   `json:"items"`    // Could reference by ID
}

// Good: Minimal, focused event
type OrderCreated struct {
    OrderID    string   `json:"orderId"`
    CustomerID string   `json:"customerId"`
    ItemIDs    []string `json:"itemIds"`
}

2. Use Meaningful Event Names

// Bad: Generic
type OrderStatusChanged struct {
    NewStatus string `json:"newStatus"`
}

// Good: Domain-specific
type OrderShipped struct {
    TrackingNumber string `json:"trackingNumber"`
    Carrier        string `json:"carrier"`
}

3. Include Correlation IDs

type Metadata struct {
    CorrelationID string `json:"correlationId"` // Trace across services
    CausationID   string `json:"causationId"`   // What caused this event
    UserID        string `json:"userId"`         // Who triggered it
}

Conclusion

Event sourcing and CQRS aren’t silver bullets, but they solve real problems:

• Perfect auditability: Every change is recorded forever
• Temporal queries: Reconstruct state at any point in time
• Scalability: Independent scaling of reads and writes
• Flexibility: New projections without changing the write model

Start simple. Begin with a single aggregate, add projections as needed, and expand gradually. The patterns are powerful, but the learning curve is real.

For protecting your event-driven APIs, see my guide on rate limiting strategies to handle traffic spikes gracefully.