sourced_postgres

# Sourced.Postgres

A PostgreSQL adapter for [Sourced](https://hex.pm/packages/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

```elixir
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`:

```elixir
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:

```elixir
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:

```console
$ mix ecto.gen.migration add_sourced_events
```

```elixir
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:

```elixir
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:

1. 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:

   ```elixir
   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)
   ```

2. Keep it short. No external calls or heavy computation: readers cannot advance past the sequence this transaction holds until it commits.

3. 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:

  ```elixir
  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`:

```console
$ 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.