Merge branch 'fix-streaming-reblog' into 'develop'

Streamer: Correctly handle reblog mutes

Closes #1129 and #1438

See merge request pleroma/pleroma!2156

lib/pleroma/web/streamer/worker.ex

@@ -138,7 +138,8 @@ defmodule Pleroma.Web.Streamer.Worker do
 
     with parent <- Object.normalize(item) || item,
          true <-
-           Enum.all?([blocked_ap_ids, muted_ap_ids, reblog_muted_ap_ids], &(item.actor not in &1)),
+           Enum.all?([blocked_ap_ids, muted_ap_ids], &(item.actor not in &1)),
+         true <- item.data["type"] != "Announce" || item.actor not in reblog_muted_ap_ids,
          true <- Enum.all?([blocked_ap_ids, muted_ap_ids], &(parent.data["actor"] not in &1)),
          true <- MapSet.disjoint?(recipients, recipient_blocks),
          %{host: item_host} <- URI.parse(item.actor),

test/web/streamer/streamer_test.exs

@@ -65,6 +65,9 @@ defmodule Pleroma.Web.StreamerTest do
       blocked = insert(:user)
       {:ok, _user_relationship} = User.block(user, blocked)
 
+      {:ok, activity} = CommonAPI.post(user, %{"status" => ":("})
+      {:ok, notif, _} = CommonAPI.favorite(activity.id, blocked)
+
       task = Task.async(fn -> refute_receive {:text, _}, @streamer_timeout end)
 
       Streamer.add_socket(
@@ -72,9 +75,6 @@ defmodule Pleroma.Web.StreamerTest do
         %{transport_pid: task.pid, assigns: %{user: user}}
       )
 
-      {:ok, activity} = CommonAPI.post(user, %{"status" => ":("})
-      {:ok, notif, _} = CommonAPI.favorite(activity.id, blocked)
-
       Streamer.stream("user:notification", notif)
       Task.await(task)
     end
@@ -83,6 +83,11 @@ defmodule Pleroma.Web.StreamerTest do
       user: user
     } do
       user2 = insert(:user)
+
+      {:ok, activity} = CommonAPI.post(user, %{"status" => "super hot take"})
+      {:ok, activity} = CommonAPI.add_mute(user, activity)
+      {:ok, notif, _} = CommonAPI.favorite(activity.id, user2)
+
       task = Task.async(fn -> refute_receive {:text, _}, @streamer_timeout end)
 
       Streamer.add_socket(
@@ -90,9 +95,6 @@ defmodule Pleroma.Web.StreamerTest do
         %{transport_pid: task.pid, assigns: %{user: user}}
       )
 
-      {:ok, activity} = CommonAPI.post(user, %{"status" => "super hot take"})
-      {:ok, activity} = CommonAPI.add_mute(user, activity)
-      {:ok, notif, _} = CommonAPI.favorite(activity.id, user2)
       Streamer.stream("user:notification", notif)
       Task.await(task)
     end
@@ -101,6 +103,11 @@ defmodule Pleroma.Web.StreamerTest do
       user: user
     } do
       user2 = insert(:user, %{ap_id: "https://hecking-lewd-place.com/user/meanie"})
+
+      {:ok, user} = User.block_domain(user, "hecking-lewd-place.com")
+      {:ok, activity} = CommonAPI.post(user, %{"status" => "super hot take"})
+      {:ok, notif, _} = CommonAPI.favorite(activity.id, user2)
+
       task = Task.async(fn -> refute_receive {:text, _}, @streamer_timeout end)
 
       Streamer.add_socket(
@@ -108,10 +115,6 @@ defmodule Pleroma.Web.StreamerTest do
         %{transport_pid: task.pid, assigns: %{user: user}}
       )
 
-      {:ok, user} = User.block_domain(user, "hecking-lewd-place.com")
-      {:ok, activity} = CommonAPI.post(user, %{"status" => "super hot take"})
-      {:ok, notif, _} = CommonAPI.favorite(activity.id, user2)
-
       Streamer.stream("user:notification", notif)
       Task.await(task)
     end
@@ -267,6 +270,8 @@ defmodule Pleroma.Web.StreamerTest do
       blocked_user = insert(:user)
       {:ok, _user_relationship} = User.block(user, blocked_user)
 
+      {:ok, activity} = CommonAPI.post(blocked_user, %{"status" => "Test"})
+
       task =
         Task.async(fn ->
           refute_receive {:text, _}, 1_000
@@ -277,8 +282,6 @@ defmodule Pleroma.Web.StreamerTest do
         user: user
       }
 
