Lightweight vs full event sourcing

Capsule overview

So far you've seen two approaches for preserving history: the audit log (capsule 03), which stores the change, and the history table (capsule 04), which stores the complete row with temporal marks. The third approach is radically different: instead of storing the state or the change, storing the events that generated the changes. The tasks table stops being the source of truth — it becomes a projection derived from events like TaskCreated, TaskTitleChanged, TaskAssigned, TaskClosed. The current state gets reconstructed by applying the events in order.

It's the approach Martin Fowler called "Event Sourcing." In its "full" form (with CQRS, a dedicated event store, asynchronous projections, snapshots for optimization), it's the foundation of systems like many financial microservices, Uber-style platforms, or any system where "what happened" matters more than "how it is now." In its "lightweight" form (an append-only table in PostgreSQL, synchronous replay to rebuild), it's something you can implement in an afternoon and use for specific cases.

This capsule has an unconventional message: for auditing a table in a monolithic API, event sourcing is typically over-engineering. Triggers (capsule 03) give you auditing with less complexity. But there are cases where event sourcing is the right answer — event-driven systems that already have an event bus, domains where the history of events is what the business models, situations where "undo" or "branching" history are features. You're going to learn to recognize those cases. You're also going to implement the lightweight version to understand the pattern in code, and you're going to see the real cost (expensive replay, snapshot complexity, evolving the event schema) that would lead you to decide when to migrate to the full approach.

By the end you'll have the judgment not to fall into the "event sourcing is what's modern, we have to use it" trap, and runnable code for the lightweight version for cases where it does apply.

Mental model: the bank account

The classic analogy for event sourcing is the bank account. A bank doesn't store "balance: $1,234.56" as the truth. It stores events: "$500 deposit on day 1," "$200 withdrawal on day 3," "$1000 deposit on day 7." The balance is a projection — the sum of the events up to today.

Why does the bank do it this way? Because the bank's business IS the events. Regulation asks for the complete list of transactions. Customers ask for statements. Auditing requires demonstrating that the balance is computed correctly from verified events. If the bank stored only "current balance" and the events as a side note, it would lose the source of truth.

Compare with TaskFlow: is your task app's business "the changes that happened to the tasks"? Almost certainly not. The business is "tasks that get done." The current state is the truth; the changes are useful metadata for auditing/debugging but they aren't the domain. That's why triggers + an audit log are enough — you preserve the metadata without inverting the model.

A simple heuristic: if describing your domain without mentioning events sounds weird or false ("the bank exists to process transactions — the balance is just a view"), event sourcing is a candidate. If describing it that way sounds forced ("the task app exists to process events about tasks"), it isn't.

The spectrum: lightweight to full

Event sourcing isn't binary. There's a spectrum of implementations, from "an events table in PostgreSQL" to "a system with a dedicated event store, CQRS, and asynchronous projections." Let's define it.

The "lightweight" version

• An append-only events table in the same DB as everything else.
• Events are immutable: never UPDATE, never DELETE.
• The "current state" still lives in a regular table (tasks), updated in the same transaction as the event's INSERT.
• Reconstructing the state from events is possible but rare (typically for debugging or specific reports).
• There's no event bus, no subscribers, no CQRS.

When it applies: when you want the guarantee "I have the complete log of events" but the domain is still modeled as current state. Useful for apps where a few specific cases (critical operations requiring replay, small domains like billing) benefit from the approach.

The "full" version

• A dedicated event store (EventStoreDB, Kafka with compaction, a custom system). The tasks table may not exist as a main table.
• The current state is a projection — a view derived from the event log, maintained by an async process.
• User commands emit events; events generate state via projections.
• CQRS (Command Query Responsibility Segregation): commands go to the write model (events); queries go to the read model (projections).
• Snapshots to optimize replaying aggregates with thousands of events.

When it applies: event-driven systems with microservices that already consume an event bus, domains where "history branching" is a feature (hypothetically: "what would happen if I reject this event"), regulation that demands strict immutability of the event log.

It isn't what this capsule teaches you. The full approach requires infrastructure, a team trained in CQRS, and an architectural commitment of months. The capsule teaches you the lightweight version and helps you recognize when (rarely) you should migrate to full.

Implementing lightweight event sourcing

Let's implement the pattern on a tasks table. You're going to see:

1. The append-only events table.
2. How to emit events atomically with the state changes.
3. How to replay events to reconstruct the state.

