commanded/commanded-ecto-projections

Read model projections for Commanded using Ecto

Commanded Ecto Projections

Read model projections for Commanded CQRS/ES applications using Ecto for persistence.

Read the Changelog for recent changes and the Hex Docs on API usage.

This README and the following guides follow the master branch which may not be the currently published version.

Overview

• Getting started
• Usage
  • Creating a read model
  • Creating a projector
  • Supervision
  • Error handling
    • error/3 callback
    • Error handling example
  • after_update/3 callback
  • Schema prefix
  • Rebuilding a projection

Example projector

defmodule MyApp.ExampleProjector do
  use Commanded.Projections.Ecto,
    application: MyApp.Application,
    repo: MyApp.Projections.Repo,
    name: "MyApp.ExampleProjector"

  project %AnEvent{} = event, _metadata, fn multi ->
    %AnEvent{name: name} = event

    projection = %ExampleProjection{name: name}

    Ecto.Multi.insert(multi, :example_projection, projection)
  end
end

Contributing

Pull requests to contribute new or improved features, and extend documentation are most welcome. Please follow the existing coding conventions.

You should include unit tests to cover any changes. Run mix test to execute the test suite:

mix deps.get
MIX_ENV=test mix setup
mix test

Contributors

• Andrey Akulov
• Ben Smith
• CptBreeza
• Florian Ebeling
• Sascha Wolf
• Tobiasz Małecki

Need help?

Please open an issue if you encounter a problem, or need assistance. You can also seek help in the #commanded channel in the official Elixir Slack.

Copyright and License

Copyright (c) 2017 Ben Smith

This library is released under the MIT License. See the LICENSE.md file for further details.

commanded-ecto-projections's Issues

ProjectionVersion: Allow to define prefix

We use commanded in an umbrella application which contains multiple applications, some of them having their own data storage. As such we want to have a separate projection version table for each application making use of CQRS.

Due to the way Ecto allows to define prefixes there seems to be no way to define a prefix for the ProjectionVersion schema. Any idea on how this could be achieved? Maybe via a config option?

Remove try / rescue from projection function

An Ecto projector is a Commanded event handler which uses a try / rescue around the handler's handle/2 function. This means that it might not be necessary to use try / rescue blocks within the Ecto projector macro. Currently these swallow the stacktrace and make it difficult to debug exceptions. Whereas the Commanded implementation logs the error and returns the stack trace.

Project macro should take a lambda containing the Ecto.Multi variable

Currently the Ecto.Multi varirable, multi, appears magically when defining your read model projection using the project macro.

To be explicit about where this variable comes from, and about its existence, a lambda should be provided as the final argument to project:

defmodule MyApp.ExampleProjector do
  use Commanded.Projections.Ecto, name: "example_projection"

  project %AnEvent{name: name}, _metadata, fn multi ->
    Ecto.Multi.insert(multi, :example_projection, %ExampleProjection{name: name})
  end

  project %AnotherEvent{name: name}, fn multi ->
    Ecto.Multi.insert(multi, :example_projection, %ExampleProjection{name: name})
  end
end

[Regression] Timestamp fields in `projection_versions` are missing microseconds.

Ecto 3.0 introduced some breaking changes:

[Ecto.Schema] :time, :naive_datetime and :utc_datetime no longer keep microseconds information. If you want to keep microseconds, use :time_usec, :naive_datetime_usec, :utc_datetime_usec

