Event Sourcing

package com.example.eventsourcing;

import com.example.shared.event.DomainEvent;

import java.util.List;

public interface EventStore {
    void append(String streamId, long expectedVersion, List<DomainEvent> events);
    List<DomainEvent> load(String streamId);
}

CREATE TABLE event_store (
    stream_id   VARCHAR(255) NOT NULL,
    version     BIGINT NOT NULL,
    event_type  VARCHAR(255) NOT NULL,
    payload     TEXT NOT NULL,
    occurred_on TIMESTAMP NOT NULL DEFAULT now(),
    PRIMARY KEY (stream_id, version)
);

package com.example.eventsourcing;

import com.example.shared.event.DomainEvent;
import org.springframework.jdbc.core.JdbcTemplate;
import org.springframework.stereotype.Component;
import org.springframework.transaction.annotation.Transactional;

import java.time.Instant;
import java.util.List;

@Component
public class SqlEventStore implements EventStore {

    private final JdbcTemplate jdbcTemplate;
    private final EventSerializer eventSerializer; // e.g. a thin Jackson ObjectMapper wrapper

    public SqlEventStore(JdbcTemplate jdbcTemplate, EventSerializer eventSerializer) {
        this.jdbcTemplate = jdbcTemplate;
        this.eventSerializer = eventSerializer;
    }

    @Override
    @Transactional
    public void append(String streamId, long expectedVersion, List<DomainEvent> events) {
        long currentVersion = currentVersionOf(streamId);
        if (currentVersion != expectedVersion) {
            throw new OptimisticConcurrencyException(streamId, expectedVersion, currentVersion);
        }
        long nextVersion = expectedVersion;
        for (DomainEvent event : events) {
            nextVersion++;
            // (stream_id, version) is the primary key -- a concurrent writer that already
            // claimed nextVersion makes this insert fail instead of silently overwriting it.
            jdbcTemplate.update(
                    "INSERT INTO event_store (stream_id, version, event_type, payload, occurred_on) "
                            + "VALUES (?, ?, ?, ?, ?)",
                    streamId, nextVersion, event.getClass().getSimpleName(),
                    eventSerializer.toJson(event), Instant.now());
        }
    }

    @Override
    public List<DomainEvent> load(String streamId) {
        return jdbcTemplate.query(
                "SELECT event_type, payload FROM event_store WHERE stream_id = ? ORDER BY version ASC",
                (rs, rowNum) -> eventSerializer.fromJson(rs.getString("event_type"), rs.getString("payload")),
                streamId);
    }

    private long currentVersionOf(String streamId) {
        Long version = jdbcTemplate.queryForObject(
                "SELECT COALESCE(MAX(version), 0) FROM event_store WHERE stream_id = ?", Long.class, streamId);
        return version == null ? 0L : version;
    }
}

class OptimisticConcurrencyException extends RuntimeException {
    OptimisticConcurrencyException(String streamId, long expectedVersion, long actualVersion) {
        super("Stream " + streamId + " expected version " + expectedVersion
                + " but was " + actualVersion);
    }
}

package com.example.eventsourcing;

import com.example.order.domain.CustomerId;
import com.example.order.domain.OrderLine;
import com.example.order.domain.OrderStatus;

import java.util.List;

public record OrderSnapshot(String streamId, long version, CustomerId customerId,
                             List<OrderLine> lines, OrderStatus status) {
}

package com.example.eventsourcing;

import com.example.order.domain.OrderId;
import com.example.order.domain.OrderLine;
import com.example.order.domain.OrderStatus;
import com.example.order.domain.event.OrderApproved;
import com.example.order.domain.event.OrderCancelled;
import com.example.order.domain.event.OrderPlaced;
import com.example.order.domain.event.OrderRejected;
import com.example.order.application.PlaceOrderCommand;
import com.example.shared.CustomerId;
import com.example.shared.Money;
import com.example.shared.event.DomainEvent;

import java.time.Instant;
import java.util.ArrayList;
import java.util.List;

public class EventSourcedOrder {

    private OrderId id;
    private CustomerId customerId;
    private final List<OrderLine> lines = new ArrayList<>();
    private OrderStatus status;
    private long version; // the event-store stream version this instance reflects

    private EventSourcedOrder() {
    }

    /** Fresh aggregate, no history yet -- decide() below produces the very first event. */
    public static EventSourcedOrder newOrder() {
        return new EventSourcedOrder();
    }

    /** Reconstitution: fold(apply, events) over a stream freshly loaded from the EventStore. */
    public static EventSourcedOrder reconstitute(List<DomainEvent> events) {
        EventSourcedOrder order = new EventSourcedOrder();
        for (DomainEvent event : events) {
            order.apply(event);
            order.version++;
        }
        return order;
    }

