live-stream: Handle client errors in the live-view process (#37)

An upstream server error now raises directly rather than resulting in a
secondary nimbleoptions validation error caused by a missing resume
message

requires new error handling config in the elixir client:
electric-sql/electric#2721

Author: magnetised

lib/phoenix/sync/live_view.ex

@@ -272,57 +272,44 @@ if Code.ensure_loaded?(Phoenix.Component) do
       pid = self()
 
       client
-      |> Electric.Client.stream(query, live: false, replica: :full)
+      |> Electric.Client.stream(query, live: false, replica: :full, errors: :stream)
       |> Stream.transform(
         fn -> {[], nil} end,
-        &live_stream_message(&1, &2, client, name, query, pid, component),
-        &update_mode(&1, client, name, query, pid, component)
+        &live_stream_message/2,
+        &update_mode(&1, {client, name, query, pid, component})
       )
     end
 
     defp live_stream_message(
            %Message.ChangeMessage{headers: %{operation: :insert}, value: value},
-           acc,
-           _client,
-           _name,
-           _query,
-           _pid,
-           _component
+           acc
          ) do
       {[value], acc}
     end
 
-    defp live_stream_message(
-           %Message.ChangeMessage{headers: %{operation: operation}} = msg,
-           {updates, resume},
-           _client,
-           _name,
-           _query,
-           _pid,
-           _component
-         )
-         when operation in [:update, :delete] do
+    defp live_stream_message(%Message.ChangeMessage{} = msg, {updates, resume}) do
       {[], {[msg | updates], resume}}
     end
 
-    defp live_stream_message(
-           %Message.ResumeMessage{} = resume,
-           {updates, nil},
-           _client,
-           _name,
-           _query,
-           _pid,
-           _component
-         ) do
+    defp live_stream_message(%Message.ControlMessage{}, acc) do
+      {[], acc}
+    end
+
+    defp live_stream_message(%Message.ResumeMessage{} = resume, {updates, nil}) do
       {[], {updates, resume}}
     end
 
-    defp live_stream_message(_message, acc, _client, _name, _query, _pid, _component) do
-      {[], acc}
+    defp live_stream_message(%Electric.Client.Error{} = error, _acc) do
+      {[], {error, nil}}
     end
 
-    defp update_mode({updates, resume}, client, name, query, pid, component) do
+    defp update_mode({%Electric.Client.Error{} = error, _resume}, _state) do
+      raise error
+    end
+
+    defp update_mode({updates, resume}, {client, name, query, pid, component}) do
       # need to send every update as a separate message.
+
       for event <- updates |> Enum.reverse() |> Enum.map(&wrap_msg(&1, name, component)),
           do: send(pid, {:sync, event})
 

mix.exs

@@ -39,7 +39,7 @@ defmodule Phoenix.Sync.MixProject do
       # require an exact version because electric moves very quickly atm
       # and a more generous specification would inevitably break.
       {:electric, "== 1.0.1", optional: true},
-      {:electric_client, "== 0.3.0"}
+      {:electric_client, ">= 0.5.0-beta-1"}
     ] ++ deps_for_env(Mix.env())
   end