deduplicate stream inserts (#3599)

Also adjusts stream_insert to prepend the insert instead of expensively
appending it.

Fixes #2689.
Closes #3596.
Closes #3598.

lib/phoenix_live_view/live_stream.ex

@@ -20,7 +20,13 @@ defmodule Phoenix.LiveView.LiveStream do
             "stream :dom_id must return a function which accepts each item, got: #{inspect(dom_id)}"
     end
 
-    items_list = for item <- items, do: {dom_id.(item), -1, item, opts[:limit]}
+    # We need to go through the items one time to map them into the proper insert tuple format.
+    # Conveniently, we reverse the list in this pass, which we need to in order to be consistent
+    # 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]
+      end
 
     %LiveStream{
       ref: ref,
@@ -61,16 +67,30 @@ defmodule Phoenix.LiveView.LiveStream do
   def insert_item(%LiveStream{} = stream, item, at, limit) do
     item_id = stream.dom_id.(item)
 
-    %{stream | inserts: stream.inserts ++ [{item_id, at, item, limit}]}
+    %{stream | inserts: [{item_id, at, item, limit} | stream.inserts]}
   end
 
   defimpl Enumerable, for: LiveStream do
     def count(%LiveStream{inserts: inserts}), do: {:ok, length(inserts)}
 
     def member?(%LiveStream{}, _item), do: raise(RuntimeError, "not implemented")
 
-    def reduce(%LiveStream{inserts: inserts} = stream, acc, fun) do
+    def reduce(%LiveStream{} = stream, acc, fun) do
       if stream.consumable? do
+        # the inserts are stored in reverse insert order, so we need to reverse them
+        # 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
+            {inserts, ids} ->
+              if MapSet.member?(ids, id) do
+                # skip duplicates
+                {inserts, ids}
+              else
+                {[insert | inserts], MapSet.put(ids, id)}
+              end
+          end
+
         do_reduce(inserts, acc, fun)
       else
         raise ArgumentError, """

test/phoenix_live_view/live_stream_test.exs

@@ -13,12 +13,12 @@ defmodule Phoenix.LiveView.LiveStreamTest do
 
   test "default dom_id" do
     stream = LiveStream.new(:users, 0, [%{id: 1}, %{id: 2}], [])
-    assert stream.inserts == [{"users-1", -1, %{id: 1}, nil}, {"users-2", -1, %{id: 2}, nil}]
+    assert stream.inserts == [{"users-2", -1, %{id: 2}, nil}, {"users-1", -1, %{id: 1}, 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-u1", -1, %{name: "u1"}, nil}, {"u-u2", -1, %{name: "u2"}, nil}]
+    assert stream.inserts == [{"u-u2", -1, %{name: "u2"}, nil}, {"u-u1", -1, %{name: "u1"}, nil}]
   end
 
   test "default dom_id without struct or map with :id" do
@@ -28,4 +28,11 @@ defmodule Phoenix.LiveView.LiveStreamTest do
       LiveStream.new(:users, 0, [%{user_id: 1}, %{user_id: 2}], [])
     end
   end
+
+  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 = %{stream | consumable?: true}
+    assert Enum.to_list(stream) == [{"users-1", %{id: 1}}, {"users-2", %{id: 2, updated: true}}]
+  end
 end