advanced · temario
7.2tema 2 de 7

Event Sourcing

Event store, rehidratación, snapshots y proyecciones.

Event Sourcing cambia el paradigma de persistencia: en lugar de almacenar el estado actual de un aggregate en una fila de base de datos, se almacena la secuencia de domain events que lo produjeron. El event store es la fuente de verdad — append-only, ordenado por aggregate y por número de secuencia. El estado actual se deriva siempre al reproducir los eventos.

La rehidratación consiste en cargar todos los eventos de un aggregate desde el store y aplicarlos en orden para reconstruir su estado en memoria. Cada evento llama a un handler interno (`on_<EventType>`) que muta el estado del aggregate. Para aggregates con miles de eventos, la rehidratación puede ser lenta; los snapshots son checkpoints periódicos que almacenan el estado serializado en un momento dado, permitiendo cargar solo el snapshot y los eventos posteriores.

Las proyecciones son handlers de eventos que construyen vistas materializadas a partir del stream de eventos para responder a consultas. Son el mecanismo de lectura en un sistema event-sourced: el event store responde a la pregunta '¿qué pasó?'; las proyecciones responden a '¿cuál es el estado actual para mostrar en pantalla?'. Las proyecciones son idempotentes y descartables — se pueden reconstruir reproduciendo el stream completo.

Los tradeoffs son reales: audit log completo, time travel (reconstruir estado en cualquier momento), integración event-driven nativa. Pero las consultas complejas requieren proyecciones adicionales, la evolución del esquema de eventos es no trivial (requiere upcasting), y depurar errores implica entender streams de eventos en lugar de filas de base de datos.

Cuándo NO usar Event Sourcing: dominios simples sin requisitos de auditoría, equipos sin experiencia en sistemas event-driven, sistemas donde el rendimiento de rehidratación es crítico y no hay estrategia de snapshots. Si el dominio no tiene el concepto de 'historial de lo que ocurrió' como requisito de negocio, el CRUD estándar es más simple y más adecuado.

structure.txt
# infrastructure/event_store/<event_store>.py — Event Store (append-only)
# domain/model/<aggregate>.py — Aggregate (reconstituted from events)
# infrastructure/projections/<view>_projection.py — Projection

@dataclass(frozen=True)
class StoredEvent:
    aggregate_id: str
    sequence: int
    event_type: str
    payload: dict
    occurred_at: datetime

class <EventStore>:
    def append(self, aggregate_id: str, events: list[<DomainEvent>], expected_version: int) -> None:
        # Optimistic concurrency: reject if current version != expected_version
        ...

    def load(self, aggregate_id: str, after_sequence: int = 0) -> list[StoredEvent]:
        # Return events in order; start after snapshot sequence if provided
        ...

class <Aggregate>:
    @classmethod
    def rehydrate(cls, events: list[<DomainEvent>]) -> '<Aggregate>':
        instance = cls.__new__(cls)
        instance._version = 0
        for event in events:
            instance._apply(event)
            instance._version += 1
        return instance

    def _apply(self, event: <DomainEvent>) -> None:
        # Dispatch to on_<EventType> handler
        handler = getattr(self, f'on_{type(event).__name__}', None)
        if handler:
            handler(event)

    def on_<SomethingHappened>(self, event: <SomethingHappened>) -> None:
        self._<attribute> = event.<value>

@dataclass
class <Snapshot>:
    aggregate_id: str
    version: int
    state: dict  # serialized aggregate state
    taken_at: datetime
project/
<project>
contexts
<bc>
domain
infrastructure

Debugging lab

Detecta y corrige el error o la violación de diseño.

0/5 tests passing0%
  1. 7.2.5.1

    # Bug: rehydrate method calls external notification service inside _apply class Aggregate: @classmethod def rehydrate(cls, events: list) -> "Aggregate": instance = cls.__new__(cls) instance._version = 0 for event in events: instance._apply(event) instance._version += 1 return instance def _apply(self, event) -> None: if isinstance(event, SomethingHappened): self.on_something_happened(event) def on_something_happened(self, event: SomethingHappened) -> None: self._value = event.new_value # Bug: calling external service during replay notification_service.notify(f"Value changed to {event.new_value}")

  2. 7.2.5.2

    # Bug: event store has an update() method that overwrites the last event payload class EventStore: def append(self, aggregate_id: str, events: list, expected_version: int) -> None: ... def load(self, aggregate_id: str) -> list: ... def update(self, aggregate_id: str, sequence: int, new_payload: dict) -> None: # Bug: allows overwriting stored event payloads self._db.execute( "UPDATE events SET payload = ? WHERE aggregate_id = ? AND sequence = ?", (new_payload, aggregate_id, sequence) )

  3. 7.2.5.3

    # Bug: no snapshot strategy — aggregate with 50,000 events loads in 8 seconds class AggregateRepository: def find(self, aggregate_id: AggregateId) -> Aggregate: # Loading all 50,000 events every time — no snapshot optimization all_events = self._event_store.load(str(aggregate_id)) return Aggregate.rehydrate(all_events)

  4. 7.2.5.4

    # Bug: projection raises ValueError if a payload field is missing class ViewProjection: def on_something_happened(self, event: SomethingHappened) -> None: # Bug: KeyError / AttributeError if field is absent in older event versions self._store.upsert(ReadModelDto( id=event.aggregate_id, label=event.label, # Will crash if field was added in a later event version category=event.category, # Same problem ))

  5. 7.2.5.5

    # Bug: event sourcing applied to a simple user preferences entity with 2 fields # UserPreferences: theme ("light"/"dark") and language ("en"/"es") # No audit requirement, no replay requirement, team of 2 developers class UserPreferencesEventStore: def append(self, user_id: str, events: list, expected_version: int) -> None: ... def load(self, user_id: str) -> list: ... class UserPreferences: @classmethod def rehydrate(cls, events: list) -> "UserPreferences": ...