Since the timestamps macro uses :naive_datetime by default (https://hexdocs.pm/ecto/Ecto.Schema.html#timestamps/1), there has been a breaking change in commanded-ecto-projections when migrating to Ecto 3.0.

schema "projection_versions" do
    field(:last_seen_event_number, :integer)

    timestamps()
end

Source: https://github.com/commanded/commanded-ecto-projections/blob/master/lib/projections/ecto.ex

Error starting using Supervisor

I'm reading the book Building Conduit and doing the example shown with the latest versions of libs and for that I'm following what is presented in the documentation, but when I try to create a Supervisor for my commanded_ecto_projections my application is not starting.

** (Mix) Could not start application conduit: Conduit.Application.start(:normal, []) returned an error: shutdown: failed to start child: Conduit.Accounts.Supervisor
    ** (EXIT) an exception was raised:
        ** (ArgumentError) The module Conduit.Accounts.Projections.User was given as a child to a supervisor
but it does not implement child_spec/1.

# conduit/accounts/supervisor.ex
defmodule Conduit.Accounts.Supervisor do
  use Supervisor

  alias Conduit.Accounts.Projections.User, as: UserProjections

  def start_link(init_arg) do
    Supervisor.start_link(__MODULE__, init_arg, name: __MODULE__)
  end

  @impl true
  def init(_init_arg) do
    children = [UserProjections]

    Supervisor.init(children, strategy: :one_for_one)
  end
end

# conduit/accounts/projectors/user.ex
defmodule Conduit.Accounts.Projectors.User do
  use Commanded.Projections.Ecto,
    application: Conduit.App,
    repo: Conduit.Repo,
    name: "Accounts.Projectors.User"

  alias Conduit.Accounts.Events.UserRegistered
  alias Conduit.Accounts.Projections.User

  project(%UserRegistered{} = registered, _metadata, fn multi ->
    multi.insert(multi, :user, %User{
      uuid: registered.user_uuid,
      username: registered.username,
      email: registered.email,
      hashed_password: registered.hashed_password,
      bio: nil,
      image: nil
    })
  end)
end

# mix.exs
  defp deps do
    [
      {:phoenix, "~> 1.6.2"},
      {:phoenix_ecto, "~> 4.4"},
      {:ecto_sql, "~> 3.6"},
      {:postgrex, ">= 0.0.0"},
      {:telemetry_metrics, "~> 0.6"},
      {:telemetry_poller, "~> 1.0"},
      {:gettext, "~> 0.18"},
      {:jason, "~> 1.2"},
      {:plug_cowboy, "~> 2.5"},
      {:commanded, "~> 1.3"},
      {:commanded_eventstore_adapter, "~> 1.2"},
      {:commanded_ecto_projections, "~> 1.2"},
      {:exconstructor, "~> 1.2"},
      {:ex_machina, "~> 2.7.0", only: :test},
      {:credo, "~> 1.5", only: [:dev, :test], runtime: false},
      {:mix_test_watch, "~> 1.0", only: [:dev, :test], runtime: false}
    ]
  end

When I start the Projector from iex as per the documentation, it works perfectly, only when I try to use the supervisor that doesn't work

{:ok, pid} = Conduit.Accounts.Projectors.User.start_link()

What could it be? and sorry for my english

Projector performance issues

I'm having trouble to execute a new projector that runs old events (start_from: :origin). They are taking to long to process each event. Just put a Logger.warn for each processed event and the results are show below:

16:05:32.903 [warn] API UP
16:05:37.823 [warn] Event X | XXX1 | 62a2e21e-88c0-4e23-9e51-fcd5b6c20032
16:05:42.924 [warn] Event X | XXX2 | 8be902db-7951-4c09-b593-d51b6ae2ac1f

What can I do to improve the execution performance?

Document :consistency option in README

It would be nice to copy-paste (or at least reference) Consistency Guarantee note in commanded-ecto-projections README.

The problem is that after reading Commands.md one would normally go into commanded-ecto-projections trying to find out from README how to turn his Ecto Projector into a "strongly consistent event handler".

Events not projected in order?

Hi there, I am fairly new to commanded, so perhaps this is a stupid question..

I pushed some events (~800,000 events w/ relatively large payload) through commanded into Postgres eventstore and now trying to use commanded-ecto-projections to project these. The order of events is important however.

Checking in Postgres for one example shows the correct and expected order:

Now when I use commanded-ecto-projections unfortunately I get the second event before the first one. My projector is super simple and no pattern matching prevents the first event from being filtered.

Is the same order of events not guaranteed as registered in event store's events table?

Thank you!

Out-of-order events are skipped when using the `:concurrency` option

Commanded.Projections.Ecto passes the :concurrency option introduced in commanded/commanded#486 through to the underlying Commanded.Event.Handler. However, out-of-order events are skipped when compared to the projection_versions table.

Dynamic schema prefix

It is currently possible to define a static schema prefix for a read model projector using global configuration or by providing a schema_prefix when defining the projector.

defmodule MyApp.ExampleProjector do
  use Commanded.Projections.Ecto,
    application: MyApp.Application,
    repo: MyApp.Projections.Repo,    
    name: "example_projection",
    schema_prefix: "example_schema_prefix"
end

We should allow a dynamic prefix to be set for a read model projector.

This might be configurable on start:

{:ok, _pid} = MyApp.ExampleProjector.start_link(schema_prefix: "tenant1")

Or by providing a one-arity function as the schema_prefix option which can be used to extract the schema from an event:

defmodule MyApp.ExampleProjector do
  use Commanded.Projections.Ecto,
    application: MyApp.Application,
    repo: MyApp.Projections.Repo,    
    name: "example_projection",
    schema_prefix: &MyApp.ExampleProjector.schema_prefix/1

  def schema_prefix(%AnEvent{tenant: tenant}), do: tenant
end

A two-arity function could also be used to receive an event and its metadata:

Document significance of `:name` option for Commanded.Projections.Ecto

That's not mentioned in the docs, and it would help understanding.

Upgrade to Ecto v3

Upgrade to the newly released Ecto v3:

• ecto {:ecto, "~> 3.0"}
• ecto_sql {:ecto_sql, "~> 3.0"}

How to insert data in different tables for one event?

Hello,

first of all, thanks a lot for this nice library :)

