Merge branch 'oban/cancel-federation' into 'develop'

Cancel queued publishing jobs when undoing an activity

See merge request pleroma/pleroma!4186

changelog.d/oban-cancel-federation.add

@@ -0,0 +1 @@
+Deleting, Unfavoriting, Unrepeating, or Unreacting will cancel undelivered publishing jobs for the original activity.

lib/pleroma/web/common_api.ex

@@ -19,6 +19,7 @@ defmodule Pleroma.Web.CommonAPI do
   alias Pleroma.Web.ActivityPub.Visibility
   alias Pleroma.Web.CommonAPI.ActivityDraft
 
+  import Ecto.Query, only: [where: 3]
   import Pleroma.Web.Gettext
   import Pleroma.Web.CommonAPI.Utils
 
@@ -156,6 +157,7 @@ defmodule Pleroma.Web.CommonAPI do
   def delete(activity_id, user) do
     with {_, %Activity{data: %{"object" => _, "type" => "Create"}} = activity} <-
            {:find_activity, Activity.get_by_id(activity_id, filter: [])},
+         {_, {:ok, _}} <- {:cancel_jobs, maybe_cancel_jobs(activity)},
          {_, %Object{} = object, _} <-
            {:find_object, Object.normalize(activity, fetch: false), activity},
          true <- User.privileged?(user, :messages_delete) || user.ap_id == object.data["actor"],
@@ -223,6 +225,7 @@ defmodule Pleroma.Web.CommonAPI do
            {:find_activity, Activity.get_by_id(id)},
          %Object{} = note <- Object.normalize(activity, fetch: false),
          %Activity{} = announce <- Utils.get_existing_announce(user.ap_id, note),
+         {_, {:ok, _}} <- {:cancel_jobs, maybe_cancel_jobs(announce)},
          {:ok, undo, _} <- Builder.undo(user, announce),
          {:ok, activity, _} <- Pipeline.common_pipeline(undo, local: true) do
       {:ok, activity}
@@ -275,6 +278,7 @@ defmodule Pleroma.Web.CommonAPI do
            {:find_activity, Activity.get_by_id(id)},
          %Object{} = note <- Object.normalize(activity, fetch: false),
          %Activity{} = like <- Utils.get_existing_like(user.ap_id, note),
+         {_, {:ok, _}} <- {:cancel_jobs, maybe_cancel_jobs(like)},
          {:ok, undo, _} <- Builder.undo(user, like),
          {:ok, activity, _} <- Pipeline.common_pipeline(undo, local: true) do
       {:ok, activity}
@@ -298,6 +302,7 @@ defmodule Pleroma.Web.CommonAPI do
 
   def unreact_with_emoji(id, user, emoji) do
     with %Activity{} = reaction_activity <- Utils.get_latest_reaction(id, user, emoji),
+         {_, {:ok, _}} <- {:cancel_jobs, maybe_cancel_jobs(reaction_activity)},
          {:ok, undo, _} <- Builder.undo(user, reaction_activity),
          {:ok, activity, _} <- Pipeline.common_pipeline(undo, local: true) do
       {:ok, activity}
@@ -671,4 +676,14 @@ defmodule Pleroma.Web.CommonAPI do
         nil
     end
   end
+
+  defp maybe_cancel_jobs(%Activity{data: %{"id" => ap_id}}) do
+    Oban.Job
+    |> where([j], j.worker == "Pleroma.Workers.PublisherWorker")
+    |> where([j], j.args["op"] == "publish_one")
+    |> where([j], j.args["params"]["id"] == ^ap_id)
+    |> Oban.cancel_all_jobs()
+  end
+
+  defp maybe_cancel_jobs(_), do: {:ok, 0}
 end

test/pleroma/web/common_api_test.exs

@@ -13,6 +13,7 @@ defmodule Pleroma.Web.CommonAPITest do
   alias Pleroma.Object
   alias Pleroma.Repo
   alias Pleroma.Rule
+  alias Pleroma.Tests.ObanHelpers
   alias Pleroma.UnstubbedConfigMock, as: ConfigMock
   alias Pleroma.User
   alias Pleroma.Web.ActivityPub.ActivityPub