    // --- decide: validate against current state, return the events that should happen ---

    public List<DomainEvent> decide(PlaceOrderCommand command) {
        if (status != null) {
            throw new IllegalStateException("Order already placed");
        }
        OrderId newId = OrderId.generate();
        Money totalAmount = command.lines().stream()
                .map(OrderLine::lineTotal)
                .reduce(Money::add)
                .orElseThrow();
        return List.of(new OrderPlaced(newId, command.customerId(), List.copyOf(command.lines()),
                totalAmount, Instant.now()));
    }

    public List<DomainEvent> decideApprove() {
        requireStatus(OrderStatus.APPROVAL_PENDING);
        return List.of(new OrderApproved(id, Instant.now()));
    }

    public List<DomainEvent> decideReject(String reason) {
        requireStatus(OrderStatus.APPROVAL_PENDING);
        return List.of(new OrderRejected(id, reason, Instant.now()));
    }

    public List<DomainEvent> decideCancel() {
        if (status != OrderStatus.APPROVAL_PENDING && status != OrderStatus.APPROVED) {
            throw new IllegalStateException("Cannot cancel order " + id + " in status " + status);
        }
        return List.of(new OrderCancelled(id, Instant.now()));
    }

    // --- apply: mutate state to reflect an event that already happened -- never fails ---

    public void apply(DomainEvent event) {
        switch (event) {
            case OrderPlaced placed -> {
                this.id = placed.orderId();
                this.customerId = placed.customerId();
                this.lines.addAll(placed.lines());
                this.status = OrderStatus.APPROVAL_PENDING;
            }
            case OrderApproved approved -> this.status = OrderStatus.APPROVED;
            case OrderRejected rejected -> this.status = OrderStatus.REJECTED;
            case OrderCancelled cancelled -> this.status = OrderStatus.CANCELLED;
            default -> throw new IllegalArgumentException("Unknown event for EventSourcedOrder: " + event);
        }
    }

    public OrderId id() {
        return id;
    }

    public OrderStatus status() {
        return status;
    }

    public long version() {
        return version;
    }

    private void requireStatus(OrderStatus expected) {
        if (status != expected) {
            throw new IllegalStateException("Expected order " + id + " to be " + expected + " but was " + status);
        }
    }
}

package com.example.eventsourcing;

import com.example.shared.event.DomainEvent;
import org.springframework.stereotype.Service;
import org.springframework.transaction.annotation.Transactional;

import java.util.List;

@Service
public class ApproveOrderEventSourcedService {

    private final EventStore eventStore;

    public ApproveOrderEventSourcedService(EventStore eventStore) {
        this.eventStore = eventStore;
    }

    @Transactional
    public void approve(String streamId) {
        EventSourcedOrder order = EventSourcedOrder.reconstitute(eventStore.load(streamId));
        List<DomainEvent> events = order.decideApprove();
        // A concurrent approve()/reject()/cancel() against the same stream loses this race --
        // append(...) throws OptimisticConcurrencyException instead of silently corrupting history.
        eventStore.append(streamId, order.version(), events);
    }
}

package com.example.eventsourcing;

import com.example.order.domain.OrderId;
import com.example.order.domain.OrderStatus;
import com.example.order.domain.event.OrderApproved;
import com.example.order.domain.event.OrderPlaced;
import com.example.shared.CustomerId;
import com.example.shared.Money;
import com.example.shared.event.DomainEvent;
import org.junit.jupiter.api.Test;

import java.math.BigDecimal;
import java.time.Instant;
import java.util.Currency;
import java.util.List;
import java.util.UUID;

import static org.assertj.core.api.Assertions.assertThat;

class EventSourcedOrderTest {

    @Test
    void approvingAPendingOrderYieldsOrderApproved() {
        // given: OrderPlaced already happened
        OrderId orderId = OrderId.generate();
        OrderPlaced given = new OrderPlaced(orderId, new CustomerId(UUID.randomUUID()), List.of(),
                new Money(BigDecimal.valueOf(75), Currency.getInstance("USD")), Instant.now());
        EventSourcedOrder order = EventSourcedOrder.reconstitute(List.of(given));

        // when: approve is decided
        List<DomainEvent> events = order.decideApprove();

        // then: exactly one OrderApproved for the same order, and applying it reaches APPROVED
        assertThat(events).hasSize(1);
        assertThat(events.get(0)).isInstanceOf(OrderApproved.class);
        assertThat(((OrderApproved) events.get(0)).orderId()).isEqualTo(orderId);

        events.forEach(order::apply);
        assertThat(order.status()).isEqualTo(OrderStatus.APPROVED);
    }
}