I was wondering how to insert data in different tables for one event? For instance, two Ecto.Multi.insert() results in seeing only the result of the last one in the table.

Configure repo at runtime

Should we support configuring repo at runtime? We can keep it in the handler state right?

Support runtime projector names

The next version of Commanded has support for specifying the name of an event handler at runtime. Currently an Ecto projector validates the name is provided at compile-time. Usage of the name must be done at runtime.

Error output on transaction failure

Hi there,

we noticed that Projections fail somewhat silently (the process just exits but yields no explanation as to why) when an Error is raised, e.g., a :not_null_violation.

This seems to be caused by the fact that Errors are not handled and bubble up unintercepted, which eventually causes the process to exit.

To improve debuggability, it'd be nice to have some logging of Errors before they are reraised and preserve the old behavior of the process dying.

I'd be happy to contribute a patch and add this, it's pretty straightforward, but I'm not sure if this would belong here or in commanded in the generic event handler and would love to hear your input on this.

Best,
Jan

Document the usage of `multi`.

It's not immediately clear what the multi passed to the projection function actually does:

project (%MyEvent{}, _, fn multi -> ... end)

How does using this multi differ from returning a new Ecto.Multi?

project (%MyEvent{}, _, fn -> Ecto.Multi.new() ... end)

Does the multi passed to the projection function already contain some other operations? Does it group some other operations together with the projection?

Thanks!

[Doc] Projection versions last seen checkpoint differs from event store subscription checkpoint

Explain why the projection versions last seen event number will often be different from the corresponding event store's last seen. This is because the projection versions table is only updated when an event is projected but the event store subscription checkpoint is updated after every event by default.

Starting as part of application supervision tree fails

Hey guys :)

I was following along this blog series but I'm stuck trying to add a Projector to the application's supervision tree.

It seems like commanded tries to read the configuration from the configured application. However, this one is not started yet when the supervision tree is initially built, causing the startup process to fail with the following error message

** (Mix) Could not start application bank_api: BankAPI.Application.start(:normal, []) returned an error: shutdown: failed to start child: BankAPI.Accounts.Supervisor
    ** (EXIT) shutdown: failed to start child: :account_opened
        ** (EXIT) an exception was raised:
            ** (RuntimeError) could not lookup :bank_api because it was not started or it does not exist
                (commanded 1.3.1) lib/commanded/application/config.ex:45: Commanded.Application.Config.lookup/1
                (commanded 1.3.1) lib/commanded/application/config.ex:15: Commanded.Application.Config.get/2
                (commanded 1.3.1) lib/commanded/registration.ex:20: Commanded.Registration.start_link/5
                (commanded 1.3.1) lib/commanded/event/handler.ex:636: Commanded.Event.Handler.start_link/4
                (stdlib 3.15.2) supervisor.erl:414: :supervisor.do_start_child_i/3
                (stdlib 3.15.2) supervisor.erl:400: :supervisor.do_start_child/2
                (stdlib 3.15.2) supervisor.erl:384: anonymous fn/3 in :supervisor.start_children/2
                (stdlib 3.15.2) supervisor.erl:1234: :supervisor.children_map/4
                (stdlib 3.15.2) supervisor.erl:350: :supervisor.init_children/2
                (stdlib 3.15.2) gen_server.erl:423: :gen_server.init_it/2
                (stdlib 3.15.2) gen_server.erl:390: :gen_server.init_it/6
                (stdlib 3.15.2) proc_lib.erl:226: :proc_lib.init_p_do_apply/3

