# `Sourced.EventStore`
[🔗](https://gitlab.com/dimakula/sourced/-/blob/0.2.0/lib/sourced/event_store.ex#L1)

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 `c: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.StoredEvent`s 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 `c:Sourced.Middleware.Domain.Event.to_type/0`, its `tags`
from `c: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.

# `opts`

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

# `stream_opts`

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

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

# `append`

```elixir
@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 `c:Sourced.EventStore.Behaviour.append/3` for the options.

# `child_spec`

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

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

Builds a store.

## Options

  * `:adapter` — *required* 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 `c:Sourced.Middleware.init/1` here.

# `query`

```elixir
@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 `c:Sourced.EventStore.Behaviour.query/2` for the options.

# `stream`

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

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

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

Cancels `subscription`, stopping further delivery.

---

*Consult [api-reference.md](api-reference.md) for complete listing*