Schema

-- The domain table (current state). Same as before.
CREATE TABLE tasks (
    id BIGSERIAL PRIMARY KEY,
    title TEXT NOT NULL,
    description TEXT NULL,
    status TEXT NOT NULL DEFAULT 'open',
    priority INTEGER NOT NULL DEFAULT 0,
    assignee_id BIGINT NULL,
    -- Internal version: increments with every event, useful for detecting desynced replay
    version INTEGER NOT NULL DEFAULT 0,
    created_at TIMESTAMPTZ NOT NULL DEFAULT NOW(),
    updated_at TIMESTAMPTZ NOT NULL DEFAULT NOW()
);

-- The events table: append-only, immutable
CREATE TABLE task_events (
    event_id BIGSERIAL PRIMARY KEY,
    -- Which entity
    task_id BIGINT NOT NULL,
    -- What kind of event (in production you'd use an ENUM, here TEXT with a CHECK for simplicity)
    event_type TEXT NOT NULL CHECK (event_type IN (
        'TaskCreated',
        'TaskTitleChanged',
        'TaskDescriptionChanged',
        'TaskStatusChanged',
        'TaskPriorityChanged',
        'TaskAssigned',
        'TaskUnassigned',
        'TaskDeleted'
    )),
    -- Payload: the event's specific data
    payload JSONB NOT NULL,
    -- Incremental version within the entity: guarantees total ordering
    version INTEGER NOT NULL,
    -- When and who
    occurred_at TIMESTAMPTZ NOT NULL DEFAULT NOW(),
    actor_id BIGINT NULL,
    -- Constraint: the version is unique per entity (there can't be two events at the same version)
    UNIQUE (task_id, version)
);

CREATE INDEX idx_task_events_task_id_version
    ON task_events (task_id, version);

CREATE INDEX idx_task_events_occurred_at
    ON task_events (occurred_at DESC);

The schema's decisions:

• version per entity: each event has an incremental version within the entity. Task #5's first event is version 1, the second is version 2, etc. This serves two purposes: detecting gaps (if the replay finds versions 1, 2, 4 but not 3, there's corruption), and concurrency control (similar to optimistic locking).
• UNIQUE (task_id, version): guarantees there aren't two events at the same version for the same entity. If two transactions try to emit an event at version 5, one commits, the other fails. That's what we want.
• payload JSONB: a flexible structure. Each event_type defines what fields it expects. TaskCreated has {title, description, priority}. TaskTitleChanged has {old_title, new_title}. Validating the payload's schema goes in the app's code, not in SQL.
• There's no FK to tasks: intentional. The event log survives even if the entity gets deleted. If you put an FK with CASCADE, you lose the log when you delete.

Emitting events atomically with the changes

The key pattern: every change to the state gets emitted as an event, in the same transaction as the change. If the transaction fails, neither the event nor the change remains. If the transaction succeeds, both remain. Atomicity guaranteed by PostgreSQL.

# app/event_sourcing.py
from typing import Optional
from sqlalchemy import text
from sqlalchemy.ext.asyncio import AsyncSession


async def emit_task_event(
    session: AsyncSession,
    task_id: int,
    event_type: str,
    payload: dict,
    actor_id: Optional[int] = None,
) -> int:
    """
    Emits an event atomically. Returns the assigned version.

    It has to be called inside the same transaction as the state change.
    """
    # Compute the next version: max(version) + 1
    result = await session.execute(text("""
        SELECT COALESCE(MAX(version), 0) + 1 AS next_version
        FROM task_events
        WHERE task_id = :tid
    """), {"tid": task_id})
    next_version = result.scalar()

    # Insert the event
    await session.execute(text("""
        INSERT INTO task_events (task_id, event_type, payload, version, actor_id)
        VALUES (:tid, :type, :payload, :version, :actor)
    """), {
        "tid": task_id,
        "type": event_type,
        "payload": payload,
        "version": next_version,
        "actor": actor_id,
    })

    # Update the version in the state table
    await session.execute(text("""
        UPDATE tasks SET version = :v, updated_at = NOW() WHERE id = :tid
    """), {"v": next_version, "tid": task_id})

    return next_version

The complete handler:

import json
from fastapi import Depends, FastAPI, HTTPException
from sqlalchemy.ext.asyncio import AsyncSession

from app.audit_context import audit_context
from app.event_sourcing import emit_task_event
from app.models import Task

app = FastAPI()


@app.post("/tasks", status_code=201)
async def create_task(
    payload: dict,
    db: AsyncSession = Depends(audit_context),
    user_id: int = 0,  # simplified for the example
):
    task = Task(
        title=payload["title"],
        description=payload.get("description"),
        priority=payload.get("priority", 0),
        version=1,
    )
    db.add(task)
    await db.flush()  # to get task.id

    await emit_task_event(
        db,
        task_id=task.id,
        event_type="TaskCreated",
        payload={
            "title": task.title,
            "description": task.description,
            "priority": task.priority,
        },
        actor_id=user_id,
    )

    await db.commit()
    return {"id": task.id, "version": 1}


@app.put("/tasks/{task_id}/title")
async def change_task_title(
    task_id: int,
    payload: dict,
    db: AsyncSession = Depends(audit_context),
    user_id: int = 0,
):
    task = await db.get(Task, task_id)
    if task is None:
        raise HTTPException(404, "Task not found")

    old_title = task.title
    new_title = payload["title"]

    if old_title == new_title:
        return {"id": task_id, "version": task.version, "no_change": True}

    task.title = new_title
    await emit_task_event(
        db,
        task_id=task_id,
        event_type="TaskTitleChanged",
        payload={"old_title": old_title, "new_title": new_title},
        actor_id=user_id,
    )
    await db.commit()
    return {"id": task_id, "version": task.version}

What matters: emit_task_event and the modification to task happen in the same transaction. If one fails (for example, if two concurrent requests try to emit at the same version), both get rolled back. Atomicity maintained.

Replay: reconstructing state from events

async def replay_task_state(session: AsyncSession, task_id: int) -> dict:
    """
    Reconstructs a task's state by applying all of its events.

    Useful for:
    - Debugging: "does the current state match the replay?"
    - Migration: rebuilding projections after a schema change
    - Advanced auditing: "show me the state as it would be without event N"
    """
    result = await session.execute(text("""
        SELECT event_type, payload, version, occurred_at, actor_id
        FROM task_events
        WHERE task_id = :tid
        ORDER BY version ASC
    """), {"tid": task_id})

    state = {
        "id": task_id,
        "title": None,
        "description": None,
        "status": None,
        "priority": None,
        "assignee_id": None,
        "version": 0,
        "deleted": False,
    }

    for row in result:
        state = apply_event(state, row.event_type, row.payload)
        state["version"] = row.version

    return state


def apply_event(state: dict, event_type: str, payload: dict) -> dict:
    """Applies an event to the state. A pure function."""
    new_state = state.copy()

    if event_type == "TaskCreated":
        new_state.update({
            "title": payload["title"],
            "description": payload.get("description"),
            "status": "open",
            "priority": payload.get("priority", 0),
        })
    elif event_type == "TaskTitleChanged":
        new_state["title"] = payload["new_title"]
    elif event_type == "TaskDescriptionChanged":
        new_state["description"] = payload["new_description"]
    elif event_type == "TaskStatusChanged":
        new_state["status"] = payload["new_status"]
    elif event_type == "TaskPriorityChanged":
        new_state["priority"] = payload["new_priority"]
    elif event_type == "TaskAssigned":
        new_state["assignee_id"] = payload["assignee_id"]
    elif event_type == "TaskUnassigned":
        new_state["assignee_id"] = None
    elif event_type == "TaskDeleted":
        new_state["deleted"] = True

    return new_state

End-to-end verification: after running several changes, compare the replay's result against the current state:

async def verify_replay_matches_state(session, task_id):
    replayed = await replay_task_state(session, task_id)
    actual = await session.get(Task, task_id)

    assert replayed["title"] == actual.title
    assert replayed["status"] == actual.status
    assert replayed["priority"] == actual.priority
    # ... etc
    assert replayed["version"] == actual.version

If this passes, the event log is consistent with the state. If not, something got desynchronized (typically a change that didn't emit an event).

The real cost of event sourcing

Here comes the important part: why this approach is NOT the module's default.

Cost 1: expensive replay

A task with 1,000 events requires reading 1,000 rows and applying 1,000 functions to reconstruct the state. An entity with 100,000 events (typical in high-activity systems) can take seconds to replay. Mitigation: periodic snapshots ("store the task's state every 50 events"), but that introduces extra complexity.

Cost 2: changing the event schema

Today you emit TaskAssigned with the payload {assignee_id: 5}. Tomorrow you decide you also have to store the assigned_at. What happens to the 50,000 old events that don't have that field? Options:

• Schema versioning: each event has a schema_version field. The apply_event function handles all the versions (if schema_version == 1: ...). It works but the complexity grows with every change.
• Event migration: reprocess all the old events to add the field. It's a heavy operation and vulnerable to errors.

Compared with an audit log or a history table, where "changing the schema" is an ALTER TABLE, event sourcing has more friction.

Cost 3: direct queries are impossible

Want a "list of tasks with priority > 5"? With an audit log or a history table, a direct SQL query. With pure event sourcing, you have to replay every task (each one requiring all its events). That's why full event sourcing always has projections (pre-computed read models). In the lightweight version, we keep tasks as a regular table and use it for queries — but then the "event source" isn't the only source of truth.

Cost 4: mental complexity

Your team has to think about the domain in terms of events. "The user changed the title" becomes "a TaskTitleChanged was emitted with old/new." Every operation gets modeled first as an event, then as a state change. It's engineering discipline that pays returns in event-driven systems, but it imposes constant overhead in systems that aren't.

Cost 5: integration with ORM frameworks

SQLAlchemy has no native support for event sourcing. Every operation that would normally be session.add(...) becomes "modify state + emit event + commit," which isn't trivial to encapsulate without losing clarity.

When you SHOULD use lightweight event sourcing

Despite the costs, there are legitimate cases:

Case 1: a domain where the events are the product.

• A billing system: "invoice issued," "payment received," "invoice cancelled." Each event is an accounting fact.
• Multi-step workflows: "document sent," "document reviewed," "document approved." The state only makes sense from the sequence.
• Legal traceability: contracts where "it was signed by A on day X" is primary information.

In these cases, modeling as events isn't overhead — it's aligning the code with the domain.

Case 2: systems with an existing event bus.

If your team already consumes Kafka, RabbitMQ, or similar for messaging between services, adding a local events table that gets published to the bus is low overhead. The event source becomes a "first-class citizen" without your app carrying all the complexity.

Case 3: "branching" or "what-if" features.

Some systems need "what would have happened if X." Event sourcing allows replay with modified or filtered events. A rare case, but it exists (financial systems, planning, simulations).

Case 4: strict immutability compliance.

Some regulatory frameworks (notably crypto, certain medical cases) require an immutable event log. Event sourcing is append-only by design; an audit log and a history table are regular tables that technically can be modified.

When NOT to use event sourcing

For auditing a table in a monolithic API: triggers + an audit log (capsule 03) are enough and simpler. Event sourcing is over-engineering.

For "seeing the previous version": history tables (capsule 04) are more direct. Event sourcing requires a replay; a history table has the version ready.

For an activity timeline in the UI: an audit log with a good query is enough. Event sourcing introduces an unnecessary layer.

When your team doesn't know CQRS or event-driven design: the learning curve + the operational cost kills the team. Only embark on it with an explicit commitment to learn.

Why does this matter in real work?

1. It's the most expensive architectural decision to reverse in this module. Migrating from an audit log to a history table is trivial (add a table, add a trigger). Migrating from either one to event sourcing requires redesigning the domain model. Only embark on it when you understand it well.

2. It's the approach the industry oversells the most. "Event sourcing is what's modern" is a recurring slogan with no context. Knowing WHEN not to use it is a senior skill — most teams that adopt event sourcing out of hype end up regretting it in 12-18 months.

3. When it applies, it's transformational. In the right cases (billing, workflows, event-driven systems), event sourcing isn't overhead — it's the right answer and it simplifies the code long term. Recognizing those cases is valuable.

4. The conversation with your team changes. "Let's audit with event sourcing" sounds tempting. "PostgreSQL triggers + an audit log is enough for our case, event sourcing would be over-engineering" requires an argument. The capsule gives you the arguments.

Traps and common mistakes

Mistake 1 (conceptual): "event sourcing gives me auditing for free, so I want it too"

Symptom: the team decides to adopt event sourcing as a replacement for the audit log because "it's the same thing and it's cleaner."

Why it's confusing: event sourcing produces an event log similar to an audit log. But the operational complexity is much higher: expensive replay, schema versioning, mental complexity.

How to tell: ask yourself honestly, "does the business model the domain as events?". If the answer is no, an audit log is the right answer.

How to fix it: keep the audit log for auditing. If the domain genuinely justifies event sourcing, embark on it aware of the cost, not as an auditing shortcut.

Mistake 2 (operational): replay without snapshots becomes unsustainable

Symptom: after months in production, reconstructing the state of some entities takes seconds. The "view full history" feature is slow.

Why it happens: entities with thousands of events require reading and applying all of them. Without an intermediate snapshot, the replay is O(N).

How to tell: measure the time of replay_task_state for entities with many events. If it exceeds 100ms for typical cases, the bug will materialize soon.

How to fix it: implement snapshots:

CREATE TABLE task_snapshots (
    snapshot_id BIGSERIAL PRIMARY KEY,
    task_id BIGINT NOT NULL,
    version INTEGER NOT NULL,
    state JSONB NOT NULL,
    created_at TIMESTAMPTZ NOT NULL DEFAULT NOW(),
    UNIQUE (task_id, version)
);

async def replay_with_snapshot(session, task_id):
    # Look for the most recent snapshot
    snapshot = await session.execute(text("""
        SELECT version, state FROM task_snapshots
        WHERE task_id = :tid ORDER BY version DESC LIMIT 1
    """), {"tid": task_id})
    snap = snapshot.first()

    if snap:
        state = snap.state
        from_version = snap.version + 1
    else:
        state = initial_state(task_id)
        from_version = 1

    # Apply events from from_version
    events = await session.execute(text("""
        SELECT event_type, payload, version FROM task_events
        WHERE task_id = :tid AND version >= :v ORDER BY version ASC
    """), {"tid": task_id, "v": from_version})

    for event in events:
        state = apply_event(state, event.event_type, event.payload)
        state["version"] = event.version

    return state

A job that creates snapshots every N events:

# Job: create a snapshot when tasks hit certain milestones
async def maybe_snapshot(session, task_id):
    last_snapshot_version = await get_last_snapshot_version(session, task_id)
    current_version = await get_current_version(session, task_id)

    if current_version - last_snapshot_version >= 50:
        state = await replay_task_state(session, task_id)
        await session.execute(text("""
            INSERT INTO task_snapshots (task_id, version, state)
            VALUES (:tid, :v, :state)
            ON CONFLICT (task_id, version) DO NOTHING
        """), {"tid": task_id, "v": current_version, "state": state})

Mistake 3 (conceptual): mutating past events

Symptom: a developer discovers a bug in an old event (e.g. a payload with a wrongly-named field) and decides to UPDATE the event to fix it.

Why it happens: the team didn't internalize "events are immutable."

How to tell: review the DB's logs. Any UPDATE task_events or DELETE FROM task_events is a serious violation of the pattern.

How to fix it: never modify past events. If an old event has an incorrect format, add compatibility logic in apply_event:

def apply_event(state, event_type, payload):
    if event_type == "TaskTitleChanged":
        # Compatibility with old events that had "title" instead of "new_title"
        new_title = payload.get("new_title") or payload.get("title")
        state["title"] = new_title
    # ...

To prevent mutation, consider granting only INSERT to the app:

REVOKE UPDATE, DELETE ON task_events FROM app_user;
GRANT INSERT, SELECT ON task_events TO app_user;

Mistake 4 (operational): losing atomicity between the event and the state

Symptom: occasionally a task's state doesn't match the replay. Investigation: there are events that exist but the state change wasn't applied (or vice versa).

Why it happens: the code emits the event and modifies the state in separate transactions. A failure in between leaves an inconsistency.

How to tell: run verify_replay_matches_state on a sample of entities. If there are divergences, this is the bug.

How to fix it: make sure both changes are in the same transaction. In SQLAlchemy:

# CORRECT: both changes in the same session, the same commit
async def update_task_status(session, task_id, new_status, user_id):
    task = await session.get(Task, task_id)
    old_status = task.status
    task.status = new_status

    await emit_task_event(
        session, task_id,
        event_type="TaskStatusChanged",
        payload={"old_status": old_status, "new_status": new_status},
        actor_id=user_id,
    )

    await session.commit()  # a single commit for both

# INCORRECT: two separate commits
async def update_task_status_BAD(session, task_id, new_status, user_id):
    task = await session.get(Task, task_id)
    task.status = new_status
    await session.commit()  # commit of the state

    await emit_task_event(...)
    await session.commit()  # commit of the event — if it fails here, inconsistency

Mistake 5 (conceptual): treating the lightweight version as "we've done event sourcing"

Symptom: the team believes it has an event-sourced architecture, proposes adding CQRS, asynchronous projections and the like — without understanding that its lightweight version isn't real event sourcing.

Why it's confusing: both have an "events table." The fundamental difference: in the lightweight version, the tasks table is still the source of truth for queries. In real event sourcing, the events are the only source of truth and tasks (if it exists) is just a projection.

How to tell: ask yourself, "could I delete the tasks table and rebuild it from events with no information lost?". If in your lightweight implementation the answer is "no" (because some fields only live in tasks), then it isn't real event sourcing, it's "an audit log with typed events."

How to fix it: adjust the expectations. The lightweight version is useful but it must not be a justification for investing in full event sourcing infrastructure. If you're going for CQRS and projections, that's a different project with an explicit architectural commitment.

Exercises

Exercise 1: classify domains

For each domain, decide whether event sourcing is appropriate, lightweight, or not applicable. Justify it.

a) A TaskFlow-style task app (managing a TODO list). b) A recurring billing system (subscriptions, charges, refunds). c) E-commerce (a product catalog + a cart). d) A version control system like Git. e) A legal workflow (document signing by multiple parties). f) Health monitoring (sensor readings every minute). g) An email account (inbox, read, archived).

