Merge branch 'priority_activities' into 'develop'

Priority activities

See merge request pleroma/pleroma!4004

changelog.d/prioritize-direct-recipients.add

@@ -0,0 +1 @@
+- Prioritize mentioned recipients (i.e., those that are not just followers) when federating.

lib/pleroma/user.ex

@@ -2681,6 +2681,8 @@ def update_last_active_at(%__MODULE__{local: true} = user) do
     |> update_and_set_cache()
   end
 
+  def update_last_active_at(user), do: user
+
   def active_user_count(days \\ 30) do
     active_after = Timex.shift(NaiveDateTime.utc_now(), days: -days)
 

lib/pleroma/web/activity_pub/publisher.ex

@@ -118,7 +118,7 @@ defp should_federate?(inbox, public) do
     end
   end
 
-  @spec recipients(User.t(), Activity.t()) :: list(User.t()) | []
+  @spec recipients(User.t(), Activity.t()) :: [[User.t()]]
   defp recipients(actor, activity) do
     followers =
       if actor.follower_address in activity.recipients do
@@ -138,7 +138,10 @@ defp recipients(actor, activity) do
           []
       end
 
-    Pleroma.Web.Federator.Publisher.remote_users(actor, activity) ++ followers ++ fetchers
+    mentioned = Pleroma.Web.Federator.Publisher.remote_users(actor, activity)
+    non_mentioned = (followers ++ fetchers) -- mentioned
+
+    [mentioned, non_mentioned]
   end
 
   defp get_cc_ap_ids(ap_id, recipients) do
@@ -195,15 +198,19 @@ def publish(%User{} = actor, %{data: %{"bcc" => bcc}} = activity)
     public = is_public?(activity)
     {:ok, data} = Transmogrifier.prepare_outgoing(activity.data)
 
-    recipients = recipients(actor, activity)
+    [priority_recipients, recipients] = recipients(actor, activity)
 
     inboxes =
+      [priority_recipients, recipients]
+      |> Enum.map(fn recipients ->
         recipients
         |> Enum.map(fn actor -> actor.inbox end)
         |> Enum.filter(fn inbox -> should_federate?(inbox, public) end)
         |> Instances.filter_reachable()
+      end)
 
     Repo.checkout(fn ->
+      Enum.each(inboxes, fn inboxes ->
         Enum.each(inboxes, fn {inbox, unreachable_since} ->
           %User{ap_id: ap_id} = Enum.find(recipients, fn actor -> actor.inbox == inbox end)
 
@@ -225,6 +232,7 @@ def publish(%User{} = actor, %{data: %{"bcc" => bcc}} = activity)
           })
         end)
       end)
+    end)
   end
 
   # Publishes an activity to all relevant peers.
@@ -239,12 +247,19 @@ def publish(%User{} = actor, %Activity{} = activity) do
     {:ok, data} = Transmogrifier.prepare_outgoing(activity.data)
     json = Jason.encode!(data)
 
+    [priority_inboxes, inboxes] =
       recipients(actor, activity)
-    |> Enum.map(fn %User{} = user ->
+      |> Enum.map(fn recipients ->
-      determine_inbox(activity, user)
+        recipients
-    end)
+        |> Enum.map(fn actor -> actor.inbox end)
-    |> Enum.uniq()
         |> Enum.filter(fn inbox -> should_federate?(inbox, public) end)
+      end)
+
+    inboxes = inboxes -- priority_inboxes
+
+    [{priority_inboxes, 0}, {inboxes, 1}]
+    |> Enum.each(fn {inboxes, priority} ->
+      inboxes
       |> Instances.filter_reachable()
       |> Enum.each(fn {inbox, unreachable_since} ->
         Pleroma.Web.Federator.Publisher.enqueue_one(
@@ -255,9 +270,13 @@ def publish(%User{} = actor, %Activity{} = activity) do
             actor_id: actor.id,
             id: activity.data["id"],
             unreachable_since: unreachable_since
-        }
+          },
+          priority: priority
         )
       end)
+    end)
+
+    :ok
   end
 
   def gather_webfinger_links(%User{} = user) do

lib/pleroma/web/federator/publisher.ex

@@ -29,11 +29,12 @@ defmodule Pleroma.Web.Federator.Publisher do
   @doc """
   Enqueue publishing a single activity.
   """
-  @spec enqueue_one(module(), Map.t()) :: :ok
+  @spec enqueue_one(module(), Map.t(), Keyword.t()) :: {:ok, %Oban.Job{}}
-  def enqueue_one(module, %{} = params) do
+  def enqueue_one(module, %{} = params, worker_args \\ []) do
     PublisherWorker.enqueue(
       "publish_one",
-      %{"module" => to_string(module), "params" => params}
+      %{"module" => to_string(module), "params" => params},
+      worker_args
     )
   end
 

test/pleroma/web/activity_pub/publisher_test.exs

@@ -331,11 +331,40 @@ test "publish to url with with different ports" do
       assert res == :ok
 
       assert called(
-               Pleroma.Web.Federator.Publisher.enqueue_one(Publisher, %{
+               Pleroma.Web.Federator.Publisher.enqueue_one(
+                 Publisher,
+                 %{
                    inbox: "https://domain.com/users/nick1/inbox",
                    actor_id: actor.id,
                    id: note_activity.data["id"]
-               })
+                 },
+                 priority: 1
+               )
+             )
+    end
+
+    test_with_mock "Publishes to directly addressed actors with higher priority.",
+                   Pleroma.Web.Federator.Publisher,
+                   [:passthrough],
+                   [] do
+      note_activity = insert(:direct_note_activity)
+
+      actor = Pleroma.User.get_by_ap_id(note_activity.data["actor"])
+
+      res = Publisher.publish(actor, note_activity)
+
+      assert res == :ok
+
+      assert called(
+               Pleroma.Web.Federator.Publisher.enqueue_one(
+                 Publisher,
+                 %{
+                   inbox: :_,
+                   actor_id: actor.id,
+                   id: note_activity.data["id"]
+                 },
+                 priority: 0
+               )
              )
     end
 
@@ -414,19 +443,27 @@ test "publish to url with with different ports" do
       assert res == :ok
 
       assert called(
-               Pleroma.Web.Federator.Publisher.enqueue_one(Publisher, %{
+               Pleroma.Web.Federator.Publisher.enqueue_one(
+                 Publisher,
+                 %{
                    inbox: "https://domain.com/users/nick1/inbox",
                    actor_id: actor.id,
                    id: delete.data["id"]
-               })
+                 },
+                 priority: 1
+               )
              )
 
       assert called(
-               Pleroma.Web.Federator.Publisher.enqueue_one(Publisher, %{
+               Pleroma.Web.Federator.Publisher.enqueue_one(
+                 Publisher,
+                 %{
                    inbox: "https://domain2.com/users/nick1/inbox",
                    actor_id: actor.id,
                    id: delete.data["id"]
-               })
+                 },
+                 priority: 1
+               )
              )
     end
   end

test/support/factory.ex

@@ -212,7 +212,7 @@ def listen_factory do
   end
 
   def direct_note_factory do
-    user2 = insert(:user)
+    user2 = insert(:user, local: false, inbox: "http://example.com/inbox")
 
     %Pleroma.Object{data: data} = note_factory()
     %Pleroma.Object{data: Map.merge(data, %{"to" => [user2.ap_id]})}