Stream function to handle direct messages.

lib/pleroma/web/activity_pub/activity_pub.ex

@@ -42,18 +42,16 @@ def insert(map, local \\ true) when is_map(map) do
   def stream_out(activity) do
     if activity.data["type"] in ["Create", "Announce"] do
       Pleroma.Web.Streamer.stream("user", activity)
-      direct? = activity.data["object"]["visibility"] == "direct"
 
-      cond do
+      visibility = Pleroma.Web.MastodonAPI.StatusView.get_visibility(activity.data["object"])
-        direct? ->
-          Pleroma.Web.Streamer.stream("direct", activity)
 
-        Enum.member?(activity.data["to"], "https://www.w3.org/ns/activitystreams#Public") ->
+      case visibility do
+        "public" ->
           Pleroma.Web.Streamer.stream("public", activity)
+          if activity.local, do: Pleroma.Web.Streamer.stream("public:local", activity)
 
-          if activity.local do
+        "direct" ->
-            Pleroma.Web.Streamer.stream("public:local", activity)
+          Pleroma.Web.Streamer.stream("direct", activity)
-          end
       end
     end
   end

lib/pleroma/web/streamer.ex

@@ -46,6 +46,19 @@ def handle_cast(%{action: :ping}, topics) do
     {:noreply, topics}
   end
 
+  def handle_cast(%{action: :stream, topic: "direct", item: item}, topics) do
+    recipient_topics =
+      User.get_recipients_from_activity(item)
+      |> Enum.map(fn %{id: id} -> "direct:#{id}" end)
+
+    Enum.each(recipient_topics || [], fn user_topic ->
+      Logger.debug("Trying to push direct message to #{user_topic}\n\n")
+      push_to_socket(topics, user_topic, item)
+    end)
+
+    {:noreply, topics}
+  end
+
   def handle_cast(%{action: :stream, topic: "user", item: %Notification{} = item}, topics) do
     topic = "user:#{item.user_id}"
 
@@ -137,8 +150,8 @@ def push_to_socket(topics, topic, item) do
     end)
   end
 
-  defp internal_topic("user", socket) do
+  defp internal_topic(topic, socket) when topic in ~w[user, direct] do
-    "user:#{socket.assigns[:user].id}"
+    "#{topic}:#{socket.assigns[:user].id}"
   end
 
   defp internal_topic(topic, _), do: topic