DOCSEARCHbyAlgolia
Submit Docs

postgresql-event-sourcing

repository·main·Indexed 23 days ago

https://github.com/eugene-khyst/postgresql-event-sourcing

A reference implementation of an event-sourced system using PostgreSQL as an event store, built with Spring Boot. It demonstrates CQRS, snapshotting, and transactional outbox patterns using a ride-hailing domain model. The project includes a core library (postgresql-event-sourcing-core) and a sample application (event-sourcing-app), featuring optimistic concurrency control, synchronous and asynchronous event handlers, and support for both polling and PostgreSQL LISTEN/NOTIFY subscription modes.

Tokens
5.7K
Snippets
13
Records
26
Agent score
80%

What's inside postgresql-event-sourcing

  1. Understand the project structure

    main

    The project is organized into two main Gradle subprojects:

    • postgresql-event-sourcing-core: A shared library containing event sourcing and PostgreSQL-related code in the eventsourcing.postgresql package.
    • event-sourcing-app: Application-specific code implementing a simplified ride-hailing sample in the com.example.eventsourcing package.

    To use the core library in your own application, add it as a dependency in your build.gradle file:

    dependencies {
    implementation project(':postgresql-event-sourcing-core')
}

  2. What is Event Sourcing?

    main

    Event sourcing is a pattern where the state of an entity (called an aggregate) is persisted as a sequence of immutable, state-changing events in an append-only stream (called a stream).

    Instead of using SQL UPDATE or DELETE to modify current state, you only use SQL INSERT to append new events. The current state of an aggregate is reconstructed by replaying all its events in the order they occurred. This provides a complete and authentic audit trail of every change in the system.

    Key terms:

    • Aggregate: The entity being tracked.
    • Stream: The sequence of events for a specific aggregate.
    • Event Store: The database used to store these event streams (e.g., PostgreSQL).

  3. What is CQRS and how does it relate to Event Sourcing?

    main

    Command-Query Responsibility Segregation (CQRS) is an architectural pattern that separates the responsibility of handling write requests (commands) from read requests (queries).

    In an event-sourced system, CQRS is used to solve the difficulty of querying append-only event streams.

    • Write Side (Command): Uses the event store to append new events. This is the primary source of truth.
    • Read Side (Query): Uses projections (read models). Projections are denormalized views of the system state derived from the event stream, optimized for fast and complex querying (e.g., in a relational or NoSQL database).

    While CQRS can be used without event sourcing, they are frequently used together: the event store acts as the write database, and a separate database acts as the read database.

  4. Domain Events vs Integration Events

    main

    When designing event-driven systems, distinguish between these two types of events:

    • Domain Events: Internal to a specific bounded context. They represent specific changes to an aggregate and are used by event handlers to update local projections or state.
    • Integration Events: Used for communication between different bounded contexts. Unlike domain events, an integration event typically represents the current state of an aggregate rather than just a single change, making it safer for external consumers to use.

  5. Synchronous vs Asynchronous Event Handlers

    main

    Event handlers process events to perform side effects like updating projections or notifying external systems. There are two main strategies:

    Synchronous Event Handlers

    Used for updating projections (read models) within the same database as the event store. The projection update happens in the same transaction as the event append, ensuring the read model is consistent with the event stream.

    Asynchronous Event Handlers

    Used for communicating with external systems (e.g., sending a message to Kafka or making an API call). These must run after the transaction that appends the event has been committed. This approach leads to eventual consistency and prevents inconsistencies that occur if an external call succeeds but the local database transaction rolls back.

  6. Reliable Event Subscription with Transaction IDs

    main

    To avoid losing events due to PostgreSQL sequence non-rollback behavior, the system uses a combination of TRANSACTION_ID and EVENT_ID for polling.

    By using pg_current_xact_id() and checking against pg_snapshot_xmin(pg_current_snapshot()), the subscriber ensures it only processes events from transactions that are guaranteed to be committed. This prevents the 'naive' outbox problem where a later transaction commits before an earlier one, causing the subscriber to skip the earlier event.

    -- 1. Acquire lock on subscription and get last processed markers
SELECT LAST_TRANSACTION_ID::text,
       LAST_EVENT_ID
  FROM ES_EVENT_SUBSCRIPTION
 WHERE SUBSCRIPTION_NAME = :subscriptionName
   FOR UPDATE SKIP LOCKED;

