Event Sourcing Implementation

Build event-sourced systems that store every state change as an immutable event. Covers event stores, aggregate reconstruction, snapshots, projections, temporal queries, and the patterns that make event sourcing practical in production.

Event sourcing is an architectural pattern where system state is derived by replaying a sequence of events rather than storing current state directly. This approach matters when you need audit trails, complex business logic, or the ability to reconstruct historical states. Implementing event sourcing requires careful attention to event storage, projection management, and consistency models.

Storing Events

Events are stored in an append-only log. The fundamental data structure is a sequence of immutable event records with a unique identifier and timestamp. Each event must include:

Event Storage Schema

CREATE TABLE events (
    id UUID PRIMARY KEY,
    stream_id UUID NOT NULL,
    version BIGINT NOT NULL,
    event_type VARCHAR(255) NOT NULL,
    payload JSONB NOT NULL,
    metadata JSONB,
    created_at TIMESTAMP WITH TIME ZONE NOT NULL DEFAULT NOW(),
    INDEX (stream_id, version),
    UNIQUE (stream_id, version)
);

Event Creation

Events are written to the store using a transaction that ensures atomicity. The event ID must be generated before writing.

INSERT INTO events (id, stream_id, version, event_type, payload, metadata)
VALUES ('123e4567-e89b-12d3-a456-426614174000', 'abc123', 1, 'UserCreated', 
        '{"name": "John Doe", "email": "[email protected]"}',
        '{"actor": "system", "correlation_id": "xyz789"}');

Event Stream Management

Each stream represents a business entity. When creating a new stream, initialize it with the first event.

-- Create stream by writing initial event
INSERT INTO events (id, stream_id, version, event_type, payload)
VALUES ('uuid-for-stream', 'stream-identifier', 1, 'StreamCreated', '{}');

Reading Events

Stream Replay

To reconstruct a stream's state, read all events for that stream in order.

SELECT * FROM events 
WHERE stream_id = 'abc123' 
ORDER BY version ASC;

Event Filtering

Filter events by type or time range using standard SQL.

SELECT * FROM events 
WHERE stream_id = 'abc123' 
  AND event_type IN ('UserCreated', 'UserUpdated')
  AND created_at BETWEEN '2023-01-01' AND '2023-12-31'
ORDER BY version ASC;

Event Batching

When reading large streams, process events in batches to avoid memory issues.

SELECT * FROM events 
WHERE stream_id = 'abc123' 
  AND version > 1000 
  AND version <= 2000
ORDER BY version ASC;

Projection Management

Projections transform event streams into queryable views. They must be idempotent and handle replay correctly.

Projection Schema

CREATE TABLE projections (
    name VARCHAR(255) PRIMARY KEY,
    last_processed_version BIGINT NOT NULL DEFAULT 0,
    last_processed_at TIMESTAMP WITH TIME ZONE NOT NULL DEFAULT NOW(),
    error_message TEXT,
    error_at TIMESTAMP WITH TIME ZONE
);

Projection Implementation

-- Example: User projection
CREATE TABLE users (
    id UUID PRIMARY KEY,
    name VARCHAR(255),
    email VARCHAR(255),
    created_at TIMESTAMP WITH TIME ZONE NOT NULL DEFAULT NOW()
);

Projection Update Logic

Projection updates must handle:

  1. Idempotency - same event processed multiple times
  2. Error recovery - partial failures during processing
  3. Consistency - atomic updates to projection and metadata
-- Update projection with error handling
BEGIN;
UPDATE projections 
SET last_processed_version = 5,
    last_processed_at = NOW()
WHERE name = 'UserProjection';

-- Apply event transformations
INSERT INTO users (id, name, email)
VALUES ('user-uuid', 'John Doe', '[email protected]');

-- Update projection metadata
UPDATE projections 
SET error_message = NULL,
    error_at = NULL
WHERE name = 'UserProjection';
COMMIT;

Event Versioning

Events must evolve over time. Each event type has a version number that allows backward compatibility.

Event Type Schema

CREATE TABLE event_types (
    name VARCHAR(255) PRIMARY KEY,
    version INT NOT NULL DEFAULT 1,
    schema JSONB,
    created_at TIMESTAMP WITH TIME ZONE NOT NULL DEFAULT NOW()
);

Migration Strategy

When changing an event type, create a new version and handle both versions during transition.

-- Create new event type version
INSERT INTO event_types (name, version, schema)
VALUES ('UserUpdated', 2, 
        '{"properties": {"name": {"type": "string"}, "email": {"type": "string"}, "updated_at": {"type": "string"}}}');

Error Handling

Replay Failures

When a projection fails during replay, it must track the error state.

UPDATE projections 
SET error_message = 'Failed to process event 123',
    error_at = NOW()
WHERE name = 'UserProjection';

Partial Success Recovery