Exercise 2: implement emit + replay for a new operation

Your app has the "assign a task to a user" operation. Implement:

a) The TaskAssigned event with its payload. b) The FastAPI handler that changes the state and emits the event atomically. c) The TaskAssigned case in apply_event so the replay works.

# app/events.py — the event's conceptual definition
TASK_ASSIGNED_PAYLOAD_SCHEMA = {
    "type": "object",
    "properties": {
        "old_assignee_id": {"type": ["integer", "null"]},
        "new_assignee_id": {"type": "integer"},
    },
    "required": ["new_assignee_id"],
}


# app/main.py — the handler
@app.put("/tasks/{task_id}/assignee")
async def assign_task(
    task_id: int,
    payload: dict,
    db: AsyncSession = Depends(audit_context),
    user_id: int = 0,
):
    new_assignee_id = payload["assignee_id"]

    task = await db.get(Task, task_id)
    if task is None:
        raise HTTPException(404, "Task not found")

    old_assignee_id = task.assignee_id

    if old_assignee_id == new_assignee_id:
        return {"id": task_id, "version": task.version, "no_change": True}

    task.assignee_id = new_assignee_id

    await emit_task_event(
        db,
        task_id=task_id,
        event_type="TaskAssigned",
        payload={
            "old_assignee_id": old_assignee_id,
            "new_assignee_id": new_assignee_id,
        },
        actor_id=user_id,
    )

    await db.commit()
    return {"id": task_id, "version": task.version, "assignee_id": new_assignee_id}


