Sourced.EventStore (sourced v0.2.0)

Copy Markdown View Source

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
end

Middleware

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.

t()

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

opts()

@type opts() :: [
  adapter: module(),
  middleware: [Sourced.Middleware.entry()],
  config: keyword()
]

stream_opts()

@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.

t()

@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

append(store, new_events, opts \\ [])

Appends new_events to store.

Raises ArgumentError if new_events is empty.

See Sourced.EventStore.Behaviour.append/3 for the options.

child_spec(event_store)

@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.

new(opts)

@spec new(opts()) :: t()

Builds a store.

Options

  • :adapterrequired the Sourced.EventStore.Behaviour implementation.
  • :config — options handed to the adapter when it starts, such as the repo or connection details it needs, and the :name it registers under.
  • :middleware — the pipeline entries, outermost first. Each entry's options run through Sourced.Middleware.init/1 here.

query(store, opts \\ [])

Reads the events of store matching the query.

See Sourced.EventStore.Behaviour.query/2 for the options.

stream(store, opts \\ [])

@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, as query/2 takes 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 to 500.

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.

subscribe(store, opts \\ [])

@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.

unsubscribe(store, subscription)

@spec unsubscribe(t(), Sourced.EventStore.Subscription.t()) :: :ok

Cancels subscription, stopping further delivery.