event-sourcing

name	event-sourcing
description	Implement event sourcing pattern where state is derived from an immutable sequence of events. Outputs event store design, aggregate patterns, projection builders, and command/event handlers.
argument-hint	["domain","aggregate types","projection requirements","event store technology"]
allowed-tools	Read, Write, Bash

Event Sourcing

Event sourcing stores every state change as an immutable event rather than overwriting current state. The current state is always derived by replaying events. This enables a complete audit log, temporal queries, and the ability to rebuild any projection from scratch.

When to Use Event Sourcing

Use when:

• Complete audit log is required (finance, healthcare, compliance)
• Temporal queries: "what was the state at 3pm last Tuesday?"
• Multiple read models needed from the same write data
• Complex domain with many state transitions

Don't use when:

• Simple CRUD with no history requirements
• Team unfamiliar with the pattern — learning curve is steep
• High write throughput + simple queries — overhead isn't worth it

Process

1. Define the domain events — past-tense, immutable facts (OrderPlaced, PaymentProcessed).
2. Design aggregates — domain objects that enforce invariants and emit events.
3. Implement the event store — append-only log with optimistic concurrency.
4. Build projections — read models derived from event streams.
5. Handle commands — validate → load aggregate → apply command → persist events.
6. Rebuild projections — replay events to create new or fix existing read models.

Output Format

Event Definitions

# domain/events.py
from dataclasses import dataclass, field
from datetime import datetime, timezone
from typing import Any
import uuid

@dataclass(frozen=True)
class DomainEvent:
    """Base class for all domain events. Immutable."""
    event_id: str = field(default_factory=lambda: str(uuid.uuid4()))
    occurred_at: datetime = field(default_factory=lambda: datetime.now(timezone.utc))
    aggregate_id: str = ""
    aggregate_version: int = 0

@dataclass(frozen=True)
class OrderPlaced(DomainEvent):
    user_id: str = ""
    items: tuple = field(default_factory=tuple)  # Immutable
    total_cents: int = 0
    shipping_address: dict = field(default_factory=dict)

@dataclass(frozen=True)
class OrderConfirmed(DomainEvent):
    confirmed_by: str = ""

@dataclass(frozen=True)
 ():
    payment_provider:  = 
    payment_id:  = 
    amount_cents:  = 


 ():
    reason:  = 
    error_code:  = 


 ():
    carrier:  = 
    tracking_number:  = 
    estimated_delivery:  = 


 ():
    reason:  = 
    cancelled_by:  = 


 ():
    refund_amount_cents:  = 
    refund_id:  = 

Aggregate

# domain/order_aggregate.py
from typing import Optional
from dataclasses import dataclass, field
from .events import *

class InvalidStateTransition(Exception):
    pass

class OrderAggregate:
    """
    Order aggregate — enforces business invariants.
    State is derived entirely from replayed events.
    Never persisted directly — only events are stored.
    """
    
    def __init__(self, order_id: str):
        self.order_id = order_id
        self.version = 0           # Optimistic concurrency version
        self._events: list[DomainEvent] = []  # Uncommitted events
        
        # State derived from events
        self.status: Optional[str] = None
        self.user_id: Optional[str] = None
        self.items: list = []
        self.total_cents: int = 0
        self.payment_id: Optional[str] = None
        self.is_paid: bool = 
    
    
    
     ():
        
         .status   :
             InvalidStateTransition()
        
        
          items:
             ValueError()
         total_cents <= :
             ValueError()
        
        ._apply(OrderPlaced(
            aggregate_id=.order_id,
            user_id=user_id,
            items=(items),
            total_cents=total_cents,
            shipping_address=shipping_address,
        ))
    
     ():
         .status != :
             InvalidStateTransition()
        
        ._apply(PaymentProcessed(
            aggregate_id=.order_id,
            payment_provider=payment_provider,
            payment_id=payment_id,
            amount_cents=amount_cents,
        ))
    
     ():
         .status != :
             InvalidStateTransition()
        
        ._apply(PaymentFailed(
            aggregate_id=.order_id,
            reason=reason,
            error_code=error_code,
        ))
    
     ():
         .status != :
             InvalidStateTransition()
        
        ._apply(OrderShipped(
            aggregate_id=.order_id,
            carrier=carrier,
            tracking_number=tracking_number,
            estimated_delivery=estimated_delivery,
        ))
    
     ():
         .status  (, , ):
             InvalidStateTransition()
        
        ._apply(OrderCancelled(
            aggregate_id=.order_id,
            reason=reason,
            cancelled_by=cancelled_by,
        ))
    
    
    
     ():
        
        
        handler = {
            OrderPlaced: ._on_order_placed,
            PaymentProcessed: ._on_payment_processed,
            PaymentFailed: ._on_payment_failed,
            OrderShipped: ._on_order_shipped,
            OrderCancelled: ._on_order_cancelled,
        }.get((event))
        
         handler:
            handler(event)
        
        .version += 
        
          is_replay:
            ._events.append(event)  
    
     ():
        .status = 
        .user_id = event.user_id
        .items = (event.items)
        .total_cents = event.total_cents
    
     ():
        .status = 
        .payment_id = event.payment_id
        .is_paid = 
    
     ():
        .status = 
    
     ():
        .status = 
    
     ():
        .status = 
    
    
    

     () -> :
        
        aggregate = cls(order_id)
         event  events:
            aggregate._apply(event, is_replay=)
         aggregate
    
     () -> [DomainEvent]:
         (._events)
    
     ():
        ._events.clear()

