An event store: an adapter, the options it runs with, and a middleware pipeline.
A store is a plain value built with new/1 and passed explicitly to every
call:
store =
Sourced.EventStore.new(
adapter: Sourced.EventStore.Postgres,
config: [repo: MyApp.Repo]
):config is handed to the adapter when it starts, and :middleware builds the
pipeline every operation runs through. Adapters name the process they register
after themselves, so running more than one store on the same adapter means
giving each a :name in its config.
Add the store to your supervision tree:
children = [{Sourced.EventStore, store}]Then pass it to the store functions:
Sourced.EventStore.append(store, events, opts)
Sourced.EventStore.query(store, opts)
Sourced.EventStore.stream(store, opts)Building a store runs each middleware's Sourced.Middleware.init/1, so it is
worth doing once rather than per call. What comes back is plain data — keep it
wherever suits the application. A module attribute moves the work to compile
time:
defmodule MyApp.EventStore do
@store Sourced.EventStore.new(
adapter: Sourced.EventStore.Postgres,
config: [repo: MyApp.Repo]
)
def store, do: @store
endMiddleware
The store itself is agnostic about how events are shaped — its only job is to
hand event maps to the adapter and return Sourced.StoredEvents from
reads.
Everything else is a Sourced.Middleware: a composable transform
wrapped around the adapter call. Pass a :middleware list to layer them on.
The first entry is the outermost wrapper.
The most common middleware is Sourced.Middleware.Domain, which
lets you append and query domain event structs directly instead of hand-built
event maps. It takes the list of event modules it should know about:
store =
Sourced.EventStore.new(
adapter: Sourced.EventStore.Postgres,
config: [repo: MyApp.Repo],
middleware: [
{Sourced.Middleware.Domain, [OrderPlaced, OrderShipped]}
]
)
{:ok, _} = Sourced.EventStore.append(store, [%OrderPlaced{id: 123}])
{:ok, %{events: [%Sourced.StoredEvent{data: %OrderPlaced{id: 123}}]}} =
Sourced.EventStore.query(store, query: [%{types: [OrderPlaced]}])A bare struct is wrapped in an event map with its type taken from
the event module's Sourced.Middleware.Domain.Event.to_type/0, its tags
from Sourced.Middleware.Domain.Event.to_tags/1, occurred_at set to the
current time, and empty metadata. To control those fields, write the event map
yourself — structs and event maps can be mixed in one append:
Sourced.EventStore.append(store, [
%OrderPlaced{id: 123},
%{
type: OrderShipped.to_type(),
data: %OrderShipped{id: 123},
tags: ["order:123"],
metadata: %{"user" => "alice"},
occurred_at: shipped_at
}
])On reads, the domain middleware decodes each stored event's data back into
its domain struct (serializing adapters return plain maps; non-serializing
adapters return the struct untouched). Events whose type is not registered in
the domain keep their raw data and log a warning, so old event types never
fail a query.
Serializing adapters also need domain events to be JSON-encodable; see
Sourced.Middleware.Domain.Event and the adapter docs.
Summary
Types
Options for a stream. The same matcher and bounds a read takes, plus the size of the batches it is read in.
A configured event store.
Functions
Appends new_events to store.
Returns the child specification for starting store under a supervisor.
Builds a store.
Reads the events of store matching the query.
Lazily reads the events of store matching the query.
Subscribes to the events of store matching the query.
Cancels subscription, stopping further delivery.
Types
@type opts() :: [ adapter: module(), middleware: [Sourced.Middleware.entry()], config: keyword() ]
@type stream_opts() :: [ query: Sourced.EventStore.Query.t(), from: non_neg_integer(), to: non_neg_integer(), batch_size: pos_integer() ]
Options for a stream. The same matcher and bounds a read takes, plus the size of the batches it is read in.
@type t() :: %Sourced.EventStore{ adapter: module(), config: keyword(), pipeline: Sourced.Middleware.pipeline() }
A configured event store.
:config are the options the adapter is started with, and :pipeline is the
initialized middleware every operation runs through.
Functions
@spec append( t(), [Sourced.EventStore.Behaviour.event() | struct()], Sourced.EventStore.Behaviour.append_opts() ) :: {:ok, non_neg_integer()} | {:error, Sourced.EventStore.Behaviour.append_error() | term()}
Appends new_events to store.
Raises ArgumentError if new_events is empty.
See Sourced.EventStore.Behaviour.append/3 for the options.
@spec child_spec(t()) :: Supervisor.child_spec()
Returns the child specification for starting store under a supervisor.
The store's :config is passed to the adapter.
Builds a store.
Options
:adapter— required theSourced.EventStore.Behaviourimplementation.:config— options handed to the adapter when it starts, such as the repo or connection details it needs, and the:nameit registers under.:middleware— the pipeline entries, outermost first. Each entry's options run throughSourced.Middleware.init/1here.
@spec query(t(), Sourced.EventStore.Behaviour.query_opts()) :: {:ok, Sourced.EventStore.Behaviour.query_result()} | {:error, term()}
Reads the events of store matching the query.
See Sourced.EventStore.Behaviour.query/2 for the options.
@spec stream(t(), stream_opts()) :: Enumerable.t()
Lazily reads the events of store matching the query.
Reads a batch at a time through query/2, advancing a cursor past the last
event of each batch, so a read model can be rebuilt over more events than fit
in memory:
store
|> Sourced.EventStore.stream(query: projection.query)
|> Enum.reduce(projection.initial_state, &Sourced.Projection.apply(projection, &2, &1))Options
:query— the matcher, asquery/2takes it.:from— the sequence to start at, inclusive.:to— the sequence to stop at, inclusive.:batch_size— how many events to read per batch. Defaults to500.
There is deliberately no :limit. The stream is lazy, so Stream.take/2
bounds the result without reading batches it does not need, and a :limit
would only collide with the one paging uses.
The stream ends where the events do rather than waiting for more, so it is a
catch-up read and not a tailing one — subscribe/2 is what tails. The two
compose into the usual rebuild, and the subscription's own catch-up read
covers whatever was appended between the last batch and the subscribe:
{state, last_sequence} =
store
|> Sourced.EventStore.stream(query: query)
|> Enum.reduce({initial_state, 0}, fn event, {state, _sequence} ->
{fold(state, event), event.sequence}
end)
{:ok, subscription} =
Sourced.EventStore.subscribe(store, query: query, from: last_sequence + 1)Unlike query/2 this raises rather than returning {:error, reason}: a lazy
stream has nowhere to put an error tuple where a caller would reliably see it.
@spec subscribe(t(), Sourced.EventStore.Subscription.opts()) :: {:ok, Sourced.EventStore.Subscription.t()} | {:error, term()}
Subscribes to the events of store matching the query.
Events arrive as {:sourced_events, subscription.ref, events}. Delivery is
at-most-once and the subscription is not durable, so a subscriber that must
not miss events monitors subscription.owner and resubscribes from its own
cursor when that process goes down — see Sourced.EventStore.Subscription.
See Sourced.EventStore.Subscription for the options.
@spec unsubscribe(t(), Sourced.EventStore.Subscription.t()) :: :ok
Cancels subscription, stopping further delivery.