Keep an onboarding history

The first service keeps customer applications in memory. To retain how an application reached its current state, store its domain events and reconstruct the application from that history.

In this separate runnable project, a customer submits an application and a reviewer approves it. PostgreSQL stores ApplicationSubmitted and ApplicationApproved. Reloading those events rebuilds the applicant details, approval state, and aggregate version.

The complete project is docs/examples/onboarding-event-store. It tests the persistence layer directly, using a real PostgreSQL container. It does not expose an HTTP API.

Run the project

Use JDK 25 and a running Docker engine. Install the framework libraries from the checkout, then run the standalone project:

./mvnw -B -ntp install
./mvnw -B -ntp -f docs/examples/onboarding-event-store/pom.xml clean verify

Testcontainers creates a PostgreSQL 15 container and initializes its events table. The four tests exercise the real event store; Docker must be available. There are no cloud credentials or external broker settings to supply.

The POM keeps the application's version separate from the framework version and declares Lombok as a provided dependency:

<?xml version="1.0" encoding="UTF-8"?>
<project xmlns="http://maven.apache.org/POM/4.0.0"
         xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
         xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 https://maven.apache.org/xsd/maven-4.0.0.xsd">
    <modelVersion>4.0.0</modelVersion>
    <parent>
        <groupId>org.fireflyframework</groupId>
        <artifactId>fireflyframework-parent</artifactId>
        <version>26.10.20-SNAPSHOT</version>
        <relativePath/>
    </parent>
    <groupId>com.example</groupId>
    <artifactId>onboarding-event-store</artifactId>
    <version>0.1.0-SNAPSHOT</version>
    <name>Customer onboarding event-store example</name>
    <dependencies>
        <dependency>
            <groupId>org.fireflyframework</groupId>
            <artifactId>fireflyframework-eventsourcing</artifactId>
        </dependency>
        <dependency>
            <groupId>org.projectlombok</groupId>
            <artifactId>lombok</artifactId>
            <scope>provided</scope>
        </dependency>
        <dependency>
            <groupId>org.postgresql</groupId>
            <artifactId>r2dbc-postgresql</artifactId>
        </dependency>
        <dependency>
            <groupId>org.postgresql</groupId>
            <artifactId>postgresql</artifactId>
            <scope>test</scope>
        </dependency>
        <dependency>
            <groupId>org.springframework.boot</groupId>
            <artifactId>spring-boot-starter-test</artifactId>
            <scope>test</scope>
        </dependency>
        <dependency>
            <groupId>io.projectreactor</groupId>
            <artifactId>reactor-test</artifactId>
            <scope>test</scope>
        </dependency>
        <dependency>
            <groupId>org.testcontainers</groupId>
            <artifactId>testcontainers-junit-jupiter</artifactId>
            <scope>test</scope>
        </dependency>
        <dependency>
            <groupId>org.testcontainers</groupId>
            <artifactId>testcontainers-postgresql</artifactId>
            <scope>test</scope>
        </dependency>
    </dependencies>
</project>

Record what happened

ApplicationSubmitted carries the customer details at submission. Its timestamp is fixed when the event is created:

package com.example.onboarding;

import com.fasterxml.jackson.annotation.JsonIgnoreProperties;
import com.fasterxml.jackson.annotation.JsonTypeName;
import java.time.Instant;
import java.util.UUID;
import org.fireflyframework.eventsourcing.annotation.DomainEvent;
import org.fireflyframework.eventsourcing.domain.Event;

@DomainEvent(value = "onboarding.application.submitted", publishable = false)
@JsonTypeName("onboarding.application.submitted")
@JsonIgnoreProperties(ignoreUnknown = true)
public record ApplicationSubmitted(UUID aggregateId, String applicantName, String email,
                                   Instant occurredAt) implements Event {
    public ApplicationSubmitted(UUID aggregateId, String applicantName, String email) {
        this(aggregateId, applicantName, email, Instant.now());
    }

    @Override public UUID getAggregateId() { return aggregateId; }
    @Override public Instant getEventTimestamp() { return occurredAt; }
}

