Patrón Event Sourcing

Resumen

El Patrón Event Sourcing almacena el estado de una aplicación como una secuencia de eventos en lugar de guardar solo el estado actual. En lugar de actualizar un registro en el sitio, añades un evento que describe lo que pasó. El estado actual se deriva reaplicando todos los eventos de una entidad, lo que te da un registro de auditoría completo, consultas temporales y la capacidad de reconstruir el estado en cualquier momento.

Cuándo usarlo

Usa el Patrón Event Sourcing cuando:

• Necesites un registro de auditoría completo de cada cambio de estado (finanzas, salud, cumplimiento)
• Quieras reconstruir estados históricos o depurar reaplicando eventos
• Ya tengas una arquitectura orientada a eventos y el event store encaje de forma natural
• Uses CQRS y puedas construir los modelos de lectura desde proyecciones de eventos
• Necesites compensar fallos reaplicando o invirtiendo eventos (consulta Saga)

Casos típicos: libros contables, inventario, seguimiento de pedidos, edición colaborativa.

Sáltalo para un CRUD sencillo. Event Sourcing añade versionado de esquemas, proyecciones y maquinaria de reproducción que una app pequeña nunca amortiza — si nadie va a preguntar “qué pasó y cuándo”, una tabla normal sale más barata.

Solución

Python

from dataclasses import dataclass
from typing import Dict, List, Optional
from datetime import datetime

@dataclass
class Event:
    type: str
    entity_id: str
    payload: dict
    timestamp: Optional[str] = None

    def __post_init__(self):
        if self.timestamp is None:
            self.timestamp = datetime.now().isoformat()

class EventStore:
    def __init__(self):
        self.streams: Dict[str, List[Event]] = {}

    def append(self, entity_id: str, event: Event):
        self.streams.setdefault(entity_id, []).append(event)

    def get_stream(self, entity_id: str) -> List[Event]:
        return list(self.streams.get(entity_id, []))

# Entidad: reconstruye estado desde eventos
class BankAccount:
    def __init__(self, account_id: str):
        self.account_id = account_id
        self.balance = 0
        self.version = 0

    def apply(self, event: Event):
        if event.type == "Deposited":
            self.balance += event.payload["amount"]
        elif event.type == "Withdrawn":
            self.balance -= event.payload["amount"]
        self.version += 1

    @classmethod
    def rehydrate(cls, account_id: str, events: List[Event]):
        account = cls(account_id)
        for e in events:
            account.apply(e)
        return account

# Uso
store = EventStore()
account_id = "ACC-123"

store.append(account_id, Event("Deposited", account_id, {"amount": 100}))
store.append(account_id, Event("Withdrawn", account_id, {"amount": 30}))
store.append(account_id, Event("Deposited", account_id, {"amount": 50}))

# Reconstruye estado
account = BankAccount.rehydrate(account_id, store.get_stream(account_id))
print(f"Balance: {account.balance}")  # 120

# Registro de auditoría completo
for e in store.get_stream(account_id):
    print(f"{e.timestamp}: {e.type} {e.payload}")

JavaScript

class Event {
  constructor(type, entityId, payload) {
    this.type = type;
    this.entityId = entityId;
    this.payload = payload;
    this.timestamp = new Date().toISOString();
  }
}

class EventStore {
  constructor() {
    this.streams = new Map();
  }

  append(entityId, event) {
    if (!this.streams.has(entityId)) this.streams.set(entityId, []);
    this.streams.get(entityId).push(event);
  }

  getStream(entityId) {
    return this.streams.get(entityId) || [];
  }
}

class BankAccount {
  constructor(accountId) {
    this.accountId = accountId;
    this.balance = 0;
    this.version = 0;
  }

  apply(event) {
    if (event.type === "Deposited") this.balance += event.payload.amount;
    if (event.type === "Withdrawn") this.balance -= event.payload.amount;
    this.version++;
  }

  static rehydrate(accountId, events) {
    const account = new BankAccount(accountId);
    events.forEach(e => account.apply(e));
    return account;
  }
}

// Uso
const store = new EventStore();
const accountId = "ACC-123";

store.append(accountId, new Event("Deposited", accountId, { amount: 100 }));
store.append(accountId, new Event("Withdrawn", accountId, { amount: 30 }));
store.append(accountId, new Event("Deposited", accountId, { amount: 50 }));

const account = BankAccount.rehydrate(accountId, store.getStream(accountId));
console.log("Balance:", account.balance); // 120

// Registro de auditoría
store.getStream(accountId).forEach(e =>
  console.log(`${e.timestamp}: ${e.type}`, e.payload)
);

Java

import java.util.*;

class Event {
    String type;
    String entityId;
    Map<String, Object> payload;
    String timestamp;

    Event(String type, String entityId, Map<String, Object> payload) {
        this.type = type;
        this.entityId = entityId;
        this.payload = payload;
        this.timestamp = new Date().toString();
    }
}

class EventStore {
    private final Map<String, List<Event>> streams = new HashMap<>();

    void append(String entityId, Event event) {
        streams.computeIfAbsent(entityId, k -> new ArrayList<>()).add(event);
    }

    List<Event> getStream(String entityId) {
        return new ArrayList<>(streams.getOrDefault(entityId, List.of()));
    }
}

class BankAccount {
    String accountId;
    double balance = 0;
    int version = 0;

    BankAccount(String accountId) {
        this.accountId = accountId;
    }

    void apply(Event event) {
        switch (event.type) {
            case "Deposited" -> balance += (double) event.payload.get("amount");
            case "Withdrawn" -> balance -= (double) event.payload.get("amount");
        }
        version++;
    }

