Event Sourcing Cqrs Patterns

Event Sourcing and CQRS Patterns

Overview

Traditional CRUD stores only current state — once you update a record, the history is gone. Event Sourcing inverts this: every state change is recorded as an immutable event appended to an event log. The current state is always derived by replaying events. CQRS (Command Query Responsibility Segregation) separates the write side (commands that emit events) from the read side (projections that build optimized query models).

When to use: Audit trail requirements, financial ledgers, collaborative editing, domain-driven systems with complex business rules, systems needing temporal queries ("what was the balance on date X?"), or microservice architectures requiring reliable async integration.

Quick Reference

| Pattern | Core Idea | Primary Red Flag | |---------|-----------|-----------------| | Event Store | Append-only log of domain events | Mutable events, missing sequence numbers | | CQRS | Separate write model (commands) from read model (queries) | Single model that is both read and written | | Projection / Read Model | Build query-optimized view from events | Projections tightly coupled to event internals | | Snapshot | Checkpoint aggregate state to skip full replay | Snapshots taken too rarely or too often | | Aggregate Replay | Rebuild aggregate by replaying its event stream | Loading entire global log instead of per-aggregate stream | | Event Versioning / Upcasting | Migrate old event schemas to new versions | Breaking schema changes without versioning | | Eventual Consistency | Accept that read models lag behind the write model | Assuming reads are always current after a write | | Idempotency / At-Least-Once | Process duplicate events safely | Missing deduplication in event handlers |

---

Patterns in Detail

1. Event Sourcing Fundamentals — Event Store and Append-Only Log

The event store is the single source of truth. Events are facts — immutable records of things that happened. The event log is append-only: you never update or delete events.

Core concepts:

• Each event has a stream ID (aggregate ID), a sequence number (position in that stream), an event type, a payload, and a timestamp.
• Optimistic concurrency: when appending, pass the expected version; the store rejects if another writer advanced it first.
• The global event log can reconstruct any state at any point in time.

Red Flags:

• Events modified or deleted after write — destroys auditability
• Missing sequence numbers — cannot detect gaps or enforce ordering
• Storing commands, not events: CreateUserCommand is a command; UserCreated is an event
• Events named with CRUD verbs (UserUpdated) instead of domain facts (UserEmailChanged)
• Storing the full aggregate state in the event payload — defeats the purpose

TypeScript — minimal event store interface:

type DomainEvent = {
  readonly streamId: string;
  readonly version: number;
  readonly type: string;
  readonly occurredAt: Date;
  readonly payload: Readonly>;
};

interface EventStore {
  append(streamId: string, events: DomainEvent[], expectedVersion: number): Promise;
  load(streamId: string, fromVersion?: number): Promise;
}

// Usage: optimistic concurrency check
async function processCommand(cmd: CreateOrderCmd, store: EventStore): Promise {
  const events = await store.load(cmd.orderId);
  const order = replayOrder(events);                   // rebuild from history
  const newEvents = order.placeOrder(cmd);             // domain logic produces events
  await store.append(cmd.orderId, newEvents, order.version); // expected version guards race conditions
}

Java — event envelope:

public record DomainEvent(
    String streamId,
    int version,
    String type,
    Instant occurredAt,
    Map payload   // immutable at runtime via Collections.unmodifiableMap
) {}

public interface EventStore {
    void append(String streamId, List events, int expectedVersion);
    List load(String streamId);
    List load(String streamId, int fromVersion);
}

Go — append with optimistic locking:

type DomainEvent struct {
    StreamID    string
    Version     int
    Type        string
    OccurredAt  time.Time
    Payload     json.RawMessage
}

type EventStore interface {
    Append(streamID string, events []DomainEvent, expectedVersion int) error
    Load(streamID string, fromVersion int) ([]DomainEvent, error)
}

// ErrConcurrencyConflict is returned when expectedVersion does not match.
var ErrConcurrencyConflict = errors.New("concurrency conflict")

---

2. CQRS — Command Query Responsibility Segregation

Commands change state and produce events. Queries read from optimized projections. The two models never share the same data structure.

Red Flags:

• A single repository used for both reads and writes — read concerns pollute the write model
• Commands returning rich query data — violates the segregation boundary
• Synchronous projection updates inside the command transaction — defeats scalability
• No command validation before dispatching — invalid commands consume resources before failing

TypeScript — command handler produces events; query hits read model:

// --- Write side ---
type PlaceOrderCommand = { orderId: string; customerId: string; items: OrderItem[] };

class OrderCommandHandler {
  constructor(private store: EventStore) {}

  async handle(cmd: PlaceOrderCommand): Promise {
    const history = await this.store.load(cmd.orderId);
    const order = OrderAggregate.replay(history);
    const events = order.place(cmd);           // domain rules, no DB reads here
    await this.store.append(cmd.orderId, events, order.version);
  }
}

// --- Read side ---
type OrderSummary = { orderId: string; status: string; total: number };

