Event Sourcing — Deep Dive

graph LR
    E1[OrderCreated] --> E2[ItemAdded]
    E2 --> E3[ItemAdded]
    E3 --> E4[ItemRemoved]
    E4 --> E5[OrderPlaced]
    E5 --> S[Current State: Order with 2 items, status=PLACED]

CREATE TABLE event_store (
    id BIGSERIAL PRIMARY KEY,
    aggregate_id UUID NOT NULL,
    aggregate_type VARCHAR(255) NOT NULL,
    event_type VARCHAR(255) NOT NULL,
    event_data JSONB NOT NULL,
    version INT NOT NULL,
    occurred_at TIMESTAMP NOT NULL,

    UNIQUE (aggregate_id, version)  -- Optimistic concurrency
);

public record StoredEvent(
    UUID id,
    UUID aggregateId,
    String aggregateType,
    String eventType,
    String eventData,        // Serialized event
    int version,             // Aggregate version
    Instant occurredAt,
    Map<String, String> metadata  // correlationId, causationId
) {}

public interface EventStore {
    // Append events (expects version for optimistic concurrency)
    void append(UUID aggregateId, List<DomainEvent> events, int expectedVersion);

    // Load events for an aggregate
    List<StoredEvent> load(UUID aggregateId);

    // Load events from a position (for projections)
    List<StoredEvent> loadFrom(long position, int batchSize);
}

public class Order {
    private OrderId id;
    private CustomerId customerId;
    private List<OrderLine> lines = new ArrayList<>();
    private OrderStatus status;
    private int version;

    // Apply events to build state
    public void apply(OrderCreated event) {
        this.id = event.orderId();
        this.customerId = event.customerId();
        this.status = OrderStatus.DRAFT;
    }

    public void apply(ItemAdded event) {
        this.lines.add(new OrderLine(event.productId(), event.quantity(), event.price()));
    }

    public void apply(ItemRemoved event) {
        this.lines.removeIf(l -> l.productId().equals(event.productId()));
    }

    public void apply(OrderPlaced event) {
        this.status = OrderStatus.PLACED;
    }

    // Reconstruct from events
    public static Order fromEvents(List<DomainEvent> events) {
        Order order = new Order();
        for (DomainEvent event : events) {
            order.apply(event);
            order.version++;
        }
        return order;
    }
}

public class EventSourcedOrderRepository {
    private final EventStore eventStore;

    public Order load(OrderId id) {
        List<StoredEvent> events = eventStore.load(id.value());
        if (events.isEmpty()) {
            throw new OrderNotFoundException(id);
        }
        return Order.fromEvents(deserializeEvents(events));
    }

    public void save(Order order) {
        List<DomainEvent> newEvents = order.getUncommittedEvents();
        eventStore.append(order.getId().value(), newEvents, order.getVersion());
        order.markEventsCommitted();
    }
}

graph LR
    subgraph Event Stream
        E1[Event 1]
        E2[Event 2]
        E3[...]
        E100[Event 100]
        S[Snapshot @ 100]
        E101[Event 101]
        E102[Event 102]
    end
    S --> R[Replay from 101]

public class SnapshotStore {
    void save(UUID aggregateId, int version, String serializedState);
    Optional<Snapshot> load(UUID aggregateId);
}

public Order load(OrderId id) {
    Optional<Snapshot> snapshot = snapshotStore.load(id.value());

    int fromVersion;
    Order order;

    if (snapshot.isPresent()) {
        order = deserialize(snapshot.get().state());
        fromVersion = snapshot.get().version() + 1;
    } else {
        order = new Order();
        fromVersion = 0;
    }

    List<StoredEvent> events = eventStore.loadFrom(id.value(), fromVersion);
    for (StoredEvent event : events) {
        order.apply(deserialize(event));
    }

    return order;
}

graph LR
    ES[(Event Store)] --> P1[Order Summary Projection]
    ES --> P2[Customer Activity Projection]
    ES --> P3[Sales Dashboard Projection]
    P1 --> DB1[(PostgreSQL)]
    P2 --> DB2[(Elasticsearch)]
    P3 --> DB3[(Redis)]

@Component
public class OrderSummaryProjection {
    private final OrderSummaryRepository repository;
    private long lastProcessedPosition;

    @Scheduled(fixedDelay = 100)
    public void process() {
        List<StoredEvent> events = eventStore.loadFrom(lastProcessedPosition, 100);

        for (StoredEvent event : events) {
            switch (event.eventType()) {
                case "OrderCreated" -> createSummary(event);
                case "OrderPlaced" -> updateStatus(event, "PLACED");
                case "OrderShipped" -> updateStatus(event, "SHIPPED");
            }
            lastProcessedPosition = event.id();
        }

        saveCheckpoint(lastProcessedPosition);
    }
}

public class EventUpcaster {
    public DomainEvent upcast(StoredEvent stored) {
        if (stored.eventType().equals("OrderCreated") && stored.version() == 1) {
            // V1 didn't have currency, default to USD
            OrderCreatedV1 v1 = deserialize(stored, OrderCreatedV1.class);
            return new OrderCreatedV2(v1.orderId(), v1.customerId(), v1.amount(), "USD");
        }
        return deserialize(stored);
    }
}

// V1 of the event
public record OrderCreated(OrderId orderId, CustomerId customerId, BigDecimal amount) {}

// V2 adds currency
public record OrderCreatedV2(OrderId orderId, CustomerId customerId, Money amount) {}

public void append(UUID aggregateId, List<Event> events, int expectedVersion) {
    int actualVersion = getLatestVersion(aggregateId);
    if (actualVersion != expectedVersion) {
        throw new ConcurrencyException("Expected version " + expectedVersion 
            + " but was " + actualVersion);
    }
    // Proceed with append
}

public void save(Order order) {
    try {
        eventStore.append(order.getId(), order.getUncommittedEvents(), order.getVersion());
    } catch (ConcurrencyException e) {
        // Reload and retry (or fail)
        Order fresh = load(order.getId());
        // Re-apply command on fresh aggregate
    }
}