stream_insert update_only aka stream_update (#3573)

Fixes #2690.
Relates to: https://elixirforum.com/t/add-stream-update-to-liveview-streams/68107

Author: SteffenDE

assets/js/phoenix_live_view/dom_patch.js

@@ -152,6 +152,14 @@ export default class DOMPatch {
           }
         },
         onBeforeNodeAdded: (el) => {
+          // don't add update_only nodes if they did not already exist
+          if (
+            this.getStreamInsert(el)?.updateOnly &&
+            !this.streamComponentRestore[el.id]
+          ) {
+            return false;
+          }
+
           DOM.maintainPrivateHooks(el, el, phxViewportTop, phxViewportBottom);
           this.trackBefore("added", el);
 
@@ -382,8 +390,8 @@ export default class DOMPatch {
 
     liveSocket.time("morphdom", () => {
       this.streams.forEach(([ref, inserts, deleteIds, reset]) => {
-        inserts.forEach(([key, streamAt, limit]) => {
-          this.streamInserts[key] = { ref, streamAt, limit, reset };
+        inserts.forEach(([key, streamAt, limit, updateOnly]) => {
+          this.streamInserts[key] = { ref, streamAt, limit, reset, updateOnly };
         });
         if (reset !== undefined) {
           DOM.all(container, `[${PHX_STREAM_REF}="${ref}"]`, (child) => {

lib/phoenix_live_view.ex

@@ -1957,6 +1957,9 @@ defmodule Phoenix.LiveView do
       here as well in order to be enforced. See `stream/4` for more information on
       limiting streams.
 
+    * `:update_only` - A boolean to only update the item in the stream. If the item does not
+      exist on the client, it will not be inserted. Defaults to `false`.
+
   ## Examples
 
   Imagine you define a stream on mount with a single item:
@@ -2006,8 +2009,9 @@ defmodule Phoenix.LiveView do
   def stream_insert(%Socket{} = socket, name, item, opts \\ []) do
     at = Keyword.get(opts, :at, -1)
     limit = Keyword.get(opts, :limit)
+    update_only = Keyword.get(opts, :update_only, false)
 
-    update_stream(socket, name, &LiveStream.insert_item(&1, item, at, limit))
+    update_stream(socket, name, &LiveStream.insert_item(&1, item, at, limit, update_only))
   end
 
   @doc """

lib/phoenix_live_view/engine.ex

@@ -84,7 +84,9 @@ defmodule Phoenix.LiveView.Comprehension do
 
   @doc false
   def __annotate__(comprehension, %Phoenix.LiveView.LiveStream{} = stream) do
-    inserts = for {id, at, _item, limit} <- stream.inserts, do: [id, at, limit]
+    inserts =
+      for {id, at, _item, limit, update_only} <- stream.inserts, do: [id, at, limit, update_only]
+
     data = [stream.ref, inserts, stream.deletes]
 
     if stream.reset? do

lib/phoenix_live_view/live_stream.ex

@@ -25,7 +25,7 @@ defmodule Phoenix.LiveView.LiveStream do
     # with manually calling stream_insert multiple times, as stream_insert prepends.
     items_list =
       for item <- items, reduce: [] do
-        items -> [{dom_id.(item), -1, item, opts[:limit]} | items]
+        items -> [{dom_id.(item), -1, item, opts[:limit], opts[:update_only]} | items]
       end
 
     %LiveStream{
@@ -64,10 +64,10 @@ defmodule Phoenix.LiveView.LiveStream do
     %{stream | deletes: [dom_id | stream.deletes]}
   end
 
-  def insert_item(%LiveStream{} = stream, item, at, limit) do
+  def insert_item(%LiveStream{} = stream, item, at, limit, update_only) do
     item_id = stream.dom_id.(item)
 
-    %{stream | inserts: [{item_id, at, item, limit} | stream.inserts]}
+    %{stream | inserts: [{item_id, at, item, limit, update_only} | stream.inserts]}
   end
 
   defimpl Enumerable, for: LiveStream do
@@ -81,7 +81,7 @@ defmodule Phoenix.LiveView.LiveStream do
         # before rendering; we also remove duplicates to only use the most recent
         # inserts, which, as the items are reversed, are first
         {inserts, _} =
-          for {id, _, _, _} = insert <- stream.inserts, reduce: {[], MapSet.new()} do
+          for {id, _, _, _, _} = insert <- stream.inserts, reduce: {[], MapSet.new()} do
             {inserts, ids} ->
               if MapSet.member?(ids, id) do
                 # skip duplicates
@@ -106,7 +106,7 @@ defmodule Phoenix.LiveView.LiveStream do
     defp do_reduce(list, {:suspend, acc}, fun), do: {:suspended, acc, &do_reduce(list, &1, fun)}
     defp do_reduce([], {:cont, acc}, _fun), do: {:done, acc}
 
-    defp do_reduce([{dom_id, _at, item, _limit} | tail], {:cont, acc}, fun) do
+    defp do_reduce([{dom_id, _at, item, _limit, _update_only} | tail], {:cont, acc}, fun) do
       do_reduce(tail, fun.({dom_id, item}, acc), fun)
     end
 

lib/phoenix_live_view/test/tree_dom.ex

@@ -425,8 +425,13 @@ defmodule Phoenix.LiveViewTest.TreeDOM do
 
     streamInserts =
       Enum.reduce(streams, %{}, fn %{ref: ref, inserts: inserts}, acc ->
-        Enum.reduce(inserts, acc, fn [id, stream_at, limit], acc ->
-          Map.put(acc, id, %{ref: ref, stream_at: stream_at, limit: limit})
+        Enum.reduce(inserts, acc, fn [id, stream_at, limit, update_only], acc ->
+          Map.put(acc, id, %{
+            ref: ref,
+            stream_at: stream_at,
+            limit: limit,
+            update_only: update_only
+          })
         end)
       end)
 
@@ -457,6 +462,10 @@ defmodule Phoenix.LiveViewTest.TreeDOM do
               # update stream item in place
               List.replace_at(acc, current_index, set_attr(node, "data-phx-stream", insert.ref))
 
+            insert[:update_only] ->
+              # skip item if it is not already in the DOM
+              acc
+
             true ->
               # stream item to be inserted at specific position
               List.insert_at(acc, insert.stream_at, set_attr(node, "data-phx-stream", insert.ref))

test/e2e/tests/streams.spec.js

@@ -921,3 +921,35 @@ test("JS commands are applied when re-joining", async ({ page }) => {
   // should still be hidden
   await expect(page.locator("#users-1")).toBeHidden();
 });
+
+test("update_only", async ({ page }) => {
+  await page.goto("/stream/reset");
+  await syncLV(page);
+
+  expect(await listItems(page)).toEqual([
+    { id: "items-a", text: "A" },
+    { id: "items-b", text: "B" },
+    { id: "items-c", text: "C" },
+    { id: "items-d", text: "D" },
+  ]);
+
+  await page.getByRole("button", { name: "Add E (update only)" }).click();
+  await syncLV(page);
+
+  expect(await listItems(page)).toEqual([
+    { id: "items-a", text: "A" },
+    { id: "items-b", text: "B" },
+    { id: "items-c", text: "C" },
+    { id: "items-d", text: "D" },
+  ]);
+
+  await page.getByRole("button", { name: "Update C (update only)" }).click();
+  await syncLV(page);
+
+  expect(await listItems(page)).toEqual([
+    { id: "items-a", text: "A" },
+    { id: "items-b", text: "B" },
+    { id: "items-c", text: expect.stringMatching(/C .*/) },
+    { id: "items-d", text: "D" },
+  ]);
+});

test/phoenix_live_view/integrations/stream_test.exs

@@ -867,6 +867,35 @@ defmodule Phoenix.LiveView.StreamTest do
            ]
   end
 
+  test "update_only", %{conn: conn} do
+    {:ok, lv, html} = live(conn, "/stream/reset")
+
+    assert ul_list_children(html) == [
+             {"items-a", "A"},
+             {"items-b", "B"},
+             {"items-c", "C"},
+             {"items-d", "D"}
+           ]
+
+    html = assert lv |> element("button", "Add E (update only)") |> render_click()
+
+    assert ul_list_children(html) == [
+             {"items-a", "A"},
+             {"items-b", "B"},
+             {"items-c", "C"},
+             {"items-d", "D"}
+           ]
+
+    html = assert lv |> element("button", "Update C (update only)") |> render_click()
+
+    assert [
+             {"items-a", "A"},
+             {"items-b", "B"},
+             {"items-c", "C " <> _},
+             {"items-d", "D"}
+           ] = ul_list_children(html)
+  end
+
   defp assert_pruned_stream(lv) do
     stream = StreamLive.run(lv, fn socket -> {:reply, socket.assigns.streams.users, socket} end)
     assert stream.inserts == []

test/phoenix_live_view/live_stream_test.exs

@@ -13,12 +13,20 @@ defmodule Phoenix.LiveView.LiveStreamTest do
 
   test "default dom_id" do
     stream = LiveStream.new(:users, 0, [%{id: 1}, %{id: 2}], [])
-    assert stream.inserts == [{"users-2", -1, %{id: 2}, nil}, {"users-1", -1, %{id: 1}, nil}]
+
+    assert stream.inserts == [
+             {"users-2", -1, %{id: 2}, nil, nil},
+             {"users-1", -1, %{id: 1}, nil, nil}
+           ]
   end
 
   test "custom dom_id" do
     stream = LiveStream.new(:users, 0, [%{name: "u1"}, %{name: "u2"}], dom_id: &"u-#{&1.name}")
-    assert stream.inserts == [{"u-u2", -1, %{name: "u2"}, nil}, {"u-u1", -1, %{name: "u1"}, nil}]
+
+    assert stream.inserts == [
+             {"u-u2", -1, %{name: "u2"}, nil, nil},
+             {"u-u1", -1, %{name: "u1"}, nil, nil}
+           ]
   end
 
   test "default dom_id without struct or map with :id" do
@@ -31,7 +39,7 @@ defmodule Phoenix.LiveView.LiveStreamTest do
 
   test "inserts are deduplicated (last insert wins)" do
     assert stream = LiveStream.new(:users, 0, [%{id: 1}, %{id: 2}], [])
-    stream = LiveStream.insert_item(stream, %{id: 2, updated: true}, -1, nil)
+    stream = LiveStream.insert_item(stream, %{id: 2, updated: true}, -1, nil, nil)
     stream = %{stream | consumable?: true}
     assert Enum.to_list(stream) == [{"users-1", %{id: 1}}, {"users-2", %{id: 2, updated: true}}]
   end

test/support/live_views/streams.ex

@@ -359,6 +359,8 @@ defmodule Phoenix.LiveViewTest.Support.StreamResetLive do
     <button phx-click="delete-insert-existing-at-one">Delete C and insert at 1</button>
     <button phx-click="prepend-existing">Prepend C</button>
     <button phx-click="append-existing">Append C</button>
+    <button phx-click="new-update-only">Add E (update only)</button>
+    <button phx-click="existing-update-only">Update C (update only)</button>
     """
   end
 
@@ -490,6 +492,18 @@ defmodule Phoenix.LiveViewTest.Support.StreamResetLive do
        at: -1
      )}
   end
+
+  def handle_event("new-update-only", _, socket) do
+    {:noreply, stream_insert(socket, :items, %{id: "e", name: "E"}, at: -1, update_only: true)}
+  end
+
+  def handle_event("existing-update-only", _, socket) do
+    {:noreply,
+     stream_insert(socket, :items, %{id: "c", name: "C #{System.unique_integer()}"},
+       at: -1,
+       update_only: true
+     )}
+  end
 end
 
 defmodule Phoenix.LiveViewTest.Support.StreamResetLCLive do