CQRS and Event Sourcing: A Practical Implementation Guide

snippet-2.sql

-- event_store schema
CREATE TABLE events (
    id           BIGSERIAL PRIMARY KEY,
    aggregate_id UUID        NOT NULL,
    aggregate_type TEXT      NOT NULL,
    event_type   TEXT        NOT NULL,
    sequence_num INT         NOT NULL,
    payload      JSONB       NOT NULL,
    occurred_at  TIMESTAMPTZ NOT NULL DEFAULT now(),
    UNIQUE (aggregate_id, sequence_num)
);

CREATE INDEX idx_events_aggregate ON events (aggregate_id, sequence_num ASC);

snippet-3.go

// order_aggregate.go
package domain

type Order struct {
    ID         string
    CustomerID string
    Items      []OrderItem
    Status     string
    version    int
}

func (o *Order) Apply(event interface{}) {
    switch e := event.(type) {
    case OrderCreatedEvent:
        o.ID = e.OrderID
        o.CustomerID = e.CustomerID
        o.Status = "pending"
    case ItemAddedEvent:
        o.Items = append(o.Items, OrderItem{
            ProductID: e.ProductID,
            Quantity:  e.Quantity,
            Price:     e.Price,
        })
    }
    o.version++
}

func (o *Order) Handle(cmd AddItemCommand) ([]interface{}, error) {
    if o.Status != "pending" {
        return nil, fmt.Errorf("cannot add items to order in status: %s", o.Status)
    }
    event := ItemAddedEvent{
        OrderID:    cmd.OrderID,
        ProductID:  cmd.ProductID,
        Quantity:   cmd.Quantity,
        Price:      cmd.Price,
        OccurredAt: time.Now().UTC(),
    }
    return []interface{}{event}, nil
}

func RehydrateOrder(events []interface{}) *Order {
    order := &Order{}
    for _, e := range events {
        order.Apply(e)
    }
    return order
}

snippet-4.go

// command_handler.go
package application

func (h *OrderCommandHandler) HandleAddItem(ctx context.Context, cmd domain.AddItemCommand) error {
    // Load aggregate by replaying all its events
    storedEvents, err := h.eventStore.Load(ctx, cmd.OrderID)
    if err != nil {
        return fmt.Errorf("load aggregate: %w", err)
    }

    order := domain.RehydrateOrder(storedEvents)
    currentVersion := order.Version()

    // Validate and produce new events
    newEvents, err := order.Handle(cmd)
    if err != nil {
        return fmt.Errorf("handle command: %w", err)
    }

    // Append to event store with optimistic concurrency check
    if err := h.eventStore.Append(ctx, cmd.OrderID, currentVersion, newEvents); err != nil {
        return fmt.Errorf("append events: %w", err)
    }

    // Publish events to the message bus for read-model projectors
    return h.publisher.Publish(ctx, newEvents...)
}

snippet-5.go

// order_projection.go
package projection

type OrderSummaryProjector struct {
    db *sql.DB
}

func (p *OrderSummaryProjector) On(event interface{}) error {
    switch e := event.(type) {
    case domain.OrderCreatedEvent:
        _, err := p.db.Exec(`
            INSERT INTO order_summaries (order_id, customer_id, total, status, created_at)
            VALUES ($1, $2, 0, 'pending', $3)
        `, e.OrderID, e.CustomerID, e.OccurredAt)
        return err

    case domain.ItemAddedEvent:
        _, err := p.db.Exec(`
            UPDATE order_summaries
            SET total = total + ($1 * $2)
            WHERE order_id = $3
        `, e.Price, e.Quantity, e.OrderID)
        return err
    }
    return nil
}

snippet-6.yaml

services:
  eventstore:
    image: postgres:16-alpine
    environment:
      POSTGRES_DB: eventstore
      POSTGRES_USER: es
      POSTGRES_PASSWORD: secret
    ports: ["5432:5432"]

  readdb:
    image: postgres:16-alpine
    environment:
      POSTGRES_DB: readmodels
      POSTGRES_USER: rm
      POSTGRES_PASSWORD: secret
    ports: ["5433:5432"]

  nats:
    image: nats:2.10-alpine
    command: ["--jetstream"]
    ports: ["4222:4222"]

  command-service:
    build: ./cmd/command-service
    environment:
      EVENT_STORE_DSN: postgres://es:secret@eventstore/eventstore
      NATS_URL: nats://nats:4222
    depends_on: [eventstore, nats]

  projection-worker:
    build: ./cmd/projection-worker
    environment:
      READ_DB_DSN: postgres://rm:secret@readdb/readmodels
      NATS_URL: nats://nats:4222
    depends_on: [readdb, nats]

snippet-7.sh

#!/usr/bin/env bash
# replay_projection.sh — rebuild the order_summaries read model

set -euo pipefail

PROJECTION=${1:?"Usage: $0 <projection-name>"}

echo "Dropping read model: $PROJECTION"
psql "$READ_DB_DSN" -c "TRUNCATE ${PROJECTION};"

echo "Triggering full event replay from position 0..."
curl -sf -X POST "http://localhost:8081/admin/replay" \
  -H "Content-Type: application/json" \
  -d "{\"projection\": \"$PROJECTION\", \"from_position\": 0}"

echo "Replay initiated. Monitor progress at /admin/replay/status"

snippet.go

// commands.go
package domain

type CreateOrderCommand struct {
    OrderID    string
    CustomerID string
    Items       []OrderItem
}

type AddItemCommand struct {
    OrderID   string
    ProductID string
    Quantity  int
    Price     float64
}

// Events are past-tense facts — they cannot be undone
type OrderCreatedEvent struct {
    OrderID    string    `json:"order_id"`
    CustomerID string    `json:"customer_id"`
    OccurredAt time.Time `json:"occurred_at"`
}

type ItemAddedEvent struct {
    OrderID   string    `json:"order_id"`
    ProductID string    `json:"product_id"`
    Quantity  int       `json:"quantity"`
    Price     float64   `json:"price"`
    OccurredAt time.Time `json:"occurred_at"`
}