Event Sourcing with Databases: Design, Patterns & Production Pitfalls

Most applications are built around a simple mental model: save the current state, overwrite it when it changes, query it when you need it. That model works until it doesn't — and the moment it stops working, you feel it hard. You can't answer 'what did this record look like three weeks ago?' You can't audit who changed what and why. You can't replay business logic against historical data. You've lost information the moment you hit UPDATE.

Event sourcing flips this model on its head. Instead of storing current state, you store an immutable, append-only log of every state transition — every event that caused the world to change. The current state becomes a derived view, computed on demand by replaying that log. This gives you a complete audit trail, time-travel debugging, the ability to rebuild any projection from scratch, and a natural fit for event-driven architectures. The database stops being a snapshot of 'now' and becomes a ledger of 'everything that ever happened'.

By the end of this article you'll know how to design an event store schema from scratch, implement snapshot strategies to keep replay times sane at scale, wire up CQRS read models as SQL projections, handle schema evolution without breaking old events, and avoid the production mistakes that bite teams six months after they ship. This is the article I wish existed when I first built an event-sourced system in production.

What is Event Sourcing with Databases?

The core idea is simple: instead of updating a row in-place to reflect the latest state, you append an event to an append-only log. Each event captures what happened, when, and why. The current state of any entity — an account, an order, a document — is computed by replaying all its events in order.

This is fundamentally different from CRUD. In CRUD, an UPDATE erases the previous state. You lose history. Event sourcing preserves every fact forever. That's hugely powerful for auditing, debugging, and building multiple read models without fighting the write schema.

But it's also a trade-off. Replaying events on every read is slow — that's where snapshots come in. Storing immutable events makes schema evolution harder — you have to handle old event formats gracefully. And because you're appending to a single table at high concurrency, your database needs to handle write throughput, not just reads.

The snippet below shows a minimal Java example. This is not what you'd use in production, but it illustrates the append-replay loop.

package io.thecodeforge.eventsourcing;

import java.util.*;
import java.util.concurrent.ConcurrentHashMap;

// Minimal in-memory event store — NOT production-ready
public final class BasicEventStore {
    private final Map<String, List<Event>> store = new ConcurrentHashMap<>();
    
    public void append(String aggregateId, Event event) {
        store.computeIfAbsent(aggregateId, k -> new ArrayList<>()).add(event);
    }
    
    public List<Event> readEvents(String aggregateId) {
        return store.getOrDefault(aggregateId, List.of());
    }
    
    // Replay all events to compute current state
    public AccountState getAccountState(String accountId) {
        AccountState state = new AccountState(accountId);
        for (Event e : readEvents(accountId)) {
            state.apply(e);
        }
        return state;
    }
    
    public record Event(String type, String data, long timestamp) {}
    
    public static class AccountState {
        private final String id;
        private double balance = 0.0;
        
        AccountState(String id) { this.id = id; }
        
        void apply(Event e) {
            switch (e.type()) {
                case "AccountOpened" -> { /* init */ }
                case "Deposited" -> balance += Double.parseDouble(e.data());
                case "Withdrawn" -> balance -= Double.parseDouble(e.data());
            }
        }
        
        public double getBalance() { return balance; }
    }
}

Event Store Schema Design

Your event store schema is the foundation of your entire system. Get it wrong and you'll fight ordering issues, slow queries, and painful migrations.

The essential columns: aggregate_id (string/ UUID), sequence_number (integer), event_type (string), event_data (JSON/JSONB), timestamp (transaction time). The primary key MUST be (aggregate_id, sequence_number). This enforces ordering per aggregate and allows fast replay.

You also need indexes for specific query patterns: projectors that need to scan events by type or by timestamp range. Consider a composite index on (event_type, timestamp) for subscriptions. But remember: you trade write throughput for read performance.

Avoid using high-latency sequences like UUIDs for sequence_number unless you can guarantee monotonic ordering. Database sequences (like SERIAL in PostgreSQL) work well for single-writer per aggregate. If you have concurrent writers for the same aggregate, you need optimistic locking with a version field.

Below is a PostgreSQL DDL for a production-grade event store table.