# app/event_sourcing.py — add to apply_event
def apply_event(state, event_type, payload):
    new_state = state.copy()

    if event_type == "TaskCreated":
        # ... as before
        pass
    elif event_type == "TaskAssigned":
        new_state["assignee_id"] = payload["new_assignee_id"]
    elif event_type == "TaskUnassigned":
        new_state["assignee_id"] = None
    # ... the rest

    return new_state

async def test_assign_emits_event_and_replays(session):
    # Create the task
    res = await client.post("/tasks", json={"title": "T1"})
    task_id = res.json()["id"]

    # Assign
    await client.put(f"/tasks/{task_id}/assignee", json={"assignee_id": 5})

    # Verify the replay matches the state
    state_replayed = await replay_task_state(session, task_id)
    assert state_replayed["assignee_id"] == 5

    state_actual = await session.get(Task, task_id)
    assert state_actual.assignee_id == 5
    assert state_actual.assignee_id == state_replayed["assignee_id"]

Exercise 3: detect inconsistency between events and state

Write a script that scans every task and verifies that the replay matches the current state. Report the divergences.

# scripts/verify_event_consistency.py
import asyncio
from sqlalchemy import select

from app.db import AsyncSessionLocal
from app.models import Task
from app.event_sourcing import replay_task_state


