Sourced.Postgres

A PostgreSQL adapter for Sourced, built on Ecto and Postgrex.

Events live in a single table keyed by a BIGINT identity column that is global and monotonic, so a reader can always resume from the last sequence it saw. Appends run at SERIALIZABLE isolation on your application's own repo, so an append can commit alongside the writes it feeds. Subscriptions are backed by LISTEN/NOTIFY, so a subscriber hears about appends from every node, not just its own.

Installation

def deps do
  [
    {:sourced_postgres, "~> 0.1"}
  ]
end

Setup

The adapter has no connection pool of its own — it reads and writes 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],
    middleware: [{Sourced.Middleware.Domain, [OrderPlaced, OrderShipped]}]
  )

Add the store to your supervision tree after the repo it runs on. append/3 and query/2 resolve the repo by looking the store up under its name, so both raise until it is started:

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

Besides :repo (required), the config takes :name — the name the store's processes register under, defaulting to the adapter module — for running more than one store on this adapter.

Migration

Generate a migration in your application and delegate to Sourced.EventStore.Postgres.Migrations, which creates the tables, indexes, and triggers, along with the functions the read watermark is computed from:

$ mix ecto.gen.migration add_sourced_events

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

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

Tables

sourced_events holds the events themselves:

sequence    BIGINT      PRIMARY KEY GENERATED BY DEFAULT AS IDENTITY
type        TEXT        NOT NULL
data        JSONB       NOT NULL
tags        TEXT[]      NOT NULL
metadata    JSONB       NOT NULL
occurred_at TIMESTAMPTZ NOT NULL

sourced_event_tags is what tag queries match against — one row per tag an event carries, kept in step with the tags column by a trigger and never written directly:

tag         TEXT        NOT NULL
sequence    BIGINT      NOT NULL  REFERENCES sourced_events (sequence)
PRIMARY KEY (tag, sequence)

Plus a btree index on sourced_events.type and one on sourced_event_tags.sequence. The side table is preferred over a GIN index on the array because it performs better for the containment match that query items compile to. Both tables are created with autovacuum tuned for an append-only workload: routine VACUUM all but disabled, ANALYZE run more often than the defaults would.

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 only takes effect after mix deps.compile postgrex --force.

Domain event structs must therefore implement the configured library's encoder protocol — @derive JSON.Encoder on each event module. Reads return data as a decoded map; turning it back into the domain struct is the job of Sourced.Middleware.Domain, via each event's from_map/1.

Concurrency

Appends run at the SERIALIZABLE isolation level. If a concurrent transaction appends events that would change the result of the query verifying the append condition, one side is aborted with a 40001:serialization_failure. For a conditional append that is translated into a Sourced.EventStore.OptimisticConcurrencyError; an unconditional append can hit the same error, and it is raised to the caller as a Postgrex.Error to retry.

Sequences come from an identity column, so rollbacks leave gaps and concurrent writers can commit out of sequence order. Reads are therefore bounded by a read watermark: the highest sequence no in-flight append can still commit beneath. Every appender publishes the last sequence that existed when it started (as a shared advisory lock, released on commit, abort, or connection death), and the watermark is the lowest of those publications — or, with nothing in flight, the last sequence the identity column drew. Events with higher sequences may already be committed, but readers do not see them until everything below them has settled.

Appending inside your own transaction

Sharing the repo means an append can be committed together with other writes, so a read model can be updated atomically with the event that feeds it. Three things to know:

The transaction must be opened SERIALIZABLE, and that must be its first statement — otherwise the append raises, since the safety of concurrent appends can no longer be guaranteed:

Repo.transact(fn ->
  Repo.query!("SET TRANSACTION ISOLATION LEVEL SERIALIZABLE")

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

Keep it short. No external calls or heavy computation: readers cannot advance past the sequence this transaction holds until it commits.
A serialization failure aborts the whole transaction, including writes made before the append. Keep the body safe to re-run.

Subscriptions

The migration installs a statement-level AFTER INSERT trigger that announces the highest sequence an append inserted on a channel; each store keeps one connection listening on it.

Subscribing starts with a catch-up read from the subscription's :from, so events already stored are delivered before any live one, and a subscriber that tracks its position can resume where it left off. From there the subscription owns its cursor: it re-queries the store rather than being handed events in the payload. The announced sequence is only an upper bound — the subscription clamps its read to it, skips the read entirely when it has already passed that sequence, and advances past events it does not match without querying for them. It clamps to the read watermark as well, which is why a subscription can lag behind an announcement it has already received. Because NOTIFY is edge-triggered, it also polls while an announcement is outstanding, bounded by how long an append takes.

Worth knowing:

Notifications fire 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 on reconnect. A subscriber that must not miss anything should track the sequence it has seen and resubscribe from there — which is what subscribe/2's :from option is for. Nothing is ever delivered twice, though: the cursor only moves forward.

Testing with `Ecto.Adapters.SQL.Sandbox`

Two things bite under the sandbox:

Every append asserts it is running SERIALIZABLE, and cannot raise the level inside a transaction the sandbox already opened. Check the connection out at that level — Ecto.Adapters.SQL.Sandbox.checkout(MyApp.Repo, isolation: "SERIALIZABLE") — or set it once for every sandbox transaction in the test repo config:
```
config :my_app, MyApp.Repo,
  after_connect: {Postgrex, :query!, ["SET default_transaction_isolation = 'serializable'", []]}
```
No live event is delivered, at any isolation level: the sandbox rolls its transaction back, and a transaction that never commits never fires its NOTIFY. Appends, queries, and the catch-up read at subscribe time all work; testing live delivery needs a connection that commits for real.

Development

The suite expects a reachable Postgres with a postgres/postgres user on localhost:

$ docker compose up -d
$ mix do ecto.create + ecto.migrate + test

The test repo is deliberately not sandboxed, for the reason above; isolation comes from truncating between tests instead. Tests are async: true; keep new test modules async. Most of the coverage comes from the shared adapter contract suite in core/contract/, which this project compiles into its test build — behaviour changes belong there rather than here.