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 almacenar solo el estado actual. En lugar de actualizar un registro in-place, añades un evento describiendo lo que sucedió. El estado actual se deriva reproduciendo todos los eventos de una entidad. Esto proporciona un trail 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 trail de auditoría completo de cada cambio de estado (finanzas, salud, cumplimiento)
• Quieras reconstruir estados históricos o depurar reproduciendo eventos
• Las arquitecturas impulsadas por eventos ya existan, haciendo que los event stores sean naturales
• CQRS esté en uso, y los modelos de lectura puedan construirse desde proyecciones de eventos
• Necesites compensar fallas reproduciendo o invirtiendo eventos (consulta Saga)
• Ejemplos: libros contables, sistemas de inventario, seguimiento de órdenes, edición colaborativa

Solución

Python

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

@dataclass
class Event:
    type: str
    entity_id: str
    payload: dict
    timestamp: 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

# Trail 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

// Trail 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 append-only:

• Event Store: Log append-only de todos los eventos de dominio por entidad
• Eventos: Registros inmutables describiendo lo que sucedió (ej.
• Rehidratación de Entidades: Reconstruir el estado actual reproduciendo todos los eventos de una entidad
• Proyecciones: Crear vistas optimizadas para lectura suscribiéndose al stream de eventos
• Snapshots: Guardar periódicamente el estado computado para evitar reproducir miles de eventos

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 costo de reproducción
Consultas Temporales	Consultar estado en cualquier momento	Depuración; reportes históricos
Reproducción de Eventos	Reproducir 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 o mutues eventos — el log es inmutable
• Usa snapshots para entidades de larga vida para evitar reproducir miles de eventos
• Versiona tus esquemas de eventos para compatibilidad hacia adelante/atrás
• Consumidores idempotentes — el mismo evento debería ser seguro de procesar múltiples veces
• Encripta campos sensibles del payload a nivel de aplicación

Errores comunes

• Usar eventos como bus de mensajes en lugar de almacén de estado (separar preocupaciones)
• Mutar o borrar eventos, rompiendo el trail de auditoría
• Olvidar manejar la evolución de esquemas de eventos (rompiendo reproducciones antiguas)
• Reproducir todos los eventos desde el inicio de los tiempos sin snapshots
• Almacenar payloads binarios grandes dentro de eventos en lugar de referencias
• No manejar la entrega duplicada de eventos en sistemas distribuidos

Notas de Producción

• Despliega gradualmente usando canary o blue-green para detectar regresiones temprano.
• Configura alertas para errores, latencia p99 y tasa de fallos antes de habilitar en producción.
• Documenta el rollback en el runbook; prueba el procedimiento en staging al menos una vez por trimestre.
• Revisa logs estructurados con correlation IDs para trazar requests end-to-end en incidentes.

Puntos Clave

• Aplica patrón event sourcing cuando necesites una solución práctica para tu caso de uso.
• Monitorea el rendimiento después de implementar; mide latencia, errores y uso de recursos antes y después.
• Revisa la sección de Troubleshooting ante errores comunes; la mayoría tienen causa raíz documentada con solución.
• Mantén dependencias actualizadas y ejecuta tests en CI para prevenir regresiones en producción.

Troubleshooting

• Pattern does not fit the problem: re-evaluate the forces (performance, scalability, team size, coupling). A pattern is only appropriate when its trade-offs match your constraints.
• Too many abstractions: if adding a pattern increases complexity without a clear benefit, simplify. Not every module needs a factory, decorator, or strategy.
• Tight coupling after refactoring: check that interfaces are stable and dependencies point inward.
• Tests break when the design changes: favor stable contracts over internal structure.
• Performance regression from indirection: measure before and after. Layers, decorators, and adapters can add latency; cache or inline hot paths if needed.

Errores Comunes en Producción

• Aplicar el patrón donde no se necesita abstracción, agregando complejidad accidental.
• Dejar que el patrón se filtre en módulos no relacionados y confundir los límites de responsabilidad.
• Sobre-ingeniería en la primera implementación en lugar de comenzar simple y medir el dolor.
• Saltar los tests de contrato, de modo que las refactorizaciones rompan consumidores en silencio.
• Ignorar modos de fallo que el patrón no cubre.
• Usar el patrón como opción por defecto en lugar de elegir la herramienta adecuada para la escala actual.
• Olvidar documentar cuándo dejar de usar el patrón y qué lo reemplaza.
• Carecer de observabilidad sobre rendimiento y propagación de errores del patrón.