If a batch of events partially succeeds, store the last successfully processed version.

-- When processing events 1-100 and event 50 fails:
UPDATE projections 
SET last_processed_version = 49,
    error_message = 'Event 50 failed',
    error_at = NOW()
WHERE name = 'UserProjection';

Consistency Models

Eventual Consistency

Projections may lag behind the event store. The system must handle queries against potentially stale data.

Transactional Writes

When writing events, ensure all related events are committed together.

BEGIN;
-- Write multiple events for a single business operation
INSERT INTO events (id, stream_id, version, event_type, payload)
VALUES 
('uuid1', 'stream1', 1, 'OrderCreated', '{}'),
('uuid2', 'stream1', 2, 'PaymentProcessed', '{}');

-- Update projection
UPDATE projections SET last_processed_version = 2 WHERE name = 'OrderProjection';
COMMIT;

Projection Rebuilding

Full Rebuild Process

Rebuilding a projection requires:

  1. Resetting the projection state
  2. Replaying all events from beginning
  3. Updating metadata when complete
-- Reset projection
UPDATE projections 
SET last_processed_version = 0,
    last_processed_at = NOW(),
    error_message = NULL,
    error_at = NULL
WHERE name = 'UserProjection';

-- Rebuild by replaying events
-- (Implementation varies by system)

Incremental Updates

Projections should support incremental updates for performance.

-- Process new events since last checkpoint
SELECT * FROM events 
WHERE stream_id = 'abc123' 
  AND version > 1000
ORDER BY version ASC;

Event Replay Operations

Time-based Replay

Replay events from a specific point in time.

SELECT * FROM events 
WHERE created_at >= '2023-06-01' 
ORDER BY version ASC;

Stream-based Replay

Replay all events for a specific stream.

SELECT * FROM events 
WHERE stream_id = 'abc123' 
ORDER BY version ASC;

Common Failure Patterns

Duplicate Event Processing

Events may be replayed multiple times due to transient failures. Projections must be idempotent.

Version Mismatch

When event schemas change, old projections may not handle new versions correctly.

Projection Locking

Concurrent projection updates can cause race conditions. Use optimistic locking or advisory locks.

-- Example of advisory lock usage
SELECT pg_advisory_lock(123);
-- Process events
SELECT pg_advisory_unlock(123);

Memory Exhaustion

Large projections can exhaust memory during replay. Implement streaming and batching.

Performance Considerations

Indexing Strategy

-- Essential indexes for event store
CREATE INDEX idx_events_stream_version ON events(stream_id, version);
CREATE INDEX idx_events_created_at ON events(created_at);
CREATE INDEX idx_events_type ON events(event_type);

Query Optimization

Use efficient queries to minimize database load.

-- Efficient stream reading
SELECT id, version, event_type, payload 
FROM events 
WHERE stream_id = 'abc123' 
  AND version BETWEEN 100 AND 200
ORDER BY version ASC;

Batch Processing

Process events in batches to reduce overhead.

-- Process 100 events at a time
SELECT * FROM events 
WHERE stream_id = 'abc123' 
  AND version > 500 
  AND version <= 600
ORDER BY version ASC;

Operational Concerns

Monitoring

Track projection lag and error rates.

-- Check projection status
SELECT name, last_processed_version, 
       NOW() - last_processed_at as lag,
       error_message
FROM projections;

Backup Strategy

Events are the source of truth. Regular backups must preserve event integrity.

Rollback Procedures

When problems occur, have clear procedures to rollback or recover.

Data Integrity

Event Validation

Validate events before storage to prevent corruption.

-- Example validation constraint
ALTER TABLE events ADD CONSTRAINT check_event_payload
CHECK (payload IS NOT NULL AND payload != '{}');

Transactional Storage

Use database transactions to ensure event consistency.

BEGIN;
INSERT INTO events (id, stream_id, version, event_type, payload)
VALUES ('uuid', 'stream1', 1, 'EventName', '{}');

-- Update projection atomically
UPDATE projections SET last_processed_version = 1 WHERE name = 'ProjectionName';
COMMIT;

Configuration Management

Projection Settings

-- Configure projection parameters
CREATE TABLE projection_config (
    name VARCHAR(255) PRIMARY KEY,
    batch_size INT DEFAULT 100,
    max_retries INT DEFAULT 3,
    retry_delay_seconds INT DEFAULT 5
);

Event Store Configuration

-- Event store tuning parameters
CREATE TABLE event_store_config (
    key VARCHAR(255) PRIMARY KEY,
    value TEXT
);

The event sourcing implementation requires careful attention to storage, consistency, and projection management. Proper error handling and replay capabilities are essential for maintaining data integrity.

This page was rewritten on 15 October 2026. It replaced a templated version that was not specific to its own title. How these pages are written and checked. If anything here is wrong, tell us at [email protected] and we will correct it.