async def main():
    async with AsyncSessionLocal() as session:
        tasks = await session.execute(select(Task))
        all_tasks = tasks.scalars().all()

        divergences = []
        for task in all_tasks:
            replayed = await replay_task_state(session, task.id)
            if (
                replayed["title"] != task.title
                or replayed["status"] != task.status
                or replayed["priority"] != task.priority
                or replayed["assignee_id"] != task.assignee_id
                or replayed["version"] != task.version
            ):
                divergences.append({
                    "task_id": task.id,
                    "actual": {
                        "title": task.title,
                        "status": task.status,
                        "priority": task.priority,
                        "assignee_id": task.assignee_id,
                        "version": task.version,
                    },
                    "replayed": replayed,
                })

        print(f"Total tasks: {len(all_tasks)}")
        print(f"Divergences found: {len(divergences)}")
        for d in divergences[:10]:  # the first 10
            print(f"  Task #{d['task_id']}:")
            print(f"    Actual:   {d['actual']}")
            print(f"    Replayed: {d['replayed']}")


if __name__ == "__main__":
    asyncio.run(main())

Exercise 4: argue against event sourcing in a code review

Your teammate proposes: "Let's re-architect the tasks module with pure event sourcing. It's more modern." Argue why it's over-engineering for TaskFlow.

