Repository: commanded/commanded
Pull request #263 (2019-03-16 22:11:45)

Implement event upcasting.
A quick stab at #258


Patch:
diff --git a/CHANGELOG.md b/CHANGELOG.md
index f928a16..7dbe65f 100644
--- a/CHANGELOG.md
+++ b/CHANGELOG.md
@@ -5,7 +5,6 @@
 ### Enhancements
 
 - Update typespec for `data` and `metadata` fields in `Commanded.EventStore.EventData` struct ([#246](https://github.com/commanded/commanded/pull/246)).
-- Add `include_execution_result` and `aggregate_version` to typespec for router command dispatch ([#262](https://github.com/commanded/commanded/pull/262)).
 - Add `.formatter.exs` to Hex package ([#247](https://github.com/commanded/commanded/pull/247)).
 
 ## v0.18.0
diff --git a/lib/commanded/event/handler.ex b/lib/commanded/event/handler.ex
index 40311b6..b220e76 100644
--- a/lib/commanded/event/handler.ex
+++ b/lib/commanded/event/handler.ex
@@ -423,10 +423,14 @@ defmodule Commanded.Event.Handler do
 
   @doc false
   def handle_info({:events, events}, %Handler{} = state) do
+    alias Commanded.Event.Upcast
     Logger.debug(fn -> describe(state) <> " received events: #{inspect(events)}" end)
 
     try do
-      state = Enum.reduce(events, state, &handle_event/2)
+      state =
+        events
+        |> Upcast.upcast_event_stream()
+        |> Enum.reduce(state, &handle_event/2)
 
       {:noreply, state}
     catch
diff --git a/lib/commanded/event/upcast.ex b/lib/commanded/event/upcast.ex
new file mode 100644
index 0000000..2d2cf8a
--- /dev/null
+++ b/lib/commanded/event/upcast.ex
@@ -0,0 +1,16 @@
+defmodule Commanded.Event.Upcast do
+  alias Commanded.Event.Upcaster
+  alias Commanded.EventStore.RecordedEvent
+
+  def upcast_event_stream(%Stream{} = event_stream) do
+    Stream.map(event_stream, &upcast_event/1)
+  end
+
+  def upcast_event_stream(event_stream) do
+    Enum.map(event_stream, &upcast_event/1)
+  end
+
+  defp upcast_event(%RecordedEvent{data: data, metadata: metadata} = event) do
+    %{event | data: Upcaster.upcast(data, metadata)}
+  end
+end
diff --git a/lib/commanded/event/upcaster.ex b/lib/commanded/event/upcaster.ex
new file mode 100644
index 0000000..f5dbb76
--- /dev/null
+++ b/lib/commanded/event/upcaster.ex
@@ -0,0 +1,27 @@
+defprotocol Commanded.Event.Upcaster do
+  @doc """
+  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).
+
+  Because the upcaster changes any historical event to the latest version,
+  consumers (aggregates, event handlers, and process managers) only need
+  to support the latest version.
+  """
+
+  @fallback_to_any true
+  @spec upcast(event :: struct(), metadata :: map()) :: 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.
+  """
+
+  def upcast(event, _metadata), do: event
+end
diff --git a/lib/commanded/event_store/event_store.ex b/lib/commanded/event_store/event_store.ex
index af7303e..c022799 100644
--- a/lib/commanded/event_store/event_store.ex
+++ b/lib/commanded/event_store/event_store.ex
@@ -170,7 +170,12 @@ defmodule Commanded.EventStore do
 
   def stream_forward(stream_uuid, start_version, read_batch_size)
       when is_binary(stream_uuid) and is_integer(start_version) and is_integer(read_batch_size) do
-    event_store_adapter().stream_forward(stream_uuid, start_version, read_batch_size)
+    alias Commanded.Event.Upcast
+
+    case event_store_adapter().stream_forward(stream_uuid, start_version, read_batch_size) do
+      {:error, _} = error -> error
+      stream -> Upcast.upcast_event_stream(stream)
+    end
   end
 
   @doc """
diff --git a/lib/commanded/process_managers/process_router.ex b/lib/commanded/process_managers/process_router.ex
index ad898b0..5960347 100644
--- a/lib/commanded/process_managers/process_router.ex
+++ b/lib/commanded/process_managers/process_router.ex
@@ -170,11 +170,16 @@ defmodule Commanded.ProcessManagers.ProcessRouter do
 
   @doc false
   def handle_info({:events, events}, %State{} = state) do
+    alias Commanded.Event.Upcast
+
     Logger.debug(fn -> describe(state) <> " received #{length(events)} event(s)" end)
 
     %State{pending_events: pending_events} = state
 
-    unseen_events = Enum.reject(events, &event_already_seen?(&1, state))
+    unseen_events =
+      events
+      |> Upcast.upcast_event_stream()
+      |> Enum.reject(&event_already_seen?(&1, state))
 
     state =
       case {pending_events, unseen_events} do


Test patch:
diff --git a/test/event/support/upcast/event_handler.ex b/test/event/support/upcast/event_handler.ex
new file mode 100644
index 0000000..a7dfead
--- /dev/null
+++ b/test/event/support/upcast/event_handler.ex
@@ -0,0 +1,15 @@
+defmodule Commanded.Event.Upcast.EventHandler do
+  use Commanded.Event.Handler, name: __MODULE__
+
+  alias Commanded.Event.Upcast.Events.{EventOne, EventTwo, EventThree, EventFour}
+
+  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)
+
+  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
new file mode 100644
index 0000000..9465090
--- /dev/null
+++ b/test/event/support/upcast/events.ex
@@ -0,0 +1,39 @@
+defmodule Commanded.Event.Upcast.Events do
+  alias Commanded.Event.Upcaster
+
+  defmodule EventOne do
+    @derive Jason.Encoder
+    defstruct [:n, :reply_to, :process_id]
+  end
+
+  defmodule EventTwo do
+    @derive Jason.Encoder
+    defstruct [:n, :reply_to, :process_id]
+
+    defimpl Upcaster do
+      def upcast(%{n: n} = event, _metadata), do: %{event | n: n * 2}
+    end
+  end
+
+  defmodule EventThree do
+    @derive Jason.Encoder
+    defstruct [:n, :reply_to, :process_id]
+  end
+
+  defmodule EventFour do
+    @derive Jason.Encoder
+    defstruct [:n, :name, :reply_to, :process_id]
+
+    defimpl Upcaster, for: EventThree do
+      def upcast(event, _metadata) do
+        data = Map.from_struct(event) |> Map.put(:name, "Chris")
+        struct(EventFour, data)
+      end
+    end
+  end
+
+  defmodule Stop do
+    @derive Jason.Encoder
+    defstruct [:process_id]
+  end
+end
diff --git a/test/event/support/upcast/process_manager.ex b/test/event/support/upcast/process_manager.ex
new file mode 100644
index 0000000..a4be3b5
--- /dev/null
+++ b/test/event/support/upcast/process_manager.ex
@@ -0,0 +1,39 @@
+defmodule Commanded.Event.Upcast.ProcessManager do
+  defmodule(Ok, do: defstruct([:id]))
+
+  defmodule Aggregate do
+    defstruct []
+    def execute(_, _), do: []
+    def apply(agg, _), do: agg
+  end
+
+  defmodule Router do
+    use Commanded.Commands.Router
+    dispatch Ok, to: Aggregate, identity: :id
+  end
+
+  @derive Jason.Encoder
+  defstruct []
+
+  alias Commanded.Event.Upcast.Events.{EventOne, EventTwo, EventThree, EventFour, Stop}
+
+  use Commanded.ProcessManagers.ProcessManager,
+    name: "UpcastProcessManager",
+    router: Router
+
+  def interested?(%EventOne{process_id: process_id}), do: {:start, process_id}
+  def interested?(%EventTwo{process_id: process_id}), do: {:start, process_id}
+  def interested?(%EventThree{process_id: process_id}), do: {:start, process_id}
+  def interested?(%EventFour{process_id: process_id}), do: {:start, process_id}
+  def interested?(%Stop{process_id: process_id}), do: {:stop, process_id}
+
+  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)
+
+  defp send_reply(%{reply_to: reply_to} = e) do
+    send(:erlang.list_to_pid(reply_to), e)
+    %Ok{id: UUID.uuid4()}
+  end
+end
diff --git a/test/event/upcaster_test.exs b/test/event/upcaster_test.exs
new file mode 100644
index 0000000..4cb134c
--- /dev/null
+++ b/test/event/upcaster_test.exs
@@ -0,0 +1,128 @@
+defmodule Event.UpcasterTest do
+  use Commanded.StorageCase
+
+  alias Commanded.EventStore
+  alias Commanded.EventStore.EventData
+  alias Commanded.Event.Upcast.Events.{EventOne, EventTwo, EventThree, EventFour, Stop}
+
+  describe "event store stream forward" do
+    setup do
+      %{
+        stream_uuid: UUID.uuid4(:hex)
+      }
+    end
+
+    test "will not upcast an event without an upcaster", %{stream_uuid: stream_uuid} do
+      %{data: %{n: n}} =
+        stream_uuid
+        |> write_events(struct(EventOne, n: 0))
+        |> read_event()
+
+      assert n == 0
+    end
+
+    test "will upcast using defined upcaster", %{stream_uuid: stream_uuid} do
+      %{data: %{n: n}} =
+        stream_uuid
+        |> write_events(struct(EventTwo, n: 10))
+        |> read_event()
+
+      assert n == 20
+    end
+
+    test "can adapt new event from old event", %{stream_uuid: stream_uuid} do
+      %{data: %EventFour{name: name}} =
+        stream_uuid
+        |> write_events(struct(EventThree, n: 0))
+        |> read_event()
+
+      assert name == "Chris"
+    end
+  end
+
+  describe "event handler" do
+    alias Commanded.Event.Upcast.EventHandler
+
+    setup do
+      {:ok, handler} = EventHandler.start_link()
+
+      %{
+        stream_uuid: UUID.uuid4(:hex),
+        handler: handler,
+        reply_to: :erlang.pid_to_list(self())
+      }
+    end
+
+    test "will receive upcasted events", %{stream_uuid: stream_uuid, reply_to: reply_to} do
+      stream_uuid
+      |> write_events([
+        struct(EventOne, n: 0, reply_to: reply_to),
+        struct(EventTwo, n: 10, reply_to: reply_to),
+        struct(EventThree, n: 0, reply_to: reply_to)
+      ])
+
+      refute_receive %EventThree{}
+
+      assert_receive %EventOne{n: 0}
+      assert_receive %EventTwo{n: 20}
+      assert_receive %EventFour{name: "Chris"}
+    end
+  end
+
+  describe "process manager" do
+    alias Commanded.Event.Upcast.ProcessManager
+
+    setup do
+      {:ok, manager} = ProcessManager.start_link()
+
+      %{
+        stream_uuid: UUID.uuid4(:hex),
+        manager: manager,
+        reply_to: :erlang.pid_to_list(self()),
+        process_id: UUID.uuid4(:hex)
+      }
+    end
+
+    test "will receive upcasted events", context do
+      %{stream_uuid: stream_uuid, reply_to: reply_to, process_id: process_id} = context
+
+      stream_uuid
+      |> write_events([
+        struct(EventOne, n: 0, reply_to: reply_to, process_id: process_id),
+        struct(EventTwo, n: 10, reply_to: reply_to, process_id: process_id),
+        struct(EventThree, n: 0, reply_to: reply_to, process_id: process_id),
+        struct(Stop, process_id: process_id)
+      ])
+
+      refute_receive %EventThree{}
+
+      assert_receive %EventOne{n: 0}
+      assert_receive %EventTwo{n: 20}
+      assert_receive %EventFour{name: "Chris"}
+    end
+  end
+
+  defp write_events(stream_uuid, events) do
+    EventStore.append_to_stream(
+      stream_uuid,
+      :any_version,
+      events |> List.wrap() |> Enum.map(&create_event/1)
+    )
+
+    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(stream_uuid)
+    |> Enum.into([])
+    |> hd()
+  end
+end