-- Production event store schema for PostgreSQL
CREATE TABLE io_thecodeforge.event_store (
    aggregate_id   VARCHAR(64) NOT NULL,
    sequence_number INTEGER NOT NULL,
    event_type     VARCHAR(128) NOT NULL,
    event_data     JSONB NOT NULL,
    created_at      TIMESTAMPTZ NOT NULL DEFAULT NOW(),
    -- Ensure ordering and uniqueness per aggregate
    PRIMARY KEY (aggregate_id, sequence_number)
);

-- For projectors that scan by event type
CREATE INDEX idx_event_store_event_type 
    ON io_thecodeforge.event_store (event_type, created_at);

-- For querying events by time range (auditing)
CREATE INDEX idx_event_store_created_at 
    ON io_thecodeforge.event_store (created_at);

-- For snapshots (separate table)
CREATE TABLE io_thecodeforge.snapshots (
    aggregate_id    VARCHAR(64) NOT NULL,
    sequence_number INTEGER NOT NULL,
    state_type      VARCHAR(128) NOT NULL,
    state_data      JSONB NOT NULL,
    created_at       TIMESTAMPTZ NOT NULL DEFAULT NOW(),
    PRIMARY KEY (aggregate_id, sequence_number)
);

Snapshot Strategies for Efficient Replay

Replaying every event from the beginning of time on every read gets expensive fast. If an aggregate has 100k events, a single read takes 100k SQL queries or a huge IN clause. Snapshots are the answer.

A snapshot captures the aggregate state at a specific point (sequence number). On read, you load the latest snapshot and replay only events after that snapshot. This keeps replay cost constant relative to the window between snapshots, not total history.

Decide when to take snapshots: after every N events, or after a time interval. Use a background process (e.g., cron job or scheduled task) that reads events since the last snapshot and writes the aggregated state. Ensure snapshot creation is idempotent: if two snapshot jobs overlap, the later one wins.

Store snapshots in a separate table with the same primary key structure but with the state data. Projections also need snapshots if they're rebuilt often.

Here's a Java snippet for snapshot creation logic.

package io.thecodeforge.eventsourcing;

import java.time.Instant;
import java.util.List;

public class SnapshotService {
    private final EventStoreRepository eventStore;
    private final SnapshotRepository snapshotRepo;
    private final int snapshotFrequency = 100; // snapshot every 100 events
    
    public SnapshotService(EventStoreRepository eventStore, SnapshotRepository snapshotRepo) {
        this.eventStore = eventStore;
        this.snapshotRepo = snapshotRepo;
    }
    
    public void createSnapshot(String aggregateId) {
        // Load latest snapshot (if any)
        Snapshot latest = snapshotRepo.findLatest(aggregateId).orElse(null);
        int fromSequence = (latest != null) ? latest.sequenceNumber() + 1 : 1;
        
        // Get current state by replaying from snapshot
        AggregateState state = (latest != null) 
            ? latest.restoreState()
            : AggregateState.initial(aggregateId);
        
        List<Event> newEvents = eventStore.readEventsSince(aggregateId, fromSequence);
        for (Event e : newEvents) {
            state.apply(e);
        }
        
        // If enough events have passed, persist snapshot
        int totalEvents = (latest != null ? latest.sequenceNumber() : 0) + newEvents.size();
        if (totalEvents - (latest != null ? latest.sequenceNumber() : 0) >= snapshotFrequency) {
            Snapshot snapshot = new Snapshot(
                aggregateId,
                totalEvents,
                state.getClass().getName(),
                state.toJson(),
                Instant.now()
            );
            snapshotRepo.save(snapshot);
        }
    }
}

Projections and CQRS Integration

Event sourcing naturally pairs with CQRS. The write side appends events, the read side builds projections — denormalised tables optimised for query patterns. Projections are updated asynchronously by subscribing to the event stream.

This decoupling lets you build multiple read models without touching the write schema. You can have a relational projection for reporting, a cache projection for APIs, and a search projection for Elasticsearch — all fed from the same event stream.

But projections introduce complexity: eventual consistency, idempotency, and rebuild capability. When a projection fails, you need to rebuild it from scratch by replaying all events. Idempotency ensures that replaying events multiple times doesn't double-count. Your projection handlers must be deterministic and handle duplicate events gracefully.

