Event-Driven Architecture Practical Guide: CQRS, Event Sourcing, Saga Patterns

// TypeScript - Domain Event basic structure
interface DomainEvent {
  eventId: string
  eventType: string
  aggregateId: string
  aggregateType: string
  timestamp: Date
  version: number
  payload: Record<string, unknown>
  metadata: {
    correlationId: string
    causationId: string
    userId?: string
  }
}

// Order creation event example
const orderPlacedEvent: DomainEvent = {
  eventId: 'evt-550e8400-e29b-41d4-a716-446655440000',
  eventType: 'OrderPlaced',
  aggregateId: 'order-12345',
  aggregateType: 'Order',
  timestamp: new Date('2026-03-14T09:00:00Z'),
  version: 1,
  payload: {
    customerId: 'cust-67890',
    items: [
      { productId: 'prod-001', quantity: 2, price: 29900 },
      { productId: 'prod-002', quantity: 1, price: 15000 },
    ],
    totalAmount: 74800,
    shippingAddress: {
      city: 'Seoul',
      district: 'Gangnam-gu',
      detail: 'Teheran-ro 123',
    },
  },
  metadata: {
    correlationId: 'corr-abc123',
    causationId: 'cmd-place-order-001',
    userId: 'user-admin-01',
  },
}

// TypeScript - CQRS Command Side

// Command definition
interface PlaceOrderCommand {
  type: 'PlaceOrder'
  customerId: string
  items: Array<{
    productId: string
    quantity: number
    price: number
  }>
  shippingAddress: string
}

// Command Handler
class PlaceOrderHandler {
  constructor(
    private orderRepository: OrderWriteRepository,
    private eventBus: EventBus,
    private inventoryService: InventoryService
  ) {}

  async handle(command: PlaceOrderCommand): Promise<string> {
    // 1. Business rule validation
    await this.inventoryService.validateStock(command.items)

    // 2. Aggregate creation
    const order = Order.create({
      customerId: command.customerId,
      items: command.items,
      shippingAddress: command.shippingAddress,
    })

    // 3. Save to write store
    await this.orderRepository.save(order)

    // 4. Publish domain events (for read model sync)
    for (const event of order.getDomainEvents()) {
      await this.eventBus.publish(event)
    }

    return order.id
  }
}

// Query Side - Read-only model
interface OrderReadModel {
  orderId: string
  customerName: string
  orderDate: string
  status: string
  totalAmount: number
  itemCount: number
  lastUpdated: string
}

// Query Handler
class GetOrdersQueryHandler {
  constructor(private readDb: ReadDatabase) {}

  async handle(query: {
    customerId: string
    status?: string
    page: number
    limit: number
  }): Promise<OrderReadModel[]> {
    // Query directly from denormalized read-only table
    return this.readDb.query(
      `SELECT order_id, customer_name, order_date, status,
              total_amount, item_count, last_updated
       FROM order_read_model
       WHERE customer_id = ?
       ${query.status ? 'AND status = ?' : ''}
       ORDER BY order_date DESC
       LIMIT ? OFFSET ?`,
      [query.customerId, query.status, query.limit, query.page * query.limit]
    )
  }
}

// TypeScript - Event-based Projection

class OrderProjection {
  constructor(private readDb: ReadDatabase) {}

  async handle(event: DomainEvent): Promise<void> {
    switch (event.eventType) {
      case 'OrderPlaced':
        await this.onOrderPlaced(event)
        break
      case 'OrderShipped':
        await this.onOrderShipped(event)
        break
      case 'OrderCancelled':
        await this.onOrderCancelled(event)
        break
    }
  }

  private async onOrderPlaced(event: DomainEvent): Promise<void> {
    const payload = event.payload as {
      customerId: string
      items: Array<{ quantity: number; price: number }>
      totalAmount: number
    }

    await this.readDb.upsert('order_read_model', {
      order_id: event.aggregateId,
      customer_id: payload.customerId,
      status: 'PLACED',
      total_amount: payload.totalAmount,
      item_count: payload.items.reduce((sum, i) => sum + i.quantity, 0),
      order_date: event.timestamp,
      last_updated: event.timestamp,
      version: event.version,
    })
  }