Is there a solution to this other than to send my supervisor an event to dynamically start the projector after the main application has started?

I uploaded my current state here

Thank you in advance :)

Use `timestamp with time zone` for timestamp fields in `projection_versions`

The timestamps inserted_at and updated_at in projection_versions table use the PostgreSQL type timestamp without time zone.

Since inserted_at and updated_at are to be implicitly interpreted as UTC, why not making it explicit and use the PostgreSQL type timestamp with time zone instead?

Allow projectors to be tested directly

Make it possible to test a projector by directly calling a function on the projector module, passing an Ecto.Multi struct, an event, and its metadata.

Examples

Test a projector by projecting a single event:

{:ok, _changes} =
  Ecto.Multi.new()
  |> MyApp.ExampleProjector.project(event, metadata)
  |> MyApp.Repo.transaction()

Test a projector by reducing a list of events:

multi = 
  Enum.reduce(events, Ecto.Multi.new(), fn event, multi ->
    MyApp.ExampleProjector.project(multi, event, metadata)
  end)

{:ok, _changes} = MyApp.Repo.transaction(multi)

Don't call the `after_update/3` function when idempotency check fails

Commanded Ecto projections uses the projection_versions table to ensure that an event cannot be projected more than once. This is needed because Commanded event handlers guarantee at-least-once event delivery (an event can be received more than once). However the after_update/3 function will still be called when the idempotency check fails due to receiving an already projected event. In this case it should not be called.

Ecto projector shutting down service when invalid event

I've propagated an event with an string value in an integer field.

While easy to correct, there seems to be a catastrophic failure within Commanded

[warn] Exercises.Evaluations.Projectors.Assignment has requested to stop: %Ecto.ChangeError{message: "value `\"20\"` for `Exercises.Evaluations.Projections.Assignment.group_id` in `insert` does not match type :integer"}
[error] GenServer {Commanded.Registration.LocalRegistry, {Commanded.Event.Handler, "Evaluations.Projectors.Assignment"}} terminating
** (stop) %Ecto.ChangeError{message: "value `\"20\"` for `Exercises.Evaluations.Projections.Assignment.group_id` in `insert` does not match type :integer"}
Last message: {:events, [%Commanded.EventStore.RecordedEvent{causation_id: "06c166e2-026e-4da9-a66e-a70c1fc5f0cc", correlation_id: "b15ed36a-cf50-4154-8efd-87049b37bf4b", created_at: ~N[2018-11-19 21:54:40.135747], data: %Exercises.Evaluations.Events.AssignmentSaved{group_id: "20", module_id: "lGH3LWcBJBlBGoN6S-qk", rubric_id: "430d19e3-fe63-44d9-bd66-cd5b9100e552", state: "waiting_for_evaluation", user_id: 5, uuid: "14ea36bb-8252-4268-b9a9-99cbd2049a48"}, event_id: "38ba6ed9-f9cb-4207-af4b-aae7d25a8278", event_number: 21, event_type: "Elixir.Exercises.Evaluations.Events.AssignmentSaved", metadata: %{}, stream_id: "14ea36bb-8252-4268-b9a9-99cbd2049a48", stream_version: 1}]}
State: %Commanded.Event.Handler{consistency: :strong, handler_module: Exercises.Evaluations.Projectors.Assignment, handler_name: "Evaluations.Projectors.Assignment", last_seen_event: nil, subscribe_from: :origin, subscription: #PID<0.462.0>}
[info] Application exercises exited: shutdown

I haven't found how to restart the service. How can I do that? Do I need to delete the ill-formed event?

Thanks!

question: Does after_update run after all the projection are executed?

I think it's not clear if the after_update runs after the single projection or after the whole transaction.
Also, would resetting the projection trigger the after_update? If so, it doesn't look a good point where to place the publishing to a channel.
Anyway thanks for this awesome project!

Ensure Ecto projections works with Commanded's event handler `error/3` callback

Defining an error/3 callback in a projection, using Commanded's event handler (#133), and returning :skip should ignore the problematic event and allow the projection to continue.