Use a dedicated projection table that tracks the last processed position (often the event store's transaction ID or sequence number). This allows the projection to resume after a crash without missing events.

Here's a simple projector example for reading account balances.

package io.thecodeforge.eventsourcing.projectors;

import io.thecodeforge.eventsourcing.*;
import java.sql.Connection;
import java.sql.PreparedStatement;

public class AccountBalanceProjector {
    private final Connection conn;
    private final EventStore eventStore;
    private long lastProcessedPosition = 0;
    
    public AccountBalanceProjector(Connection conn, EventStore eventStore) {
        this.conn = conn;
        this.eventStore = eventStore;
    }
    
    public void processNextBatch() {
        List<Event> events = eventStore.readEventsAfter(lastProcessedPosition, 100);
        for (Event e : events) {
            switch (e.type()) {
                case "Deposited" -> applyDeposit(e);
                case "Withdrawn" -> applyWithdrawal(e);
                // ignore other event types
            }
            lastProcessedPosition = e.globalPosition();
        }
        // Save checkpoint (normally in a transaction)
        saveCheckpoint(lastProcessedPosition);
    }
    
    private void applyDeposit(Event e) {
        String accountId = e.aggregateId();
        double amount = Double.parseDouble(e.data());
        try (PreparedStatement stmt = conn.prepareStatement(
            "UPDATE account_balances SET balance = balance + ? WHERE account_id = ?")) {
            stmt.setDouble(1, amount);
            stmt.setString(2, accountId);
            stmt.executeUpdate();
        } catch (SQLException ex) { throw new RuntimeException(ex); }
    }
    // ... withdrawal method similar
}

Schema Evolution: Handling Event Versioning

Events are immutable, but your understanding of the domain changes. You'll need to add fields, rename them, or change structure. The challenge: you have millions of events in the old format. You can't go back and update them — they're immutable.

Two strategies: versioned event types and upcasters.

Versioned event types: each event type carries a version number (e.g., OrderPlacedV1, OrderPlacedV2). New events are written in the new version. On replay, you check the version and apply the appropriate handler. Old events stay as-is.

Upcasters: conversion functions that transform an old event into the new format on-the-fly during replay. This centralizes the transformation and keeps application code dealing only with the latest version. Upcasters run after deserialization but before handler application.

Choose upcasters over versioned types if the number of event types grows large — you avoid handler proliferation. Choose versioned types if different handlers need different logic per version.

Real-world tools like EventStoreDB or Axon Framework have built-in upcaster support. For custom databases, you implement upcasters in your application layer.

Here's an upcaster example in Java.

package io.thecodeforge.eventsourcing.upcaster;

import com.fasterxml.jackson.databind.JsonNode;
import com.fasterxml.jackson.databind.ObjectMapper;
import com.fasterxml.jackson.databind.node.ObjectNode;

// Converts OrderPlacedV1 to OrderPlacedV2 (adds currency field)
public class OrderPlacedUpcaster {
    private static final ObjectMapper mapper = new ObjectMapper();
    
    public static JsonNode upcast(JsonNode oldEvent) {
        if (!"OrderPlacedV1".equals(oldEvent.get("eventType").asText()))
            return oldEvent; // not for us
        
        ObjectNode v2 = oldEvent.deepCopy();
        v2.put("eventType", "OrderPlacedV2");
        v2.put("currency", "USD"); // default for old events
        // Optionally add a field to mark it was upcasted
        v2.put("_upcasted", true);
        return v2;
    }
}

Production Pitfalls: Real-World Mistakes and Fixes

After running event sourcing in production for months, teams hit predictable but painful problems. Here are the three that hurt most.

1. Missing per-aggregate ordering. As covered in the incident report, without a correct sequence number and single-writer per aggregate, events get reordered. The fix: enforce that all events for an aggregate are appended by the same thread/process, or use optimistic locking with a version field in the aggregate.
2. Snapshot corruption due to concurrent writes. If your snapshot job reads events while a new event is being written, it may snap an incomplete state. The snapshot appears valid but when replaying after it, you get duplicate or missing events. Solution: run snapshots on a read replica with a known lag, or use snapshot isolation level.
3. Projection rebuild timeout. When a projection fails and needs a full rebuild, replaying millions of events can exceed HTTP timeout for APIs or exhaust memory. Test rebuild times regularly. Consider rebuilding via a background job with progress tracking, and serve stale read models until rebuild completes.

Below is a checklist for avoiding these pitfalls.

-- Pitfall prevention checklist (run after deployment)

-- 1. Check for ordering gaps per aggregate
SELECT aggregate_id, count(*) as event_count,
       max(sequence_number) - min(sequence_number) + 1 - count(*) as gaps
FROM io_thecodeforge.event_store
GROUP BY aggregate_id
HAVING gaps > 0;

-- 2. Verify latest snapshot sequence_number <= latest event sequence_number
SELECT s.aggregate_id, s.sequence_number as snapshot_seq, 
       max(e.sequence_number) as latest_event_seq
FROM io_thecodeforge.snapshots s
JOIN io_thecodeforge.event_store e ON s.aggregate_id = e.aggregate_id
GROUP BY s.aggregate_id, s.sequence_number
HAVING s.sequence_number > max(e.sequence_number);

-- 3. Find projection lag: difference between event store max sequence and projection checkpoint
-- (Requires a projection_checkpoint table)
SELECT projector_name, 
       (SELECT max(sequence_number) FROM io_thecodeforge.event_store) - checkpoint_seq AS lag
FROM io_thecodeforge.projection_checkpoints;

Why Your Standard CRUD Setup Is Begging for Event Sourcing

Your typical CRUD app is a lie. You store the current state, overwrite it on every update, and pretend nothing happened before. That works fine until you need to explain how a customer's account went from $10,000 to zero in three minutes.

Event sourcing doesn't store state. It stores facts. Every change is an immutable event appended to a log. Want to know the balance at any point in time? Replay the events. Need to debug a race condition? Read the event stream. Auditors knocking? Hand them the entire history.

The tradeoff is real: you're trading simple reads for complex reconstruction. If your problem isn't auditability, temporal queries, or high-write contention, don't touch this pattern. But if you're building financial systems, inventory management, or anything where "how did we get here" matters more than "where are we now," CRUD is a liability.

Event sourcing swaps 'latest state' for 'immutable truth.' That's a powerful shift, but it demands discipline. You need idempotent events, careful versioning, and a strategy for snapshots. Skip any of those and you'll be debugging replay logic at 3 AM.

// io.thecodeforge — database tutorial
// CRUD approach vs event sourcing for account transfer

-- CRUD: only current state, no history
CREATE TABLE accounts (
    account_id UUID PRIMARY KEY,
    balance DECIMAL(12,2) NOT NULL,
    updated_at TIMESTAMPTZ DEFAULT NOW()
);

-- UPDATE wipes prior state forever
UPDATE accounts SET balance = 0.00 WHERE account_id = 'a1b2c3';

-- Event sourcing: every change is a row
CREATE TABLE account_events (
    event_id BIGSERIAL PRIMARY KEY,
    aggregate_id UUID NOT NULL,
    event_type VARCHAR(50) NOT NULL,
    payload JSONB NOT NULL,
    version INT NOT NULL,
    recorded_at TIMESTAMPTZ DEFAULT NOW()
);

INSERT INTO account_events (aggregate_id, event_type, payload, version)
VALUES ('a1b2c3', 'FundsWithdrawn', '{"amount": 5000}', 1);

Event Sourcing Is Not CQRS — Stop Confusing the Two

Every junior who reads a blog post about event sourcing immediately starts drawing boxes labeled 'Command' and 'Query.' That's CQRS. It's a separate pattern that often pairs with event sourcing, but they are not the same thing.

Event sourcing is about storage: how you persist facts. CQRS is about architecture: how you separate reads from writes. You can do event sourcing without CQRS by materializing state directly from events on read. Ugly but functional. You can also do CQRS with a plain relational database — just point your queries at a read replica.

Here's the confusion that causes real damage: teams slap CQRS on top of event sourcing too early, before they understand their query patterns. They build expensive projections for queries that could be answered by a simple snapshot. Then they complain event sourcing is slow.

Start with a single store. Write events, build snapshots, and query from snapshots. Only introduce separate read models when you have data — not guesses — proving you need them. Premature CQRS is the microservices of event sourcing: a solution in search of a problem.

When you do need CQRS (high read/write disparity, different read models for different consumers), design your projections as idempotent subscribers to the event stream. Miss an event? Reprocess from last checkpoint. That's fault tolerance, not failure.

// io.thecodeforge — database tutorial
// Projection populated from events, no separate write model

CREATE MATERIALIZED VIEW account_balances AS
WITH event_audit AS (
    SELECT
        aggregate_id,
        SUM(
            CASE
                WHEN event_type = 'FundsDeposited' THEN (payload->>'amount')::NUMERIC
                WHEN event_type = 'FundsWithdrawn' THEN -(payload->>'amount')::NUMERIC
                ELSE 0
            END
        ) AS balance
    FROM account_events
    GROUP BY aggregate_id
)
SELECT * FROM event_audit;

-- Query snapshot directly
SELECT * FROM account_balances WHERE aggregate_id = 'a1b2c3';

What Are Event-Driven Microservices in Kafka?

Event-driven microservices decouple services by communicating through immutable event streams instead of direct HTTP calls. Kafka acts as the durable, ordered log where each service produces and consumes events. Unlike a shared database, Kafka stores events as topics — each event is a fact that any service can replay. This matches event sourcing's append-only nature: the Kafka log becomes your event store. Each microservice owns its state by reading from shared topics and projecting its local view. The WHY: synchronous coupling kills scalability. Event-driven architectures let services fail independently, scale separately, and evolve without coordinated deployments. The HOW: producers write events to partitioned topics, consumers track offsets, and brokers guarantee ordering per partition. Combine this with a schema registry to enforce event versions. Kafka handles replay via consumer group rebalancing — add instances to scale reads. Production trap: never treat Kafka as a message queue with ephemeral state; events are permanent records.

// io.thecodeforge — database tutorial

CREATE TABLE event_stream (
    stream_id   UUID NOT NULL,
    version     BIGINT NOT NULL,
    event_type  VARCHAR(255) NOT NULL,
    payload     JSONB NOT NULL,
    created_at  TIMESTAMPTZ NOT NULL DEFAULT NOW(),
    PRIMARY KEY (stream_id, version)
);

CREATE INDEX idx_events_stream_type 
    ON event_stream (stream_id, created_at);

CREATE TABLE kafka_event_record (
    topic       VARCHAR(255) NOT NULL,
    partition   INT NOT NULL,
    offset      BIGINT NOT NULL,
    stream_id   UUID NOT NULL,
    event_data  JSONB NOT NULL,
    recorded_at TIMESTAMPTZ NOT NULL DEFAULT NOW(),
    PRIMARY KEY (topic, partition, offset)
);

Integrating with Webhooks and Database

Webhooks push event notifications to external systems when state changes. The WHY: polling is wasteful and slow — webhooks give instant, targeted delivery. The HOW: after appending an event to your event store, a background process (or database trigger) publishes the event to registered webhook endpoints. PostgreSQL LISTEN/NOTIFY or logical replication slots can notify subscribers without polling. For high throughput, push events into a queue (e.g., Redis, RabbitMQ) and have workers dispatch HTTP POSTs to URLs with retries. The event store remains the source of truth — webhooks are a downstream projection. Production trap: webhooks fail silently; implement exponential backoff and dead-letter queues. Never send raw internal event structures; map to a stable external schema. Use idempotency keys so external systems can safely retry. The key rule: persist before notify — the event store commit must be durable before any webhook fires.

// io.thecodeforge — database tutorial

CREATE TABLE webhook_subscription (
    id          UUID PRIMARY KEY,
    url         TEXT NOT NULL,
    event_type  VARCHAR(255) NOT NULL,
    retry_count INT DEFAULT 0,
    created_at  TIMESTAMPTZ DEFAULT NOW()
);

CREATE TABLE webhook_delivery (
    id           UUID PRIMARY KEY,
    sub_id       UUID REFERENCES webhook_subscription(id),
    event_id     UUID NOT NULL,
    status       VARCHAR(20) DEFAULT 'pending',
    next_retry_at TIMESTAMPTZ,
    payload      JSONB
);

CREATE INDEX idx_delivery_retry 
    ON webhook_delivery (status, next_retry_at)
    WHERE status = 'pending';

Event Store Implementation: Dedicated vs General-Purpose DB

Choosing between a dedicated event store (like EventStoreDB) and a general-purpose database (like PostgreSQL) for event sourcing depends on your scalability needs, operational complexity, and team expertise. Dedicated event stores are optimized for append-only logs, built-in projections, and event versioning, but introduce a new technology stack. General-purpose databases offer familiarity, transactional consistency, and reduced operational overhead, but require careful schema design.

Dedicated Event Store (e.g., EventStoreDB) - Pros: Native support for event streams, subscriptions, and projections; automatic indexing on event metadata; built-in idempotency and concurrency handling. - Cons: Additional infrastructure to manage; learning curve for event store-specific APIs; potential vendor lock-in.

General-Purpose DB (e.g., PostgreSQL) - Pros: Leverage existing skills; ACID transactions across event store and other business tables; rich ecosystem of tools (e.g., pglogical for replication). - Cons: Need to implement event versioning, snapshotting, and stream ordering yourself; risk of accidental updates/deletes if not careful.

Schema Design in PostgreSQL A typical event store table: ``sql CREATE TABLE events ( id BIGSERIAL PRIMARY KEY, stream_id UUID NOT NULL, stream_version INT NOT NULL, event_type VARCHAR(255) NOT NULL, event_data JSONB NOT NULL, metadata JSONB, created_at TIMESTAMPTZ DEFAULT NOW(), UNIQUE(stream_id, stream_version) ); CREATE INDEX idx_events_stream_id ON events(stream_id); CREATE INDEX idx_events_created_at ON events(created_at); ` The UNIQUE(stream_id, stream_version) constraint ensures optimistic concurrency control. Use JSONB` for flexible event data and metadata.

When to Choose Which - Use a dedicated event store if you need high-throughput event ingestion, built-in projections, or multi-region replication. - Use a general-purpose DB if you already have a strong PostgreSQL team, need tight integration with existing transactional data, or want to minimize new infrastructure.

For most startups and mid-sized applications, starting with PostgreSQL is pragmatic. You can always migrate to a dedicated event store later if needed.

CREATE TABLE events (
  id BIGSERIAL PRIMARY KEY,
  stream_id UUID NOT NULL,
  stream_version INT NOT NULL,
  event_type VARCHAR(255) NOT NULL,
  event_data JSONB NOT NULL,
  metadata JSONB,
  created_at TIMESTAMPTZ DEFAULT NOW(),
  UNIQUE(stream_id, stream_version)
);
CREATE INDEX idx_events_stream_id ON events(stream_id);
CREATE INDEX idx_events_created_at ON events(created_at);

Snapshot Strategies: Rebuilding State from Event Streams

Rebuilding aggregate state from an event stream by replaying all events can become prohibitively slow as the stream grows. Snapshots capture the state at a point in time, allowing you to start replay from the snapshot instead of the beginning. Effective snapshot strategies balance storage overhead with replay speed.

Snapshot Frequency - Interval-based: Take a snapshot every N events (e.g., every 100 events). Simple but may waste storage if state changes little. - Time-based: Snapshot every hour/day. Good for predictable load. - Threshold-based: Snapshot when state size exceeds a limit (e.g., 1MB). Useful for large aggregates.

Snapshot Storage Store snapshots in a separate table: ``sql CREATE TABLE snapshots ( stream_id UUID NOT NULL, stream_version INT NOT NULL, state JSONB NOT NULL, created_at TIMESTAMPTZ DEFAULT NOW(), PRIMARY KEY (stream_id, stream_version) ); `` When rebuilding, fetch the latest snapshot and replay events after that version.

Rebuilding Logic ```sql -- Fetch latest snapshot SELECT state, stream_version FROM snapshots WHERE stream_id = $1 ORDER BY stream_version DESC LIMIT 1;

-- Replay events after snapshot version SELECT * FROM events WHERE stream_id = $1 AND stream_version > $snapshot_version ORDER BY stream_version ASC; ``` Apply events to the snapshot state to get current state.

Strategies - Full snapshots: Store entire state. Simple but large. - Incremental snapshots: Store only changes since last snapshot. More complex but space-efficient. - Hybrid: Use full snapshots periodically with incremental in between.

Production Considerations - Snapshots should be created asynchronously (e.g., via a background job) to avoid slowing down event writes. - Clean up old snapshots to save storage (keep only the latest N or those within a time window). - For high-throughput systems, consider snapshotting only when the event count exceeds a threshold.

Snapshots are essential for any event-sourced system expected to handle long-lived aggregates or high event volumes.

-- Create snapshots table
CREATE TABLE snapshots (
  stream_id UUID NOT NULL,
  stream_version INT NOT NULL,
  state JSONB NOT NULL,
  created_at TIMESTAMPTZ DEFAULT NOW(),
  PRIMARY KEY (stream_id, stream_version)
);

-- Rebuild state: fetch latest snapshot and replay events
WITH latest_snapshot AS (
  SELECT state, stream_version FROM snapshots
  WHERE stream_id = $1
  ORDER BY stream_version DESC LIMIT 1
)
SELECT e.* FROM latest_snapshot ls
JOIN events e ON e.stream_id = $1 AND e.stream_version > ls.stream_version
ORDER BY e.stream_version ASC;

Event Sourcing with PostgreSQL: Outbox Pattern and Logical Replication

Combining event sourcing with PostgreSQL's outbox pattern and logical replication enables reliable event publishing and real-time data synchronization. The outbox pattern ensures events are atomically written to both the event store and an outbox table within the same transaction, preventing dual-write issues. Logical replication then streams these outbox events to other systems.

Outbox Pattern When an aggregate produces events, insert them into both the events table and an outbox table in a single transaction: ```sql BEGIN; -- Insert event into event store INSERT INTO events (stream_id, stream_version, event_type, event_data, metadata) VALUES ($1, $2, $3, $4, $5);

-- Insert into outbox for publishing INSERT INTO outbox (event_id, event_type, payload, created_at) VALUES ($6, $3, $4, NOW()); COMMIT; ``` The outbox table acts as a queue. A separate process (e.g., a worker) reads from the outbox, publishes events to a message broker (e.g., Kafka, RabbitMQ), and deletes processed rows.

Logical Replication PostgreSQL logical replication allows you to replicate the outbox table to another database or stream changes to external systems. Set up a publication on the outbox table: ``sql CREATE PUBLICATION outbox_pub FOR TABLE outbox; ` Then create a subscription on the consumer database: `sql CREATE SUBSCRIPTION outbox_sub CONNECTION 'host=primary_db port=5432 dbname=mydb' PUBLICATION outbox_pub; `` This replicates outbox rows in near real-time. The consumer can then process events without polling the primary database.

Alternative: Using pgoutput Plugin For direct streaming to Kafka, use the pgoutput plugin with Debezium or similar tools. Debezium captures changes from PostgreSQL's write-ahead log (WAL) and streams them to Kafka topics.

Production Considerations - Ensure the outbox table has a primary key (e.g., event_id) for idempotent processing. - Monitor replication lag to avoid stale data. - Clean up processed outbox rows periodically to prevent table bloat. - Use pg_stat_replication to check replication status.

This approach gives you the reliability of PostgreSQL transactions with the scalability of message streaming.

-- Outbox table
CREATE TABLE outbox (
  id BIGSERIAL PRIMARY KEY,
  event_id UUID NOT NULL,
  event_type VARCHAR(255) NOT NULL,
  payload JSONB NOT NULL,
  created_at TIMESTAMPTZ DEFAULT NOW(),
  processed_at TIMESTAMPTZ
);

-- Transactional write to events and outbox
BEGIN;
INSERT INTO events (stream_id, stream_version, event_type, event_data, metadata)
VALUES ($1, $2, $3, $4, $5);
INSERT INTO outbox (event_id, event_type, payload)
VALUES ($6, $3, $4);
COMMIT;

-- Create publication for logical replication
CREATE PUBLICATION outbox_pub FOR TABLE outbox;

-- On subscriber
CREATE SUBSCRIPTION outbox_sub
CONNECTION 'host=primary port=5432 dbname=mydb'
PUBLICATION outbox_pub;