  private async onOrderShipped(event: DomainEvent): Promise<void> {
    const payload = event.payload as { trackingNumber: string }

    // Idempotency guarantee: version check
    await this.readDb.updateWhere(
      'order_read_model',
      {
        status: 'SHIPPED',
        tracking_number: payload.trackingNumber,
        last_updated: event.timestamp,
        version: event.version,
      },
      { order_id: event.aggregateId, version: event.version - 1 }
    )
  }

  private async onOrderCancelled(event: DomainEvent): Promise<void> {
    const payload = event.payload as { reason: string }

    await this.readDb.updateWhere(
      'order_read_model',
      {
        status: 'CANCELLED',
        cancellation_reason: payload.reason,
        last_updated: event.timestamp,
        version: event.version,
      },
      { order_id: event.aggregateId }
    )
  }
}

// TypeScript - Event Sourcing Aggregate

abstract class EventSourcedAggregate {
  private uncommittedEvents: DomainEvent[] = []
  protected version: number = 0

  abstract get id(): string

  // Apply event (state change)
  protected apply(event: DomainEvent): void {
    this.when(event)
    this.version++
    this.uncommittedEvents.push(event)
  }

  // Event handler (implemented by subclass)
  protected abstract when(event: DomainEvent): void

  // Restore state from event stream
  loadFromHistory(events: DomainEvent[]): void {
    for (const event of events) {
      this.when(event)
      this.version++
    }
  }

  getUncommittedEvents(): DomainEvent[] {
    return [...this.uncommittedEvents]
  }

  clearUncommittedEvents(): void {
    this.uncommittedEvents = []
  }
}

// Order Aggregate
class Order extends EventSourcedAggregate {
  private _id: string = ''
  private _customerId: string = ''
  private _items: OrderItem[] = []
  private _status: OrderStatus = OrderStatus.DRAFT
  private _totalAmount: number = 0

  get id(): string {
    return this._id
  }

  // Factory method (create command)
  static create(params: { orderId: string; customerId: string; items: OrderItem[] }): Order {
    const order = new Order()
    const totalAmount = params.items.reduce((sum, item) => sum + item.price * item.quantity, 0)

    order.apply({
      eventId: crypto.randomUUID(),
      eventType: 'OrderPlaced',
      aggregateId: params.orderId,
      aggregateType: 'Order',
      timestamp: new Date(),
      version: 1,
      payload: {
        customerId: params.customerId,
        items: params.items,
        totalAmount,
      },
      metadata: {
        correlationId: crypto.randomUUID(),
        causationId: 'create',
      },
    })

    return order
  }

  // Confirm order command
  confirm(): void {
    if (this._status !== OrderStatus.PLACED) {
      throw new Error(`Cannot confirm order in status: ${this._status}`)
    }

    this.apply({
      eventId: crypto.randomUUID(),
      eventType: 'OrderConfirmed',
      aggregateId: this._id,
      aggregateType: 'Order',
      timestamp: new Date(),
      version: this.version + 1,
      payload: { confirmedAt: new Date().toISOString() },
      metadata: {
        correlationId: crypto.randomUUID(),
        causationId: 'confirm',
      },
    })
  }

  // Cancel order command (used in compensating transactions)
  cancel(reason: string): void {
    if (this._status === OrderStatus.CANCELLED) {
      return // Idempotency guarantee
    }

    this.apply({
      eventId: crypto.randomUUID(),
      eventType: 'OrderCancelled',
      aggregateId: this._id,
      aggregateType: 'Order',
      timestamp: new Date(),
      version: this.version + 1,
      payload: { reason, cancelledAt: new Date().toISOString() },
      metadata: {
        correlationId: crypto.randomUUID(),
        causationId: 'cancel',
      },
    })
  }