ApplicationApproved records the reviewer responsible for the approval:

package com.example.onboarding;

import com.fasterxml.jackson.annotation.JsonIgnoreProperties;
import com.fasterxml.jackson.annotation.JsonTypeName;
import java.time.Instant;
import java.util.UUID;
import org.fireflyframework.eventsourcing.annotation.DomainEvent;
import org.fireflyframework.eventsourcing.domain.Event;

@DomainEvent(value = "onboarding.application.approved", publishable = false)
@JsonTypeName("onboarding.application.approved")
@JsonIgnoreProperties(ignoreUnknown = true)
public record ApplicationApproved(UUID aggregateId, String reviewer, Instant occurredAt) implements Event {
    public ApplicationApproved(UUID aggregateId, String reviewer) {
        this(aggregateId, reviewer, Instant.now());
    }

    @Override public UUID getAggregateId() { return aggregateId; }
    @Override public Instant getEventTimestamp() { return occurredAt; }
}

Both implement Jafly's Event contract. @DomainEvent supplies the event name. The test configures Jackson's subtype registration explicitly so loading an event returns its concrete Java type. Persisted event names and fields form a compatibility contract; plan their evolution before changing them.

These records contain customer names and email addresses. A deployed service needs its own access controls, retention policy, and protection for stored personal data.

Put state rules in the aggregate

The aggregate permits submission from DRAFT and approval from SUBMITTED. Each operation applies a new event. The on methods update state during both a new operation and replay:

package com.example.onboarding;

import java.util.UUID;
import lombok.Getter;
import org.fireflyframework.eventsourcing.aggregate.AggregateRoot;

@Getter
public class OnboardingAggregate extends AggregateRoot {
    public static final String TYPE = "CustomerOnboarding";

    public enum Status { DRAFT, SUBMITTED, APPROVED }

    private Status status = Status.DRAFT;
    private String applicantName;
    private String email;
    private String reviewer;

    public OnboardingAggregate(UUID id) {
        super(id, TYPE);
    }

    public void submit(String applicantName, String email) {
        if (status != Status.DRAFT) {
            throw new IllegalStateException("Only a draft application can be submitted");
        }
        requireText(applicantName, "Applicant name");
        requireText(email, "Email");
        applyChange(new ApplicationSubmitted(getId(), applicantName, email));
    }

    public void approve(String reviewer) {
        if (status != Status.SUBMITTED) {
            throw new IllegalStateException("Only a submitted application can be approved");
        }
        requireText(reviewer, "Reviewer");
        applyChange(new ApplicationApproved(getId(), reviewer));
    }

    private void on(ApplicationSubmitted event) {
        applicantName = event.applicantName();
        email = event.email();
        status = Status.SUBMITTED;
    }

    private void on(ApplicationApproved event) {
        reviewer = event.reviewer();
        status = Status.APPROVED;
    }

    private static void requireText(String value, String label) {
        if (value == null || value.isBlank()) {
            throw new IllegalArgumentException(label + " is required");
        }
    }
}

A new aggregate starts at version -1. Submission produces version 0; approval produces version 1. These are aggregate history positions. The event schema version is a separate concept.

The aggregate checks required text and allowed transitions. The service must still validate HTTP input, identify the reviewer, and authorize the operation before calling it.

Save and reconstruct the application

OnboardingEventService belongs to this example. It connects the aggregate to the framework's EventStore interface:

package com.example.onboarding;

import java.util.List;
import java.util.UUID;
import lombok.RequiredArgsConstructor;
import org.fireflyframework.eventsourcing.store.EventStore;
import reactor.core.publisher.Mono;

@RequiredArgsConstructor
public class OnboardingEventService {
    private final EventStore eventStore;