Event Store

# infrastructure/event_store.py
import json
import asyncpg
from datetime import datetime, timezone
from typing import Type

class OptimisticConcurrencyError(Exception):
    pass

class EventStore:
    """
    Append-only event store backed by PostgreSQL.
    Optimistic concurrency via expected_version.
    """
    
    def __init__(self, pool: asyncpg.Pool):
        self.pool = pool
    
    async def append_events(
        self,
        aggregate_id: str,
        events: list,
        expected_version: int,
    ):
        """
        Append events to the store.
        expected_version: the version the caller believes the aggregate is at.
        Raises OptimisticConcurrencyError if another writer has appended events since.
        """
        async with self.pool.acquire() as conn:
            async with conn.transaction():
                # Check current version (optimistic lock)
                current_version = await conn.fetchval(
                    "SELECT COALESCE(MAX(version), 0) FROM events WHERE aggregate_id = $1",
                    aggregate_id
                )
                
                if current_version != expected_version:
                    raise OptimisticConcurrencyError(
                        f"Concurrency conflict: expected version , "
                        
                    )
                
                
                 i, event  (events):
                    version = expected_version + i + 
                     conn.execute(
                        ,
                        event.event_id,
                        aggregate_id,
                        version,
                        (event).__name__,
                        json.dumps(._serialize(event)),
                        event.occurred_at,
                    )
    
      () -> :
        
          .pool.acquire()  conn:
            rows =  conn.fetch(
                ,
                aggregate_id, from_version
            )
        
         [._deserialize(row[], row[])  row  rows]
    
      () -> :
        
          .pool.acquire()  conn:
            query = 
            params = [event_type]
            
             after:
                query += 
                params.append(after)
            
            query += 
            rows =  conn.fetch(query, *params)
        
         [._deserialize(event_type, row[])  row  rows]
    
     () -> :
        
         dataclasses
         {k: v  k, v  dataclasses.asdict(event).items()
                 k   (, , , )}
    
     () -> :
        
        event_classes = {
            : OrderPlaced,
            : PaymentProcessed,
            : PaymentFailed,
            : OrderShipped,
            : OrderCancelled,
        }
        cls = event_classes.get(event_type)
          cls:
             ValueError()
         cls(**json.loads(data))



CREATE_EVENTS_TABLE = 

Projections

# projections/order_read_model.py
class OrderProjection:
    """
    Read model for order queries.
    Built by processing event stream — completely denormalized for fast reads.
    Can be rebuilt from scratch by replaying all events.
    """
    
    def __init__(self, db):
        self.db = db
    
    async def handle(self, event):
        """Route events to appropriate handlers."""
        handlers = {
            "OrderPlaced": self.on_order_placed,
            "PaymentProcessed": self.on_payment_processed,
            "PaymentFailed": self.on_payment_failed,
            "OrderShipped": self.on_order_shipped,
            "OrderCancelled": self.on_order_cancelled,
        }
        handler = handlers.get(type(event).__name__)
        if handler:
            await handler(event)
    
    async def on_order_placed(self, event: OrderPlaced):
        await self.db.execute(
            """
            INSERT INTO order_read_model (
                order_id, user_id, status, total_cents,
                items, created_at, updated_at
            ) VALUES ($1, $2, 'pending', $3, $4, $5, $5)
            ON CONFLICT (order_id) DO NOTHING
            """,
            event.aggregate_id, event.user_id, event.total_cents,
            json.dumps(list(event.items)), event.occurred_at,
        )
    
    async def ():
         .db.execute(
            ,
            event.aggregate_id, event.payment_id, event.occurred_at,
        )
    
      ():
         .db.execute(
            ,
            event.aggregate_id, event.reason, event.occurred_at,
        )
    
      ():
        
         .db.execute()
        
        all_events =  event_store.load_events_by_type(, limit=)
        
        
        
         event  all_events:
             .handle(event)
        
        ()

Rules

• Events are past-tense, immutable facts — OrderPlaced, not PlaceOrder; never modify stored events.
• Aggregates emit events, never mutate directly — all state changes flow through event application.
• Optimistic concurrency at the event store — the expected_version check prevents lost updates.
• Projections are disposable — they're derived; rebuild them anytime from the event log.
• Separate command model from query model — aggregates handle writes; projections handle reads (CQRS).
• Event schema migration is hard — version your events from day one (OrderPlaced.v2).
• Snapshots for long-lived aggregates — store periodic snapshots to avoid replaying 10,000 events on every load.
• Idempotent event handlers — projections may receive the same event twice; handle duplicates gracefully.
• Don't put behavior in projections — they transform events to read models; no business logic.
• Test by replaying — event sourcing tests replay events and assert on derived state, not mocked calls.