  // Event handler - state change logic
  protected when(event: DomainEvent): void {
    switch (event.eventType) {
      case 'OrderPlaced': {
        const p = event.payload as {
          customerId: string
          items: OrderItem[]
          totalAmount: number
        }
        this._id = event.aggregateId
        this._customerId = p.customerId
        this._items = p.items
        this._totalAmount = p.totalAmount
        this._status = OrderStatus.PLACED
        break
      }
      case 'OrderConfirmed':
        this._status = OrderStatus.CONFIRMED
        break
      case 'OrderCancelled':
        this._status = OrderStatus.CANCELLED
        break
    }
  }
}

enum OrderStatus {
  DRAFT = 'DRAFT',
  PLACED = 'PLACED',
  CONFIRMED = 'CONFIRMED',
  SHIPPED = 'SHIPPED',
  CANCELLED = 'CANCELLED',
}

interface OrderItem {
  productId: string
  quantity: number
  price: number
}

# Python - Snapshot-based Event Store

from dataclasses import dataclass, field
from datetime import datetime
from typing import Any
import json


@dataclass
class Snapshot:
    aggregate_id: str
    aggregate_type: str
    version: int
    state: dict
    created_at: datetime = field(default_factory=datetime.utcnow)


class EventStore:
    SNAPSHOT_INTERVAL = 100  # Create snapshot every 100 events

    def __init__(self, db_connection):
        self.db = db_connection

    async def save_events(
        self, aggregate_id: str, events: list[dict], expected_version: int
    ) -> None:
        """Save events with optimistic concurrency control"""
        async with self.db.transaction():
            # Check current version (optimistic locking)
            current_version = await self._get_current_version(aggregate_id)
            if current_version != expected_version:
                raise ConcurrencyError(
                    f"Expected version {expected_version}, "
                    f"but current version is {current_version}"
                )

            # Batch insert events
            for i, event in enumerate(events):
                version = expected_version + i + 1
                await self.db.execute(
                    """INSERT INTO event_store
                       (event_id, aggregate_id, aggregate_type,
                        event_type, version, payload, metadata, created_at)
                       VALUES (?, ?, ?, ?, ?, ?, ?, ?)""",
                    (
                        event["event_id"],
                        aggregate_id,
                        event["aggregate_type"],
                        event["event_type"],
                        version,
                        json.dumps(event["payload"]),
                        json.dumps(event["metadata"]),
                        datetime.utcnow(),
                    ),
                )

            # Check snapshot creation condition
            new_version = expected_version + len(events)
            if new_version % self.SNAPSHOT_INTERVAL == 0:
                await self._create_snapshot(aggregate_id, new_version)

    async def load_aggregate(self, aggregate_id: str) -> tuple[list[dict], int]:
        """Load events from snapshot to restore state"""
        # 1. Query most recent snapshot
        snapshot = await self._get_latest_snapshot(aggregate_id)

        if snapshot:
            # 2. Load only events after snapshot
            events = await self.db.fetch_all(
                """SELECT * FROM event_store
                   WHERE aggregate_id = ? AND version > ?
                   ORDER BY version ASC""",
                (aggregate_id, snapshot.version),
            )
            return events, snapshot
        else:
            # 3. Load all events if no snapshot
            events = await self.db.fetch_all(
                """SELECT * FROM event_store
                   WHERE aggregate_id = ?
                   ORDER BY version ASC""",
                (aggregate_id,),
            )
            return events, None

    async def _create_snapshot(
        self, aggregate_id: str, version: int
    ) -> None:
        """Save current state as snapshot"""
        events, _ = await self.load_aggregate(aggregate_id)
        # Rebuild aggregate and serialize state
        aggregate = self._rebuild_aggregate(events)
        await self.db.execute(
            """INSERT INTO snapshots
               (aggregate_id, aggregate_type, version, state, created_at)
               VALUES (?, ?, ?, ?, ?)""",
            (
                aggregate_id,
                aggregate.aggregate_type,
                version,
                json.dumps(aggregate.to_dict()),
                datetime.utcnow(),
            ),
        )

    async def _get_current_version(self, aggregate_id: str) -> int:
        result = await self.db.fetch_one(
            "SELECT MAX(version) FROM event_store WHERE aggregate_id = ?",
            (aggregate_id,),
        )
        return result[0] if result[0] else 0