-- 2. Read new 'safe' events (committed and visible)
SELECT e.ID,
       e.TRANSACTION_ID::text,
       e.EVENT_TYPE,
       e.JSON_DATA
  FROM ES_EVENT e
  JOIN ES_AGGREGATE a on a.ID = e.AGGREGATE_ID
 WHERE a.AGGREGATE_TYPE = :aggregateType
   AND (e.TRANSACTION_ID, e.ID) > (:lastProcessedTransactionId::xid8, :lastProcessedEventId)
   AND e.TRANSACTION_ID < pg_snapshot_xmin(pg_current_snapshot())
 ORDER BY e.TRANSACTION_ID ASC, e.ID ASC;

-- 3. Update subscription progress
UPDATE ES_EVENT_SUBSCRIPTION
   SET LAST_TRANSACTION_ID = :lastProcessedTransactionId::xid8,
       LAST_EVENT_ID = :lastProcessedEventId
 WHERE SUBSCRIPTION_NAME = :subscriptionName;

  7. Understand Eventual Consistency and At-Least-Once Delivery

    main

    When using asynchronous event handlers (subscriptions) with PostgreSQL, be aware of the following architectural constraints:

    • At-Least-Once Delivery: Asynchronous handlers might process the same event more than once (e.g., if the service crashes after processing but before recording progress). Consumers of integration events must be idempotent.
    • Eventual Consistency: There is a lag between the write model (the event store) and the integration events sent to message brokers. This lag is determined by the polling interval or the efficiency of the LISTEN/NOTIFY mechanism.
    • Long-Running Transactions: A long-running transaction in the database can pause event handlers because the subscription processor waits for all transactions with lower IDs to commit before processing newer ones.

  8. How Snapshotting optimizes state restoration

    main

    For entities that accumulate a large number of events (e.g., bank accounts), replaying every event to restore state becomes inefficient. Snapshotting is an optimization technique where you periodically save the aggregate's state and its current version.

    To restore an aggregate using snapshots:

    1. Read the latest available snapshot.
    2. Read all events from the stream that occurred after the version recorded in that snapshot.
    3. Replay only those remaining events to reach the current state.

  9. Restore an Aggregate to a Specific Revision

    main

    To restore an aggregate to a specific version (or the latest version), follow these two steps:

    1. Read the latest snapshot that is less than or equal to the target version.
    2. Replay events from the event stream starting from the version the snapshot represents up to the target version.

    This allows you to reconstruct the state of an aggregate at any point in time.

    -- 1. Read latest valid snapshot
SELECT a.AGGREGATE_TYPE,
       s.JSON_DATA
  FROM ES_AGGREGATE_SNAPSHOT s
  JOIN ES_AGGREGATE a ON a.ID = s.AGGREGATE_ID
 WHERE s.AGGREGATE_ID = :aggregateId
   AND (:version IS NULL OR s.VERSION <= :version)
 ORDER BY s.VERSION DESC
 LIMIT 1;

-- 2. Read forward from snapshot version to target version
SELECT ID,
       TRANSACTION_ID::text,
       EVENT_TYPE,
       JSON_DATA
  FROM ES_EVENT
 WHERE AGGREGATE_ID = :aggregateId
   AND (:fromVersion IS NULL OR VERSION > :fromVersion)
   AND (:toVersion IS NULL OR VERSION <= :toVersion)
 ORDER BY VERSION ASC;

  11. Run the event-sourcing sample

    main

    Follow these steps to build and run the full stack (PostgreSQL, Kafka, and the application) using Docker and Gradle:

    1. Install Prerequisites:

    2. Build the project:

      ./gradlew clean build jibDockerBuild -i

    3. Start the infrastructure:

      docker compose --env-file gradle.properties up -d --scale event-sourcing-app=2

      Wait a few minutes for services to initialize.

    4. Monitor logs:

      docker compose logs -f event-sourcing-app

    5. Run E2E tests (optional):

      E2E_TESTING=true ./gradlew clean test -i

    6. Database Access: Explore the database using Adminer at http://localhost:8181. Credentials can be found in the docker-compose.yml file.

    ./gradlew clean build jibDockerBuild -i

docker compose --env-file gradle.properties up -d --scale event-sourcing-app=2

docker compose logs -f event-sourcing-app

E2E_TESTING=true ./gradlew clean test -i

  12. Implement Optimistic Concurrency Control for Aggregates

    main

    To prevent lost updates when appending events, use optimistic concurrency control by checking the aggregate version in the ES_AGGREGATE table before inserting the new event. This must be performed within a single transaction using two SQL statements:

    1. Update the aggregate version only if the current version matches the expected version.
    2. Insert the new event into the ES_EVENT table.

    Use pg_current_xact_id() to capture the transaction ID for reliable event tracking.