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.3"}
  ]
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

  def up, do: Sourced.EventStore.Postgres.Migrations.up(version: 2)
  def down, do: Sourced.EventStore.Postgres.Migrations.down(version: 1)
end

The schema is versioned, and up/1 runs every step between the version the database is at and the one asked for — the latest by default. When a release adds a version, upgrade an existing install with a second migration pinned to it. down(version: 2) reverts version 2 and anything above it, leaving the database at version 1:

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

  def up, do: Sourced.EventStore.Postgres.Migrations.up(version: 2)
  def down, do: Sourced.EventStore.Postgres.Migrations.down(version: 2)
end

Version	Adds
1	`sourced_events`, `sourced_event_tags`, their indexes and triggers, and the read watermark functions — every 0.1.x and 0.2.x release
2	`sourced_checkpoints`, for durable subscribers

The version is recorded as a comment on sourced_events. A database migrated by a release before versioning has no comment and is read as version 1.

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.

sourced_checkpoints holds the positions of durable subscribers (see below), one row per name:

name        TEXT        PRIMARY KEY
sequence    BIGINT      NOT NULL
fingerprint TEXT        NOT NULL
inserted_at TIMESTAMP   NOT NULL
updated_at  TIMESTAMP   NOT NULL

The two timestamps are timestamp without time zone, written in UTC.

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.

Durable subscribers

Sourced.EventStore.Postgres.Subscriber is the resumable version: a GenServer behaviour that keeps the last sequence it processed in a sourced_checkpoints row and subscribes from it on start. handle_events/2 runs inside a transaction on the store's repo and the checkpoint commits in that same transaction, so a read model written through the repo can never disagree with its position — a crash before the commit rolls both back and the next start redelivers the batch. Nothing reaches a committed handler twice, so it need not be idempotent.

defmodule MyApp.OrderProjector do
  use Sourced.EventStore.Postgres.Subscriber

  @impl true
  def handle_events(events, state) do
    Enum.each(events, &MyApp.Orders.project/1)
    {:ok, state}
  end
end

children = [
  MyApp.Repo,
  {Sourced.EventStore, store},
  {MyApp.OrderProjector, store: store, name: "orders", query: [%{tags: ["order"]}]}
]

Handlers that do not write to the repo

A handler whose work is not in the database — an email, a call to another service — gains nothing from the transaction: an HTTP call cannot be rolled back, and the pooled connection sits checked out for however long the call takes. The transaction?/2 callback decides, per batch, whether handle_events/2 runs inside the transaction that commits the checkpoint. It defaults to true; override it to run outside one:

defmodule MyApp.EmailNotifier do
  use Sourced.EventStore.Postgres.Subscriber

  @impl true
  def transaction?(_events, _state), do: false

  @impl true
  def handle_events(events, state) do
    Enum.each(events, &MyApp.Mailer.deliver_for/1)
    {:ok, state}
  end
end

Outside a transaction the checkpoint is advanced in one short statement of its own, only after the handler has returned {:ok, _}. That is at-least-once delivery — a crash between the handler finishing and the checkpoint committing redelivers the batch on the next start — so such a handler must be idempotent. It receives the events about to be handled, so a subscriber can also decide by what is in them. Only return false for handlers that do not write through the store's repo; one that does would silently lose the atomicity described above.

A new name starts from the beginning of the store by default; start_from: :latest starts it at the current read watermark instead, and start_from: sequence at that event. Either applies only the first time the name is seen — once a position exists it is resumed from regardless.

The name carries two guarantees. A fingerprint of the query is stored with the position, and starting the same name with a different query raises rather than resuming against a slice it was never computed over. And two processes under one name are detected rather than tolerated: each moves the checkpoint only from the position it last committed, so the second to commit a batch stops instead of doubling up.

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.