Event Sourcing — Event-Driven Data Persistence for Microservices

# event_sourcing_basics.py
# Event sourcing fundamentals

def event_sourcing_basics():
    print("Event Sourcing Fundamentals")
    print("=" * 40)
    print()
    
    concepts = [
        {
            "concept": "Event Store",
            "desc": "Append-only database of events. Events are never updated or deleted.",
            "analogy": "Like a bank statement showing every transaction"
        },
        {
            "concept": "Event",
            "desc": "Immutable record of something that happened. Contains type, timestamp, data, and metadata.",
            "example": "OrderPlaced { orderId: '123', total: 59.99 }"
        },
        {
            "concept": "Aggregate",
            "desc": "A cluster of domain objects treated as a unit. Events are recorded per aggregate.",
            "example": "An Order aggregate: events are OrderPlaced, ItemAdded, PaymentReceived"
        },
        {
            "concept": "Projection",
            "desc": "Current state derived by replaying events. Can have multiple projections from the same events.",
            "example": "OrderSummary projection, CustomerSpend projection (same events, different views)"
        },
        {
            "concept": "Snapshot",
            "desc": "Periodic save of current state to avoid replaying all events from the beginning.",
            "example": "Snapshot after every 100 events. Replay from snapshot instead of event 1."
        },
    ]
    
    for c in concepts:
        print(f"{c['concept']:20s}")
        print(f"  {c['desc']}")
        if 'analogy' in c:
            print(f"  Analogy: {c['analogy']}")
        if 'example' in c:
            print(f"  Example: {c['example']}")
        print()

event_sourcing_basics()

# event_store.py
# Event store implementation

def event_store_impl():
    print("Event Store Implementation")
    print("=" * 40)
    print()
    
    store_code = """
import json
import uuid
from datetime import datetime, timezone

class EventStore:
    def __init__(self):
        self.events = []  # In-memory; use PostgreSQL/Kafka in production
    
    def append(self, aggregate_type, aggregate_id, 
               event_type, data, expected_version=None):
        """Append event with optimistic concurrency check."""
        version = self.get_current_version(aggregate_type, aggregate_id)
        
        if expected_version is not None and version != expected_version:
            raise ConcurrencyException(
                f"Expected version {expected_version}, got {version}"
            )
        
        event = {
            "id": str(uuid.uuid4()),
            "aggregate_type": aggregate_type,
            "aggregate_id": aggregate_id,
            "event_type": event_type,
            "version": version + 1,
            "data": data,
            "timestamp": datetime.now(timezone.utc).isoformat(),
        }
        
        self.events.append(event)
        return event
    
    def get_events(self, aggregate_type, aggregate_id):
        """Get all events for an aggregate."""
        return [
            e for e in self.events
            if e["aggregate_type"] == aggregate_type 
            and e["aggregate_id"] == aggregate_id
        ]
    
    def get_current_version(self, aggregate_type, aggregate_id):
        events = self.get_events(aggregate_type, aggregate_id)
        return len(events)
    
    def get_all_events_since(self, timestamp):
        """For building projections."""
        return [
            e for e in self.events
            if e["timestamp"] >= timestamp
        ]

class ConcurrencyException(Exception):
    pass

# Usage
store = EventStore()
store.append("Order", "ORD-123", "OrderPlaced", 
             {"customer_id": "CUST-1", "total": 59.99})
store.append("Order", "ORD-123", "PaymentReceived",
             {"amount": 59.99, "method": "card"})
store.append("Order", "ORD-123", "OrderShipped",
             {"carrier": "UPS", "tracking": "1Z999AA10123456784"})

events = store.get_events("Order", "ORD-123")
print(f"Order ORD-123 has {len(events)} events")
"""
    print(store_code)

event_store_impl()

# state_rebuild.py
# Rebuilding state from events

