SideEffects / ChatView: Add an unread cache.

This is to prevent wrong values in the stream.

lib/pleroma/web/activity_pub/side_effects.ex

@@ -142,6 +142,11 @@ def handle_object_creation(%{"type" => "ChatMessage"} = object, meta) do
           {:ok, chat} = Chat.bump_or_create(user.id, other_user.ap_id)
           {:ok, cm_ref} = ChatMessageReference.create(chat, object, user.ap_id != actor.ap_id)
 
+          # We add a cache of the unread value here so that it doesn't change when being streamed out
+          chat =
+            chat
+            |> Map.put(:unread, ChatMessageReference.unread_count_for_chat(chat))
+
           Streamer.stream(
             ["user", "user:pleroma_chat"],
             {user, %{cm_ref | chat: chat, object: object}}

lib/pleroma/web/pleroma_api/views/chat_view.ex

@@ -20,7 +20,7 @@ def render("show.json", %{chat: %Chat{} = chat} = opts) do
     %{
       id: chat.id |> to_string(),
       account: AccountView.render("show.json", Map.put(opts, :user, recipient)),
-      unread: ChatMessageReference.unread_count_for_chat(chat),
+      unread: Map.get(chat, :unread) || ChatMessageReference.unread_count_for_chat(chat),
       last_message:
         last_message &&
           ChatMessageReferenceView.render("show.json", chat_message_reference: last_message),

lib/pleroma/web/streamer/streamer.ex

@@ -90,34 +90,20 @@ def remove_socket(topic) do
     if should_env_send?(), do: Registry.unregister(@registry, topic)
   end
 
-  def stream(topics, item) when is_list(topics) do
+  def stream(topics, items) do
     if should_env_send?() do
-      Enum.each(topics, fn t ->
-        spawn(fn -> do_stream(t, item) end)
+      List.wrap(topics)
+      |> Enum.each(fn topic ->
+        List.wrap(items)
+        |> Enum.each(fn item ->
+          spawn(fn -> do_stream(topic, item) end)
+        end)
       end)
     end
 
     :ok
   end
 
-  def stream(topic, items) when is_list(items) do
-    if should_env_send?() do
-      Enum.each(items, fn i ->
-        spawn(fn -> do_stream(topic, i) end)
-      end)
-
-      :ok
-    end
-  end
-
-  def stream(topic, item) do
-    if should_env_send?() do
-      spawn(fn -> do_stream(topic, item) end)
-    end
-
-    :ok
-  end
-
   def filtered_by_user?(%User{} = user, %Activity{} = item) do
     %{block: blocked_ap_ids, mute: muted_ap_ids, reblog_mute: reblog_muted_ap_ids} =
       User.outgoing_relationships_ap_ids(user, [:block, :mute, :reblog_mute])

lib/pleroma/web/views/streamer_view.ex

@@ -55,6 +55,8 @@ def render("chat_update.json", %{chat_message_reference: cm_ref}) do
     # Explicitly giving the cmr for the object here, so we don't accidentally
     # send a later 'last_message' that was inserted between inserting this and
     # streaming it out
+    # 
+    # It also contains the chat with a cache of the correct unread count
     Logger.debug("Trying to stream out #{inspect(cm_ref)}")
 
     representation =

test/web/pleroma_api/views/chat_view_test.exs

@@ -16,6 +16,21 @@ defmodule Pleroma.Web.PleromaAPI.ChatViewTest do
 
   import Pleroma.Factory
 
+  test "giving a chat with an 'unread' field, it uses that" do
+    user = insert(:user)
+    recipient = insert(:user)
+
+    {:ok, chat} = Chat.get_or_create(user.id, recipient.ap_id)
+
+    chat =
+      chat
+      |> Map.put(:unread, 5)
+
+    represented_chat = ChatView.render("show.json", chat: chat)
+
+    assert represented_chat[:unread] == 5
+  end
+
   test "it represents a chat" do
     user = insert(:user)
     recipient = insert(:user)