Implementar Event Sourcing en Arquitecturas Serverless

Visión general

Los sistemas tradicionales almacenan el estado actual. Una orden está “enviada,” y la fila de la base de datos dice status = shipped. Si un usuario pregunta “¿cuándo cambió el estado a enviado?” la base de datos no tiene respuesta — el valor anterior fue sobrescrito. Si un analista pregunta “¿cuántas órdenes fueron canceladas y re-enviadas el mes pasado?” el sistema no puede responder sin agregar columnas de auditoría que rastreen cada cambio a mano.

Event sourcing almacena cada cambio de estado como un evento inmutable en un log append-only. El estado actual se computa reproduciendo los eventos. El estado de una orden no es una fila — es la secuencia [OrderCreated, ItemAdded, PaymentProcessed, Shipped]. Eso te da un audit trail completo, consultas temporales (“¿cuál era el estado ayer a las 3pm?”) y la capacidad de reconstruir proyecciones desde cero. En arquitecturas serverless, los eventos se capturan vía DynamoDB streams, SQS o EventBridge, y las funciones Lambda proyectan el modelo de lectura.

La implementación de abajo usa DynamoDB como event store, DynamoDB Streams para disparar la Lambda de proyección y replay en memoria para reconstruir el agregado.

Cuándo usarlo

Usa esta receta cuando:

• El historial completo de auditoría de todos los cambios es un requisito de negocio. Consulta Event-Driven Functions para arquitecturas event-driven.
• Necesitas responder preguntas temporales sobre estados pasados
• Reconstruir modelos de lectura desde cero es una capacidad necesaria. Consulta Serverless Orchestration para gestionar workflows con estado.
• El modelo de escritura es complejo y el modelo de lectura necesita optimizarse por separado
• Requisitos de compliance o regulatorios exigen logs de cambios inmutables. Consulta el patrón CQRS para separar modelos de lectura y escritura.

Solución

Event Store con DynamoDB y Streams

interface DomainEvent {
  eventId: string;
  aggregateId: string;
  eventType: string;
  payload: Record<string, unknown>;
  timestamp: string;
  version: number;
}

class OrderEventStore {
  constructor(private tableName: string, private client: DynamoDBDocument) {}

  async appendEvents(aggregateId: string, events: DomainEvent[]): Promise<void> {
    const currentVersion = await this.getCurrentVersion(aggregateId);

    const transactItems = events.map((event, index) => ({
      Put: {
        TableName: this.tableName,
        Item: {
          pk: `ORDER#${aggregateId}`,
          sk: `EVENT#${(currentVersion + index + 1).toString().padStart(10, '0')}`,
          eventId: event.eventId,
          eventType: event.eventType,
          payload: event.payload,
          timestamp: new Date().toISOString(),
          version: currentVersion + index + 1,
        },
        ConditionExpression: 'attribute_not_exists(pk)',
      },
    }));

    await this.client.transactWrite({ TransactItems: transactItems });
  }

  async getEvents(aggregateId: string): Promise<DomainEvent[]> {
    const result = await this.client.query({
      TableName: this.tableName,
      KeyConditionExpression: 'pk = :pk AND begins_with(sk, :sk)',
      ExpressionAttributeValues: {
        ':pk': `ORDER#${aggregateId}`,
        ':sk': 'EVENT#',
      },
      ScanIndexForward: true,
    });

    return (result.Items || []).map(item => ({
      eventId: item.eventId,
      aggregateId,
      eventType: item.eventType,
      payload: item.payload,
      timestamp: item.timestamp,
      version: item.version,
    }));
  }

  private async getCurrentVersion(aggregateId: string): Promise<number> {
    const events = await this.getEvents(aggregateId);
    return events.length > 0 ? events[events.length - 1].version : 0;
  }
}

Handler de Proyección en Lambda

export const handler = async (event: DynamoDBStreamEvent): Promise<void> => {
  for (const record of event.Records) {
    if (record.eventName !== 'INSERT') continue;

    const newImage = unmarshall(record.dynamodb?.NewImage as any);
    const domainEvent: DomainEvent = {
      eventId: newImage.eventId,
      aggregateId: newImage.aggregateId,
      eventType: newImage.eventType,
      payload: newImage.payload,
      timestamp: newImage.timestamp,
      version: newImage.version,
    };

    await projectEvent(domainEvent);
  }
};