class ConcurrencyError(Exception):
    pass

// TypeScript - Saga Orchestrator (Order Processing)

interface SagaStep {
  name: string
  action: () => Promise<void>
  compensation: () => Promise<void>
}

class OrderSagaOrchestrator {
  private completedSteps: SagaStep[] = []
  private sagaLog: SagaLogEntry[] = []

  constructor(
    private paymentService: PaymentService,
    private inventoryService: InventoryService,
    private shippingService: ShippingService,
    private sagaStore: SagaStore
  ) {}

  async execute(orderId: string, orderData: OrderData): Promise<SagaResult> {
    const sagaId = crypto.randomUUID()

    const steps: SagaStep[] = [
      {
        name: 'ProcessPayment',
        action: async () => {
          await this.paymentService.processPayment({
            orderId,
            amount: orderData.totalAmount,
            customerId: orderData.customerId,
          })
        },
        compensation: async () => {
          await this.paymentService.refundPayment({
            orderId,
            amount: orderData.totalAmount,
          })
        },
      },
      {
        name: 'ReserveInventory',
        action: async () => {
          await this.inventoryService.reserve({
            orderId,
            items: orderData.items,
          })
        },
        compensation: async () => {
          await this.inventoryService.releaseReservation({
            orderId,
            items: orderData.items,
          })
        },
      },
      {
        name: 'CreateShipment',
        action: async () => {
          await this.shippingService.createShipment({
            orderId,
            address: orderData.shippingAddress,
            items: orderData.items,
          })
        },
        compensation: async () => {
          await this.shippingService.cancelShipment({ orderId })
        },
      },
    ]

    try {
      for (const step of steps) {
        await this.logStep(sagaId, step.name, 'STARTED')

        try {
          await step.action()
          this.completedSteps.push(step)
          await this.logStep(sagaId, step.name, 'COMPLETED')
        } catch (error) {
          await this.logStep(sagaId, step.name, 'FAILED', error)
          // Execute compensating transactions
          await this.compensate(sagaId)
          return {
            success: false,
            sagaId,
            failedStep: step.name,
            error: (error as Error).message,
          }
        }
      }

      await this.sagaStore.markCompleted(sagaId)
      return { success: true, sagaId }
    } catch (compensationError) {
      // When compensation transaction also fails - manual intervention needed
      await this.sagaStore.markRequiresIntervention(sagaId)
      throw new SagaCompensationFailedError(sagaId, compensationError as Error)
    }
  }

  private async compensate(sagaId: string): Promise<void> {
    // Compensate completed steps in reverse order
    const stepsToCompensate = [...this.completedSteps].reverse()

    for (const step of stepsToCompensate) {
      try {
        await this.logStep(sagaId, step.name, 'COMPENSATING')
        await step.compensation()
        await this.logStep(sagaId, step.name, 'COMPENSATED')
      } catch (error) {
        await this.logStep(sagaId, step.name, 'COMPENSATION_FAILED', error)
        // Register in retry queue on compensation failure
        await this.sagaStore.enqueueRetry(sagaId, step.name)
        throw error
      }
    }
  }

  private async logStep(
    sagaId: string,
    stepName: string,
    status: string,
    error?: unknown
  ): Promise<void> {
    const entry: SagaLogEntry = {
      sagaId,
      stepName,
      status,
      timestamp: new Date(),
      error: error ? (error as Error).message : undefined,
    }
    this.sagaLog.push(entry)
    await this.sagaStore.appendLog(entry)
  }
}

interface SagaResult {
  success: boolean
  sagaId: string
  failedStep?: string
  error?: string
}

interface SagaLogEntry {
  sagaId: string
  stepName: string
  status: string
  timestamp: Date
  error?: string
}

# Python - Choreography-based Saga (event subscription approach)

from abc import ABC, abstractmethod
from dataclasses import dataclass
from enum import Enum
import asyncio
import json


