Add support for keyword-based sync_stream views (#95)

Fixes #94

Author: magnetised

lib/phoenix/sync/live_view.ex

@@ -17,12 +17,15 @@ if Code.ensure_loaded?(Phoenix.Component) do
         {:noreply, sync_stream_update(socket, event)}
       end
     end
+
+    See `sync_stream/4` for more details.
     ```
     """
 
     use Phoenix.Component
 
     alias Electric.Client.Message
+    alias Phoenix.Sync.PredefinedShape
 
     require Record
 
@@ -190,6 +193,40 @@ if Code.ensure_loaded?(Phoenix.Component) do
             {:ok, Phoenix.Sync.LiveView.sync_stream(socket, :users, User)}
           end
         end
+
+    ## Keyword-based Shapes
+
+    `Ecto` is not required to use `sync_stream/4`. [Keyword-based
+    shapes](../../../README.md#using-a-keyword-list) are possible but
+    work a little differently.
+
+        def mount(_params, _session, socket) do
+          socket =
+            Phoenix.Sync.LiveView.sync_stream(
+              socket,
+              :admins,
+              table: "users",
+              where: "admin = true"
+            )
+          {:ok, socket}
+        end
+
+    Without an underlying `Ecto.Schema` module to map the stream values, you
+    will receive simple map values with string keys (plus a special
+    `__sync_key__` value used for the live view
+    [`dom_id`](https://hexdocs.pm/phoenix_live_view/Phoenix.LiveView.html#stream_configure/3)
+    function.
+
+    So to use these in your template, you should be careful to use the
+    `value["key"]` syntax to retrieve values. Using a keyword-based shape, The
+    first example above becomes:
+
+          <div phx-update="stream">
+            <div :for={{id, item} <- @streams.items} id={id}>
+              <%= item["value"] %>
+            </div>
+          </div>
+
     """
     @spec sync_stream(
             socket :: Phoenix.LiveView.Socket.t(),
@@ -215,8 +252,9 @@ if Code.ensure_loaded?(Phoenix.Component) do
             end
           end)
 
-        Phoenix.LiveView.stream(
-          socket,
+        socket
+        |> configure_live_stream(name, query)
+        |> Phoenix.LiveView.stream(
           name,
           client_live_stream(client, name, query, component),
           stream_opts
@@ -226,6 +264,36 @@ if Code.ensure_loaded?(Phoenix.Component) do
       end
     end
 
+    defp configure_live_stream(socket, name, query) do
+      if PredefinedShape.is_queryable?(query) do
+        # if the shape is an Ecto.Struct then we can just fallback to the default
+        # dom_id function (which is `value.id`)
+        socket
+      else
+        # since we don't have control over the `mount` callback, but want to use
+        # `stream_configure` correctly which means that the stream should only be
+        # configured once, we do our own tracking of configured streams in the
+        # socket assigns
+        case socket.assigns do
+          %{__sync_stream_config__: %{^name => true}} ->
+            socket
+
+          assigns ->
+            assigns =
+              Map.update(
+                assigns,
+                :__sync_stream_config__,
+                %{name => true},
+                &Map.put(&1, name, true)
+              )
+
+            Phoenix.LiveView.stream_configure(%{socket | assigns: assigns}, name,
+              dom_id: &Map.fetch!(&1, :__sync_key__)
+            )
+        end
+      end
+    end
+
     @doc """
     Handle Electric events within a LiveView.
 
@@ -277,15 +345,42 @@ if Code.ensure_loaded?(Phoenix.Component) do
     defp client_live_stream(client, name, query, component) do
       pid = self()
 
+      shape =
+        query
+        |> PredefinedShape.new!()
+        |> PredefinedShape.to_shape()
+
       client
-      |> Electric.Client.stream(query, live: false, replica: :full, errors: :stream)
+      |> Electric.Client.stream(shape, live: false, replica: :full, errors: :stream)
+      |> keyed_stream(query)
       |> Stream.transform(
         fn -> {[], nil} end,
         &live_stream_message/2,
-        &update_mode(&1, {client, name, query, pid, component})
+        &update_mode(&1, {client, name, query, shape, pid, component})
       )
     end
 
+    # when the stream is based off an Ecto.Schema struct then the default id
+    # based dom-id function works. otherwise we can use the message key because
+    # we **know** that is unique for a shape
+    defp keyed_stream(base_stream, query) do
+      if PredefinedShape.is_queryable?(query) do
+        base_stream
+      else
+        Stream.map(base_stream, fn
+          %Message.ChangeMessage{key: key, value: value} = msg ->
+            %{msg | value: Map.put(value, :__sync_key__, sanitise_key(key))}
+
+          msg ->
+            msg
+        end)
+      end
+    end
+
+    defp sanitise_key(key) do
+      key |> String.replace("\"", "") |> String.replace(" ", "%20")
+    end
+
     defp live_stream_message(
            %Message.ChangeMessage{headers: %{operation: :insert}, value: value},
            acc
@@ -313,7 +408,7 @@ if Code.ensure_loaded?(Phoenix.Component) do
       raise error
     end
 
-    defp update_mode({updates, resume}, {client, name, query, pid, component}) do
+    defp update_mode({updates, resume}, {client, name, query, shape, pid, component}) do
       # need to send every update as a separate message.
 
       for event <- updates |> Enum.reverse() |> Enum.map(&wrap_msg(&1, name, component)),
@@ -323,7 +418,8 @@ if Code.ensure_loaded?(Phoenix.Component) do
 
       Task.start_link(fn ->
         client
-        |> Electric.Client.stream(query, resume: resume, replica: :full)
+        |> Electric.Client.stream(shape, resume: resume, replica: :full)
+        |> keyed_stream(query)
         |> Stream.each(&send_live_event(&1, pid, name, component))
         |> Stream.run()
       end)

lib/phoenix/sync/predefined_shape.ex

@@ -156,6 +156,11 @@ defmodule Phoenix.Sync.PredefinedShape do
     |> ShapeDefinition.params(format: :keyword)
   end
 
+  @doc false
+  def to_shape(%__MODULE__{} = predefined_shape) do
+    to_shape_definition(predefined_shape)
+  end
+
   @doc false
   def to_stream_params(%__MODULE__{} = predefined_shape) do
     {to_shape_definition(predefined_shape), predefined_shape.stream_config}

test/phoenix/sync/live_view_test.exs

@@ -34,7 +34,7 @@ defmodule Phoenix.Sync.LiveViewTest do
 
       Client.Mock.async_response(client,
         status: 200,
-        schema: %{id: %{type: "int8"}, name: %{type: "text"}},
+        schema: %{id: %{type: "uuid"}, name: %{type: "text"}},
         last_offset: Client.Offset.first(),
         shape_handle: "users-1",
         body: Client.Mock.transaction(users, operation: :insert)
@@ -121,7 +121,7 @@ defmodule Phoenix.Sync.LiveViewTest do
 
       Client.Mock.async_response(client,
         status: 200,
-        schema: %{id: %{type: "int8"}, name: %{type: "text"}},
+        schema: %{id: %{type: "uuid"}, name: %{type: "text"}},
         last_offset: Client.Offset.first(),
         shape_handle: "users-1",
         body: body
@@ -158,7 +158,7 @@ defmodule Phoenix.Sync.LiveViewTest do
 
       Client.Mock.async_response(client,
         status: 200,
-        schema: %{id: %{type: "int8"}, name: %{type: "text"}},
+        schema: %{id: %{type: "uuid"}, name: %{type: "text"}},
         last_offset: Client.Offset.first(),
         shape_handle: "users-1",
         body: Client.Mock.transaction(users, operation: :insert)
@@ -224,6 +224,70 @@ defmodule Phoenix.Sync.LiveViewTest do
           assert %Electric.Client.Error{message: %{"message" => "this is wrong"}} = error
       end
     end
+
+    defp add_message_keys(txn) do
+      Enum.map(txn, fn
+        %{"value" => %{"id" => id}} = msg ->
+          Map.put(msg, "key", ~s|"public"."users"/"#{id}"|)
+
+        msg ->
+          msg
+      end)
+    end
+
+    test "allows for keyword-based shapes", %{conn: conn} do
+      {:ok, client} = Client.Mock.new()
+
+      users = [
+        %{id: "6dfea52e-1096-4b62-aafd-838ddd49477d", name: "User 1"},
+        %{id: "9fc8f0a7-42e9-4473-9981-43a1904cd88a", name: "User 2"},
+        %{id: "4252d858-8764-4069-bb8c-e670f899b80a", name: "User 3"}
+      ]
+
+      Client.Mock.async_response(client,
+        status: 200,
+        schema: %{id: %{type: "uuid", pk_position: 0}, name: %{type: "text"}},
+        last_offset: Client.Offset.first(),
+        shape_handle: "users-1",
+        # TODO: update the mock client to add the key by default
+        body: Client.Mock.transaction(users, operation: :insert) |> add_message_keys()
+      )
+
+      conn =
+        conn
+        |> put_private(:electric_client, client)
+        |> put_private(:test_pid, self())
+
+      {:ok, lv, html} = live(conn, "/stream/keyword")
+
+      for %{name: name} <- users do
+        assert html =~ name
+      end
+
+      users2 = [
+        %{id: "53183977-bd54-4171-9697-51e13b0ff7ca", name: "User 4"},
+        %{id: "92d42f40-cf16-4d51-a663-171d9fa1a21a", name: "User 5"},
+        %{id: "04e15019-010e-4aa1-8eb0-33132099d05b", name: "User 6"}
+      ]
+
+      {:ok, _req} =
+        Client.Mock.response(client,
+          status: 200,
+          last_offset: Client.Offset.new(3, 2),
+          shape_handle: "users-1",
+          body: Client.Mock.transaction(users2, lsn: 3, operation: :insert) |> add_message_keys()
+        )
+
+      for _ <- users2 do
+        assert_receive {:sync, _}
+      end
+
+      html = render(lv)
+
+      for %{name: name} <- users ++ users2 do
+        assert html =~ name
+      end
+    end
   end
 
   describe "electric_client_configuration/1" do

test/support/live_views/stream.ex

@@ -245,3 +245,44 @@ defmodule Phoenix.Sync.LiveViewTest.StreamSandbox do
     {:noreply, Phoenix.Sync.LiveView.sync_stream_update(socket, event)}
   end
 end
+
+defmodule Phoenix.Sync.LiveViewTest.StreamLiveKeyword do
+  use Phoenix.LiveView
+
+  def run(lv, func) do
+    GenServer.call(lv.pid, {:run, func})
+  end
+
+  def render(assigns) do
+    ~H"""
+    <div id="users" phx-update="stream">
+      <div :for={{id, user} <- @streams.users} id={id}>
+        <%= user["name"] %>
+      </div>
+    </div>
+    """
+  end
+
+  def mount(_params, _session, socket) do
+    client =
+      get_in(socket.private.connect_info.private, [:electric_client]) ||
+        raise "missing client configuration"
+
+    parent =
+      get_in(socket.private.connect_info.private, [:test_pid]) ||
+        raise "missing parent pid configuration"
+
+    {:ok,
+     socket
+     |> assign(:test_pid, parent)
+     |> Phoenix.Sync.LiveView.sync_stream(:users, [table: "users", where: "name != ''"],
+       client: client
+     )}
+  end
+
+  def handle_info({:sync, event}, socket) do
+    # send message to test pid, just for sync
+    send(socket.assigns.test_pid, {:sync, event})
+    {:noreply, Phoenix.Sync.LiveView.sync_stream_update(socket, event)}
+  end
+end

test/support/router.ex

@@ -24,6 +24,7 @@ defmodule Phoenix.Sync.LiveViewTest.Router do
     live "/stream", StreamLive
     live "/stream/with-component", StreamLiveWithComponent
     live "/stream/sandbox", StreamSandbox
+    live "/stream/keyword", StreamLiveKeyword
   end
 
   scope "/" do