async function projectEvent(event: DomainEvent): Promise<void> {
  switch (event.eventType) {
    case 'OrderCreated':
      await createOrderProjection(event.aggregateId, event.payload);
      break;
    case 'ItemAdded':
      await addItemToOrderProjection(event.aggregateId, event.payload);
      break;
    case 'OrderShipped':
      await updateOrderStatus(event.aggregateId, 'shipped');
      break;
  }
}

Reconstrucción del Agregado de Orden

class OrderAggregate {
  private status: string = 'pending';
  private items: OrderItem[] = [];
  private total: number = 0;

  applyEvent(event: DomainEvent): void {
    switch (event.eventType) {
      case 'OrderCreated':
        this.status = 'created';
        this.total = event.payload.total as number;
        break;
      case 'ItemAdded':
        this.items.push(event.payload.item as OrderItem);
        this.total += (event.payload.item as OrderItem).price;
        break;
      case 'OrderShipped':
        this.status = 'shipped';
        break;
      case 'OrderCancelled':
        this.status = 'cancelled';
        break;
    }
  }

  static fromEvents(events: DomainEvent[]): OrderAggregate {
    const order = new OrderAggregate();
    for (const event of events) {
      order.applyEvent(event);
    }
    return order;
  }

  toSnapshot(): OrderSnapshot {
    return {
      status: this.status,
      items: this.items,
      total: this.total,
    };
  }
}

Código companion: el repo companion de event-sourcing-serverless tiene estos tres archivos TypeScript listos para integrar en un proyecto Lambda.

Explicación

• Event store: un log append-only — los eventos nunca se actualizan ni se borran. Cada evento lleva un ID único, un aggregate ID, un tipo, un payload y una versión. La versión te da ordenamiento y control de concurrencia optimista vía ConditionExpression.
• Reconstrucción del agregado: el estado actual no se guarda directamente. Cargas todos los eventos de un agregado y los reproducís en orden; el agregado arranca vacío y aplica cada evento. Es determinista — la misma secuencia siempre produce el mismo estado.
• Proyecciones (modelos de lectura): se construyen suscribiéndose al stream de eventos. Cuando llega un evento, la Lambda disparada por el stream actualiza la vista optimizada para lectura. Podés tener varias proyecciones sobre los mismos eventos — una para el dashboard del cliente, otra para analytics, otra para indexación de búsqueda.
• Snapshots: reproducir miles de eventos para un agregado de larga vida es lento. Los snapshots guardan el estado del agregado en una versión concreta; al cargar, traés el último snapshot y reproducís solo lo posterior. Tomalos cada 50–100 eventos y de forma asíncrona, para que el write path siga rápido.

Variantes

Enfoque	Store	Proyecciones	Mejor para
DynamoDB + Streams	DynamoDB	Lambda	Nativo AWS, escala moderada
EventStoreDB	EventStoreDB	Subscriptions	Alto volumen, dominios complejos
Kafka + KTables	Kafka	Kafka Streams	Stream processing, replay
S3 + Athena	S3	Queries de Athena	Audit, compliance, analytics
Aurora + Outbox	PostgreSQL	CDC	Event sourcing relacional

Event store en Python (boto3)

El mismo event store en Python — transact_write_items mantiene la concurrencia optimista y TypeSerializer se encarga del ruido de los atributos tipados:

import boto3
from boto3.dynamodb.types import TypeSerializer
from boto3.dynamodb.conditions import Key
from datetime import datetime, timezone

serializer = TypeSerializer()