    static BankAccount rehydrate(String accountId, List<Event> events) {
        BankAccount account = new BankAccount(accountId);
        events.forEach(account::apply);
        return account;
    }
}

// Uso
EventStore store = new EventStore();
String accountId = "ACC-123";

store.append(accountId, new Event("Deposited", accountId, Map.of("amount", 100.0)));
store.append(accountId, new Event("Withdrawn", accountId, Map.of("amount", 30.0)));
store.append(accountId, new Event("Deposited", accountId, Map.of("amount", 50.0)));

BankAccount account = BankAccount.rehydrate(accountId, store.getStream(accountId));
System.out.println("Balance: " + account.balance); // 120.0

Explicación

Event Sourcing reemplaza el modelo CRUD tradicional con un log de eventos de solo-adición:

• Event Store: log inmutable con todos los eventos de dominio por entidad
• Eventos: registros inmutables de lo que pasó (ej. MoneyDeposited, OrderShipped)
• Rehidratación de entidades: reconstruir el estado actual reaplicando todos los eventos de una entidad
• Proyecciones: vistas optimizadas para lectura construidas suscribiéndose al flujo de eventos
• Snapshots: el estado calculado guardado periódicamente para no reaplicar miles de eventos

Los trade-offs importan tanto como la mecánica:

• El coste de reproducción crece con el stream. Rehidratar una entidad es O(número de eventos), así que las entidades de larga vida necesitan snapshots o las lecturas se convierten en tu cuello de botella.
• Los cambios de esquema son para siempre. Renombrar un campo rompe la reproducción de eventos antiguos salvo que añadas un upcaster o conserves la rama vieja en apply. Versiona los tipos de evento desde el primer día.
• Las lecturas son eventualmente consistentes. Una proyección puede ir por detrás del lado de escritura; una lectura justo después de escribir puede devolver datos obsoletos si no versionas o esperas al consumidor.
• El log solo crece. El coste de almacenamiento, la retención y el borrado GDPR son decisiones de diseño, no detalles a posteriori.

Variantes

Variante	Descripción	Caso de uso
Event Sourcing Completo	Todo el estado proviene de eventos; sin DB mutable	Máxima auditabilidad; cumplimiento
Híbrido	Eventos + snapshot de estado actual	Rendimiento; reducir coste de reproducción
Consultas Temporales	Consultar estado en cualquier momento	Depuración; reportes históricos
Reproducción de Eventos	Reaplicar eventos para reconstruir o migrar	Migraciones de esquema; recuperación de bugs

Lo que funciona

• Los eventos deben describir intención de negocio (ej. OrderPlaced), no acciones técnicas (RowInserted)
• Nunca borres ni mutes eventos — el log es inmutable
• Usa snapshots en entidades de larga vida para no reaplicar miles de eventos
• Versiona los esquemas de eventos para compatibilidad hacia adelante y hacia atrás
• Haz los consumidores idempotentes — el mismo evento debe ser seguro de procesar más de una vez
• Encripta los campos sensibles del payload a nivel de aplicación

Errores comunes

• Usar eventos como bus de mensajes en lugar de almacén de estado (son responsabilidades distintas)
• Mutar o borrar eventos, rompiendo el registro de auditoría
• Olvidar la evolución de esquemas y romper la reproducción de eventos antiguos
• Reaplicar todos los eventos desde el principio de los tiempos sin snapshots
• Guardar payloads binarios grandes dentro de los eventos en lugar de referencias
• No manejar la entrega duplicada de eventos en sistemas distribuidos

Notas de Producción

• Despliega los cambios de esquema en dos pasos: primero código capaz de leer la versión vieja y la nueva, después empieza a emitir la nueva. Nunca rompas la reproducción de streams históricos.
• Alerta por el lag de las proyecciones, no solo por errores: un consumidor que se queda atrás parece pérdida de datos para el usuario.
• Define una cadencia de snapshots por tipo de entidad (ej. cada 50-100 eventos) y mide la rehidratación en staging con volúmenes reales.
• Presupuesta el crecimiento del almacenamiento: los event stores solo crecen. Decide retención, archivado y límites de payload desde el inicio.
• Encripta los datos personales por sujeto: el crypto-shredding (borrar la clave del sujeto) es la respuesta práctica al borrado GDPR cuando la historia no se puede reescribir.

Troubleshooting

• La reproducción da un estado incorrecto: un esquema antiguo ya no encaja con el apply actual. Añade un upcaster para esa versión o conserva la rama vieja en apply.
• El modelo de lectura devuelve datos obsoletos: el consumidor de la proyección se quedó atrás respecto al event store. Revisa el lag y haz que las escrituras devuelvan la versión aplicada para que la UI espere o recargue.
• Eventos duplicados en el stream: el productor reintentó tras un timeout del almacén. Haz los appends idempotentes con un ID único o un check de versión esperada.
• La rehidratación se vuelve lenta con el tiempo: el stream superó la cadencia de snapshots. Acorta el intervalo o pagina el stream.
• Datos personales acabaron en los payloads: los eventos no se pueden borrar. Rota la clave de cifrado del sujeto o agrega un evento marcador DataForgotten y respétalo en las proyecciones.

Puntos Clave

• El estado es un fold sobre los eventos: no guardes “la fila”, guarda lo que pasó y deriva el resto.
• El log de eventos es inmutable y versionado — la evolución de esquema es una restricción desde el primer día.
• Añade snapshots cuando el coste de reproducción duela, no por defecto.
• Event Sourcing no es un modelo de lectura; combínalo con CQRS cuando lectura y escritura diverjan.

Ver también

• Event Sourcing — Martin Fowler
• Patrón Event Sourcing — microservices.io
• Event Sourcing — Azure Architecture Center