Exercise 5: implement snapshots to speed up replay

Implement the snapshot system: the table, the function that creates a snapshot, the replay that uses snapshots.

-- The snapshots table
CREATE TABLE task_snapshots (
    snapshot_id BIGSERIAL PRIMARY KEY,
    task_id BIGINT NOT NULL,
    version INTEGER NOT NULL,
    state JSONB NOT NULL,
    created_at TIMESTAMPTZ NOT NULL DEFAULT NOW(),
    UNIQUE (task_id, version)
);

CREATE INDEX idx_task_snapshots_task_id_version
    ON task_snapshots (task_id, version DESC);

# app/event_sourcing.py
SNAPSHOT_INTERVAL = 50  # take a snapshot every 50 events


async def create_snapshot(session: AsyncSession, task_id: int):
    """Creates a snapshot of the current state rebuilt from events."""
    state = await replay_task_state_from_scratch(session, task_id)

    await session.execute(text("""
        INSERT INTO task_snapshots (task_id, version, state)
        VALUES (:tid, :v, :state)
        ON CONFLICT (task_id, version) DO NOTHING
    """), {"tid": task_id, "v": state["version"], "state": state})


async def replay_task_state(session: AsyncSession, task_id: int) -> dict:
    """
    Replay with snapshots: starts from the most recent snapshot and applies
    later events. Much faster than replaying from scratch for entities
    with many events.
    """
    # 1. Look for the most recent snapshot
    snapshot_result = await session.execute(text("""
        SELECT version, state FROM task_snapshots
        WHERE task_id = :tid
        ORDER BY version DESC
        LIMIT 1
    """), {"tid": task_id})
    snapshot_row = snapshot_result.first()

    if snapshot_row:
        state = dict(snapshot_row.state)  # a mutable copy
        from_version = snapshot_row.version + 1
    else:
        state = {
            "id": task_id, "title": None, "description": None,
            "status": None, "priority": None, "assignee_id": None,
            "version": 0, "deleted": False,
        }
        from_version = 1

    # 2. Apply events from from_version
    events_result = await session.execute(text("""
        SELECT event_type, payload, version
        FROM task_events
        WHERE task_id = :tid AND version >= :v
        ORDER BY version ASC
    """), {"tid": task_id, "v": from_version})

    for row in events_result:
        state = apply_event(state, row.event_type, row.payload)
        state["version"] = row.version

    return state


