Event Sourcing

Event Sourcing

# Store events not current state
INSERT INTO order_events (order_id,event_type,data) VALUES
(1,'OrderCreated', '{"user_id":5,"items":[...]}'),
(1,'PaymentTaken', '{"amount":99.99}'),
(1,'ItemsShipped', '{"tracking":"DHL-123"}');

# Rebuild state by replaying events
def get_order_state(order_id: int) -> dict:
    events = db.query(
        'SELECT * FROM order_events WHERE order_id=%s ORDER BY id',
        [order_id]
    )
    state = {}
    for e in events:
        if e.type == 'OrderCreated':
            state = {'status':'pending','items':e.data['items']}
        elif e.type == 'PaymentTaken':
            state['status'] = 'paid'
        elif e.type == 'ItemsShipped':
            state['status'] = 'shipped'
    return state

# Benefits: full audit trail, time travel, replay events