interface OrderQueryRepository {
  findById(orderId: string): Promise;
  findByCustomer(customerId: string): Promise;
}

// Controller keeps the two sides separate
class OrderController {
  constructor(
    private commands: OrderCommandHandler,
    private queries: OrderQueryRepository,
  ) {}

  async placeOrder(cmd: PlaceOrderCommand): Promise {
    await this.commands.handle(cmd);
    // Return 202 Accepted — the read model will update asynchronously
  }

  async getOrder(id: string): Promise {
    return this.queries.findById(id);          // never touches the event store
  }
}

Java — using MediatR-style dispatch:

public record PlaceOrderCommand(String orderId, String customerId, List items)
    implements Command {}

@Component
public class PlaceOrderHandler implements CommandHandler {
    private final EventStore store;

    @Override
    public void handle(PlaceOrderCommand cmd) {
        var history = store.load(cmd.orderId());
        var order = OrderAggregate.replay(history);
        var events = order.place(cmd);
        store.append(cmd.orderId(), events, order.version());
    }
}

// Query side — completely separate Spring Data repository projecting onto a read table
public interface OrderSummaryRepository extends JpaRepository {
    List findByCustomerId(String customerId);
}

---

3. Projections, Read Models, and View Models

A projection is a function from an event stream to a read-optimized data structure. It subscribes to events and updates the read model (view model) stored in a queryable store (SQL table, Redis, Elasticsearch, etc.).

Red Flags:

• Projection directly reads from the event store on every query — defeats the purpose
• Projection contains business logic — projections should only reshape data
• No idempotency in projection handlers — replaying events corrupts the read model
• Projection deletes and rebuilds the entire read model on every event — does not scale
• Single projection coupled to multiple aggregates' internal details

TypeScript — idempotent projection handler:

type OrderPlaced = { orderId: string; customerId: string; total: number; status: 'placed' };
type OrderShipped = { orderId: string; shippedAt: string; status: 'shipped' };

class OrderProjection {
  constructor(private db: Database) {}