async def replay_task_state_from_scratch(session: AsyncSession, task_id: int) -> dict:
    """Replay from the first event, without using snapshots. Useful for creating the snapshot."""
    state = {
        "id": task_id, "title": None, "description": None,
        "status": None, "priority": None, "assignee_id": None,
        "version": 0, "deleted": False,
    }
    events = await session.execute(text("""
        SELECT event_type, payload, version FROM task_events
        WHERE task_id = :tid ORDER BY version ASC
    """), {"tid": task_id})
    for row in events:
        state = apply_event(state, row.event_type, row.payload)
        state["version"] = row.version
    return state


async def maybe_create_snapshot(session: AsyncSession, task_id: int):
    """Call after emit_task_event. Creates a snapshot if N events have passed."""
    last_snap_result = await session.execute(text("""
        SELECT COALESCE(MAX(version), 0) AS last_version
        FROM task_snapshots WHERE task_id = :tid
    """), {"tid": task_id})
    last_snap_version = last_snap_result.scalar()

    current_version_result = await session.execute(text("""
        SELECT COALESCE(MAX(version), 0) AS current
        FROM task_events WHERE task_id = :tid
    """), {"tid": task_id})
    current_version = current_version_result.scalar()

    if current_version - last_snap_version >= SNAPSHOT_INTERVAL:
        await create_snapshot(session, task_id)