def state_rebuild():
    print("Rebuilding State from Events")
    print("=" * 40)
    print()
    
    projection_code = """
class OrderProjection:
    """Build current order state from events."""
    
    def __init__(self, event_store):
        self.event_store = event_store
    
    def get_order(self, order_id):
        events = self.event_store.get_events("Order", order_id)
        return self._reconstruct(events)
    
    def _reconstruct(self, events):
        state = {
            "order_id": None,
            "customer_id": None,
            "items": [],
            "total": 0.0,
            "status": "pending",
            "payments": [],
            "shipments": [],
        }
        
        for event in events:
            if event["event_type"] == "OrderPlaced":
                state["order_id"] = event["aggregate_id"]
                state["customer_id"] = event["data"]["customer_id"]
                state["items"] = event["data"].get("items", [])
                state["total"] = event["data"]["total"]
                state["status"] = "placed"
            
            elif event["event_type"] == "ItemAdded":
                state["items"].append(event["data"]["item"])
                state["total"] += event["data"]["price"]
            
            elif event["event_type"] == "PaymentReceived":
                state["payments"].append(event["data"])
                state["status"] = "paid"
            
            elif event["event_type"] == "OrderShipped":
                state["shipments"].append(event["data"])
                state["status"] = "shipped"
            
            elif event["event_type"] == "OrderCancelled":
                state["status"] = "cancelled"
                state["cancel_reason"] = event["data"].get("reason")
        
        return state
    
    def get_all_active_orders(self):
        """Get all orders that are not delivered/cancelled."""
        all_orders = []
        # In production, maintain a separate index for this
        for event in self.event_store.events:
            if event["event_type"] == "OrderPlaced":
                state = self.get_order(event["aggregate_id"])
                if state["status"] not in ("delivered", "cancelled"):
                    all_orders.append(state)
        return all_orders

# Time travel: get order state as of yesterday
def get_order_at_time(events, target_timestamp):
    relevant = [e for e in events if e["timestamp"] <= target_timestamp]
    return OrderProjection._reconstruct(relevant)
"""
    print(projection_code)

state_rebuild()

# when_to_use.py
# Event sourcing suitability

def when_to_use():
    print("When to Use Event Sourcing")
    print("=" * 40)
    print()
    
    criteria = [
        {
            "scenario": "Audit trail required",
            "suitable": "Yes",
            "reason": "Every state change preserved forever with full context"
        },
        {
            "scenario": "Temporal queries needed",
            "suitable": "Yes",
            "reason": "Can reconstruct state at any point in time"
        },
        {
            "scenario": "Complex business logic",
            "suitable": "Yes",
            "reason": "Events model business processes naturally"
        },
        {
            "scenario": "Simple CRUD application",
            "suitable": "No",
            "reason": "Event sourcing adds complexity with no benefit"
        },
        {
            "scenario": "High write throughput",
            "suitable": "Maybe",
            "reason": "Append-only writes are fast but projections add read cost"
        },
        {
            "scenario": "Data deletion required (GDPR)",
            "suitable": "No",
            "reason": "Immutable event log conflicts with right-to-deletion"
        },
    ]
    
    print(f"{'Scenario':35s} {'Suitable':10s}")
    print("-" * 48)
    for c in criteria:
        print(f"{c['scenario']:35s} {c['suitable']:10s}  {c['reason']}")

when_to_use()

def inventory_event_sourcing():
    print("Inventory Management - Event Sourced")
    print("=" * 40)
    print()
    print("Events:")
    print("  ProductRegistered { sku, name, category, reorder_point }")
    print("  StockAdded        { sku, quantity, location }")
    print("  StockDeducted     { sku, quantity, order_id }")
    print("  StockAdjusted     { sku, new_count, reason }")
    print("  StockTransferred  { sku, from_location, to_location, qty }")
    print()
    print("Projections:")
    print("  CurrentStock      - Replay all events, latest count per SKU")
    print("  MovementHistory   - Replay events filtered by SKU")
    print("  LowStockAlerts    - Replay, check count < reorder_point")
    print()
    print("Snapshot: Every 100 events save CurrentState")
    print("Recovery: Load latest snapshot, replay remaining events")

inventory_event_sourcing()