  async on(event: DomainEvent): Promise {
    // Idempotent: use UPSERT with the event version as the cursor
    switch (event.type) {
      case 'OrderPlaced': {
        const p = event.payload as OrderPlaced;
        await this.db.upsert('order_summary', {
          order_id: p.orderId,
          customer_id: p.customerId,
          total: p.total,
          status: p.status,
          last_event_version: event.version,
        }, { conflictOn: 'order_id' });
        break;
      }
      case 'OrderShipped': {
        const p = event.payload as OrderShipped;
        await this.db.update('order_summary',
          { status: p.status, shipped_at: p.shippedAt, last_event_version: event.version },
          { where: 'order_id = ? AND last_event_version  = { streamId: string; version: number; state: T };

interface SnapshotStore {
  save(snapshot: Snapshot): Promise;
  load(streamId: string): Promise | null>;
}

async function loadAggregate(
  id: string,
  eventStore: EventStore,
  snapshotStore: SnapshotStore,
): Promise {
  const snapshot = await snapshotStore.load(id);
  const fromVersion = snapshot ? snapshot.version + 1 : 0;
  const events = await eventStore.load(id, fromVersion);

  let aggregate: OrderAggregate;
  if (snapshot) {
    aggregate = OrderAggregate.fromSnapshot(snapshot.state, snapshot.version);
  } else {
    aggregate = OrderAggregate.empty();
  }
  return aggregate.replay(events);
}

// Save a snapshot every N events
const SNAPSHOT_THRESHOLD = 50;
async function saveIfNeeded(agg: OrderAggregate, store: SnapshotStore): Promise {
  if (agg.version % SNAPSHOT_THRESHOLD === 0) {
    await store.save({ streamId: agg.id, version: agg.version, state: agg.toState() });
  }
}

Java — aggregate replay with snapshot:

public class OrderAggregate {
    private String id;
    private String status;
    private int version;

    public static OrderAggregate replay(List events) {
        return events.stream().reduce(
            new OrderAggregate(),
            OrderAggregate::apply,
            (a, b) -> b   // combiner unused in sequential stream
        );
    }

    public static OrderAggregate fromSnapshot(OrderSnapshot snap) {
        var agg = new OrderAggregate();
        agg.id = snap.orderId();
        agg.status = snap.status();
        agg.version = snap.version();
        return agg;
    }

    private OrderAggregate apply(DomainEvent event) {
        // Return new instance — immutable evolution
        var next = new OrderAggregate();
        next.id = this.id;
        next.version = event.version();
        next.status = switch (event.type()) {
            case "OrderPlaced" -> "placed";
            case "OrderShipped" -> "shipped";
            default -> this.status;
        };
        return next;
    }
}

---

5. Event Schema Evolution and Upcasting / Event Versioning

Events are permanent. When the schema changes, you cannot alter old events — you upcast them to the new format at read time.

Red Flags:

• Breaking schema change (rename/remove field) on an existing event type — crashes replays
• No version field on events — cannot distinguish v1 from v2 payloads
• Upcast logic scattered across multiple projections — each must be updated independently
• Upcasting mutates the stored event — violates immutability of the log
• Treating schema evolution as an infrastructure problem instead of a domain problem

TypeScript — upcaster pipeline:

type EventV1 = { type: 'UserRegistered'; version: 1; email: string };
type EventV2 = { type: 'UserRegistered'; version: 2; email: string; username: string };

function upcastUserRegistered(raw: EventV1 | EventV2): EventV2 {
  if (raw.version === 2) return raw;
  // v1 → v2: derive username from email local part
  return { ...raw, version: 2, username: raw.email.split('@')[0] };
}

// Apply upcasters before handing events to aggregates or projections
function upcast(event: DomainEvent): DomainEvent {
  if (event.type === 'UserRegistered') {
    const payload = upcastUserRegistered(event.payload as EventV1 | EventV2);
    return { ...event, payload };
  }
  return event;
}

// Load pipeline: raw events → upcast → replay
async function loadUser(id: string, store: EventStore): Promise {
  const raw = await store.load(id);
  const upcasted = raw.map(upcast);
  return UserAggregate.replay(upcasted);
}

Java — versioned event with upcaster registry:

public interface Upcaster {
    String eventType();
    int fromVersion();
    Map upcast(Map payload);
}

@Component
public class UserRegisteredV1ToV2 implements Upcaster {
    public String eventType() { return "UserRegistered"; }
    public int fromVersion() { return 1; }

    public Map upcast(Map payload) {
        var result = new LinkedHashMap<>(payload);
        result.put("version", 2);
        result.computeIfAbsent("username",
            k -> ((String) payload.get("email")).split("@")[0]);
        return Collections.unmodifiableMap(result);
    }
}

Go — version tag in payload:

type RawEvent struct {
    Type    string          `json:"type"`
    Version int             `json:"version"`
    Payload json.RawMessage `json:"payload"`
}

func upcastAll(raw []RawEvent) []RawEvent {
    result := make([]RawEvent, len(raw))
    for i, e := range raw {
        result[i] = upcastOne(e)  // returns new RawEvent, never modifies in place
    }
    return result
}

Cross-reference: architectural-patterns — Strangler Fig: use event versioning when migrating event schemas incrementally alongside system evolution.

---

6. Eventual Consistency and Idempotency / At-Least-Once Delivery

Read models are eventually consistent — there is a lag between a command being processed and the projection updating. Message brokers guarantee at-least-once delivery, so projections must be idempotent.

Red Flags:

• Read-your-own-writes assumption — querying the read model immediately after a write and expecting the new state
• No deduplication — same event applied twice causes double-counting or duplicate rows
• No event ordering guarantee enforced — out-of-order events corrupt the read model
• Missing idempotency key on external API calls triggered by events — double charges, double emails
• Projection handler that fails partially and cannot be safely retried

TypeScript — idempotent event handler with deduplication table:

async function handleEvent(event: DomainEvent, db: Database): Promise {
  const dedupeKey = `${event.streamId}:${event.version}`;

  await db.transaction(async (tx) => {
    // Guard: skip if already processed (idempotent at-least-once safety)
    const existing = await tx.queryOne(
      'SELECT 1 FROM processed_events WHERE dedupe_key = ?', [dedupeKey]
    );
    if (existing) return;  // duplicate delivery — safe to skip

    // Apply to read model
    await applyToReadModel(event, tx);

    // Mark as processed within same transaction
    await tx.execute(
      'INSERT INTO processed_events (dedupe_key, processed_at) VALUES (?, NOW())',
      [dedupeKey]
    );
  });
}

Java — idempotent consumer with Spring and JPA:

@Transactional
public void onEvent(DomainEvent event) {
    var key = event.streamId() + ":" + event.version();
    if (processedEventRepository.existsByDedupeKey(key)) return;

    applyToReadModel(event);
    processedEventRepository.save(new ProcessedEvent(key, Instant.now()));
}

Go — at-least-once consumer with explicit acknowledgment:

func (c *Consumer) Process(ctx context.Context, msg Message) error {
    key := fmt.Sprintf("%s:%d", msg.StreamID, msg.Version)
    ok, err := c.dedupe.AlreadyProcessed(ctx, key)
    if err != nil {
        return fmt.Errorf("dedupe check: %w", err)
    }
    if ok {
        return nil  // idempotent skip — ack without reprocessing
    }

…

## Source & license

This open-source skill is cataloged on AgentStack and links to its original source — we do not rehost the code.

- **Author:** [mickeyyaya](https://github.com/mickeyyaya)
- **Source:** [mickeyyaya/refactoring-skills](https://github.com/mickeyyaya/refactoring-skills)
- **License:** MIT

Install and usage instructions live in the source repository linked above.