Pipeline: Always run common_pipeline in a transaction for now.

lib/pleroma/web/activity_pub/pipeline.ex

@@ -5,6 +5,7 @@
 defmodule Pleroma.Web.ActivityPub.Pipeline do
   alias Pleroma.Activity
   alias Pleroma.Object
+  alias Pleroma.Repo
   alias Pleroma.Web.ActivityPub.ActivityPub
   alias Pleroma.Web.ActivityPub.MRF
   alias Pleroma.Web.ActivityPub.ObjectValidator
@@ -14,6 +15,16 @@ defmodule Pleroma.Web.ActivityPub.Pipeline do
   @spec common_pipeline(map(), keyword()) ::
           {:ok, Activity.t() | Object.t(), keyword()} | {:error, any()}
   def common_pipeline(object, meta) do
+    case Repo.transaction(fn -> do_common_pipeline(object, meta) end) do
+      {:ok, value} ->
+        value
+
+      {:error, e} ->
+        {:error, e}
+    end
+  end
+
+  def do_common_pipeline(object, meta) do
     with {_, {:ok, validated_object, meta}} <-
            {:validate_object, ObjectValidator.validate(object, meta)},
          {_, {:ok, mrfd_object}} <- {:mrf_object, MRF.filter(validated_object)},

lib/pleroma/web/activity_pub/transmogrifier/chat_message_handling.ex

@@ -3,24 +3,18 @@
 # SPDX-License-Identifier: AGPL-3.0-only
 
 defmodule Pleroma.Web.ActivityPub.Transmogrifier.ChatMessageHandling do
-  alias Pleroma.Repo
   alias Pleroma.Web.ActivityPub.Pipeline
 
   def handle_incoming(
         %{"type" => "Create", "object" => %{"type" => "ChatMessage"}} = data,
         _options
       ) do
-    # Create has to be run inside a transaction because the object is created as a side effect.
-    # If this does not work, we need to roll back creating the activity.
-    case Repo.transaction(fn -> Pipeline.common_pipeline(data, local: false) end) do
-      {:ok, {:ok, activity, _}} ->
+    case Pipeline.common_pipeline(data, local: false) do
+      {:ok, activity, _} ->
         {:ok, activity}
 
-      {:ok, e} ->
+      e ->
         e
-
-      {:error, e} ->
-        {:error, e}
     end
   end
 end

lib/pleroma/web/common_api/common_api.ex

@@ -9,7 +9,6 @@ defmodule Pleroma.Web.CommonAPI do
   alias Pleroma.FollowingRelationship
   alias Pleroma.Formatter
   alias Pleroma.Object
-  alias Pleroma.Repo
   alias Pleroma.ThreadMute
   alias Pleroma.User
   alias Pleroma.UserRelationship
@@ -26,36 +25,27 @@ defmodule Pleroma.Web.CommonAPI do
   require Logger
 
   def post_chat_message(%User{} = user, %User{} = recipient, content) do
-    transaction =
-      Repo.transaction(fn ->
-        with {_, true} <-
-               {:content_length,
-                String.length(content) <= Pleroma.Config.get([:instance, :chat_limit])},
-             {_, {:ok, chat_message_data, _meta}} <-
-               {:build_object,
-                Builder.chat_message(
-                  user,
-                  recipient.ap_id,
-                  content |> Formatter.html_escape("text/plain")
-                )},
-             {_, {:ok, create_activity_data, _meta}} <-
-               {:build_create_activity,
-                Builder.create(user, chat_message_data, [recipient.ap_id])},
-             {_, {:ok, %Activity{} = activity, _meta}} <-
-               {:common_pipeline,
-                Pipeline.common_pipeline(create_activity_data,
-                  local: true
-                )} do
-          {:ok, activity}
-        else
-          {:content_length, false} -> {:error, :content_too_long}
-          e -> e
-        end
-      end)
-
-    case transaction do
-      {:ok, value} -> value
-      error -> error
+    with {_, true} <-
+           {:content_length,
+            String.length(content) <= Pleroma.Config.get([:instance, :chat_limit])},
+         {_, {:ok, chat_message_data, _meta}} <-
+           {:build_object,
+            Builder.chat_message(
+              user,
+              recipient.ap_id,
+              content |> Formatter.html_escape("text/plain")
+            )},
+         {_, {:ok, create_activity_data, _meta}} <-
+           {:build_create_activity, Builder.create(user, chat_message_data, [recipient.ap_id])},
+         {_, {:ok, %Activity{} = activity, _meta}} <-
+           {:common_pipeline,
+            Pipeline.common_pipeline(create_activity_data,
+              local: true
+            )} do
+      {:ok, activity}
+    else
+      {:content_length, false} -> {:error, :content_too_long}
+      e -> e
     end
   end
 

test/web/activity_pub/transmogrifier/chat_message_test.exs

@@ -68,6 +68,7 @@ defmodule Pleroma.Web.ActivityPub.Transmogrifier.ChatMessageTest do
       recipient = insert(:user, ap_id: List.first(data["to"]), local: true)
 
       {:ok, %Activity{} = activity} = Transmogrifier.handle_incoming(data)
+      assert activity.local == false
 
       assert activity.actor == author.ap_id
       assert activity.recipients == [recipient.ap_id, author.ap_id]