Send delete event over Mastodon streaming api

Closes #116

lib/pleroma/web/activity_pub/activity_pub.ex

@@ -92,7 +92,7 @@ def insert(map, local \\ true) when is_map(map) do
   def stream_out(activity) do
     public = "https://www.w3.org/ns/activitystreams#Public"
 
-    if activity.data["type"] in ["Create", "Announce"] do
+    if activity.data["type"] in ["Create", "Announce", "Delete"] do
       Pleroma.Web.Streamer.stream("user", activity)
       Pleroma.Web.Streamer.stream("list", activity)
 
@@ -103,6 +103,7 @@ def stream_out(activity) do
           Pleroma.Web.Streamer.stream("public:local", activity)
         end
 
+        if activity.data["type"] in ["Create"] do
           activity.data["object"]
           |> Map.get("tag", [])
           |> Enum.filter(fn tag -> is_bitstring(tag) end)
@@ -115,6 +116,7 @@ def stream_out(activity) do
               Pleroma.Web.Streamer.stream("public:local:media", activity)
             end
           end
+        end
       else
         if !Enum.member?(activity.data["cc"] || [], public) &&
              !Enum.member?(

lib/pleroma/web/streamer.ex

@@ -205,6 +205,15 @@ def push_to_socket(topics, topic, %Activity{data: %{"type" => "Announce"}} = ite
     end)
   end
 
+  def push_to_socket(topics, topic, %Activity{id: id, data: %{"type" => "Delete"}}) do
+    Enum.each(topics[topic] || [], fn socket ->
+      send(
+        socket.transport_pid,
+        {:text, %{event: "delete", payload: to_string(id)} |> Jason.encode!()}
+      )
+    end)
+  end
+
   def push_to_socket(topics, topic, item) do
     Enum.each(topics[topic] || [], fn socket ->
       # Get the current user so we have up-to-date blocks etc.

test/web/streamer_test.exs

@@ -6,7 +6,8 @@ defmodule Pleroma.Web.StreamerTest do
   use Pleroma.DataCase
 
   alias Pleroma.Web.Streamer
-  alias Pleroma.{List, User}
+  alias Pleroma.List
+  alias Pleroma.User
   alias Pleroma.Web.CommonAPI
   import Pleroma.Factory
 
@@ -35,6 +36,28 @@ test "it sends to public" do
     Streamer.push_to_socket(topics, "public", activity)
 
     Task.await(task)
+
+    task =
+      Task.async(fn ->
+        assert_receive {:text, _}, 4_000
+      end)
+
+    fake_socket = %{
+      transport_pid: task.pid,
+      assigns: %{
+        user: user
+      }
+    }
+
+    {:ok, activity} = CommonAPI.delete(activity.id, other_user)
+
+    topics = %{
+      "public" => [fake_socket]
+    }
+
+    Streamer.push_to_socket(topics, "public", activity)
+
+    Task.await(task)
   end
 
   test "it doesn't send to blocked users" do