class OrderEventStore:
    def __init__(self, table_name: str):
        self.table_name = table_name
        self.client = boto3.client("dynamodb")
        self.table = boto3.resource("dynamodb").Table(table_name)

    def append_events(self, aggregate_id: str, events: list[dict]) -> None:
        current = self.current_version(aggregate_id)
        items = []
        for i, event in enumerate(events):
            version = current + i + 1
            item = {
                "pk": f"ORDER#{aggregate_id}",
                "sk": f"EVENT#{version:010d}",
                "eventId": event["event_id"],
                "eventType": event["event_type"],
                "payload": event["payload"],
                "timestamp": datetime.now(timezone.utc).isoformat(),
                "version": version,
            }
            items.append({
                "Put": {
                    "TableName": self.table_name,
                    "Item": {k: serializer.serialize(v) for k, v in item.items()},
                    "ConditionExpression": "attribute_not_exists(pk)",
                }
            })
        self.client.transact_write_items(TransactItems=items)

    def get_events(self, aggregate_id: str) -> list[dict]:
        resp = self.table.query(
            KeyConditionExpression=(
                Key("pk").eq(f"ORDER#{aggregate_id}") & Key("sk").begins_with("EVENT#")
            ),
            ScanIndexForward=True,
        )
        return resp.get("Items", [])

    def current_version(self, aggregate_id: str) -> int:
        events = self.get_events(aggregate_id)
        return events[-1]["version"] if events else 0

El handler de proyección del lado de Python es un handler Lambda común sobre event["Records"] — mismo filtro de INSERT, misma idea de unmarshalling vía TypeDeserializer.

Lo que funciona

• Versiona cada evento: incluí una versión monotónica creciente por agregado. Evita lost updates cuando dos usuarios modifican el mismo agregado a la vez.
• Hacé los eventos inmutables y autocontenidos: un evento debería llevar todos los datos necesarios para entenderlo, no solo deltas. OrderCreated debería incluir customer ID, dirección de envío e items — no solo “la orden 123 fue creada”. Los consumidores futuros no deberían tener que consultar otros sistemas para interpretar el evento.
• Usá correlation IDs a lo largo de la cadena de eventos: cuando un evento dispara otro (p. ej., OrderShipped dispara InventoryDecremented), propagá el correlation ID. Hace posible el tracing end-to-end y el debugging en cadenas de eventos distribuidas.
• Implementá proyecciones idempotentes: Lambda reintenta ante fallos, y una proyección que incrementa un contador en cada invocación va a contar de más. Escribí el event ID en la fila de la proyección y salteá si ya fue procesado.
• Archivá eventos viejos a cold storage: DynamoDB se pone caro cuando guardás millones de eventos a largo plazo. Mové los eventos de más de 90 días a S3 con TTL de DynamoDB o export jobs.

Errores comunes

• Guardar el estado actual junto a los eventos: mantener a la vez un log de eventos y una tabla de estado actual permite que diverjan — un bug en la proyección escribe el estado A mientras el log contiene eventos del estado B. El event store es la fuente de verdad; las proyecciones son derivadas, nunca primarias.
• Exponer los tipos de evento internos a sistemas externos: los consumidores externos no deberían depender de tus schemas internos. Definí un contrato público de eventos (p. ej., OrderConfirmed) y mapeá los eventos internos sobre él, para que los refactors no rompan integraciones.
• Ignorar la evolución del schema de eventos: cuando un tipo de evento gana un campo, los eventos viejos del log no lo tienen. El agregado tiene que manejar campos faltantes con gracia — o mejor, versioná los eventos y escribí upcasters.
• Reproducir desde el principio en cada query: reproducir 10.000 eventos por cada GET /order/123 destruye el rendimiento. Usá snapshots para agregados de larga vida y cargá desde ellos.
• Asumir entrega ordenada y exactamente una vez: los streams son at-least-once y solo ordenados por shard. Revisá los sequence numbers y hacé las proyecciones idempotentes en vez de confiar en el orden de llegada.

See Also

• AWS — DynamoDB Streams — documentación oficial sobre stream records, ordenamiento por shard y triggers de Lambda.
• Martin Fowler — Event Sourcing — la descripción canónica del patrón y sus trade-offs.
• Event sourcing en una base de datos relacional — el mismo patrón con PostgreSQL/MySQL en vez de DynamoDB.
• Event sourcing con CQRS en Python — agregados, proyecciones y snapshots en Python.