A PostgreSQL implementation of Sourced.EventStore backed by your application's Ecto repo.

Setup

Appends and reads have no connection pool of their own: they run through a repo you already run. Point the store at one with :repo:

store =
  Sourced.EventStore.new(
    adapter: Sourced.EventStore.Postgres,
    config: [repo: MyApp.Repo]
  )

Then add it to your supervision tree, after the repo it runs on:

children = [
  MyApp.Repo,
  {Sourced.EventStore, store}
]

append/3 and query/2 resolve the repo by looking the store up under its name, so both raise until it is started. What the store supervises is the one connection it does open for itself: a Postgrex.Notifications listener, which subscribe/2 is delivered on.

Generate a migration to add the tables, indexes, and triggers, along with the functions the read watermark described under "Concurrency" is computed from:

defmodule MyApp.Repo.Migrations.AddSourcedEvents do
  use Ecto.Migration

  defdelegate up, to: Sourced.EventStore.Postgres.Migrations
  defdelegate down, to: Sourced.EventStore.Postgres.Migrations
end

Ecto.Adapters.SQL.Sandbox.checkout(MyApp.Repo, isolation: "SERIALIZABLE")

config :my_app, MyApp.Repo,
  after_connect: {Postgrex, :query!, ["SET default_transaction_isolation = 'serializable'", []]}

JSON

Event data and metadata are encoded to jsonb by Postgrex, using whichever library is configured as its :json_library. Postgrex defaults to Jason. To use the JSON module that ships with Elixir 1.18 and later instead:

config :postgrex, :json_library, JSON

Postgrex reads that setting while it compiles, so a change to it only takes effect after mix deps.compile postgrex --force.

When appending domain event structs (see Sourced.Middleware.Domain), each struct must implement the configured library's encoder protocol. Reads return the data as a decoded map; converting it back into the domain struct is handled by the store's domain, if configured.

Concurrency

All appends are executed under the Serializable Isolation Level. This ensures that if a concurrent transaction appends events that would change the result of the query that verifies the append condition, then one or more of the concurrent transactions would be aborted with a 40001 :serialization_failure. In the case of a conditional append, this error is converted into a Sourced.EventStore.OptimisticConcurrencyError. It's also possible for an unconditional append to fail with the same error, in which case it is raised to the caller as a Postgrex error that can be retried.

Event sequences are generated as an auto-incrementing identity column, which can create gaps in the case of rollbacks. With multiple concurrent writers, there are also no guarantees that the events will be committed in increasing sequence order. The store doesn't prevent out of sequence appends, but to ensure that readers and subscribers receive events in sequential order, it maintains a read watermark: the highest sequence that no append still in flight can commit beneath. Every appender publishes the last sequence that existed when it started, and the watermark is the lowest of those publications; with nothing in flight it is the last sequence the identity column drew. Reads are bounded by it. So even if events with later sequences have already been committed, readers will not read them until all previous events in the sequence have been committed first. The one exception to this is that a transaction can always read the events it appended itself.

Appending inside your own transaction

Sharing the repo allows an append to be committed along with other writes, so a read model can be updated atomically with the event that feeds it. However, there are several considerations to ensure that this works correctly and safely:

1. The top-level transaction must be started under the SERIALIZABLE isolation level, otherwise all appends will raise because the safety of concurrent appends can no longer be guaranteed:

Repo.transact(fn ->
  # 1. The transaction must be SERIALIZABLE, and this must be its first statement.
  Repo.query!("SET TRANSACTION ISOLATION LEVEL SERIALIZABLE")

  {:ok, sequence} = EventStore.append(store, [%OrderPlaced{id: 123}])
  Repo.insert!(%Order{id: 123, sequence: sequence})
end)

2. The transaction should be short. Avoid external calls or any other heavy computation that would delay when the transaction gets committed. Readers won't be able to read any events committed by any other appends until this transaction commits and allows them to read all events in the correct sequence.

3. A serialization failure aborts the entire transaction, even if the failure was not due to a concurrency error on the store. Keep the transaction body safe to re-run in case a serialization failure needs to be retried.

Subscriptions

subscribe/2 is backed by Postgres' LISTEN/NOTIFY. The migration installs a statement-level AFTER INSERT trigger that announces the highest sequence an append inserted; each store keeps one connection listening on the channel.

Subscribing starts with a catch-up read from the subscription's :from, so events already in the store are delivered before any live one is, and a subscriber that tracks its position can resume where it left off rather than only seeing what arrives next.

From there a subscription owns the cursor it reads from — it re-queries the store from the last sequence it was told about, rather than being handed the events themselves. The announced sequence is only an upper bound: a subscription clamps its read to it, skips the read entirely when the sequence is one it has already passed, and advances its cursor past events it does not match without having to query for them.

It clamps to a second bound as well: the read watermark, the highest sequence no in-flight append can still commit beneath. An announced sequence is the highest one some append inserted, which says nothing about whether a different append is still in flight holding a lower one, so a cursor advanced to the announcement alone could skip that lower sequence for good once it commits. The watermark is what stops that, and it is why a subscription can lag behind an announcement it has already received.

Because a sequence can be announced before the watermark covers it, and NOTIFY is edge-triggered, a subscription also polls: an announcement it cannot yet act on would otherwise have no second edge to be delivered on. Polling only runs while there is an announcement outstanding, so it is bounded by how long an append takes rather than running continuously.

The trade-offs worth knowing:

• Notifications are delivered on commit, so a subscriber never sees an event that was rolled back, and never sees one out of sequence order.

• Delivery is at-most-once and not durable. Events appended while the listening connection is down are not replayed when it reconnects. A subscriber that must not miss anything should track the sequence it has seen and re-subscribe from there. Nothing is ever delivered twice, though: the cursor only moves forward, so a delivery is never repeated and a retry cannot duplicate one.

• No live event is delivered under Ecto.Adapters.SQL.Sandbox. The sandbox wraps each test in a transaction it rolls back, and a transaction that never commits never fires its NOTIFY. The catch-up read at subscribe time still returns what the test has already appended, and appends and queries work fine there; it is live delivery that needs a test committing for real.