    public Mono<OnboardingAggregate> save(OnboardingAggregate application) {
        return Mono.defer(() -> {
            var pending = List.copyOf(application.getUncommittedEvents());
            if (pending.isEmpty()) {
                return Mono.just(application);
            }
            long expectedVersion = application.getCurrentVersion() - pending.size();
            return eventStore.appendEvents(application.getId(), OnboardingAggregate.TYPE,
                            pending, expectedVersion)
                    .map(stream -> {
                        application.markEventsAsCommitted();
                        return application;
                    });
        });
    }

    public Mono<OnboardingAggregate> load(UUID id) {
        return eventStore.loadEventStream(id, OnboardingAggregate.TYPE)
                .flatMap(stream -> {
                    if (stream.isEmpty()) {
                        return Mono.empty();
                    }
                    var application = new OnboardingAggregate(id);
                    application.loadFromHistory(stream.getEvents());
                    return Mono.just(application);
                });
    }
}

Saving sends only uncommitted events. The expected stored version is the current aggregate version minus its pending event count. The service clears pending events after the append succeeds.

Loading creates a new aggregate and passes its ordered history to loadFromHistory. An unknown ID returns an empty Mono. The test verifies that the reloaded object is a different instance with the same applicant details and the correct approval state.

Configure the PostgreSQL boundary

The test creates one PostgresqlConnectionFactory and uses it for DatabaseClient, R2dbcEntityTemplate, R2dbcTransactionManager, and TransactionalOperator. Those objects are passed to R2dbcEventStore. The test also registers the concrete event types with its Jackson mapper.

Its SQL fixture follows the event store's table contract:

CREATE TABLE events (
    event_id UUID PRIMARY KEY,
    aggregate_id UUID NOT NULL,
    aggregate_type VARCHAR(255) NOT NULL,
    aggregate_version BIGINT NOT NULL,
    global_sequence BIGSERIAL UNIQUE,
    event_type VARCHAR(255) NOT NULL,
    event_data TEXT NOT NULL,
    metadata TEXT,
    created_at TIMESTAMP WITH TIME ZONE DEFAULT NOW(),
    CONSTRAINT unique_aggregate_version UNIQUE(aggregate_id, aggregate_version)
);
CREATE INDEX idx_events_aggregate ON events(aggregate_id, aggregate_type);

The unique (aggregate_id, aggregate_version) constraint protects history positions. Event data and metadata use TEXT, matching the module's initial PostgreSQL migration. event_type has a 255-character limit.

This example initializes a disposable test schema. A deployed application must manage its database migrations and indexes. Review the module's migration resources before configuring additional features.

Test success and failure

The integration test covers four outcomes:

Scenario	Expected result
Submit, reload, approve, reload	Applicant data survives reconstruction; status is APPROVED; stored version is 1
Two reviewers load version 0	The first approval commits; the stale writer receives ConcurrencyException; its event is not stored
One event exceeds the SQL event-type limit	PostgreSQL returns SQLSTATE 22001; the append leaves no event rows
Approve before submission or load an unknown ID	The invalid transition is rejected; the unknown application stays absent

package com.example.onboarding;

import com.fasterxml.jackson.databind.ObjectMapper;
import com.fasterxml.jackson.databind.jsontype.NamedType;
import com.fasterxml.jackson.datatype.jsr310.JavaTimeModule;
import io.r2dbc.postgresql.PostgresqlConnectionConfiguration;
import io.r2dbc.postgresql.PostgresqlConnectionFactory;
import io.r2dbc.spi.R2dbcException;
import java.time.Duration;
import java.util.List;
import java.util.UUID;
import org.fireflyframework.eventsourcing.config.EventSourcingProperties;
import org.fireflyframework.eventsourcing.domain.Event;
import org.fireflyframework.eventsourcing.store.ConcurrencyException;
import org.fireflyframework.eventsourcing.store.r2dbc.R2dbcEventStore;
import org.junit.jupiter.api.BeforeEach;
import org.junit.jupiter.api.Test;
import org.springframework.data.r2dbc.core.R2dbcEntityTemplate;
import org.springframework.r2dbc.connection.R2dbcTransactionManager;
import org.springframework.r2dbc.core.DatabaseClient;
import org.springframework.transaction.reactive.TransactionalOperator;
import org.testcontainers.containers.PostgreSQLContainer;
import org.testcontainers.junit.jupiter.Container;
import org.testcontainers.junit.jupiter.Testcontainers;
import reactor.test.StepVerifier;

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