@@ -22,7 +23,7 @@ defmodule Pleroma.Web.CommonAPITest do
   alias Pleroma.Web.CommonAPI
   alias Pleroma.Workers.PollWorker
 
-  import Ecto.Query, only: [from: 2]
+  import Ecto.Query, only: [from: 2, where: 3]
   import Mock
   import Mox
   import Pleroma.Factory
@@ -1920,4 +1921,217 @@ defmodule Pleroma.Web.CommonAPITest do
       assert [] = announces
     end
   end
+
+  describe "Oban jobs are cancelled" do
+    setup do
+      clear_config([:instance, :federating], true)
+
+      local_user = insert(:user)
+
+      remote_one =
+        insert(:user, %{
+          local: false,
+          nickname: "nick1@domain.com",
+          ap_id: "https://domain.com/users/nick1",
+          inbox: "https://domain.com/users/nick1/inbox",
+          shared_inbox: "https://domain.com/inbox"
+        })
+
+      remote_two =
+        insert(:user, %{
+          local: false,
+          nickname: "nick2@example.com",
+          ap_id: "https://example.com/users/nick2",
+          inbox: "https://example.com/users/nick2/inbox",
+          shared_inbox: "https://example.com/inbox"
+        })
+
+      %{local_user: local_user, remote_one: remote_one, remote_two: remote_two}
+    end
+
+    test "when deleting posts", %{
+      local_user: local_user,
+      remote_one: remote_one,
+      remote_two: remote_two
+    } do
+      {:ok, _, _} = Pleroma.User.follow(remote_one, local_user)
+      {:ok, _, _} = Pleroma.User.follow(remote_two, local_user)
+
+      {:ok, %{data: %{"id" => ap_id}} = activity} =
+        CommonAPI.post(local_user, %{status: "Happy Friday everyone!"})
+
+      # Generate the publish_one jobs
+      ObanHelpers.perform_all()
+
+      publish_one_jobs =
+        all_enqueued()
+        |> Enum.filter(fn job ->
+          match?(
+            %{
+              state: "available",
+              queue: "federator_outgoing",
+              worker: "Pleroma.Workers.PublisherWorker",
+              args: %{"op" => "publish_one", "params" => %{"id" => ^ap_id}}
+            },
+            job
+          )
+        end)
+
+      assert length(publish_one_jobs) == 2
+
+      # The delete should have triggered cancelling the publish_one jobs
+      assert {:ok, _delete} = CommonAPI.delete(activity.id, local_user)
+
+      # all_enqueued/1 will not return cancelled jobs
+      cancelled_jobs =
+        Oban.Job
+        |> where([j], j.worker == "Pleroma.Workers.PublisherWorker")
+        |> where([j], j.state == "cancelled")
+        |> where([j], j.args["op"] == "publish_one")
+        |> where([j], j.args["params"]["id"] == ^ap_id)
+        |> Pleroma.Repo.all()
+
+      assert length(cancelled_jobs) == 2
+    end
+
+    test "when unfavoriting posts", %{
+      local_user: local_user,
+      remote_one: remote_user
+    } do
+      {:ok, activity} =
+        CommonAPI.post(remote_user, %{status: "I like turtles!"})
+
+      {:ok, %{data: %{"id" => ap_id}} = _favorite} =
+        CommonAPI.favorite(local_user, activity.id)
+
+      # Generate the publish_one jobs
+      ObanHelpers.perform_all()
+
+      publish_one_jobs =
+        all_enqueued()
+        |> Enum.filter(fn job ->
+          match?(
+            %{
+              state: "available",
+              queue: "federator_outgoing",
+              worker: "Pleroma.Workers.PublisherWorker",
+              args: %{"op" => "publish_one", "params" => %{"id" => ^ap_id}}
+            },
+            job
+          )
+        end)
+
+      assert length(publish_one_jobs) == 1
+
+      # The unfavorite should have triggered cancelling the publish_one jobs
+      assert {:ok, _unfavorite} = CommonAPI.unfavorite(activity.id, local_user)
+
+      # all_enqueued/1 will not return cancelled jobs
+      cancelled_jobs =
+        Oban.Job
+        |> where([j], j.worker == "Pleroma.Workers.PublisherWorker")
+        |> where([j], j.state == "cancelled")
+        |> where([j], j.args["op"] == "publish_one")
+        |> where([j], j.args["params"]["id"] == ^ap_id)
+        |> Pleroma.Repo.all()
+
+      assert length(cancelled_jobs) == 1
+    end
+
+    test "when unboosting posts", %{
+      local_user: local_user,
+      remote_one: remote_one,
+      remote_two: remote_two
+    } do
+      {:ok, _, _} = Pleroma.User.follow(remote_one, local_user)
+      {:ok, _, _} = Pleroma.User.follow(remote_two, local_user)
+
+      {:ok, activity} =
+        CommonAPI.post(remote_one, %{status: "This is an unpleasant post"})
+
+      {:ok, %{data: %{"id" => ap_id}} = _repeat} =
+        CommonAPI.repeat(activity.id, local_user)
+
+      # Generate the publish_one jobs
+      ObanHelpers.perform_all()
+
+      publish_one_jobs =
+        all_enqueued()
+        |> Enum.filter(fn job ->
+          match?(
+            %{
+              state: "available",
+              queue: "federator_outgoing",
+              worker: "Pleroma.Workers.PublisherWorker",
+              args: %{"op" => "publish_one", "params" => %{"id" => ^ap_id}}
+            },
+            job
+          )
+        end)
+
+      assert length(publish_one_jobs) == 2
+
+      # The unrepeat should have triggered cancelling the publish_one jobs
+      assert {:ok, _unfavorite} = CommonAPI.unrepeat(activity.id, local_user)
+
+      # all_enqueued/1 will not return cancelled jobs
+      cancelled_jobs =
+        Oban.Job
+        |> where([j], j.worker == "Pleroma.Workers.PublisherWorker")
+        |> where([j], j.state == "cancelled")
+        |> where([j], j.args["op"] == "publish_one")
+        |> where([j], j.args["params"]["id"] == ^ap_id)
+        |> Pleroma.Repo.all()
+
+      assert length(cancelled_jobs) == 2
+    end
+
+    test "when unreacting to posts", %{
+      local_user: local_user,
+      remote_one: remote_one,
+      remote_two: remote_two
+    } do
+      {:ok, _, _} = Pleroma.User.follow(remote_one, local_user)
+      {:ok, _, _} = Pleroma.User.follow(remote_two, local_user)
+
+      {:ok, activity} =
+        CommonAPI.post(remote_one, %{status: "Gang gang!!!!"})
+
+      {:ok, %{data: %{"id" => ap_id}} = _react} =
+        CommonAPI.react_with_emoji(activity.id, local_user, "👍")
+
+      # Generate the publish_one jobs
+      ObanHelpers.perform_all()
+
+      publish_one_jobs =
+        all_enqueued()
+        |> Enum.filter(fn job ->
+          match?(
+            %{
+              state: "available",
+              queue: "federator_outgoing",
+              worker: "Pleroma.Workers.PublisherWorker",
+              args: %{"op" => "publish_one", "params" => %{"id" => ^ap_id}}
+            },
+            job
+          )
+        end)
+
+      assert length(publish_one_jobs) == 2
+
+      # The unreact should have triggered cancelling the publish_one jobs
+      assert {:ok, _unreact} = CommonAPI.unreact_with_emoji(activity.id, local_user, "👍")
+
+      # all_enqueued/1 will not return cancelled jobs
+      cancelled_jobs =
+        Oban.Job
+        |> where([j], j.worker == "Pleroma.Workers.PublisherWorker")
+        |> where([j], j.state == "cancelled")
+        |> where([j], j.args["op"] == "publish_one")
+        |> where([j], j.args["params"]["id"] == ^ap_id)
+        |> Pleroma.Repo.all()
+
+      assert length(cancelled_jobs) == 2
+    end
+  end
 end