class EventBus:
    """Simple in-memory event bus (use Kafka/RabbitMQ in production)"""

    def __init__(self):
        self._handlers: dict[str, list] = {}

    def subscribe(self, event_type: str, handler):
        if event_type not in self._handlers:
            self._handlers[event_type] = []
        self._handlers[event_type].append(handler)

    async def publish(self, event_type: str, payload: dict):
        handlers = self._handlers.get(event_type, [])
        for handler in handlers:
            await handler(payload)


# Payment Service - Choreography approach
class PaymentService:
    def __init__(self, event_bus: EventBus):
        self.event_bus = event_bus
        # Subscribe to order creation event
        event_bus.subscribe("OrderCreated", self.on_order_created)
        # Subscribe to inventory failure event (for compensation)
        event_bus.subscribe("InventoryReservationFailed", self.on_inventory_failed)

    async def on_order_created(self, payload: dict):
        """Process payment on order creation"""
        try:
            order_id = payload["order_id"]
            amount = payload["total_amount"]

            # Payment processing logic
            payment_result = await self._process_payment(
                order_id, amount
            )

            # Publish success event
            await self.event_bus.publish("PaymentProcessed", {
                "order_id": order_id,
                "payment_id": payment_result["payment_id"],
                "amount": amount,
            })
        except Exception as e:
            # Publish failure event
            await self.event_bus.publish("PaymentFailed", {
                "order_id": payload["order_id"],
                "reason": str(e),
            })

    async def on_inventory_failed(self, payload: dict):
        """Refund payment on inventory failure (compensating transaction)"""
        order_id = payload["order_id"]
        await self._refund_payment(order_id)
        await self.event_bus.publish("PaymentRefunded", {
            "order_id": order_id,
        })

    async def _process_payment(self, order_id: str, amount: int) -> dict:
        # Actual payment processing logic
        return {"payment_id": f"pay-{order_id}"}

    async def _refund_payment(self, order_id: str) -> None:
        # Actual refund processing logic
        pass


# Inventory Service - Choreography approach
class InventoryService:
    def __init__(self, event_bus: EventBus):
        self.event_bus = event_bus
        # Subscribe to payment completion event
        event_bus.subscribe("PaymentProcessed", self.on_payment_processed)

    async def on_payment_processed(self, payload: dict):
        """Reserve inventory on payment completion"""
        try:
            order_id = payload["order_id"]
            # Check and reserve inventory
            await self._reserve_inventory(order_id)

            await self.event_bus.publish("InventoryReserved", {
                "order_id": order_id,
            })
        except InsufficientStockError:
            await self.event_bus.publish("InventoryReservationFailed", {
                "order_id": payload["order_id"],
                "reason": "insufficient_stock",
            })


class InsufficientStockError(Exception):
    pass

# docker-compose.yml - EventStoreDB cluster setup
version: '3.8'
services:
  eventstoredb:
    image: eventstore/eventstore:24.2
    container_name: eventstoredb
    environment:
      - EVENTSTORE_CLUSTER_SIZE=1
      - EVENTSTORE_RUN_PROJECTIONS=All
      - EVENTSTORE_START_STANDARD_PROJECTIONS=true
      - EVENTSTORE_INSECURE=true
      - EVENTSTORE_ENABLE_ATOM_PUB_OVER_HTTP=true
      - EVENTSTORE_MEM_DB=false
      - EVENTSTORE_DB=/var/lib/eventstore-data
      - EVENTSTORE_INDEX=/var/lib/eventstore-index
      - EVENTSTORE_LOG=/var/log/eventstore
    ports:
      - '2113:2113' # HTTP/gRPC
      - '1113:1113' # TCP (legacy)
    volumes:
      - eventstore-data:/var/lib/eventstore-data
      - eventstore-index:/var/lib/eventstore-index
      - eventstore-logs:/var/log/eventstore

  # Kafka - for event streaming (combined with EventStoreDB)
  kafka:
    image: confluentinc/cp-kafka:7.6.0
    container_name: kafka
    environment:
      KAFKA_NODE_ID: 1
      KAFKA_PROCESS_ROLES: broker,controller
      KAFKA_CONTROLLER_QUORUM_VOTERS: 1@kafka:9093
      KAFKA_LISTENERS: PLAINTEXT://0.0.0.0:9092,CONTROLLER://0.0.0.0:9093
      KAFKA_ADVERTISED_LISTENERS: PLAINTEXT://kafka:9092
      KAFKA_CONTROLLER_LISTENER_NAMES: CONTROLLER
      KAFKA_LISTENER_SECURITY_PROTOCOL_MAP: CONTROLLER:PLAINTEXT,PLAINTEXT:PLAINTEXT
      KAFKA_LOG_DIRS: /var/lib/kafka/data
      KAFKA_AUTO_CREATE_TOPICS_ENABLE: 'false'
      KAFKA_NUM_PARTITIONS: 6
      KAFKA_DEFAULT_REPLICATION_FACTOR: 1
      CLUSTER_ID: 'MkU3OEVBNTcwNTJENDM2Qk'
    ports:
      - '9092:9092'
    volumes:
      - kafka-data:/var/lib/kafka/data

