defmodule Commanded.Projections.Ecto do
@moduledoc """
Read model projections for Commanded using Ecto.
## Example usage
defmodule Projector do
use Commanded.Projections.Ecto,
application: MyApp.Application,
name: "my-projection",
repo: MyApp.Repo,
schema_prefix: "my-prefix",
timeout: :infinity
project %Event{}, _metadata, fn multi ->
Ecto.Multi.insert(multi, :my_projection, %MyProjection{...})
end
project %AnotherEvent{}, fn multi ->
Ecto.Multi.insert(multi, :my_projection, %MyProjection{...})
end
end
## Guides
- [Getting started](getting-started.html)
- [Usage](usage.html)
"""
defmacro __using__(opts) do
opts = opts || []
schema_prefix =
opts[:schema_prefix] || Application.get_env(:commanded_ecto_projections, :schema_prefix)
quote location: :keep do
@behaviour Commanded.Projections.Ecto
@opts unquote(opts)
@repo @opts[:repo] || Application.compile_env(:commanded_ecto_projections, :repo) ||
raise("Commanded Ecto projections expects :repo to be configured in environment")
@timeout @opts[:timeout] || :infinity
# Pass through any other configuration to the event handler
@handler_opts Keyword.drop(@opts, [:repo, :schema_prefix, :timeout])
unquote(__include_schema_prefix__(schema_prefix))
unquote(__include_projection_version_schema__())
use Ecto.Schema
use Commanded.Event.Handler, @handler_opts
import Ecto.Query
import unquote(__MODULE__)
def update_projection(event, metadata, multi_fn) do
projection_name = Map.fetch!(metadata, :handler_name)
event_number = Map.fetch!(metadata, :event_number)
projection_version = %ProjectionVersion{
projection_name: projection_name,
last_seen_event_number: event_number
}
prefix = schema_prefix(event, metadata)
# Query to update an existing projection version with the last seen event number with
# a check to ensure that the event has not already been projected.
update_projection_version =
from(pv in ProjectionVersion,
where:
pv.projection_name == ^projection_name and pv.last_seen_event_number < ^event_number,
update: [set: [last_seen_event_number: ^event_number]]
)
multi =
Ecto.Multi.new()
|> Ecto.Multi.run(:track_projection_version, fn repo, _changes ->
try do
repo.insert(projection_version,
prefix: prefix,
on_conflict: update_projection_version,
conflict_target: [:projection_name]
)
rescue
exception in Ecto.StaleEntryError ->
# Attempted to insert a projection version for an already seen event
{:error, :already_seen_event}
exception ->
reraise exception, __STACKTRACE__
end
end)
with %Ecto.Multi{} = multi <- apply(multi_fn, [multi]),
{:ok, changes} <- transaction(multi) do
if function_exported?(__MODULE__, :after_update, 3) do
apply(__MODULE__, :after_update, [event, metadata, changes])
else
:ok
end
else
{:error, :track_projection_version, :already_seen_event, _changes} -> :ok
{:error, _stage, error, _changes} -> {:error, error}
{:error, _error} = reply -> reply
end
end
defp transaction(%Ecto.Multi{} = multi) do
@repo.transaction(multi, timeout: @timeout, pool_timeout: @timeout)
end
defoverridable schema_prefix: 1, schema_prefix: 2
end
end
## User callbacks
@optional_callbacks [after_update: 3, schema_prefix: 1, schema_prefix: 2]
@doc """
The optional `after_update/3` callback function defined in a projector is
called after each projected event.
The function receives the event, its metadata, and all changes from the
`Ecto.Multi` struct that were executed within the database transaction.
You could use this function to notify subscribers that the read model has been
updated, such as by publishing changes via Phoenix PubSub channels.
## Example
defmodule MyApp.ExampleProjector do
use Commanded.Projections.Ecto,
application: MyApp.Application,
repo: MyApp.Projections.Repo,
name: "MyApp.ExampleProjector"
project %AnEvent{name: name}, fn multi ->
Ecto.Multi.insert(multi, :example_projection, %ExampleProjection{name: name})
end
@impl Commanded.Projections.Ecto
def after_update(event, metadata, changes) do
# Use the event, metadata, or `Ecto.Multi` changes and return `:ok`
:ok
end
end
"""
@callback after_update(event :: struct, metadata :: map, changes :: Ecto.Multi.changes()) ::
:ok | {:error, any}
@doc """
The optional `schema_prefix/1` callback function defined in a projector is
used to set the schema of the `projection_versions` table used by the
projector for idempotency checks.
It is passed the event and its metadata and must return the schema name, as a
string, or `nil`.
"""
@callback schema_prefix(event :: struct) :: String.t() | nil
@doc """
The optional `schema_prefix/2` callback function defined in a projector is
used to set the schema of the `projection_versions` table used by the
projector for idempotency checks.
It is passed the event and its metadata, and must return the schema name, as a
string, or `nil`
"""
@callback schema_prefix(event :: struct(), metadata :: map()) :: String.t() | nil
defp __include_schema_prefix__(schema_prefix) do
quote do
cond do
is_nil(unquote(schema_prefix)) ->
def schema_prefix(_event), do: nil
def schema_prefix(event, _metadata), do: schema_prefix(event)
is_binary(unquote(schema_prefix)) ->
def schema_prefix(_event), do: nil
def schema_prefix(_event, _metadata), do: unquote(schema_prefix)
is_function(unquote(schema_prefix), 1) ->
def schema_prefix(event), do: nil
def schema_prefix(event, _metadata), do: apply(unquote(schema_prefix), [event])
is_function(unquote(schema_prefix), 2) ->
def schema_prefix(event), do: nil
def schema_prefix(event, metadata), do: apply(unquote(schema_prefix), [event, metadata])
true ->
raise ArgumentError,
message:
"expected :schema_prefix option to be a string or a one-arity or two-arity function, but got: " <>
inspect(unquote(schema_prefix))
end
end
end
defp __include_projection_version_schema__ do
quote do
defmodule ProjectionVersion do
@moduledoc false
use Ecto.Schema
@primary_key {:projection_name, :string, []}
schema "projection_versions" do
field(:last_seen_event_number, :integer)
timestamps(type: :naive_datetime_usec)
end
end
end
end
defmacro project(event, do: block) do
IO.warn(
"project macro with \"do end\" block is deprecated; use project/2 with function instead",
Macro.Env.stacktrace(__ENV__)
)
quote do
def handle(unquote(event) = event, metadata) do
update_projection(event, metadata, fn var!(multi) ->
unquote(block)
end)
end
end
end
@doc """
Project a domain event into a read model by appending one or more operations
to the `Ecto.Multi` struct passed to the projection function you define
The operations will be executed in a database transaction including an
idempotency check to guarantee an event cannot be projected more than once.
## Example
project %AnEvent{}, fn multi ->
Ecto.Multi.insert(multi, :my_projection, %MyProjection{...})
end
"""
defmacro project(event, lambda) do
quote do
def handle(unquote(event) = event, metadata) do
update_projection(event, metadata, unquote(lambda))
end
end
end
defmacro project(event, metadata, do: block) do
IO.warn(
"project macro with \"do end\" block is deprecated; use project/3 with function instead",
Macro.Env.stacktrace(__ENV__)
)
quote do
def handle(unquote(event) = event, unquote(metadata) = metadata) do
update_projection(event, metadata, fn var!(multi) ->
unquote(block)
end)
end
end
end
@doc """
Project a domain event and its metadata map into a read model by appending one
or more operations to the `Ecto.Multi` struct passed to the projection
function you define.
The operations will be executed in a database transaction including an
idempotency check to guarantee an event cannot be projected more than once.
## Example
project %AnEvent{}, metadata, fn multi ->
Ecto.Multi.insert(multi, :my_projection, %MyProjection{...})
end
"""
defmacro project(event, metadata, lambda) do
quote do
def handle(unquote(event) = event, unquote(metadata) = metadata) do
update_projection(event, metadata, unquote(lambda))
end
end
end
end