Sourced

Event sourcing for Elixir using the Dynamic Context Boundary (DCB) model — a single flat event stream, events tagged with the domain concepts they concern, and consistency enforced by optimistic locking scoped to whatever slice of the stream a decision actually read.

This package defines the store API, the tag/query model, projections and decision models, the middleware pipeline, and an in-memory adapter. It depends only on :telemetry. For persistence, add an adapter — e.g. https://hex.pm/packages/sourced_postgres.

Installation

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

Getting started

Define your domain events. use Sourced.Middleware.Domain.Event gives you a stored type derived from the module name and a decoder for reading them back; to_tags/1 is what makes an event reachable by anything other than its type.

defmodule OrderPlaced do
  use Sourced.Middleware.Domain.Event

  @derive JSON.Encoder
  defstruct [:id, :customer_id]

  @impl Sourced.Middleware.Domain.Event
  def to_tags(event), do: ["order:#{event.id}", "customer:#{event.customer_id}"]
end

Build a store, naming its adapter and any middleware. The first middleware entry is the outermost, so telemetry placed above the domain middleware reports decoded events:

store =
  Sourced.EventStore.new(
    adapter: Sourced.EventStore.InMemory,
    middleware: [
      Sourced.Middleware.Telemetry,
      {Sourced.Middleware.Domain, [OrderPlaced, OrderShipped]}
    ]
  )

The store is a plain value: keep it in a module attribute, in your application state, or wherever suits. Add it to your supervision tree, then append, query, and subscribe by passing it in:

children = [{Sourced.EventStore, store}]

{:ok, 1} = Sourced.EventStore.append(store, [%OrderPlaced{id: 1, customer_id: 42}])

{:ok, %{events: events, last_sequence: 1}} =
  Sourced.EventStore.query(store, query: [%{tags: ["customer:42"]}])

{:ok, %Sourced.EventStore.Subscription{ref: ref}} =
  Sourced.EventStore.subscribe(store, query: [%{types: [OrderPlaced]}])

receive do
  {:sourced_events, ^ref, events} -> handle(events)
end

Swapping Sourced.EventStore.InMemory for a persistent adapter is the only change needed to run the same code against a real database.

What's included

The store.Sourced.EventStore.new/1 returns a %Sourced.EventStore{} — an adapter, its :config, and an initialized middleware pipeline — that you pass to append/3, query/2, subscribe/2, and unsubscribe/2. The :config is what the adapter is started with, including the :name it registers under, which defaults to the adapter module. The store itself is deliberately thin — it builds a Sourced.Operation and hands it to the middleware pipeline, whose innermost terminal calls the adapter.

Events. An append takes plain maps — a type, its data, when it occurred_at, and optionally tags and metadata, as described by t:Sourced.EventStore.Behaviour.event/0 — or domain structs under the domain middleware, and returns {:ok, last_sequence}. Reads return Sourced.StoredEvent structs, carrying the store-assigned sequence. Metadata keys are strings so they round-trip through JSON-backed stores. Tags are plain strings, conventionally "prefix:id"; keep personal data out of them.

Queries. A query is a plain list of items — maps with an optional :types and an optional :tags — following the DCB specification: an event matches if it matches any item (OR); within an item its type must be one of types (OR) and its tags must include all of tags (AND). An empty query matches everything. :from, :to, and :limit bound the result as a whole. Sourced.EventStore.Query.apply/2 is the reference in-memory implementation that adapters loading events into memory can reuse.

Consistency.append/3 takes :expected_sequence and an optional :query scoping the check to matching events — compared against the last matching event's sequence, or 0 when none match. A mismatch returns {:error, %Sourced.EventStore.OptimisticConcurrencyError{}}. Passing a query without an expected sequence raises, since a query alone on append means nothing.

Projections and decision models.Sourced.Projection is a fold together with the query it folds over: an :initial_state, :handlers keyed by event type — those keys are the query's types — and the :tags narrowing it to the domain concepts it is about. compose/1 combines projections into one that folds to a map keyed the same way and queries the union, feeding each part only the events its own query matches, and returns a projection itself, so composites nest. Sourced.DecisionModel.build/2 runs a projection, or a map of them, against a store: it reads the slice, folds it, and keeps the sequence that slice was at, which append/3 turns into the :query and :expected_sequence for the append that follows. refresh/2 folds a stale model forward over just the events that beat it, so a lost append costs a bounded read rather than another full one. The same projection also drives a long-lived read model — subscribe with its query and fold each batch into the carried state with project/3.

Deciders.Sourced.Decider.decide/4 closes the read-decide-append loop: it builds the model, hands the folded state to a function returning {:ok, events}, :noop or {:error, reason}, and appends. A losing append is retried against a model refreshed past whatever beat it — bounded :attempts, and no backoff. The decision function must be pure, since it runs once per attempt, and decide/4 cannot be called from inside a transaction you opened yourself.

Subscriptions.subscribe/2 returns a Sourced.EventStore.Subscription handle and delivers the events already stored — exactly what query/2 would have returned — as the first {:sourced_events, ref, events} message, keyed by that handle's ref. Every later matching event arrives the same way, batched per append and in ascending sequence order. Adapters must take that snapshot and register the subscription atomically, so no event falls between the two: it is either in the first batch or in a later one, never both and never neither. Sourced.EventStore.Subscription owns matching, middleware, monitoring, and delivery, so an adapter only decides when to offer a batch.

Middleware. A Plug-style pipeline (Sourced.Middleware) is how the store stays agnostic of domain and cross-cutting concerns. Each entry's options run through c:Sourced.Middleware.init/1 when the store is built, so validation and precomputation happen once rather than per operation. Two are shipped:

• Sourced.Middleware.Domain — wraps domain structs into event maps on the way in (deriving type and tags), decodes data back into structs on the way out, and resolves event modules used as query criteria. Unregistered types are logged and left raw rather than raising, so retired event types never break a query.
• Sourced.Middleware.Telemetry — one :telemetry.span/3 per operation under [:sourced, :event_store, :append | :query | :subscribe | :notify], with the operation's result as :outcome.

Writing your own means implementing call/3: transform the operation on the way in, call Sourced.Middleware.next/2, and optionally rewrite the result on the way out.

Adapters.Sourced.EventStore.Behaviour is the contract, and Sourced.EventStore.InMemory is a GenServer implementation of it for dev and test (it keeps data as-is, so domain structs come back without a decode round trip).

contract/ holds a shared, use-able suite that drives every assertion end-to-end through the store API:

defmodule MyApp.EventStore.SomeAdapterTest do
  use ExUnit.Case, async: true

  use Sourced.EventStore.Contract,
    adapter: MyApp.EventStore.SomeAdapter,
    config: [some: :opt]

  setup do
    start_supervised!({Sourced.EventStore, @store})
    start_supervised!({Sourced.EventStore, @domain_store})
    :ok
  end
end

It is compiled only in :test, so it never reaches the published package; adapters in other projects pull it in through their own elixirc_paths. New adapters should opt into it rather than hand-rolling tests, and changes to the store contract belong there so every adapter is held to them.