volumes:
  eventstore-data:
  eventstore-index:
  eventstore-logs:
  kafka-data:

// TypeScript - Order Saga implemented with Temporal Workflow

import { proxyActivities, defineSignal, setHandler, condition, sleep } from '@temporalio/workflow'

// Activity proxies
const payment = proxyActivities<PaymentActivities>({
  startToCloseTimeout: '30s',
  retry: {
    maximumAttempts: 3,
    initialInterval: '1s',
    backoffCoefficient: 2,
    maximumInterval: '30s',
  },
})

const inventory = proxyActivities<InventoryActivities>({
  startToCloseTimeout: '10s',
  retry: { maximumAttempts: 3 },
})

const shipping = proxyActivities<ShippingActivities>({
  startToCloseTimeout: '15s',
  retry: { maximumAttempts: 3 },
})

const notification = proxyActivities<NotificationActivities>({
  startToCloseTimeout: '5s',
  retry: { maximumAttempts: 5 },
})

// Signal definition (send messages to workflow from outside)
const cancelOrderSignal = defineSignal<[string]>('cancelOrder')

// Order Processing Saga Workflow
export async function orderSagaWorkflow(input: OrderSagaInput): Promise<OrderSagaResult> {
  let isCancelled = false
  let cancelReason = ''

  // Cancel signal handler
  setHandler(cancelOrderSignal, (reason: string) => {
    isCancelled = true
    cancelReason = reason
  })

  const compensations: Array<() => Promise<void>> = []

  try {
    // Step 1: Process payment
    if (isCancelled) throw new SagaCancelledError(cancelReason)

    const paymentResult = await payment.processPayment({
      orderId: input.orderId,
      amount: input.totalAmount,
      customerId: input.customerId,
    })

    compensations.push(async () => {
      await payment.refundPayment({
        orderId: input.orderId,
        paymentId: paymentResult.paymentId,
        amount: input.totalAmount,
      })
    })

    // Step 2: Reserve inventory
    if (isCancelled) throw new SagaCancelledError(cancelReason)

    await inventory.reserveInventory({
      orderId: input.orderId,
      items: input.items,
    })

    compensations.push(async () => {
      await inventory.releaseReservation({
        orderId: input.orderId,
        items: input.items,
      })
    })

    // Step 3: Create shipment
    if (isCancelled) throw new SagaCancelledError(cancelReason)

    const shipmentResult = await shipping.createShipment({
      orderId: input.orderId,
      address: input.shippingAddress,
      items: input.items,
    })

    compensations.push(async () => {
      await shipping.cancelShipment({
        orderId: input.orderId,
        shipmentId: shipmentResult.shipmentId,
      })
    })

    // All steps succeeded - send confirmation notification (Saga unaffected by failure)
    await notification
      .sendOrderConfirmation({
        orderId: input.orderId,
        customerId: input.customerId,
      })
      .catch(() => {
        /* Notification failure is ignored */
      })

    return {
      success: true,
      orderId: input.orderId,
      paymentId: paymentResult.paymentId,
      shipmentId: shipmentResult.shipmentId,
    }
  } catch (error) {
    // Execute compensating transactions (reverse order)
    for (const compensate of compensations.reverse()) {
      try {
        await compensate()
      } catch (compError) {
        // Temporal automatically manages retries
        // Final failure sends to Dead Letter Queue
        console.error('Compensation failed:', compError)
      }
    }

    return {
      success: false,
      orderId: input.orderId,
      error: (error as Error).message,
    }
  }
}

