Behaviour defining the contract for a Dynamic Context Boundary (DCB) event store.
Implementations are process-based and can be started under a supervisor. Call
functions on the implementing module, passing the Sourced.EventStore as the
first argument — its :name is the name the implementation registered its
process under.
Summary
Types
A new event to append: a plain map carrying a type, its data, and when it
occurred_at, optionally alongside tags and metadata.
Options for a read.
The result of a query response.
A configured store, as built by Sourced.EventStore.new/1.
Callbacks
Appends one or more events atomically to the store.
Returns a supervisor child specification for the event store.
Loads all matching events into memory.
Subscribes to events matching a query, both stored and future.
Cancels a subscription, stopping further delivery.
Types
@type append_error() :: Sourced.EventStore.OptimisticConcurrencyError.t()
@type append_opts() :: [ expected_sequence: non_neg_integer(), query: Sourced.EventStore.Query.t() ]
@type event() :: %{ :type => String.t(), :data => Sourced.StoredEvent.data(), :occurred_at => DateTime.t(), optional(:tags) => [Sourced.StoredEvent.tag()], optional(:metadata) => Sourced.StoredEvent.metadata() }
A new event to append: a plain map carrying a type, its data, and when it
occurred_at, optionally alongside tags and metadata.
%{
type: "OrderPlaced",
data: %{"order_id" => 1, "customer_id" => 42},
tags: ["order:1", "customer:42"],
occurred_at: DateTime.utc_now()
}Omitted tags and metadata default to [] and %{} before an adapter sees
the event. The store assigns a sequence, and reads return
Sourced.StoredEvent structs.
With a Sourced.Middleware.Domain configured, domain event structs can be
appended directly and are converted into this shape for you.
@type query_opts() :: [ query: Sourced.EventStore.Query.t(), from: non_neg_integer(), to: non_neg_integer(), limit: pos_integer() ]
Options for a read.
:query is the matcher; :from, :to and :limit bound the result of this
particular read and are deliberately not part of Sourced.EventStore.Query,
so a matcher reused for an append or a subscription cannot carry them along.
@type query_result() :: %{ events: [Sourced.StoredEvent.t()], last_sequence: non_neg_integer() }
The result of a query response.
:last_sequence is the sequence of the last event in the returned batch, or
0 when nothing matched — so a bounded read reports the last sequence it
saw, not the last matching sequence in the stream.
@type store() :: Sourced.EventStore.t()
A configured store, as built by Sourced.EventStore.new/1.
Callbacks
@callback append(store(), [event()], append_opts()) :: {:ok, non_neg_integer()} | {:error, append_error() | term()}
Appends one or more events atomically to the store.
Returns {:ok, last_sequence} — the sequence assigned to the last of
new_events. The events themselves are not returned; read them back with
query/2 if you need them.
new_events is never empty: Sourced.EventStore.append/3 raises before an
adapter is reached, so adapters need no clause for it. This follows the DCB
specification, whose Events argument MUST not be empty.
Options
:expected_sequence— the sequence the caller expects the matched slice to be at. Returnsappend_error()if anything matching has arrived after it, enabling optimistic concurrency control. Omit to append unconditionally.:query— when combined with:expected_sequence, scopes the concurrency check to events matching the query. Without one the check covers the whole stream.
The check is strictly-greater, not equality: an expected_sequence ahead of
the last matching event asserts nothing the store can contradict, and appends.
Only a sequence the store has already moved past is a conflict.
@callback child_spec(opts :: keyword()) :: Supervisor.child_spec()
Returns a supervisor child specification for the event store.
Receives the store's :config, with the store's :name added to it.
@callback query(store(), query_opts()) :: {:ok, query_result()} | {:error, term()}
Loads all matching events into memory.
Events are returned in ascending sequence order.
Options
:query— an optionalQuerydescribing the desired subset of the event stream.:from— only events at or after this sequence (inclusive).:to— only events at or before this sequence (inclusive).:limit— at most this many events, taken from the start of the match.
@callback subscribe(store(), Sourced.EventStore.Subscription.t()) :: {:ok, Sourced.EventStore.Subscription.t()} | {:error, term()}
Subscribes to events matching a query, both stored and future.
Adapters should use Subscription.deliver/2 to deliver events to the subscriber
in the correct format. Events must be delivered in ascending sequence order.
Note that the filtering of events must be handled on the adapter level prior
to calling deliver/2.
If the adapter's subscription mechanism only supports reading live events, the adapter is responsible for delivering all matching events currently in the store.
Subscriptions should be cancelled automatically when the subscriber exits. Use
Process.monitor/1 to monitor for :DOWN messages from the subscriber pid.
Returns the subscription with :owner set to the process that delivers on it,
which is what a subscriber monitors to learn the subscription has ended. It
must be a process whose death means no further delivery: the per-subscription
process where there is one, otherwise whichever process holds the
subscription. Adapters must not report a process that outlives delivery.
@callback unsubscribe(store(), Sourced.EventStore.Subscription.t()) :: :ok
Cancels a subscription, stopping further delivery.
Idempotent: unsubscribing an unknown or already-cancelled subscription is
:ok.
Subscriptions are also cancelled automatically when the subscriber exits.