import time

async def benchmark(session, task_id):
    start = time.time()
    state_with = await replay_task_state(session, task_id)
    time_with = time.time() - start

    start = time.time()
    state_without = await replay_task_state_from_scratch(session, task_id)
    time_without = time.time() - start

    print(f"With snapshot: {time_with*1000:.1f}ms")
    print(f"Without snapshot: {time_without*1000:.1f}ms")
    assert state_with == state_without  # the same result

Summary and next step

In this capsule you learned:

• Event sourcing models the domain as a sequence of events, not as current state. The state gets rebuilt by replaying events. It's the right approach when the business models events as the product.
• The implementation spectrum runs from "an append-only table in PostgreSQL" (lightweight) to "CQRS + an event store + asynchronous projections" (full). This capsule teaches you the lightweight version and gives you criteria for not embarking on full without cause.
• For auditing a table in a monolithic API, event sourcing is over-engineering. Triggers + an audit log (capsule 03) are simpler and enough. Event sourcing's costs (replay, schema versioning, mental complexity) aren't justified for that use case.
• Cases where event sourcing DOES apply: domains where events are the product (billing, legal workflows), systems with an existing event bus, "history branching" features, strict immutability compliance.
• The lightweight version's key detail: atomicity between the event and the state change. The same session, the same commit. Any decoupling causes inconsistencies.
• Events are immutable. Never UPDATE or DELETE. If you need to modify an old event, add compatibility logic in apply_event.
• Snapshots are an essential tool for keeping replay from getting expensive. Implement them once the events per entity exceed ~50.
• Verifying consistency between the replay and the current state has to be a recurring job. Without that verification, the divergences accumulate silently.

Before moving on you should be able to:

• Recognize whether a domain is a candidate for event sourcing (state-based vs event-based).
• Implement the lightweight version (an events table + emit + replay) for a new operation.
• Argue why event sourcing is over-engineering for typical auditing cases.
• Implement snapshots to optimize replay when necessary.
• Detect inconsistencies between events and state with a verification script.

Next capsule — Auditing with SQLAlchemy event listeners. So far the three approaches (audit log, history table, lightweight event sourcing) live in SQL — PostgreSQL triggers do the work. But there are cases where triggers aren't an option: managed databases that restrict custom functions, teams that prefer keeping all the logic in Python, tests that need to mock the behavior. You're going to learn to implement auditing with SQLAlchemy 2.0's event.listens_for(Session, "after_flush"), its differences from triggers, and when each approach wins.

Resources

1. Martin Fowler — "Event Sourcing" — the canonical article. Required reading. Spoiler: Fowler presents it with caution, not as "the universal solution."
2. Greg Young — "CQRS, Task Based UIs, Event Sourcing agh!" — the coiner of the term introduces event sourcing and CQRS. Dense material but clarifying.
3. Microsoft Docs — Event Sourcing pattern — a description of the pattern with explicit trade-offs. Useful for recognizing when it applies and when it doesn't.
4. EventStoreDB Documentation — the best-known dedicated event store implementation. Useful for understanding what the "full" version is and what problems it solves.
5. PostgreSQL Documentation — JSONB — the reference for designing event payloads. Operators like ->, ->>, ? are basics.
6. Vlad Mihalcea — "Event Sourcing Pattern" — an implementation in Java/JPA. Useful for comparing approaches across stacks and seeing the common patterns.
7. Confluent — "Event Sourcing in Practice" — the event sourcing perspective with Kafka as the event bus. Useful for understanding what happens when event sourcing goes beyond a single DB.
8. Postgres Audit Trigger sample — a reference implementation of auditing with triggers; useful for contrasting with event sourcing and seeing why triggers are simpler for typical cases.

Module 3 — SQL Patterns for Production APIs Guide

Next capsule: Auditing with SQLAlchemy event listeners — when triggers aren't an option.