diff --git a/.formatter.exs b/.formatter.exs index 74ca3339..7221197b 100644 --- a/.formatter.exs +++ b/.formatter.exs @@ -6,7 +6,8 @@ locals_without_parens = [ middleware: 1, project: 2, project: 3, - router: 1 + router: 1, + project_batch: 1 ] [ diff --git a/guides/howtos/building-read-models-with-ecto.md b/guides/howtos/building-read-models-with-ecto.md index 73255b24..364d998b 100644 --- a/guides/howtos/building-read-models-with-ecto.md +++ b/guides/howtos/building-read-models-with-ecto.md @@ -81,6 +81,61 @@ Supervisor.start_link([ ], strategy: :one_for_one) ``` +### Subscription options + +You can control how events are consumed from the event store using the `:subscription_opts` option. This provides a convenient way to group subscription-related configuration together: + +```elixir +defmodule MyApp.ExampleProjector do + use Commanded.Projections.Ecto, + application: MyApp.Application, + repo: MyApp.Projections.Repo, + name: "example_projection", + subscription_opts: [ + start_from: :origin, # Start from the first event + subscribe_to: :all, # Subscribe to all streams + concurrency: 4 # Process with 4 workers + ] + + project %AnEvent{}, _metadata, fn multi -> + # Your projection logic + end +end +``` + +#### Available subscription options + +- **`:start_from`** - Where to begin reading events + - `:origin` - Start from the very first event (use for new projections) + - `:current` - Start from the current position (skip historical events) + - Positive integer - Start from a specific event number + +- **`:subscribe_to`** - Which event stream(s) to subscribe to + - `:all` - Subscribe to all events (default) + - Stream name (string) - Subscribe to a specific stream + +- **`:concurrency`** - Number of concurrent workers for parallel processing + - Positive integer - Enable concurrent event processing + - **Note:** Mutually exclusive with `:batch_size` + +#### Alternative: Top-level options + +You can also pass these options at the top level instead of nesting them under `:subscription_opts`. Both formats are equivalent: + +```elixir +defmodule MyApp.ExampleProjector do + use Commanded.Projections.Ecto, + application: MyApp.Application, + repo: MyApp.Projections.Repo, + name: "example_projection", + start_from: :origin, + subscribe_to: :all, + concurrency: 4 +end +``` + +If you provide both nested and top-level options, the top-level ones take precedence. + ### Using the `project` macro The `project/3` macro expects the domain event, metadata, and a single-arity function that takes and returns an `Ecto.Multi` data structure for grouping multiple Repo operations. These will all be executed within a single transaction. You can use `Ecto.Multi` to insert, update, and delete data. @@ -116,6 +171,35 @@ project %ItemUpdated{uuid: uuid} = event, _metadata, fn multi -> end ``` +### Using the `project_batch` macro + +You can use `project_batch` to receive events in batches. To enable batching, you need to set the `batch_size` and use the `project_batch/1` macro. `project_batch/1` receives a function that takes a list of `{event, metadata}` tuples for all the events in the batch and an `Ecto.Multi` structure, similar to `project/3`. + +Note that there is currently no built in way to target a single type of event to be projected, and as such a single `project_batch` macro is expected to gracefully handle (or ignore) any events that it may receive. + +#### Example + +```elixir +defmodule MyApp.Projections.BatchProjector do + use Commanded.Projections.Ecto, + application: MyApp.Application, + repo: MyApp.Projections.Repo, + name: "example_batch_projection", + batch_size: 10 + + project_batch fn events, multi -> + projections = events + |> Enum.map(fn + {%AnEvent{name: name}, _metadata} -> %{name: name} + _ -> nil + end) + |> Enum.reject(&is_nil/1) + + Ecto.Multi.insert_all(multi, :example_batch_projection, Projection, projections) + end +end +``` + ## Supervision Your projector module must be included in your application supervision tree: @@ -217,6 +301,50 @@ end You could use this function to notify subscribers that the read model has been updated (e.g. pub/sub to Phoenix channels). +#### ⚠️ Transaction Semantics + +**CRITICAL:** The `after_update/3` callback executes **AFTER** the database transaction has been committed. This means: + +- **Errors cannot rollback the transaction** - If this callback returns an error or raises an exception, the projection data is already persisted in the database. +- **Use for side effects only** - This callback is designed for notifications, pub/sub, external API calls, or other side effects that should happen after successful projection updates. +- **Error handling implications** - Returning `{:error, reason}` or raising an exception will propagate the error up, but the database changes are permanent. The event handler may retry, potentially causing duplicate side effects. + +If you need to perform validation or operations that should prevent the projection from being saved, do so within your `project/2` function (inside the `Ecto.Multi`), not in this callback. + +### `after_update_batch/2` callback + +Similarly for batching projectors, you can define an `after_update_batch/2` callback function in a projector to be called after a batch of events has been projected. The function receives a list of `{event, metadata}` tuples for each processed event and all changes from the `Ecto.Multi` struct. + +```elixir +defmodule MyApp.BatchProjector do + use Commanded.Projections.Ecto, + application: MyApp.Application, + repo: MyApp.Projections.Repo, + name: "MyApp.BatchProjector", + batch_size: 10 + + project_batch fn events, multi -> + # Your batch projection logic + end + + @impl Commanded.Projections.Ecto + def after_update_batch(events, changes) do + # Notify subscribers about the batch update + :ok + end +end +``` + +#### ⚠️ Transaction Semantics + +**CRITICAL:** The `after_update_batch/2` callback executes **AFTER** the database transaction has been committed. This means: + +- **Errors cannot rollback the transaction** - If this callback returns an error or raises an exception, the projection data is already persisted in the database. +- **Use for side effects only** - This callback is designed for notifications, pub/sub, external API calls, or other side effects that should happen after successful projection updates. +- **Error handling implications** - Returning `{:error, reason}` or raising an exception will propagate the error up, but the database changes are permanent. The event handler may retry, potentially causing duplicate side effects. + +If you need to perform validation or operations that should prevent the projection from being saved, do so within your `project_batch/2` function (inside the `Ecto.Multi`), not in this callback. + ## Schema prefix When using a prefix for your Ecto schemas you might also want to change the prefix for the `ProjectionVersion` schema. There are a number of options to do this: @@ -225,7 +353,8 @@ When using a prefix for your Ecto schemas you might also want to change the pref ```elixir # config/config.exs - config :commanded_ecto_projections, schema_prefix: "example_schema_prefix" + config :commanded, Commanded.Projections.Ecto, + schema_prefix: "example_schema_prefix" ``` 2. Provide a static `schema_prefix` as a projector option: diff --git a/lib/commanded/projections/ecto.ex b/lib/commanded/projections/ecto.ex index f9ab148d..ea3206cb 100644 --- a/lib/commanded/projections/ecto.ex +++ b/lib/commanded/projections/ecto.ex @@ -3,24 +3,63 @@ if Code.ensure_loaded?(Ecto) do @moduledoc """ Read model projections for Commanded using Ecto. - ## Example usage + Build read models (projections) from events in your Commanded application with + automatic idempotency, transaction support, and optional batch processing. - defmodule Projector do + ## Basic Usage + + defmodule MyApp.Projectors.AccountProjector do use Commanded.Projections.Ecto, application: MyApp.Application, - name: "my-projection", - repo: MyApp.Repo, - schema_prefix: "my-prefix", - timeout: :infinity + name: "account_projector", + repo: MyApp.Repo + + project %AccountOpened{account_id: account_id, name: name}, _metadata, fn multi -> + Ecto.Multi.insert(multi, :account, %Account{ + id: account_id, + name: name, + balance: 0 + }) + end - project %Event{}, _metadata, fn multi -> - Ecto.Multi.insert(multi, :my_projection, %MyProjection{...}) + project %MoneyDeposited{account_id: account_id, amount: amount}, _metadata, fn multi -> + Ecto.Multi.update_all( + multi, + :account, + where(Account, id: ^account_id), + inc: [balance: ^amount] + ) end + end + + ## Batch Processing + + Process multiple events in a single database transaction for improved throughput: + + defmodule MyApp.Projectors.HighVolumeBatchProjector do + use Commanded.Projections.Ecto, + application: MyApp.Application, + name: "high_volume_batch_projector", + repo: MyApp.Repo, + batch_size: 50 + + # Use project_batch/1 instead of project/3 for batch projectors + project_batch fn events, multi -> + # Process all events in a single transaction + Enum.reduce(events, multi, fn + {%EventA{id: id}, _metadata}, multi -> + Ecto.Multi.insert(multi, {:event_a, id}, %ReadModel{...}) - project %AnotherEvent{}, fn multi -> - Ecto.Multi.insert(multi, :my_projection, %MyProjection{...}) + {%EventB{id: id}, _metadata}, multi -> + Ecto.Multi.update(multi, {:event_b, id}, ...) + end) + end end - end + + **Batch Limitations:** + - Cannot use `:concurrency` option (mutually exclusive) + - Only static `:schema_prefix` strings allowed (no dynamic functions) + - Uses watermark-based idempotency for efficient batch processing ## Guides @@ -29,35 +68,39 @@ if Code.ensure_loaded?(Ecto) do """ - defmacro __using__(opts) do - opts = opts || [] - - schema_prefix = - opts[:schema_prefix] || - Application.get_env(:commanded, Commanded.Projections.Ecto, []) - |> Keyword.get(:schema_prefix) - - quote location: :keep do - @behaviour Commanded.Projections.Ecto + @doc false + def validate_mutual_exclusivity(opts) do + batch_size = Keyword.get(opts, :batch_size) + concurrency = Keyword.get(opts, :concurrency) - @opts unquote(opts) - @repo @opts[:repo] || - Application.compile_env(:commanded, [Commanded.Projections.Ecto, :repo]) || - raise("Commanded Ecto projections expects :repo to be configured in environment") - @timeout @opts[:timeout] || :infinity + case {batch_size, concurrency} do + {nil, _} -> + :ok - # Pass through any other configuration to the event handler - @handler_opts Keyword.drop(@opts, [:repo, :schema_prefix, :timeout]) + {_, nil} -> + :ok - unquote(__include_schema_prefix__(schema_prefix)) - unquote(__include_projection_version_schema__()) - - use Ecto.Schema - use Commanded.Event.Handler, @handler_opts + {_batch_size, _concurrency} -> + {:error, + "cannot use both :batch_size and :concurrency options - they are mutually exclusive. " <> + "Use :batch_size for batch processing OR :concurrency for concurrent processing, but not both."} + end + end - import Ecto.Query - import unquote(__MODULE__) + @doc false + def validate_batch_schema_prefix_compatibility(batch_size, schema_prefix) do + if batch_size && is_function(schema_prefix) do + {:error, + "cannot use :batch_size with dynamic :schema_prefix (function) - " <> + "batch projectors only support static string prefixes. " <> + "Either remove :batch_size or change :schema_prefix to a string."} + else + :ok + end + end + defp __define_update_projection__ do + quote do def update_projection(event, metadata, multi_fn) do projection_name = Map.fetch!(metadata, :handler_name) event_number = Map.fetch!(metadata, :event_number) @@ -112,18 +155,260 @@ if Code.ensure_loaded?(Ecto) do {:error, _error} = reply -> reply end end + end + end + + defp __define_update_projection_batch__ do + quote do + def update_projection_batch([], _multi_fn) do + # Empty batches are idempotent - nothing to process + :ok + end + + def update_projection_batch([{first_event, first_event_metadata} | _] = events, multi_fn) + when is_map(first_event_metadata) and is_map_key(first_event_metadata, :handler_name) and + is_map_key(first_event_metadata, :event_number) do + # Safe to use first event's handler_name for entire batch because: + # All events in a batch come from the same handler process (received via {:events, events} message). + # The handler enriches all events with its own handler_name when calling enrich_metadata/2. + # It's architecturally impossible for events with different handler_name values to appear in the same batch. + %{handler_name: projection_name} = first_event_metadata + prefix = schema_prefix(first_event, first_event_metadata) + + multi = build_batch_projection_multi(events, projection_name, prefix) + + # Apply user's projection function and merge into our transaction + execute_batch_projection(multi, multi_fn) + end + + def update_projection_batch(invalid_batch, _multi_fn) do + {:error, + {:invalid_batch_format, + "Expected list of {event, metadata} tuples, got: #{inspect(invalid_batch, limit: 3)}"}} + end + + unquote(__define_batch_helpers__()) + end + end + + defp __define_batch_helpers__ do + quote do + defp build_batch_projection_multi(events, projection_name, prefix) do + Ecto.Multi.new() + |> Ecto.Multi.run(:lock_and_filter, fn repo, _changes -> + lock_and_filter_batch_events(repo, events, projection_name, prefix) + end) + |> Ecto.Multi.run(:track_projection_version, fn repo, %{lock_and_filter: result} -> + track_batch_projection_version(repo, result, projection_name, prefix) + end) + end + + defp lock_and_filter_batch_events(repo, events, projection_name, prefix) do + try do + # Lock the projection_version row for this projector + # This ensures no other process can update it concurrently + current_last_seen = + repo.one( + from(pv in ProjectionVersion, + where: pv.projection_name == ^projection_name, + select: pv.last_seen_event_number, + lock: "FOR UPDATE" + ), + prefix: prefix + ) || 0 + + # Validate all events have required metadata fields + invalid_events = + Enum.reject(events, fn + {_event, metadata} when is_map(metadata) -> + is_map_key(metadata, :event_number) and is_map_key(metadata, :handler_name) + + _ -> + false + end) + + if invalid_events != [] do + {:error, + {:invalid_batch_structure, + "All events must be {event, metadata} tuples with :event_number and :handler_name in metadata"}} + else + # Filter events based on FRESH data from inside the transaction + unseen_events = + Enum.filter(events, fn {_event, metadata} -> + metadata.event_number > current_last_seen + end) + + case unseen_events do + [] -> + # All events already seen + {:error, :already_seen_event} + + unseen_events -> + # Get last event in batch (events are guaranteed to be ordered by Commanded) + # Commanded's subscription always delivers events in ascending event_number order + {_last_event, last_metadata} = List.last(unseen_events) + %{event_number: last_unseen_number} = last_metadata + + # Advance watermark to highest event number in this batch + # Provides defense-in-depth: ensures idempotency even if Commanded's + # subscription position is lost or corrupted during crash recovery + {:ok, {current_last_seen, unseen_events, last_unseen_number}} + end + end + rescue + exception -> + reraise exception, __STACKTRACE__ + end + end + + defp track_batch_projection_version( + repo, + {_current_last_seen, _unseen_events, last_event_number}, + projection_name, + prefix + ) do + projection_version = %ProjectionVersion{ + projection_name: projection_name, + last_seen_event_number: last_event_number + } + + # Update watermark BEFORE user projection executes. + # This is safe because we hold a FOR UPDATE lock from :lock_and_filter step: + # 1. The lock ensures no other process can read or write this row until our + # transaction commits or rolls back, preventing all race conditions + # 2. The conditional WHERE clause provides defense-in-depth: + # only batches with higher event_numbers can update the watermark + # 3. If the user projection fails, the entire transaction rolls back atomically, + # including this watermark update, maintaining consistency + # 4. The early update prevents wasted work in concurrent transactions + # that might be processing older events (they'll see the lock and wait) + update_projection_version = + from(pv in ProjectionVersion, + where: + pv.projection_name == ^projection_name and + pv.last_seen_event_number < ^last_event_number, + update: [set: [last_seen_event_number: ^last_event_number]] + ) + + repo.insert( + projection_version, + prefix: prefix, + on_conflict: update_projection_version, + conflict_target: [:projection_name] + ) + end + + defp execute_batch_projection(multi, multi_fn) do + # This ensures EVERYTHING happens in ONE atomic transaction + with %Ecto.Multi{} = multi <- + Ecto.Multi.run(multi, :prepare_user_multi, fn _repo, + %{ + lock_and_filter: + {_current, unseen_events, + _last} + } -> + user_multi = multi_fn.(unseen_events, Ecto.Multi.new()) + + case user_multi do + %Ecto.Multi{} -> + {:ok, {unseen_events, user_multi}} + + other -> + {:error, {:invalid_multi_return, other}} + end + end), + %Ecto.Multi{} = multi <- + Ecto.Multi.merge(multi, fn %{prepare_user_multi: {_unseen_events, user_multi}} -> + user_multi + end), + {:ok, changes} <- transaction(multi) do + # Get the unseen events from our preparation step + {unseen_events, _user_multi} = changes.prepare_user_multi + + after_update_batch(unseen_events, changes) + else + {:error, :lock_and_filter, :already_seen_event, _changes} -> + :ok + + {:error, _stage, error, _changes} -> + {:error, error} + + {:error, _error} = reply -> + reply + end + end + end + end + + defp __define_helper_functions__ do + quote do + def after_update(_event, _metadata, _changes), do: :ok + def after_update_batch(_events, _changes), do: :ok defp transaction(%Ecto.Multi{} = multi) do @repo.transaction(multi, timeout: @timeout, pool_timeout: @timeout) end + end + end + + defmacro __using__(opts) do + opts = opts || [] + batch_size = opts[:batch_size] + + schema_prefix = + opts[:schema_prefix] || + Application.get_env(:commanded, Commanded.Projections.Ecto, []) + |> Keyword.get(:schema_prefix) + + # Validate mutual exclusivity + case validate_mutual_exclusivity(opts) do + :ok -> :ok + {:error, message} -> raise CompileError, description: message + end + + # Validate batch schema prefix compatibility + case validate_batch_schema_prefix_compatibility(batch_size, schema_prefix) do + :ok -> :ok + {:error, message} -> raise CompileError, description: message + end + + quote location: :keep do + @behaviour Commanded.Projections.Ecto - defoverridable schema_prefix: 1, schema_prefix: 2 + @opts unquote(opts) + @repo @opts[:repo] || + Application.compile_env(:commanded, [Commanded.Projections.Ecto, :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__) + + unquote(__define_update_projection__()) + unquote(__define_update_projection_batch__()) + unquote(__define_helper_functions__()) + + defoverridable after_update: 3, after_update_batch: 2, schema_prefix: 1, schema_prefix: 2 end end ## User callbacks - @optional_callbacks [after_update: 3, schema_prefix: 1, schema_prefix: 2] + @optional_callbacks [ + after_update: 3, + after_update_batch: 2, + schema_prefix: 1, + schema_prefix: 2 + ] @doc """ The optional `after_update/3` callback function defined in a projector is @@ -135,6 +420,24 @@ if Code.ensure_loaded?(Ecto) do You could use this function to notify subscribers that the read model has been updated, such as by publishing changes via Phoenix PubSub channels. + ## ⚠️ Transaction Semantics + + **CRITICAL:** This callback executes **AFTER** the database transaction has been + committed. This means: + + - **Errors cannot rollback the transaction** - If this callback returns an error + or raises an exception, the projection data is already persisted in the database. + - **Use for side effects only** - This callback is designed for notifications, + pub/sub, external API calls, or other side effects that should happen after + successful projection updates. + - **Error handling implications** - Returning `{:error, reason}` or raising an + exception will propagate the error up, but the database changes are permanent. + The event handler may retry, potentially causing duplicate side effects. + + If you need to perform validation or operations that should prevent the projection + from being saved, do so within your `project/3` function (inside the `Ecto.Multi`), + not in this callback. + ## Example defmodule MyApp.ExampleProjector do @@ -143,7 +446,7 @@ if Code.ensure_loaded?(Ecto) do repo: MyApp.Projections.Repo, name: "MyApp.ExampleProjector" - project %AnEvent{name: name}, fn multi -> + project %AnEvent{name: name}, _metadata, fn multi -> Ecto.Multi.insert(multi, :example_projection, %ExampleProjection{name: name}) end @@ -158,6 +461,58 @@ if Code.ensure_loaded?(Ecto) do @callback after_update(event :: struct, metadata :: map, changes :: Ecto.Multi.changes()) :: :ok | {:ok, any} | {:error, any} + @doc """ + The optional `after_update_batch/2` callback function defined in a projector is + called after a batch of projected events. + + The function receives the events, their metadata, and all changes from the + `Ecto.Multi` struct that were executed within the database transaction. + + ## ⚠️ Transaction Semantics + + **CRITICAL:** This callback executes **AFTER** the database transaction has been + committed. This means: + + - **Errors cannot rollback the transaction** - If this callback returns an error + or raises an exception, the projection data is already persisted in the database. + - **Use for side effects only** - This callback is designed for notifications, + pub/sub, external API calls, or other side effects that should happen after + successful projection updates. + - **Error handling implications** - Returning `{:error, reason}` or raising an + exception will propagate the error up, but the database changes are permanent. + The event handler may retry, potentially causing duplicate side effects. + + If you need to perform validation or operations that should prevent the projection + from being saved, do so within your `project_batch/2` function (inside the + `Ecto.Multi`), not in this callback. + + ## Example + + defmodule MyApp.ExampleProjector do + use Commanded.Projections.Ecto, + application: MyApp.Application, + repo: MyApp.Projections.Repo, + name: "MyApp.ExampleProjector", + batch_size: 10 + + project_batch fn events, multi -> + Enum.reduce(events, multi, fn {event, _metadata}, multi -> + # Process each event in the batch + handle_event(event, multi) + end) + end + + @impl Commanded.Projections.Ecto + def after_update_batch(events, changes) do + # Use the events, metadata, or `Ecto.Multi` changes and return `:ok` + :ok + end + end + + """ + @callback after_update_batch(events :: list(tuple), 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 @@ -186,15 +541,15 @@ if Code.ensure_loaded?(Ecto) do def schema_prefix(event, _metadata), do: schema_prefix(event) is_binary(unquote(schema_prefix)) -> - def schema_prefix(_event), do: nil + def schema_prefix(_event), do: unquote(schema_prefix) 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), do: apply(unquote(schema_prefix), [event]) 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), do: nil def schema_prefix(event, metadata), do: apply(unquote(schema_prefix), [event, metadata]) @@ -299,5 +654,55 @@ if Code.ensure_loaded?(Ecto) do end end end + + @doc """ + Project a batch of domain events and their metadata into a read model by appending + one or more operations to the `Ecto.Multi` struct passed to the projection function. + + This macro is used when `:batch_size` is configured. Events are received as a list + of `{event, metadata}` tuples and processed in a single database transaction. + + The projection function receives two arguments: + - `events`: List of `{event, metadata}` tuples + - `multi`: An `Ecto.Multi` struct to build your projection operations + + ## Example + + project_batch fn events, multi -> + Enum.reduce(events, multi, fn + {%EventA{id: id}, _metadata}, multi -> + Ecto.Multi.insert(multi, {:event_a, id}, %ReadModel{...}) + + {%EventB{id: id}, _metadata}, multi -> + Ecto.Multi.update(multi, {:event_b, id}, ...) + end) + end + + """ + defmacro project_batch(lambda) do + quote do + if Module.defines?(__MODULE__, {:handle_batch, 1}) do + raise CompileError, + description: """ + project_batch can only be called once per projector. + + To handle multiple event types, use pattern matching inside your handler: + + project_batch fn events, multi -> + Enum.reduce(events, multi, fn {event, metadata}, multi -> + case event do + %EventA{} -> ... + %EventB{} -> ... + end + end) + end + """ + end + + def handle_batch(all_events) do + update_projection_batch(all_events, unquote(lambda)) + end + end + end end end diff --git a/test/projections/commanded_batch_integration_test.exs b/test/projections/commanded_batch_integration_test.exs new file mode 100644 index 00000000..bbbad8ae --- /dev/null +++ b/test/projections/commanded_batch_integration_test.exs @@ -0,0 +1,117 @@ +defmodule Commanded.Projections.CommandedBatchIntegrationTest do + use ExUnit.Case + + import Commanded.Projections.ProjectionAssertions + import Ecto.Query + + alias Commanded.Projections.Events.AnEvent + alias Commanded.Projections.Projection + alias Commanded.Projections.Repo + alias Ecto.Adapters.SQL.Sandbox + + defmodule IntegrationBatchProjector do + use Commanded.Projections.Ecto, + application: TestApplication, + repo: Commanded.Projections.Repo, + name: "IntegrationBatchProjector", + batch_size: 5 + + project_batch fn events, multi -> + projections = + Enum.map(events, fn + {%Commanded.Projections.Events.AnEvent{name: name}, _metadata} -> %{name: name} + end) + + Ecto.Multi.insert_all(multi, :projections, Commanded.Projections.Projection, projections) + end + end + + setup do + start_supervised!(TestApplication) + Sandbox.checkout(Repo) + + :ok + end + + describe "Commanded batch integration" do + test "handle_batch function is defined" do + assert function_exported?(IntegrationBatchProjector, :handle_batch, 1) + end + + test "handle_batch processes single batch correctly" do + events = [ + {%AnEvent{name: "e1"}, %{handler_name: "IntegrationBatchProjector", event_number: 1}}, + {%AnEvent{name: "e2"}, %{handler_name: "IntegrationBatchProjector", event_number: 2}}, + {%AnEvent{name: "e3"}, %{handler_name: "IntegrationBatchProjector", event_number: 3}} + ] + + assert :ok == IntegrationBatchProjector.handle_batch(events) + + assert_projections(Projection, ["e1", "e2", "e3"]) + + last_seen = + Repo.one( + from(pv in IntegrationBatchProjector.ProjectionVersion, + where: pv.projection_name == "IntegrationBatchProjector", + select: pv.last_seen_event_number + ) + ) + + assert last_seen == 3 + end + + test "handle_batch processes multiple batches with correct watermark" do + assert :ok == + IntegrationBatchProjector.handle_batch([ + {%AnEvent{name: "batch1_e1"}, + %{handler_name: "IntegrationBatchProjector", event_number: 1}}, + {%AnEvent{name: "batch1_e2"}, + %{handler_name: "IntegrationBatchProjector", event_number: 2}} + ]) + + assert :ok == + IntegrationBatchProjector.handle_batch([ + {%AnEvent{name: "batch2_e3"}, + %{handler_name: "IntegrationBatchProjector", event_number: 3}}, + {%AnEvent{name: "batch2_e4"}, + %{handler_name: "IntegrationBatchProjector", event_number: 4}} + ]) + + assert_projections(Projection, ["batch1_e1", "batch1_e2", "batch2_e3", "batch2_e4"]) + + last_seen = + Repo.one( + from(pv in IntegrationBatchProjector.ProjectionVersion, + where: pv.projection_name == "IntegrationBatchProjector", + select: pv.last_seen_event_number + ) + ) + + assert last_seen == 4 + end + + test "handle_batch maintains idempotency across multiple calls" do + events = [ + {%AnEvent{name: "idem_e1"}, + %{handler_name: "IntegrationBatchProjector", event_number: 10}}, + {%AnEvent{name: "idem_e2"}, + %{handler_name: "IntegrationBatchProjector", event_number: 11}} + ] + + assert :ok == IntegrationBatchProjector.handle_batch(events) + assert :ok == IntegrationBatchProjector.handle_batch(events) + + assert_projections(Projection, ["idem_e1", "idem_e2"]) + + last_seen = + Repo.one( + from(pv in IntegrationBatchProjector.ProjectionVersion, + where: pv.projection_name == "IntegrationBatchProjector", + select: pv.last_seen_event_number + ) + ) + + assert last_seen == 11 + end + end +end diff --git a/test/projections/ecto_projection_batch_test.exs b/test/projections/ecto_projection_batch_test.exs new file mode 100644 index 00000000..e5448b4d --- /dev/null +++ b/test/projections/ecto_projection_batch_test.exs @@ -0,0 +1,264 @@ +defmodule Commanded.Projections.EctoProjectionBatchTest do + use ExUnit.Case + + import Commanded.Projections.ProjectionAssertions + + alias Commanded.Projections.Events.{AnEvent, AnotherEvent, ErrorEvent} + alias Commanded.Projections.Projection + alias Commanded.Projections.Repo + alias Ecto.Adapters.SQL.Sandbox + + defmodule BatchProjector do + use Commanded.Projections.Ecto, + application: TestApplication, + name: "BatchProjector", + repo: Commanded.Projections.Repo, + batch_size: 1 + + project_batch fn events, multi -> + projections = + Enum.map(events, fn + {%AnEvent{name: name}, _metadata} -> %{name: name} + {%AnotherEvent{name: name}, _metadata} -> %{name: name} + {%ErrorEvent{}, _metadata} -> :error + end) + + if Enum.any?(projections, &(&1 == :error)) do + Ecto.Multi.error(multi, :projection, :failure) + else + Ecto.Multi.insert_all(multi, :projection, Projection, projections) + end + end + end + + setup do + start_supervised!(TestApplication) + Sandbox.checkout(Repo) + end + + test "should handle multiple projected events" do + assert :ok == + BatchProjector.handle_batch([ + {%AnEvent{}, %{handler_name: "BatchProjector", event_number: 1}}, + {%AnEvent{}, %{handler_name: "BatchProjector", event_number: 2}} + ]) + + assert_projections(Projection, ["AnEvent", "AnEvent"]) + assert_seen_event("BatchProjector", 2) + end + + test "should handle two different types of projected events" do + assert :ok == + BatchProjector.handle_batch([ + {%AnEvent{}, %{handler_name: "BatchProjector", event_number: 1}}, + {%AnotherEvent{}, %{handler_name: "BatchProjector", event_number: 2}} + ]) + + assert_projections(Projection, ["AnEvent", "AnotherEvent"]) + assert_seen_event("BatchProjector", 2) + end + + test "should ignore already projected batch" do + assert :ok == + BatchProjector.handle_batch([ + {%AnEvent{}, %{handler_name: "BatchProjector", event_number: 1}} + ]) + + assert :ok == + BatchProjector.handle_batch([ + {%AnEvent{}, %{handler_name: "BatchProjector", event_number: 1}} + ]) + + assert_projections(Projection, ["AnEvent"]) + assert_seen_event("BatchProjector", 1) + end + + test "partial batch already seen should return an :ok" do + assert :ok == + BatchProjector.handle_batch([ + {%AnEvent{name: "e1"}, %{handler_name: "BatchProjector", event_number: 1}}, + {%AnEvent{name: "e2"}, %{handler_name: "BatchProjector", event_number: 2}}, + {%AnEvent{name: "e3"}, %{handler_name: "BatchProjector", event_number: 3}} + ]) + + assert_projections(Projection, ["e1", "e2", "e3"]) + assert_seen_event("BatchProjector", 3) + + assert :ok == + BatchProjector.handle_batch([ + {%AnEvent{name: "e2"}, %{handler_name: "BatchProjector", event_number: 2}}, + {%AnEvent{name: "e3"}, %{handler_name: "BatchProjector", event_number: 3}}, + {%AnEvent{name: "e4"}, %{handler_name: "BatchProjector", event_number: 4}} + ]) + + assert_projections(Projection, ["e1", "e2", "e3", "e4"]) + assert_seen_event("BatchProjector", 4) + end + + test "entire batches already seen should return an :ok" do + assert :ok == + BatchProjector.handle_batch([ + {%AnEvent{name: "e1"}, %{handler_name: "BatchProjector", event_number: 1}}, + {%AnEvent{name: "e2"}, %{handler_name: "BatchProjector", event_number: 2}}, + {%AnEvent{name: "e3"}, %{handler_name: "BatchProjector", event_number: 3}} + ]) + + assert :ok == + BatchProjector.handle_batch([ + {%AnEvent{name: "e1"}, %{handler_name: "BatchProjector", event_number: 1}}, + {%AnEvent{name: "e2"}, %{handler_name: "BatchProjector", event_number: 2}}, + {%AnEvent{name: "e3"}, %{handler_name: "BatchProjector", event_number: 3}} + ]) + end + + test "should return an error on failure" do + assert {:error, :failure} == + BatchProjector.handle_batch([ + {%ErrorEvent{}, %{handler_name: "BatchProjector", event_number: 1}} + ]) + + assert_projections(Projection, []) + end + + defmodule BatchProjectorAfterUpdateCallback do + use Commanded.Projections.Ecto, + application: TestApplication, + name: "BatchProjectorAfterUpdateCallback", + repo: Commanded.Projections.Repo, + batch_size: 1 + + project_batch fn events, multi -> + projections = + Enum.map(events, fn + {%{name: name}, _metadata} -> %{name: name} + end) + + Ecto.Multi.insert_all(multi, :projection, Projection, projections) + end + + def after_update_batch(events, changes) do + {%{pid: pid}, _metadata} = List.first(events) + + send(pid, {:after_update_batch, length(events), changes}) + + :ok + end + end + + test "should call after_update_batch/2 callback" do + assert :ok == + BatchProjectorAfterUpdateCallback.handle_batch([ + {%AnEvent{pid: self()}, + %{handler_name: "BatchProjectorAfterUpdateCallback", event_number: 1}}, + {%AnEvent{pid: self()}, + %{handler_name: "BatchProjectorAfterUpdateCallback", event_number: 2}}, + {%AnEvent{pid: self()}, + %{handler_name: "BatchProjectorAfterUpdateCallback", event_number: 3}} + ]) + + assert_receive {:after_update_batch, 3, _changes} + end + + defmodule BatchProjectorCallbackWithChanges do + use Commanded.Projections.Ecto, + application: TestApplication, + name: "BatchProjectorCallbackWithChanges", + repo: Commanded.Projections.Repo, + batch_size: 1 + + project_batch fn events, multi -> + projections = + Enum.map(events, fn + {%{name: name}, _metadata} -> %{name: name} + end) + + Ecto.Multi.insert_all(multi, :projection, Projection, projections) + end + + def after_update_batch(events, changes) do + {%{pid: pid}, _metadata} = List.first(events) + + send(pid, {:after_update_batch_changes, changes}) + + :ok + end + end + + test "after_update_batch/2 callback receives correct changes from Ecto.Multi" do + assert :ok == + BatchProjectorCallbackWithChanges.handle_batch([ + {%AnEvent{pid: self()}, + %{handler_name: "BatchProjectorCallbackWithChanges", event_number: 1}}, + {%AnEvent{pid: self()}, + %{handler_name: "BatchProjectorCallbackWithChanges", event_number: 2}} + ]) + + assert_receive {:after_update_batch_changes, changes} + assert Map.has_key?(changes, :lock_and_filter) + assert Map.has_key?(changes, :track_projection_version) + assert Map.has_key?(changes, :prepare_user_multi) + assert Map.has_key?(changes, :projection) + end + + defmodule BatchProjectorCallbackReturnsError do + use Commanded.Projections.Ecto, + application: TestApplication, + name: "BatchProjectorCallbackReturnsError", + repo: Commanded.Projections.Repo, + batch_size: 1 + + project_batch fn events, multi -> + projections = + Enum.map(events, fn + {%{name: name}, _metadata} -> %{name: name} + end) + + Ecto.Multi.insert_all(multi, :projection, Projection, projections) + end + + def after_update_batch(_events, _changes) do + {:error, :callback_failed} + end + end + + test "after_update_batch/2 callback error does not rollback transaction" do + assert {:error, :callback_failed} == + BatchProjectorCallbackReturnsError.handle_batch([ + {%AnEvent{}, + %{handler_name: "BatchProjectorCallbackReturnsError", event_number: 1}} + ]) + + assert_projections(Projection, ["AnEvent"]) + end + + defmodule BatchProjectorCallbackRaises do + use Commanded.Projections.Ecto, + application: TestApplication, + name: "BatchProjectorCallbackRaises", + repo: Commanded.Projections.Repo, + batch_size: 1 + + project_batch fn events, multi -> + projections = + Enum.map(events, fn + {%{name: name}, _metadata} -> %{name: name} + end) + + Ecto.Multi.insert_all(multi, :projection, Projection, projections) + end + + def after_update_batch(_events, _changes) do + raise "Intentional callback exception" + end + end + + test "after_update_batch/2 callback exception does not rollback transaction" do + assert_raise RuntimeError, "Intentional callback exception", fn -> + BatchProjectorCallbackRaises.handle_batch([ + {%AnEvent{}, %{handler_name: "BatchProjectorCallbackRaises", event_number: 1}} + ]) + end + + assert_projections(Projection, ["AnEvent"]) + end +end diff --git a/test/projections/projection_version_schema_prefix_test.exs b/test/projections/projection_version_schema_prefix_test.exs index bf326f5e..834436ff 100644 --- a/test/projections/projection_version_schema_prefix_test.exs +++ b/test/projections/projection_version_schema_prefix_test.exs @@ -219,6 +219,119 @@ defmodule Commanded.Projections.ProjectionVersionSchemaPrefixTest do end end + describe "schema_prefix/1 callback (1-arity)" do + test "should return static schema prefix value for 1-arity callback" do + defmodule StaticPrefixOneArityProjector do + use Commanded.Projections.Ecto, + application: TestApplication, + name: "static_prefix_one_arity_projector", + schema_prefix: "my_static_prefix", + repo: Commanded.Projections.Repo + end + + event = %SchemaEvent{schema: "test"} + + # Test 1-arity callback directly + assert StaticPrefixOneArityProjector.schema_prefix(event) == "my_static_prefix" + + # Test 2-arity callback for consistency + assert StaticPrefixOneArityProjector.schema_prefix(event, %{}) == "my_static_prefix" + end + + test "should invoke 1-arity function for 1-arity callback" do + defmodule OneArityFunctionPrefixProjector do + use Commanded.Projections.Ecto, + application: TestApplication, + name: "one_arity_function_prefix_projector", + schema_prefix: fn %SchemaEvent{schema: schema} -> "dynamic_#{schema}" end, + repo: Commanded.Projections.Repo + end + + event = %SchemaEvent{schema: "tenant123"} + + # Test 1-arity callback directly + assert OneArityFunctionPrefixProjector.schema_prefix(event) == "dynamic_tenant123" + + # Test 2-arity callback for consistency + assert OneArityFunctionPrefixProjector.schema_prefix(event, %{}) == "dynamic_tenant123" + end + + test "should return nil for 1-arity callback when schema_prefix is 2-arity function" do + defmodule TwoArityFunctionPrefixProjector do + use Commanded.Projections.Ecto, + application: TestApplication, + name: "two_arity_function_prefix_projector", + schema_prefix: fn _event, %{tenant: tenant} -> tenant end, + repo: Commanded.Projections.Repo + end + + event = %SchemaEvent{schema: "test"} + + # 1-arity should return nil (can't invoke 2-arity function) + assert TwoArityFunctionPrefixProjector.schema_prefix(event) == nil + + # 2-arity should invoke the function + assert TwoArityFunctionPrefixProjector.schema_prefix(event, %{tenant: "tenant456"}) == + "tenant456" + end + + test "should return nil for 1-arity callback when no schema_prefix configured" do + defmodule NoPrefixOneArityProjector do + use Commanded.Projections.Ecto, + application: TestApplication, + name: "no_prefix_one_arity_projector", + repo: Commanded.Projections.Repo + end + + event = %SchemaEvent{schema: "test"} + + # Both should return nil + assert NoPrefixOneArityProjector.schema_prefix(event) == nil + assert NoPrefixOneArityProjector.schema_prefix(event, %{}) == nil + end + + test "1-arity callback with 1-arity function should handle different events" do + defmodule PerEventOneArityProjector do + use Commanded.Projections.Ecto, + application: TestApplication, + name: "per_event_one_arity_projector", + schema_prefix: fn %SchemaEvent{schema: schema} -> "prefix_#{schema}" end, + repo: Commanded.Projections.Repo + end + + # Test that 1-arity callback returns correct value for different events + assert PerEventOneArityProjector.schema_prefix(%SchemaEvent{schema: "tenant1"}) == + "prefix_tenant1" + + assert PerEventOneArityProjector.schema_prefix(%SchemaEvent{schema: "tenant2"}) == + "prefix_tenant2" + + assert PerEventOneArityProjector.schema_prefix(%SchemaEvent{schema: "tenant3"}) == + "prefix_tenant3" + end + + test "1-arity callback should work with app config schema prefix" do + Application.put_env(:commanded, Commanded.Projections.Ecto, + schema_prefix: "app_config_prefix" + ) + + defmodule AppConfigOneArityProjector do + use Commanded.Projections.Ecto, + application: TestApplication, + name: "app_config_one_arity_projector", + repo: Commanded.Projections.Repo + end + + event = %SchemaEvent{schema: "test"} + + # Test 1-arity callback directly + assert AppConfigOneArityProjector.schema_prefix(event) == "app_config_prefix" + + # Test 2-arity callback for consistency + assert AppConfigOneArityProjector.schema_prefix(event, %{}) == "app_config_prefix" + end + end + defp assert_schema_prefix(projector, expected_prefix) do prefix = schema_prefix(projector, %SchemaEvent{}, %{})