Refactor Default

Author: robacourt

packages/sync-service/lib/electric/shapes/consumer/event_handler/default.ex

@@ -6,7 +6,6 @@ defmodule Electric.Shapes.Consumer.EventHandler.Default do
   alias Electric.Replication.Changes
   alias Electric.Replication.Changes.Transaction
   alias Electric.Shapes.Consumer.LogOp
-  alias Electric.Shapes.Consumer.Materializer
   alias Electric.Shapes.Consumer.Plan
   alias Electric.Shapes.Shape
   alias Electric.Utils.ResultStream
@@ -28,28 +27,16 @@ defmodule Electric.Shapes.Consumer.EventHandler.Default do
         last_log_offset: last_log_offset,
         changes: changes
       }) do
-    extra_refs =
-      if Shape.are_deps_filled(state.shape) do
-        refs = Materializer.get_all_as_refs(state.shape, state.stack_id)
-        {refs, refs}
-      end
-
-    result =
-      changes
-      |> Stream.map(fn
-        %Changes.TruncatedRelation{} -> {:error, :truncate}
-        change -> {:ok, change}
-      end)
-      |> ResultStream.flat_map(
-        &Shape.convert_change(state.shape, &1,
-          stack_id: state.stack_id,
-          shape_handle: state.shape_handle,
-          extra_refs: extra_refs
-        )
+    changes
+    |> Stream.map(&error_on_truncate/1)
+    |> ResultStream.flat_map(
+      &Shape.convert_change(state.shape, &1,
+        stack_id: state.stack_id,
+        shape_handle: state.shape_handle
       )
-      |> ResultStream.to_list()
-
-    case result do
+    )
+    |> ResultStream.to_list()
+    |> case do
       {:error, :truncate} ->
         {:stop, {:truncate, xid}}
 
@@ -70,25 +57,16 @@ defmodule Electric.Shapes.Consumer.EventHandler.Default do
     {:ok, state, %Plan{}}
   end
 
-  def handle_event(_state, {:pg_snapshot_known, _snapshot}) do
-    raise ArgumentError, "received {:pg_snapshot_known, snapshot} in Default handler"
-  end
-
-  def handle_event(_state, {:query_move_in_complete, _rows, _move_in_lsn}) do
-    raise ArgumentError, "received {:query_move_in_complete, ...} in Default handler"
-  end
-
-  def handle_event(_state, {:materializer_changes, _dep_handle, _payload}) do
-    raise ArgumentError, "received {:materializer_changes, ...} in Default handler"
-  end
-
   @impl true
   def routing_views(_state), do: %{}
 
   defp mark_last_change([]), do: []
 
   defp mark_last_change(changes) do
-    {last, rest} = List.pop_at(changes, -1)
-    rest ++ [%{last | last?: true}]
+    [last | rest] = Enum.reverse(changes)
+    Enum.reverse([%{last | last?: true} | rest])
   end
+
+  defp error_on_truncate(%Changes.TruncatedRelation{}), do: {:error, :truncate}
+  defp error_on_truncate(change), do: {:ok, change}
 end