@Testcontainers
class OnboardingEventStoreTest {
    private static final Duration TIMEOUT = Duration.ofSeconds(10);

    @Container
    static final PostgreSQLContainer<?> postgres = new PostgreSQLContainer<>("postgres:15-alpine")
            .withDatabaseName("onboarding")
            .withUsername("onboarding_test")
            .withPassword("onboarding_test")
            .withInitScript("events.sql");

    private DatabaseClient database;
    private R2dbcEventStore store;
    private OnboardingEventService service;

    @BeforeEach
    void connectToPostgres() {
        var connectionFactory = new PostgresqlConnectionFactory(PostgresqlConnectionConfiguration.builder()
                .host(postgres.getHost()).port(postgres.getFirstMappedPort())
                .database(postgres.getDatabaseName()).username(postgres.getUsername())
                .password(postgres.getPassword()).build());
        database = DatabaseClient.create(connectionFactory);
        var manager = new R2dbcTransactionManager(connectionFactory);
        var mapper = new ObjectMapper().registerModule(new JavaTimeModule());
        mapper.registerSubtypes(new NamedType(ApplicationSubmitted.class, "onboarding.application.submitted"),
                new NamedType(ApplicationApproved.class, "onboarding.application.approved"));
        store = new R2dbcEventStore(database, new R2dbcEntityTemplate(connectionFactory), mapper,
                new EventSourcingProperties(), manager, TransactionalOperator.create(manager), connectionFactory);
        service = new OnboardingEventService(store);
    }

    @Test
    void reconstructsTheApplicationAndItsApprovalFromStoredEvents() {
        UUID id = UUID.randomUUID();
        var application = new OnboardingAggregate(id);
        application.submit("Ada Lovelace", "ada@example.com");
        service.save(application).block(TIMEOUT);
        assertThat(application.hasUncommittedEvents()).isFalse();

        var submitted = service.load(id).block(TIMEOUT);
        assertThat(submitted).isNotSameAs(application);
        assertThat(submitted.getApplicantName()).isEqualTo("Ada Lovelace");
        assertThat(submitted.getEmail()).isEqualTo("ada@example.com");
        assertThat(submitted.getStatus()).isEqualTo(OnboardingAggregate.Status.SUBMITTED);
        assertThat(submitted.getCurrentVersion()).isZero();
        submitted.approve("reviewer-1");
        service.save(submitted).block(TIMEOUT);

        var reloaded = service.load(id).block(TIMEOUT);
        assertThat(reloaded.getStatus()).isEqualTo(OnboardingAggregate.Status.APPROVED);
        assertThat(reloaded.getReviewer()).isEqualTo("reviewer-1");
        assertThat(reloaded.getCurrentVersion()).isEqualTo(1);
        assertThat(reloaded.hasUncommittedEvents()).isFalse();
        var stream = store.loadEventStream(id, OnboardingAggregate.TYPE).block(TIMEOUT);
        assertThat(stream.getEvents()).hasSize(2);
        assertThat(stream.getEvents().get(0).getEvent()).isInstanceOf(ApplicationSubmitted.class);
        assertThat(stream.getEvents().get(1).getEvent()).isInstanceOf(ApplicationApproved.class);
        assertThat(rowCount(id)).isEqualTo(2);
    }