interface OrderSagaInput {
  orderId: string
  customerId: string
  totalAmount: number
  shippingAddress: string
  items: Array<{
    productId: string
    quantity: number
    price: number
  }>
}

interface OrderSagaResult {
  success: boolean
  orderId: string
  paymentId?: string
  shipmentId?: string
  error?: string
}

class SagaCancelledError extends Error {
  constructor(reason: string) {
    super(`Saga cancelled: ${reason}`)
    this.name = 'SagaCancelledError'
  }
}

# AWS CloudFormation - DynamoDB Event Store Table

AWSTemplateFormatVersion: '2010-09-09'
Description: Event Store on DynamoDB

Resources:
  EventStoreTable:
    Type: AWS::DynamoDB::Table
    Properties:
      TableName: event-store
      BillingMode: PAY_PER_REQUEST
      AttributeDefinitions:
        - AttributeName: aggregateId
          AttributeType: S
        - AttributeName: version
          AttributeType: N
        - AttributeName: eventType
          AttributeType: S
        - AttributeName: timestamp
          AttributeType: S
      KeySchema:
        - AttributeName: aggregateId
          KeyType: HASH
        - AttributeName: version
          KeyType: RANGE
      GlobalSecondaryIndexes:
        - IndexName: eventType-timestamp-index
          KeySchema:
            - AttributeName: eventType
              KeyType: HASH
            - AttributeName: timestamp
              KeyType: RANGE
          Projection:
            ProjectionType: ALL
      StreamSpecification:
        StreamViewType: NEW_AND_OLD_IMAGES
      PointInTimeRecoverySpecification:
        PointInTimeRecoveryEnabled: true
      Tags:
        - Key: Environment
          Value: production
        - Key: Service
          Value: event-store

  SnapshotTable:
    Type: AWS::DynamoDB::Table
    Properties:
      TableName: event-store-snapshots
      BillingMode: PAY_PER_REQUEST
      AttributeDefinitions:
        - AttributeName: aggregateId
          AttributeType: S
        - AttributeName: version
          AttributeType: N
      KeySchema:
        - AttributeName: aggregateId
          KeyType: HASH
        - AttributeName: version
          KeyType: RANGE
      Tags:
        - Key: Environment
          Value: production

  SagaStateTable:
    Type: AWS::DynamoDB::Table
    Properties:
      TableName: saga-state
      BillingMode: PAY_PER_REQUEST
      AttributeDefinitions:
        - AttributeName: sagaId
          AttributeType: S
        - AttributeName: createdAt
          AttributeType: S
      KeySchema:
        - AttributeName: sagaId
          KeyType: HASH
      GlobalSecondaryIndexes:
        - IndexName: createdAt-index
          KeySchema:
            - AttributeName: createdAt
              KeyType: HASH
          Projection:
            ProjectionType: ALL
      TimeToLiveSpecification:
        AttributeName: ttl
        Enabled: true

// TypeScript - Idempotency guarantee pattern

class IdempotentEventHandler {
  constructor(
    private processedEvents: ProcessedEventStore,
    private handler: EventHandler
  ) {}

  async handle(event: DomainEvent): Promise<void> {
    // Check if event was already processed
    const isProcessed = await this.processedEvents.exists(event.eventId)
    if (isProcessed) {
      console.log(`Event ${event.eventId} already processed, skipping`)
      return
    }

    try {
      // Process event
      await this.handler.handle(event)

      // Record completion (auto-cleanup with TTL)
      await this.processedEvents.markAsProcessed(event.eventId, {
        processedAt: new Date(),
        ttl: 60 * 60 * 24 * 7, // Expire after 7 days
      })
    } catch (error) {
      // Not recorded on failure so retry is possible
      throw error
    }
  }
}