-      {:ok, activity} = CommonAPI.post(blocked_user, %{"status" => "Test"})
-
       topics = %{
         "public" => [fake_socket]
       }
@@ -335,6 +338,12 @@ defmodule Pleroma.Web.StreamerTest do
     {:ok, list} = List.create("Test", user_a)
     {:ok, list} = List.follow(list, user_b)
 
+    {:ok, activity} =
+      CommonAPI.post(user_b, %{
+        "status" => "@#{user_c.nickname} Test",
+        "visibility" => "direct"
+      })
+
     task =
       Task.async(fn ->
         refute_receive {:text, _}, 1_000
@@ -345,12 +354,6 @@ defmodule Pleroma.Web.StreamerTest do
       user: user_a
     }
 
-    {:ok, activity} =
-      CommonAPI.post(user_b, %{
-        "status" => "@#{user_c.nickname} Test",
-        "visibility" => "direct"
-      })
-
     topics = %{
       "list:#{list.id}" => [fake_socket]
     }
@@ -367,6 +370,12 @@ defmodule Pleroma.Web.StreamerTest do
     {:ok, list} = List.create("Test", user_a)
     {:ok, list} = List.follow(list, user_b)
 
+    {:ok, activity} =
+      CommonAPI.post(user_b, %{
+        "status" => "Test",
+        "visibility" => "private"
+      })
+
     task =
       Task.async(fn ->
         refute_receive {:text, _}, 1_000
@@ -377,12 +386,6 @@ defmodule Pleroma.Web.StreamerTest do
       user: user_a
     }
 
-    {:ok, activity} =
-      CommonAPI.post(user_b, %{
-        "status" => "Test",
-        "visibility" => "private"
-      })
-
     topics = %{
       "list:#{list.id}" => [fake_socket]
     }
@@ -401,6 +404,12 @@ defmodule Pleroma.Web.StreamerTest do
     {:ok, list} = List.create("Test", user_a)
     {:ok, list} = List.follow(list, user_b)
 
+    {:ok, activity} =
+      CommonAPI.post(user_b, %{
+        "status" => "Test",
+        "visibility" => "private"
+      })
+
     task =
       Task.async(fn ->
         assert_receive {:text, _}, 1_000
@@ -411,12 +420,6 @@ defmodule Pleroma.Web.StreamerTest do
       user: user_a
     }
 
-    {:ok, activity} =
-      CommonAPI.post(user_b, %{
-        "status" => "Test",
-        "visibility" => "private"
-      })
-
     Streamer.add_socket(
       "list:#{list.id}",
       fake_socket
@@ -433,6 +436,9 @@ defmodule Pleroma.Web.StreamerTest do
     user3 = insert(:user)
     CommonAPI.hide_reblogs(user1, user2)
 
+    {:ok, create_activity} = CommonAPI.post(user3, %{"status" => "I'm kawen"})
+    {:ok, announce_activity, _} = CommonAPI.repeat(create_activity.id, user2)
+
     task =
       Task.async(fn ->
         refute_receive {:text, _}, 1_000
@@ -443,9 +449,6 @@ defmodule Pleroma.Web.StreamerTest do
       user: user1
     }
 
-    {:ok, create_activity} = CommonAPI.post(user3, %{"status" => "I'm kawen"})
-    {:ok, announce_activity, _} = CommonAPI.repeat(create_activity.id, user2)
-
     topics = %{
       "public" => [fake_socket]
     }
@@ -455,6 +458,34 @@ defmodule Pleroma.Web.StreamerTest do
     Task.await(task)
   end
 
+  test "it does send non-reblog notification for reblog-muted actors" do
+    user1 = insert(:user)
+    user2 = insert(:user)
+    user3 = insert(:user)
+    CommonAPI.hide_reblogs(user1, user2)
+
+    {:ok, create_activity} = CommonAPI.post(user3, %{"status" => "I'm kawen"})
+    {:ok, favorite_activity, _} = CommonAPI.favorite(create_activity.id, user2)
+
+    task =
+      Task.async(fn ->
+        assert_receive {:text, _}, 1_000
+      end)
+
+    fake_socket = %StreamerSocket{
+      transport_pid: task.pid,
+      user: user1
+    }
+
+    topics = %{
+      "public" => [fake_socket]
+    }
+
+    Worker.push_to_socket(topics, "public", favorite_activity)
+
+    Task.await(task)
+  end
+
   test "it doesn't send posts from muted threads" do
     user = insert(:user)
     user2 = insert(:user)