[#1364] Improved control over generation / sending of notifications. Fixed blocking / muting users notifications issue. Added tests.

lib/pleroma/activity.ex

@@ -95,6 +95,16 @@ defmodule Pleroma.Activity do
     |> preload([activity, object: object], object: object)
   end
 
+  def user_actor(%Activity{actor: nil}), do: nil
+
+  def user_actor(%Activity{} = activity) do
+    with %User{} <- activity.user_actor do
+      activity.user_actor
+    else
+      _ -> User.get_cached_by_ap_id(activity.actor)
+    end
+  end
+
   def with_joined_user_actor(query, join_type \\ :inner) do
     join(query, join_type, [activity], u in User,
       on: u.ap_id == activity.actor,

lib/pleroma/notification.ex

@@ -10,6 +10,7 @@ defmodule Pleroma.Notification do
   alias Pleroma.Object
   alias Pleroma.Pagination
   alias Pleroma.Repo
+  alias Pleroma.ThreadMute
   alias Pleroma.User
   alias Pleroma.Web.CommonAPI.Utils
   alias Pleroma.Web.Push
@@ -17,6 +18,7 @@ defmodule Pleroma.Notification do
 
   import Ecto.Query
   import Ecto.Changeset
+
   require Logger
 
   @type t :: %__MODULE__{}
@@ -101,7 +103,7 @@ defmodule Pleroma.Notification do
 
     query
     |> where([n, a], a.actor not in ^notification_muted_ap_ids)
-    |> join(:left, [n, a], tm in Pleroma.ThreadMute,
+    |> join(:left, [n, a], tm in ThreadMute,
       on: tm.user_id == ^user.id and tm.context == fragment("?->>'context'", a.data)
     )
     |> where([n, a, o, tm], is_nil(tm.user_id))
@@ -284,58 +286,108 @@ defmodule Pleroma.Notification do
   def create_notifications(%Activity{data: %{"to" => _, "type" => "Create"}} = activity) do
     object = Object.normalize(activity)
 
-    unless object && object.data["type"] == "Answer" do
+    if object && object.data["type"] == "Answer" do
-      users = get_notified_from_activity(activity)
-      notifications = Enum.map(users, fn user -> create_notification(activity, user) end)
-      {:ok, notifications}
-    else
       {:ok, []}
+    else
+      do_create_notifications(activity)
     end
   end
 
   def create_notifications(%Activity{data: %{"type" => type}} = activity)
       when type in ["Like", "Announce", "Follow", "Move", "EmojiReact"] do
-    notifications =
+    do_create_notifications(activity)
-      activity
-      |> get_notified_from_activity()
-      |> Enum.map(&create_notification(activity, &1))
-
-    {:ok, notifications}
   end
 
   def create_notifications(_), do: {:ok, []}
 
+  defp do_create_notifications(%Activity{} = activity) do
+    {enabled_receivers, disabled_receivers} = get_notified_from_activity(activity)
+    potential_receivers = enabled_receivers ++ disabled_receivers
+
+    notifications =
+      Enum.map(potential_receivers, fn user ->
+        do_send = user in enabled_receivers
+        create_notification(activity, user, do_send)
+      end)
+
+    {:ok, notifications}
+  end
+
   # TODO move to sql, too.
-  def create_notification(%Activity{} = activity, %User{} = user) do
+  def create_notification(%Activity{} = activity, %User{} = user, do_send \\ true) do
     unless skip?(activity, user) do
       notification = %Notification{user_id: user.id, activity: activity}
       {:ok, notification} = Repo.insert(notification)
 
-      ["user", "user:notification"]
+      if do_send do
-      |> Streamer.stream(notification)
+        Streamer.stream(["user", "user:notification"], notification)
-
         Push.send(notification)
+      end
+
       notification
     end
   end
 
+  @doc """
+  Returns a tuple with 2 elements:
+    {enabled notification receivers, currently disabled receivers (blocking / [thread] muting)}
+  """
   def get_notified_from_activity(activity, local_only \\ true)
 
   def get_notified_from_activity(%Activity{data: %{"type" => type}} = activity, local_only)
       when type in ["Create", "Like", "Announce", "Follow", "Move", "EmojiReact"] do
+    potential_receiver_ap_ids =
       []
       |> Utils.maybe_notify_to_recipients(activity)
       |> Utils.maybe_notify_mentioned_recipients(activity)
       |> Utils.maybe_notify_subscribers(activity)
       |> Utils.maybe_notify_followers(activity)
       |> Enum.uniq()
+
+    notification_enabled_ap_ids =
+      potential_receiver_ap_ids
+      |> exclude_relation_restricting_ap_ids(activity)
+      |> exclude_thread_muter_ap_ids(activity)
+
+    potential_receivers =
+      potential_receiver_ap_ids
+      |> Enum.uniq()
       |> User.get_users_from_set(local_only)
+
+    notification_enabled_users =
+      Enum.filter(potential_receivers, fn u -> u.ap_id in notification_enabled_ap_ids end)
+
+    {notification_enabled_users, potential_receivers -- notification_enabled_users}
   end
 
-  def get_notified_from_activity(_, _local_only), do: []
+  def get_notified_from_activity(_, _local_only), do: {[], []}
+
+  @doc "Filters out AP IDs of users basing on their relationships with activity actor user"
+  def exclude_relation_restricting_ap_ids([], _activity), do: []
+
+  def exclude_relation_restricting_ap_ids(ap_ids, %Activity{} = activity) do
+    relation_restricted_ap_ids =
+      activity
+      |> Activity.user_actor()
+      |> User.incoming_relations_ungrouped_ap_ids([
+        :block,
+        :notification_mute
+      ])
+
+    Enum.uniq(ap_ids) -- relation_restricted_ap_ids
+  end
+
+  @doc "Filters out AP IDs of users who mute activity thread"
+  def exclude_thread_muter_ap_ids([], _activity), do: []
+
+  def exclude_thread_muter_ap_ids(ap_ids, %Activity{} = activity) do
+    thread_muter_ap_ids = ThreadMute.muter_ap_ids(activity.data["context"])
+
+    Enum.uniq(ap_ids) -- thread_muter_ap_ids
+  end
 
   @spec skip?(Activity.t(), User.t()) :: boolean()
-  def skip?(activity, user) do
+  def skip?(%Activity{} = activity, %User{} = user) do
     [
       :self,
       :followers,
@@ -344,18 +396,20 @@ defmodule Pleroma.Notification do
       :non_follows,
       :recently_followed
     ]
-    |> Enum.any?(&skip?(&1, activity, user))
+    |> Enum.find(&skip?(&1, activity, user))
   end
 
+  def skip?(_, _), do: false
+
   @spec skip?(atom(), Activity.t(), User.t()) :: boolean()
-  def skip?(:self, activity, user) do
+  def skip?(:self, %Activity{} = activity, %User{} = user) do
     activity.data["actor"] == user.ap_id
   end
 
   def skip?(
         :followers,
-        activity,
+        %Activity{} = activity,
-        %{notification_settings: %{followers: false}} = user
+        %User{notification_settings: %{followers: false}} = user
       ) do
     actor = activity.data["actor"]
     follower = User.get_cached_by_ap_id(actor)
@@ -364,15 +418,19 @@ defmodule Pleroma.Notification do
 
   def skip?(
         :non_followers,
-        activity,
+        %Activity{} = activity,
-        %{notification_settings: %{non_followers: false}} = user
+        %User{notification_settings: %{non_followers: false}} = user
       ) do
     actor = activity.data["actor"]
     follower = User.get_cached_by_ap_id(actor)
     !User.following?(follower, user)
   end
 
-  def skip?(:follows, activity, %{notification_settings: %{follows: false}} = user) do
+  def skip?(
+        :follows,
+        %Activity{} = activity,
+        %User{notification_settings: %{follows: false}} = user
+      ) do
     actor = activity.data["actor"]
     followed = User.get_cached_by_ap_id(actor)
     User.following?(user, followed)
@@ -380,15 +438,16 @@ defmodule Pleroma.Notification do
 
   def skip?(
         :non_follows,
-        activity,
+        %Activity{} = activity,
-        %{notification_settings: %{non_follows: false}} = user
+        %User{notification_settings: %{non_follows: false}} = user
       ) do
     actor = activity.data["actor"]
     followed = User.get_cached_by_ap_id(actor)
     !User.following?(user, followed)
   end
 
-  def skip?(:recently_followed, %{data: %{"type" => "Follow"}} = activity, user) do
+  # To do: consider defining recency in hours and checking FollowingRelationship with a single SQL
+  def skip?(:recently_followed, %Activity{data: %{"type" => "Follow"}} = activity, %User{} = user) do
     actor = activity.data["actor"]
 
     Notification.for_user(user)

lib/pleroma/thread_mute.ex

@@ -9,7 +9,8 @@ defmodule Pleroma.ThreadMute do
   alias Pleroma.ThreadMute
   alias Pleroma.User
 
-  require Ecto.Query
+  import Ecto.Changeset
+  import Ecto.Query
 
   schema "thread_mutes" do
     belongs_to(:user, User, type: FlakeId.Ecto.CompatType)
@@ -18,19 +19,43 @@ defmodule Pleroma.ThreadMute do
 
   def changeset(mute, params \\ %{}) do
     mute
-    |> Ecto.Changeset.cast(params, [:user_id, :context])
+    |> cast(params, [:user_id, :context])
-    |> Ecto.Changeset.foreign_key_constraint(:user_id)
+    |> foreign_key_constraint(:user_id)
-    |> Ecto.Changeset.unique_constraint(:user_id, name: :unique_index)
+    |> unique_constraint(:user_id, name: :unique_index)
   end
 
   def query(user_id, context) do
     {:ok, user_id} = FlakeId.Ecto.CompatType.dump(user_id)
 
     ThreadMute
-    |> Ecto.Query.where(user_id: ^user_id)
+    |> where(user_id: ^user_id)
-    |> Ecto.Query.where(context: ^context)
+    |> where(context: ^context)
   end
 
+  def muters_query(context) do
+    ThreadMute
+    |> join(:inner, [tm], u in assoc(tm, :user))
+    |> where([tm], tm.context == ^context)
+    |> select([tm, u], u.ap_id)
+  end
+
+  def muter_ap_ids(context, ap_ids \\ nil)
+
+  def muter_ap_ids(context, ap_ids) when context not in [nil, ""] do
+    context
+    |> muters_query()
+    |> maybe_filter_on_ap_id(ap_ids)
+    |> Repo.all()
+  end
+
+  def muter_ap_ids(_context, _ap_ids), do: []
+
+  defp maybe_filter_on_ap_id(query, ap_ids) when is_list(ap_ids) do
+    where(query, [tm, u], u.ap_id in ^ap_ids)
+  end
+
+  defp maybe_filter_on_ap_id(query, _ap_ids), do: query
+
   def add_mute(user_id, context) do
     %ThreadMute{}
     |> changeset(%{user_id: user_id, context: context})

lib/pleroma/user.ex

@@ -149,22 +149,26 @@ defmodule Pleroma.User do
            {outgoing_relation, outgoing_relation_target},
            {incoming_relation, incoming_relation_source}
          ]} <- @user_relationships_config do
-      # Definitions of `has_many :blocker_blocks`, `has_many :muter_mutes` etc.
+      # Definitions of `has_many` relations: :blocker_blocks, :muter_mutes, :reblog_muter_mutes,
+      #   :notification_muter_mutes, :subscribee_subscriptions
       has_many(outgoing_relation, UserRelationship,
         foreign_key: :source_id,
         where: [relationship_type: relationship_type]
       )
 
-      # Definitions of `has_many :blockee_blocks`, `has_many :mutee_mutes` etc.
+      # Definitions of `has_many` relations: :blockee_blocks, :mutee_mutes, :reblog_mutee_mutes,
+      #   :notification_mutee_mutes, :subscriber_subscriptions
       has_many(incoming_relation, UserRelationship,
         foreign_key: :target_id,
         where: [relationship_type: relationship_type]
       )
 
-      # Definitions of `has_many :blocked_users`, `has_many :muted_users` etc.
+      # Definitions of `has_many` relations: :blocked_users, :muted_users, :reblog_muted_users,
+      #   :notification_muted_users, :subscriber_users
       has_many(outgoing_relation_target, through: [outgoing_relation, :target])
 
-      # Definitions of `has_many :blocker_users`, `has_many :muter_users` etc.
+      # Definitions of `has_many` relations: :blocker_users, :muter_users, :reblog_muter_users,
+      #   :notification_muter_users, :subscribee_users
       has_many(incoming_relation_source, through: [incoming_relation, :source])
     end
 
@@ -184,7 +188,9 @@ defmodule Pleroma.User do
 
   for {_relationship_type, [{_outgoing_relation, outgoing_relation_target}, _]} <-
         @user_relationships_config do
-    # Definitions of `blocked_users_relation/1`, `muted_users_relation/1`, etc.
+    # `def blocked_users_relation/2`, `def muted_users_relation/2`,
+    #   `def reblog_muted_users_relation/2`, `def notification_muted_users/2`,
+    #   `def subscriber_users/2`
     def unquote(:"#{outgoing_relation_target}_relation")(user, restrict_deactivated? \\ false) do
       target_users_query = assoc(user, unquote(outgoing_relation_target))
 
@@ -195,7 +201,8 @@ defmodule Pleroma.User do
       end
     end
 
-    # Definitions of `blocked_users/1`, `muted_users/1`, etc.
+    # `def blocked_users/2`, `def muted_users/2`, `def reblog_muted_users/2`,
+    #   `def notification_muted_users/2`, `def subscriber_users/2`
     def unquote(outgoing_relation_target)(user, restrict_deactivated? \\ false) do
       __MODULE__
       |> apply(unquote(:"#{outgoing_relation_target}_relation"), [
@@ -205,7 +212,8 @@ defmodule Pleroma.User do
       |> Repo.all()
     end
 
-    # Definitions of `blocked_users_ap_ids/1`, `muted_users_ap_ids/1`, etc.
+    # `def blocked_users_ap_ids/2`, `def muted_users_ap_ids/2`, `def reblog_muted_users_ap_ids/2`,
+    #   `def notification_muted_users_ap_ids/2`, `def subscriber_users_ap_ids/2`
     def unquote(:"#{outgoing_relation_target}_ap_ids")(user, restrict_deactivated? \\ false) do
       __MODULE__
       |> apply(unquote(:"#{outgoing_relation_target}_relation"), [
@@ -1217,7 +1225,9 @@ defmodule Pleroma.User do
   E.g. `outgoing_relations_ap_ids(user, [:block])` -> `%{block: ["https://some.site/users/userapid"]}`
   """
   @spec outgoing_relations_ap_ids(User.t(), list(atom())) :: %{atom() => list(String.t())}
-  def outgoing_relations_ap_ids(_, []), do: %{}
+  def outgoing_relations_ap_ids(_user, []), do: %{}
+
+  def outgoing_relations_ap_ids(nil, _relationship_types), do: %{}
 
   def outgoing_relations_ap_ids(%User{} = user, relationship_types)
       when is_list(relationship_types) do
@@ -1238,6 +1248,30 @@ defmodule Pleroma.User do
     )
   end
 
+  def incoming_relations_ungrouped_ap_ids(user, relationship_types, ap_ids \\ nil)
+
+  def incoming_relations_ungrouped_ap_ids(_user, [], _ap_ids), do: []
+
+  def incoming_relations_ungrouped_ap_ids(nil, _relationship_types, _ap_ids), do: []
+
+  def incoming_relations_ungrouped_ap_ids(%User{} = user, relationship_types, ap_ids)
+      when is_list(relationship_types) do
+    user
+    |> assoc(:incoming_relationships)
+    |> join(:inner, [user_rel], u in assoc(user_rel, :source))
+    |> where([user_rel, u], user_rel.relationship_type in ^relationship_types)
+    |> maybe_filter_on_ap_id(ap_ids)
+    |> select([user_rel, u], u.ap_id)
+    |> distinct(true)
+    |> Repo.all()
+  end
+
+  defp maybe_filter_on_ap_id(query, ap_ids) when is_list(ap_ids) do
+    where(query, [user_rel, u], u.ap_id in ^ap_ids)
+  end
+
+  defp maybe_filter_on_ap_id(query, _ap_ids), do: query
+
   def deactivate_async(user, status \\ true) do
     BackgroundWorker.enqueue("deactivate_user", %{"user_id" => user.id, "status" => status})
   end

lib/pleroma/user_relationship.ex

@@ -21,15 +21,18 @@ defmodule Pleroma.UserRelationship do
   end
 
   for relationship_type <- Keyword.keys(UserRelationshipTypeEnum.__enum_map__()) do
-    # Definitions of `create_block/2`, `create_mute/2` etc.
+    # `def create_block/2`, `def create_mute/2`, `def create_reblog_mute/2`,
+    #   `def create_notification_mute/2`, `def create_inverse_subscription/2`
     def unquote(:"create_#{relationship_type}")(source, target),
       do: create(unquote(relationship_type), source, target)
 
-    # Definitions of `delete_block/2`, `delete_mute/2` etc.
+    # `def delete_block/2`, `def delete_mute/2`, `def delete_reblog_mute/2`,
+    #   `def delete_notification_mute/2`, `def delete_inverse_subscription/2`
     def unquote(:"delete_#{relationship_type}")(source, target),
       do: delete(unquote(relationship_type), source, target)
 
-    # Definitions of `block_exists?/2`, `mute_exists?/2` etc.
+    # `def block_exists?/2`, `def mute_exists?/2`, `def reblog_mute_exists?/2`,
+    #   `def notification_mute_exists?/2`, `def inverse_subscription_exists?/2`
     def unquote(:"#{relationship_type}_exists?")(source, target),
       do: exists?(unquote(relationship_type), source, target)
   end

lib/pleroma/web/activity_pub/transmogrifier.ex

@@ -1108,13 +1108,11 @@ defmodule Pleroma.Web.ActivityPub.Transmogrifier do
   end
 
   def add_mention_tags(object) do
-    mentions =
+    {enabled_receivers, disabled_receivers} = Utils.get_notified_from_object(object)
-      object
+    potential_receivers = enabled_receivers ++ disabled_receivers
-      |> Utils.get_notified_from_object()
+    mentions = Enum.map(potential_receivers, &build_mention_tag/1)
-      |> Enum.map(&build_mention_tag/1)
 
     tags = object["tag"] || []
-
     Map.put(object, "tag", tags ++ mentions)
   end
 

test/notification_test.exs

@@ -6,12 +6,14 @@ defmodule Pleroma.NotificationTest do
   use Pleroma.DataCase
 
   import Pleroma.Factory
+  import Mock
 
   alias Pleroma.Notification
   alias Pleroma.Tests.ObanHelpers
   alias Pleroma.User
   alias Pleroma.Web.ActivityPub.Transmogrifier
   alias Pleroma.Web.CommonAPI
+  alias Pleroma.Web.Push
   alias Pleroma.Web.Streamer
 
   describe "create_notifications" do
@@ -382,7 +384,7 @@ defmodule Pleroma.NotificationTest do
     end
   end
 
-  describe "notification target determination" do
+  describe "notification target determination / get_notified_from_activity/2" do
     test "it sends notifications to addressed users in new messages" do
       user = insert(:user)
       other_user = insert(:user)
@@ -392,7 +394,9 @@ defmodule Pleroma.NotificationTest do
           "status" => "hey @#{other_user.nickname}!"
         })
 
-      assert other_user in Notification.get_notified_from_activity(activity)
+      {enabled_receivers, _disabled_receivers} = Notification.get_notified_from_activity(activity)
+
+      assert other_user in enabled_receivers
     end
 
     test "it sends notifications to mentioned users in new messages" do
@@ -420,7 +424,9 @@ defmodule Pleroma.NotificationTest do
 
       {:ok, activity} = Transmogrifier.handle_incoming(create_activity)
 
-      assert other_user in Notification.get_notified_from_activity(activity)
+      {enabled_receivers, _disabled_receivers} = Notification.get_notified_from_activity(activity)
+
+      assert other_user in enabled_receivers
     end
 
     test "it does not send notifications to users who are only cc in new messages" do
@@ -442,7 +448,9 @@ defmodule Pleroma.NotificationTest do
 
       {:ok, activity} = Transmogrifier.handle_incoming(create_activity)
 
-      assert other_user not in Notification.get_notified_from_activity(activity)
+      {enabled_receivers, _disabled_receivers} = Notification.get_notified_from_activity(activity)
+
+      assert other_user not in enabled_receivers
     end
 
     test "it does not send notification to mentioned users in likes" do
@@ -457,7 +465,10 @@ defmodule Pleroma.NotificationTest do
 
       {:ok, activity_two, _} = CommonAPI.favorite(activity_one.id, third_user)
 
-      assert other_user not in Notification.get_notified_from_activity(activity_two)
+      {enabled_receivers, _disabled_receivers} =
+        Notification.get_notified_from_activity(activity_two)
+
+      assert other_user not in enabled_receivers
     end
 
     test "it does not send notification to mentioned users in announces" do
@@ -472,7 +483,96 @@ defmodule Pleroma.NotificationTest do
 
       {:ok, activity_two, _} = CommonAPI.repeat(activity_one.id, third_user)
 
-      assert other_user not in Notification.get_notified_from_activity(activity_two)
+      {enabled_receivers, _disabled_receivers} =
+        Notification.get_notified_from_activity(activity_two)
+
+      assert other_user not in enabled_receivers
+    end
+
+    test_with_mock "it returns blocking recipient in disabled recipients list",
+                   Push,
+                   [:passthrough],
+                   [] do
+      user = insert(:user)
+      other_user = insert(:user)
+      {:ok, _user_relationship} = User.block(other_user, user)
+
+      {:ok, activity} = CommonAPI.post(user, %{"status" => "hey @#{other_user.nickname}!"})
+
+      {enabled_receivers, disabled_receivers} = Notification.get_notified_from_activity(activity)
+
+      assert [] == enabled_receivers
+      assert [other_user] == disabled_receivers
+
+      assert 1 == length(Repo.all(Notification))
+      refute called(Push.send(:_))
+    end
+
+    test_with_mock "it returns notification-muting recipient in disabled recipients list",
+                   Push,
+                   [:passthrough],
+                   [] do
+      user = insert(:user)
+      other_user = insert(:user)
+      {:ok, _user_relationships} = User.mute(other_user, user)
+
+      {:ok, activity} = CommonAPI.post(user, %{"status" => "hey @#{other_user.nickname}!"})
+
+      {enabled_receivers, disabled_receivers} = Notification.get_notified_from_activity(activity)
+
+      assert [] == enabled_receivers
+      assert [other_user] == disabled_receivers
+
+      assert 1 == length(Repo.all(Notification))
+      refute called(Push.send(:_))
+    end
+
+    test_with_mock "it returns thread-muting recipient in disabled recipients list",
+                   Push,
+                   [:passthrough],
+                   [] do
+      user = insert(:user)
+      other_user = insert(:user)
+
+      {:ok, activity} = CommonAPI.post(user, %{"status" => "hey @#{other_user.nickname}!"})
+
+      {:ok, _} = CommonAPI.add_mute(other_user, activity)
+
+      {:ok, same_context_activity} =
+        CommonAPI.post(user, %{
+          "status" => "hey-hey-hey @#{other_user.nickname}!",
+          "in_reply_to_status_id" => activity.id
+        })
+
+      {enabled_receivers, disabled_receivers} =
+        Notification.get_notified_from_activity(same_context_activity)
+
+      assert [other_user] == disabled_receivers
+      refute other_user in enabled_receivers
+
+      [pre_mute_notification, post_mute_notification] =
+        Repo.all(from(n in Notification, where: n.user_id == ^other_user.id, order_by: n.id))
+
+      pre_mute_notification_id = pre_mute_notification.id
+      post_mute_notification_id = post_mute_notification.id
+
+      assert called(
+               Push.send(
+                 :meck.is(fn
+                   %Notification{id: ^pre_mute_notification_id} -> true
+                   _ -> false
+                 end)
+               )
+             )
+
+      refute called(
+               Push.send(
+                 :meck.is(fn
+                   %Notification{id: ^post_mute_notification_id} -> true
+                   _ -> false
+                 end)
+               )
+             )
     end
   end