diff --git a/guides/explanations/events.md b/guides/explanations/events.md index 17f73204..be582123 100644 --- a/guides/explanations/events.md +++ b/guides/explanations/events.md @@ -223,38 +223,6 @@ You request the consistency guarantee, either `:strong` or `:eventual`, when dis An event handler is a `GenServer` process that subscribes to the configured event store. For each event persisted to the store the `handle/2` callback is called, passing the domain event and its metadata. -## Upcasting events - -Commanded supports upcasting of events at runtime using the `Commanded.Event.Upcaster` protocol. - -By implementing the upcaster protocol you can transform an event before it is used by a consumer. This might be an aggregate or an event handler. Because the upcaster changes the event at runtime, handlers only need to support the latest version. You can also use upcasting to change the type of event. - -### Examples - -Change the shape of an event by renaming a field: - -```elixir -defimpl Commanded.Event.Upcaster, for: AnEvent do - def upcast(%AnEvent{} = event, _metadata) do - %AnEvent{name: name} = event - - %AnEvent{event | first_name: name} - end -end -``` - -Change the type of event by replacing a historical event with a new event: - -```elixir -defimpl Commanded.Event.Upcaster, for: HistoricalEvent do - def upcast(%HistoricalEvent{} = event, _metadata) do - %HistoricalEvent{id: id, name: name} = event - - %NewEvent{id: id, name: name} - end -end -``` - ## Reset an EventHandler An event handler can be reset (using a mix task), it will restart the event store subscription from the configured diff --git a/guides/explanations/fork-differences.md b/guides/explanations/fork-differences.md index 7d085f8c..19b27758 100644 --- a/guides/explanations/fork-differences.md +++ b/guides/explanations/fork-differences.md @@ -11,9 +11,9 @@ This fork maintains an independent release cycle to introduce new features and i --- -## Breaking Changes +## Breaking Changes ❌ -### **Removed ProcessManager Support** ❌ +### **Removed ProcessManager Support** [PR #24](https://github.com/straw-hat-team/commanded/pull/24) **Changes:** @@ -26,7 +26,18 @@ This fork maintains an independent release cycle to introduce new features and i --- -### **EnrichedMetadata Struct** 🔄 +### **Removed Upcasting Support** + +**Changes:** +- Removed `Commanded.Event.Upcast` and `Commanded.Event.Upcaster` modules +- Removed upcasting from event handlers and aggregates +- Removed upcasting documentation + +**Reason:** Event schema transformations can be handled explicitly in event handlers and aggregates using pattern matching, eliminating the need for a global upcasting component. + +--- + +### **EnrichedMetadata Struct** [PR #11](https://github.com/straw-hat-team/commanded/pull/11) **Changes:** @@ -37,9 +48,9 @@ This fork maintains an independent release cycle to introduce new features and i --- -## Features Added +## Features Added ✨ -### **UUIDv7 Support** ✨ +### **UUIDv7 Support** [PR #22](https://github.com/straw-hat-team/commanded/pull/22) **Changes:** @@ -48,7 +59,7 @@ This fork maintains an independent release cycle to introduce new features and i --- -### **Configurable UUID Provider** ✨ +### **Configurable UUID Provider** **Changes:** - Integrated `Uniq.UUID` library for UUID generation @@ -57,7 +68,7 @@ This fork maintains an independent release cycle to introduce new features and i --- -### **Aggregate Telemetry for Version Conflicts** ✨ +### **Aggregate Telemetry for Version Conflicts** [PR #8](https://github.com/straw-hat-team/commanded/pull/8) **Changes:** @@ -66,7 +77,7 @@ This fork maintains an independent release cycle to introduce new features and i --- -### **Custom Event ID Support** ✨ +### **Custom Event ID Support** [PR #2](https://github.com/straw-hat-team/commanded/pull/2) **Changes:** @@ -76,7 +87,7 @@ This fork maintains an independent release cycle to introduce new features and i --- -### **EventStore Adapter** ✨ +### **EventStore Adapter** [PR #1](https://github.com/straw-hat-team/commanded/pull/1) **Changes:** diff --git a/guides/howtos/migrating-from-v1-to-v2.md b/guides/howtos/migrating-from-v1-to-v2.md index 6e6b6e59..543d19b3 100644 --- a/guides/howtos/migrating-from-v1-to-v2.md +++ b/guides/howtos/migrating-from-v1-to-v2.md @@ -94,18 +94,6 @@ end end ``` -4. **Update Upcaster implementations:** - - ```elixir - defimpl Commanded.Event.Upcaster, for: MyEvent do - alias Commanded.EventStore.EnrichedMetadata - - def upcast(event, %EnrichedMetadata{} = metadata) do - # Implementation - end - end - ``` - ## System Metadata Fields The following fields are now directly accessible on the `EnrichedMetadata` struct: diff --git a/lib/commanded/aggregates/aggregate.ex b/lib/commanded/aggregates/aggregate.ex index 82613710..37e7a514 100644 --- a/lib/commanded/aggregates/aggregate.ex +++ b/lib/commanded/aggregates/aggregate.ex @@ -119,7 +119,6 @@ defmodule Commanded.Aggregates.Aggregate do alias Commanded.Aggregates.ExecutionContext alias Commanded.Application.Config alias Commanded.Event.Mapper - alias Commanded.Event.Upcast alias Commanded.EventStore alias Commanded.EventStore.RecordedEvent alias Commanded.Registration @@ -392,15 +391,12 @@ defmodule Commanded.Aggregates.Aggregate do @doc false @impl GenServer def handle_info({:events, events}, %Aggregate{} = state) do - %Aggregate{application: application} = state - Logger.debug(describe(state) <> " received events: " <> inspect(events)) try do state = events |> Enum.reject(&event_already_seen?(&1, state)) - |> Upcast.upcast_event_stream(additional_metadata: %{application: application}) |> Enum.reduce(state, &handle_event/2) noreply_with_lifespan(state) diff --git a/lib/commanded/event/handler.ex b/lib/commanded/event/handler.ex index 972050ea..3253a1e0 100644 --- a/lib/commanded/event/handler.ex +++ b/lib/commanded/event/handler.ex @@ -407,7 +407,6 @@ defmodule Commanded.Event.Handler do alias Commanded.Event.ErrorHandler alias Commanded.Event.FailureContext alias Commanded.Event.Handler - alias Commanded.Event.Upcast alias Commanded.EventStore.RecordedEvent alias Commanded.EventStore.Subscription alias Commanded.Subscriptions @@ -952,7 +951,7 @@ defmodule Commanded.Event.Handler do @doc false @impl GenServer def handle_info({:events, events}, state) do - %Handler{application: application, handler_callback: callback} = state + %Handler{handler_callback: callback} = state processor = case callback do @@ -963,10 +962,7 @@ defmodule Commanded.Event.Handler do Logger.debug(describe(state) <> " received events: #{inspect(events)}") try do - state = - events - |> Upcast.upcast_event_stream(additional_metadata: %{application: application}) - |> processor.(state) + state = processor.(events, state) {:noreply, state} catch @@ -1417,8 +1413,7 @@ defmodule Commanded.Event.Handler do handler_name, handler_module ) do - %RecordedEvent{data: data} = - event = Upcast.upcast_event(event, additional_metadata: %{application: application}) + %RecordedEvent{data: data} = event metadata = RecordedEvent.enrich_metadata(event, diff --git a/lib/commanded/event/upcast.ex b/lib/commanded/event/upcast.ex deleted file mode 100644 index 3bc71adb..00000000 --- a/lib/commanded/event/upcast.ex +++ /dev/null @@ -1,22 +0,0 @@ -defmodule Commanded.Event.Upcast do - @moduledoc false - - alias Commanded.Event.Upcaster - alias Commanded.EventStore.RecordedEvent - - def upcast_event_stream(event_stream, opts \\ []) - - def upcast_event_stream(%Stream{} = event_stream, opts), - do: Stream.map(event_stream, &upcast_event(&1, opts)) - - def upcast_event_stream(event_stream, opts), - do: Enum.map(event_stream, &upcast_event(&1, opts)) - - def upcast_event(%RecordedEvent{} = event, opts) do - %RecordedEvent{data: data} = event - - enriched_metadata = RecordedEvent.enrich_metadata(event, opts) - - %RecordedEvent{event | data: Upcaster.upcast(data, enriched_metadata)} - end -end diff --git a/lib/commanded/event/upcaster.ex b/lib/commanded/event/upcaster.ex deleted file mode 100644 index ddf8e2c4..00000000 --- a/lib/commanded/event/upcaster.ex +++ /dev/null @@ -1,67 +0,0 @@ -defprotocol Commanded.Event.Upcaster do - @moduledoc """ - Protocol to allow an event to be transformed before being passed to a - consumer. - - You can use an upcaster to change the shape of an event (e.g. add a new field - with a default, rename a field) or rename an event. - - Upcaster will run for new events and for historical events. - - Because the upcaster changes any historical event to the latest version, - consumers (aggregates and event handlers) only need - to support the latest version. - - ## Example - - alias Commanded.EventStore.EnrichedMetadata - - defimpl Commanded.Event.Upcaster, for: AnEvent do - def upcast(%AnEvent{} = event, %EnrichedMetadata{} = metadata) do - %AnEvent{name: name} = event - - %AnEvent{event | first_name: name} - end - end - - ## Metadata - - The `upcast/2` function receives the domain event and an `EnrichedMetadata` struct - associated with that event. - - The `EnrichedMetadata` struct contains the following fields: - - - `application` - the `Commanded.Application` used to read the event. - - `event_id` - a globally unique UUID to identify the event. - - `event_number` - a globally unique, monotonically incrementing integer - used to order the event amongst all events. - - `stream_id` - the stream identity for the event. - - `stream_version` - the version of the stream for the event. - - `causation_id` - an optional UUID identifier used to identify which - command caused the event. - - `correlation_id` - an optional UUID identifier used to correlate related - commands/events. - - `created_at` - the datetime, in UTC, indicating when the event was - created. - - `metadata` - the user-provided metadata as a string-keyed map. - - """ - - alias Commanded.EventStore.EnrichedMetadata - - @fallback_to_any true - @spec upcast(event :: struct(), metadata :: EnrichedMetadata.t()) :: struct() - def upcast(event, metadata) -end - -defimpl Commanded.Event.Upcaster, for: Any do - @moduledoc """ - The default implementation of the `Commanded.Event.Upcaster`. - - This will return an event unchanged. - """ - - alias Commanded.EventStore.EnrichedMetadata - - def upcast(event, %EnrichedMetadata{}), do: event -end diff --git a/lib/commanded/event_store.ex b/lib/commanded/event_store.ex index ec356f48..d2a5f9f3 100644 --- a/lib/commanded/event_store.ex +++ b/lib/commanded/event_store.ex @@ -25,7 +25,6 @@ defmodule Commanded.EventStore do """ alias Commanded.Application - alias Commanded.Event.Upcast @type application :: Application.t() @type config :: Keyword.t() @@ -80,7 +79,7 @@ defmodule Commanded.EventStore do error stream -> - Upcast.upcast_event_stream(stream, additional_metadata: %{application: application}) + stream end end) end diff --git a/mix.exs b/mix.exs index a0a73720..2d374290 100644 --- a/mix.exs +++ b/mix.exs @@ -131,7 +131,6 @@ defmodule Commanded.Mixfile do Commanded.Event.FailureContext, Commanded.Event.Handler, Commanded.Event.Mapper, - Commanded.Event.Upcaster, Commanded.Event.EventId ], "Event Store": [ diff --git a/test/event/support/upcast/batch_event_handler.ex b/test/event/support/upcast/batch_event_handler.ex deleted file mode 100644 index ff0a5641..00000000 --- a/test/event/support/upcast/batch_event_handler.ex +++ /dev/null @@ -1,13 +0,0 @@ -defmodule Commanded.Event.Upcast.BatchEventHandler do - use Commanded.Event.Handler, - application: Commanded.DefaultApp, - name: __MODULE__, - batch_size: 5 - - def handle_batch([{_first, _metadata} | _rest] = events), do: send_reply(events) - - defp send_reply([{%{reply_to: reply_to}, _metadata} | _rest] = events) do - send(:erlang.list_to_pid(reply_to), events) - :ok - end -end diff --git a/test/event/support/upcast/event_handler.ex b/test/event/support/upcast/event_handler.ex deleted file mode 100644 index ee7bbf47..00000000 --- a/test/event/support/upcast/event_handler.ex +++ /dev/null @@ -1,18 +0,0 @@ -defmodule Commanded.Event.Upcast.EventHandler do - use Commanded.Event.Handler, - application: Commanded.DefaultApp, - name: __MODULE__ - - alias Commanded.Event.Upcast.Events.{EventFive, EventFour, EventOne, EventThree, EventTwo} - - def handle(%EventOne{} = e, _), do: send_reply(e) - def handle(%EventTwo{} = e, _), do: send_reply(e) - def handle(%EventThree{} = e, _), do: send_reply(e) - def handle(%EventFour{} = e, _), do: send_reply(e) - def handle(%EventFive{} = e, _), do: send_reply(e) - - defp send_reply(%{reply_to: reply_to} = e) do - send(:erlang.list_to_pid(reply_to), e) - :ok - end -end diff --git a/test/event/support/upcast/events.ex b/test/event/support/upcast/events.ex deleted file mode 100644 index fb413e5c..00000000 --- a/test/event/support/upcast/events.ex +++ /dev/null @@ -1,54 +0,0 @@ -defmodule Commanded.Event.Upcast.Events do - alias Commanded.Event.Upcaster - alias Commanded.EventStore.EnrichedMetadata - - defmodule EventOne do - @derive Jason.Encoder - defstruct [:version, :reply_to, :process_id] - end - - defmodule EventTwo do - @derive Jason.Encoder - defstruct [:version, :reply_to, :process_id] - - defimpl Upcaster do - def upcast(%EventTwo{} = event, %EnrichedMetadata{}) do - %EventTwo{event | version: 2} - end - end - end - - defmodule EventThree do - @derive Jason.Encoder - defstruct [:version, :reply_to, :process_id] - end - - defmodule EventFour do - @derive Jason.Encoder - defstruct [:version, :name, :reply_to, :process_id] - end - - defmodule EventFive do - @derive Jason.Encoder - defstruct [:version, :reply_to, :process_id, :event_metadata] - - defimpl Upcaster do - def upcast(%EventFive{} = event, %EnrichedMetadata{} = metadata) do - %EventFive{event | version: 2, event_metadata: metadata} - end - end - end - - defimpl Upcaster, for: EventThree do - def upcast(%EventThree{} = event, %EnrichedMetadata{}) do - data = Map.from_struct(event) |> Map.put(:name, "Chris") |> Map.put(:version, 2) - - struct(EventFour, data) - end - end - - defmodule Stop do - @derive Jason.Encoder - defstruct [:process_id] - end -end diff --git a/test/event/support/upcast/upcast_aggregate.ex b/test/event/support/upcast/upcast_aggregate.ex deleted file mode 100644 index 8938c715..00000000 --- a/test/event/support/upcast/upcast_aggregate.ex +++ /dev/null @@ -1,15 +0,0 @@ -defmodule Commanded.Event.Upcast.UpcastAggregate do - @moduledoc false - defstruct [:id] - - alias Commanded.Event.Upcast.Events.EventFour - alias Commanded.Event.Upcast.UpcastAggregate - - def apply(%UpcastAggregate{} = aggregate, %EventFour{} = event) do - %EventFour{reply_to: reply_to} = event - - send(:erlang.list_to_pid(reply_to), event) - - aggregate - end -end diff --git a/test/event/upcaster_test.exs b/test/event/upcaster_test.exs deleted file mode 100644 index 2fa81fd4..00000000 --- a/test/event/upcaster_test.exs +++ /dev/null @@ -1,182 +0,0 @@ -defmodule Event.UpcasterTest do - use ExUnit.Case - - alias Commanded.Aggregates.Aggregate - alias Commanded.{DefaultApp, EventStore} - alias Commanded.EventStore.{EventData, RecordedEvent} - alias Commanded.Event.Upcast.Events.{EventFive, EventFour, EventOne, EventThree, EventTwo} - alias Commanded.Event.Upcast.UpcastAggregate - alias Commanded.UUID - - setup do - start_supervised!(DefaultApp) - - :ok - end - - describe "upcast events from event store stream forward" do - test "will not upcast an event without an upcaster" do - assert %EventOne{version: version} = - write_events(DefaultApp, struct(EventOne, version: 1)) |> read_event() - - assert version == 1 - end - - test "will upcast using defined upcaster" do - assert %EventTwo{version: version} = - write_events(DefaultApp, struct(EventTwo, version: 1)) |> read_event() - - assert version == 2 - end - - test "upcasted event can include event metadata" do - assert %EventFive{event_metadata: metadata} = - write_events(DefaultApp, struct(EventFive, version: 1)) |> read_event() - - assert_metadata(metadata) - end - - test "can adapt new event from old event" do - assert %EventFour{name: name, version: version} = - write_events(DefaultApp, struct(EventThree, version: 1)) |> read_event() - - assert name == "Chris" - assert version == 2 - end - end - - describe "upcast events received by aggregate" do - test "should receive upcasted events" do - {:ok, pid} = - Aggregate.start_link([application: DefaultApp], - aggregate_module: UpcastAggregate, - aggregate_uuid: "upcast" - ) - - event = %RecordedEvent{ - event_id: UUID.uuid4(), - event_number: 1, - stream_id: "upcast", - stream_version: 1, - data: %EventThree{version: 1, reply_to: :erlang.pid_to_list(self())}, - metadata: %{} - } - - # Send event which needs to be upcast - send(pid, {:events, [event]}) - - assert_receive %EventFour{version: 2, name: "Chris"} - end - end - - describe "upcast events received by event handler" do - alias Commanded.Event.Upcast.EventHandler - - setup do - handler = start_supervised!(EventHandler) - - [ - handler: handler, - reply_to: :erlang.pid_to_list(self()) - ] - end - - test "will receive upcasted events", %{reply_to: reply_to} do - write_events( - DefaultApp, - [ - struct(EventOne, version: 1, reply_to: reply_to), - struct(EventTwo, version: 1, reply_to: reply_to), - struct(EventThree, version: 1, reply_to: reply_to), - struct(EventFive, version: 1, reply_to: reply_to) - ] - ) - - assert_receive %EventOne{version: 1} - assert_receive %EventTwo{version: 2} - assert_receive %EventFour{version: 2, name: "Chris"} - assert_receive %EventFive{version: 2, event_metadata: metadata} - refute_receive %EventThree{} - - assert_metadata(metadata) - end - end - - describe "upcast events received by batch event handler" do - alias Commanded.Event.Upcast.BatchEventHandler - - setup do - handler = start_supervised!(BatchEventHandler) - - [ - handler: handler, - reply_to: :erlang.pid_to_list(self()) - ] - end - - test "will receive upcasted events", %{reply_to: reply_to} do - write_events( - DefaultApp, - [ - struct(EventOne, version: 1, reply_to: reply_to), - struct(EventTwo, version: 1, reply_to: reply_to), - struct(EventThree, version: 1, reply_to: reply_to), - struct(EventFive, version: 1, reply_to: reply_to) - ] - ) - - assert_receive [{%EventOne{version: 1}, metadata1}] - assert_receive [{%EventTwo{version: 2}, metadata2}] - assert_receive [{%EventFour{version: 2, name: "Chris"}, metadata4}] - assert_receive [{%EventFive{version: 2, event_metadata: event_metadata}, metadata5}] - refute_receive [{%EventThree{}, _metadata3}] - - assert_metadata(metadata1) - assert_metadata(metadata2) - assert_metadata(metadata4) - assert_metadata(metadata5) - assert_metadata(event_metadata) - end - end - - defp assert_metadata(metadata) do - assert %{ - application: Commanded.DefaultApp, - causation_id: nil, - correlation_id: nil, - created_at: %DateTime{}, - event_id: event_id, - event_number: event_number, - stream_id: stream_id, - stream_version: stream_version - } = metadata - - assert is_binary(event_id) - assert is_number(event_number) - assert is_binary(stream_id) - assert is_number(stream_version) - end - - defp write_events(application, events) do - stream_uuid = UUID.uuid4() - events = events |> List.wrap() |> Enum.map(&create_event/1) - - :ok = EventStore.append_to_stream(application, stream_uuid, :any_version, events) - - stream_uuid - end - - defp create_event(%{__struct__: event_type} = data) do - %EventData{ - event_type: to_string(event_type), - data: data, - metadata: %{} - } - end - - defp read_event(stream_uuid) do - EventStore.stream_forward(DefaultApp, stream_uuid) - |> Enum.at(0) - |> Map.get(:data) - end -end