    @Test
    void rejectsAStaleWriterWithoutOverwritingTheCommittedApproval() {
        UUID id = UUID.randomUUID();
        var application = new OnboardingAggregate(id);
        application.submit("Grace Hopper", "grace@example.com");
        service.save(application).block(TIMEOUT);
        var firstReviewer = service.load(id).block(TIMEOUT);
        var staleReviewer = service.load(id).block(TIMEOUT);
        firstReviewer.approve("reviewer-a");
        staleReviewer.approve("reviewer-b");
        service.save(firstReviewer).block(TIMEOUT);

        StepVerifier.create(service.save(staleReviewer))
                .expectErrorSatisfies(error -> {
                    assertThat(error).isInstanceOf(ConcurrencyException.class);
                    var conflict = (ConcurrencyException) error;
                    assertThat(conflict.getExpectedVersion()).isZero();
                    assertThat(conflict.getActualVersion()).isEqualTo(1);
                }).verify(TIMEOUT);
        assertThat(staleReviewer.hasUncommittedEvents()).isTrue();
        assertThat(service.load(id).block(TIMEOUT).getReviewer()).isEqualTo("reviewer-a");
        assertThat(rowCount(id)).isEqualTo(2);
    }

    @Test
    void rollsBackTheWholeAppendWhenOneEventViolatesTheSchema() {
        UUID id = UUID.randomUUID();
        var submitted = new ApplicationSubmitted(id, "Ada Lovelace", "ada@example.com");
        var invalid = new InvalidStoredEvent(id);
        StepVerifier.create(store.appendEvents(id, OnboardingAggregate.TYPE,
                        List.of(submitted, invalid), -1L))
                .expectErrorSatisfies(error -> {
                    Throwable cause = error;
                    while (cause != null && !(cause instanceof R2dbcException)) {
                        cause = cause.getCause();
                    }
                    assertThat(cause).isInstanceOf(R2dbcException.class);
                    assertThat(((R2dbcException) cause).getSqlState()).isEqualTo("22001");
                }).verify(TIMEOUT);
        assertThat(rowCount(id)).isZero();
        assertThat(store.getAggregateVersion(id, OnboardingAggregate.TYPE).block(TIMEOUT)).isEqualTo(-1);
    }

    @Test
    void refusesApprovalBeforeSubmissionAndLeavesUnknownApplicationsAbsent() {
        UUID id = UUID.randomUUID();
        var application = new OnboardingAggregate(id);
        assertThatThrownBy(() -> application.approve("reviewer-1"))
                .isInstanceOf(IllegalStateException.class);
        assertThat(application.hasUncommittedEvents()).isFalse();
        StepVerifier.create(service.load(id)).verifyComplete();
        assertThat(rowCount(id)).isZero();
    }

    private long rowCount(UUID id) {
        return database.sql("SELECT count(*) AS total FROM events WHERE aggregate_id = :id")
                .bind("id", id).map((row, metadata) -> row.get("total", Long.class))
                .one().block(TIMEOUT);
    }

    record InvalidStoredEvent(UUID aggregateId) implements Event {
        @Override public UUID getAggregateId() { return aggregateId; }
        @Override public String getEventType() { return "x".repeat(256); }
    }
}

After a version conflict, reload the history and reconsider the requested operation against its current state. Repeating the old write blindly can violate a business rule even if a later database append succeeds.

Keep the transaction claim precise

The rollback test proves that a failed batch append leaves no event rows when this store uses the configured PostgreSQL connection factory and transaction manager. It does not prove atomicity with a separate database, HTTP call, or message broker.

No outbox service or broker publisher is configured in this project. The event-sourcing module has optional outbox support, but enabling and testing that path is separate work. This example also uses no snapshots or projection worker: replay rebuilds the aggregate itself.

Continue with event sourcing for the module's other APIs, or business coordination when onboarding spans several operations with recovery or compensating steps.