// Redis-based duplicate detection
class RedisProcessedEventStore implements ProcessedEventStore {
  constructor(private redis: Redis) {}

  async exists(eventId: string): Promise<boolean> {
    const result = await this.redis.exists(`processed:${eventId}`)
    return result === 1
  }

  async markAsProcessed(
    eventId: string,
    options: { processedAt: Date; ttl: number }
  ): Promise<void> {
    await this.redis.setex(`processed:${eventId}`, options.ttl, options.processedAt.toISOString())
  }
}

// TypeScript - Event Upcaster (Schema Evolution)

class EventUpcaster {
  private upcasters: Map<string, Map<number, (event: DomainEvent) => DomainEvent>> = new Map()

  register(
    eventType: string,
    fromVersion: number,
    upcaster: (event: DomainEvent) => DomainEvent
  ): void {
    if (!this.upcasters.has(eventType)) {
      this.upcasters.set(eventType, new Map())
    }
    this.upcasters.get(eventType)!.set(fromVersion, upcaster)
  }

  upcast(event: DomainEvent): DomainEvent {
    const typeUpcasters = this.upcasters.get(event.eventType)
    if (!typeUpcasters) return event

    let currentEvent = event
    let schemaVersion = (event.metadata as any).schemaVersion || 1

    while (typeUpcasters.has(schemaVersion)) {
      const upcasterFn = typeUpcasters.get(schemaVersion)!
      currentEvent = upcasterFn(currentEvent)
      schemaVersion++
    }

    return currentEvent
  }
}

// Usage example: OrderPlaced event v1 -> v2 upcasting
const upcaster = new EventUpcaster()

upcaster.register('OrderPlaced', 1, (event) => {
  // v1 had shippingAddress as a string
  // v2 changed to a structured object
  const payload = event.payload as any
  return {
    ...event,
    payload: {
      ...payload,
      shippingAddress: {
        full: payload.shippingAddress,
        city: '',
        zipCode: '',
      },
    },
    metadata: {
      ...event.metadata,
      schemaVersion: 2,
    },
  }
})

# Python - Projection recovery script

import asyncio
from datetime import datetime


class ProjectionRecovery:
    def __init__(self, event_store, read_db, projection):
        self.event_store = event_store
        self.read_db = read_db
        self.projection = projection

    async def recover_from_checkpoint(self) -> dict:
        """Checkpoint-based projection recovery"""
        # 1. Check last checkpoint
        checkpoint = await self.read_db.get_checkpoint(
            self.projection.name
        )
        last_position = checkpoint.get("position", 0) if checkpoint else 0

        print(
            f"Recovering projection '{self.projection.name}' "
            f"from position {last_position}"
        )

        # 2. Load events after checkpoint
        events = await self.event_store.read_all_from(last_position)
        processed = 0
        errors = 0

        for event in events:
            try:
                await self.projection.handle(event)
                processed += 1

                # Update checkpoint every 100 events
                if processed % 100 == 0:
                    await self.read_db.save_checkpoint(
                        self.projection.name,
                        {"position": event["global_position"]},
                    )
            except Exception as e:
                errors += 1
                print(
                    f"Error processing event "
                    f"{event['event_id']}: {e}"
                )
                # Continue after logging error (skip & log)
                continue

        # 3. Save final checkpoint
        if events:
            await self.read_db.save_checkpoint(
                self.projection.name,
                {"position": events[-1]["global_position"]},
            )

        return {
            "projection": self.projection.name,
            "processed": processed,
            "errors": errors,
            "recovered_at": datetime.utcnow().isoformat(),
        }

    async def full_rebuild(self) -> dict:
        """Full rebuild of read model"""
        print(
            f"Full rebuild of projection '{self.projection.name}'"
        )

        # 1. Delete existing read model
        await self.read_db.drop_projection_data(self.projection.name)

        # 2. Reset checkpoint
        await self.read_db.save_checkpoint(
            self.projection.name, {"position": 0}
        )

        # 3. Replay all events
        return await self.recover_from_checkpoint()