From 59309a9eff5c2e61b2195945eca21c5126eb3f5f Mon Sep 17 00:00:00 2001 From: Mark Felder Date: Sun, 28 Jul 2024 20:41:21 -0400 Subject: Publisher job simplification Publisher jobs now store the the activity id instead of inserting duplicate JSON data in the Oban queue for each delivery. --- lib/pleroma/web/activity_pub/publisher.ex | 41 +++++++++++++++++-------------- 1 file changed, 22 insertions(+), 19 deletions(-) (limited to 'lib') diff --git a/lib/pleroma/web/activity_pub/publisher.ex b/lib/pleroma/web/activity_pub/publisher.ex index c8bdf2250..cb436e8d8 100644 --- a/lib/pleroma/web/activity_pub/publisher.ex +++ b/lib/pleroma/web/activity_pub/publisher.ex @@ -80,13 +80,26 @@ defmodule Pleroma.Web.ActivityPub.Publisher do parameters set: * `inbox`: the inbox to publish to - * `json`: the JSON message body representing the ActivityPub message * `actor`: the actor which is signing the message - * `id`: the ActivityStreams URI of the message + * `activity_id`: the internal activity id + * `cc`: the cc recipients relevant to this inbox (optional) """ - def publish_one(%{inbox: inbox, json: json, actor: %User{} = actor, id: id} = params) do - Logger.debug("Federating #{id} to #{inbox}") + def publish_one(%{inbox: inbox, actor: %User{} = actor, activity_id: activity_id} = params) do + activity = Activity.get_by_id(activity_id) + + ap_id = activity.data["id"] + Logger.debug("Federating #{ap_id} to #{inbox}") uri = %{path: path} = URI.parse(inbox) + + {:ok, data} = Transmogrifier.prepare_outgoing(activity.data) + + cc = Map.get(params, :cc) + + json = + data + |> Map.put("cc", cc) + |> Jason.encode!() + digest = "SHA-256=" <> (:crypto.hash(:sha256, json) |> Base.encode64()) date = Pleroma.Signature.signed_date() @@ -119,7 +132,7 @@ defmodule Pleroma.Web.ActivityPub.Publisher do else {_post_result, %{status: code} = response} = e -> unless params[:unreachable_since], do: Instances.set_unreachable(inbox) - Logger.metadata(activity: id, inbox: inbox, status: code) + Logger.metadata(id: activity_id, inbox: inbox, status: code) Logger.error("Publisher failed to inbox #{inbox} with status #{code}") case response do @@ -136,7 +149,7 @@ defmodule Pleroma.Web.ActivityPub.Publisher do e -> unless params[:unreachable_since], do: Instances.set_unreachable(inbox) - Logger.metadata(activity: id, inbox: inbox) + Logger.metadata(activity: activity_id, inbox: inbox) Logger.error("Publisher failed to inbox #{inbox} #{inspect(e)}") {:error, e} end @@ -251,7 +264,6 @@ defmodule Pleroma.Web.ActivityPub.Publisher do def publish(%User{} = actor, %{data: %{"bcc" => bcc}} = activity) when is_list(bcc) and bcc != [] do public = public?(activity) - {:ok, data} = Transmogrifier.prepare_outgoing(activity.data) [priority_recipients, recipients] = recipients(actor, activity) @@ -276,16 +288,11 @@ defmodule Pleroma.Web.ActivityPub.Publisher do # instance would only accept a first message for the first recipient and ignore the rest. cc = get_cc_ap_ids(ap_id, recipients) - json = - data - |> Map.put("cc", cc) - |> Jason.encode!() - __MODULE__.enqueue_one(%{ inbox: inbox, - json: json, + cc: cc, actor_id: actor.id, - id: activity.data["id"], + activity_id: activity.id, unreachable_since: unreachable_since }) end) @@ -302,9 +309,6 @@ defmodule Pleroma.Web.ActivityPub.Publisher do Relay.publish(activity) end - {:ok, data} = Transmogrifier.prepare_outgoing(activity.data) - json = Jason.encode!(data) - [priority_inboxes, inboxes] = recipients(actor, activity) |> Enum.map(fn recipients -> @@ -326,9 +330,8 @@ defmodule Pleroma.Web.ActivityPub.Publisher do __MODULE__.enqueue_one( %{ inbox: inbox, - json: json, actor_id: actor.id, - id: activity.data["id"], + activity_id: activity.id, unreachable_since: unreachable_since }, priority: priority -- cgit v1.2.3 From 74072622e08dd1efdc7bf69c3278250ea1efb22e Mon Sep 17 00:00:00 2001 From: Mark Felder Date: Mon, 29 Jul 2024 09:52:13 -0400 Subject: Remove actor and actor_id from the job as it can be inferred by the activity --- lib/pleroma/web/activity_pub/publisher.ex | 17 +++-------------- 1 file changed, 3 insertions(+), 14 deletions(-) (limited to 'lib') diff --git a/lib/pleroma/web/activity_pub/publisher.ex b/lib/pleroma/web/activity_pub/publisher.ex index cb436e8d8..373bf6e41 100644 --- a/lib/pleroma/web/activity_pub/publisher.ex +++ b/lib/pleroma/web/activity_pub/publisher.ex @@ -80,12 +80,12 @@ defmodule Pleroma.Web.ActivityPub.Publisher do parameters set: * `inbox`: the inbox to publish to - * `actor`: the actor which is signing the message * `activity_id`: the internal activity id * `cc`: the cc recipients relevant to this inbox (optional) """ - def publish_one(%{inbox: inbox, actor: %User{} = actor, activity_id: activity_id} = params) do - activity = Activity.get_by_id(activity_id) + def publish_one(%{inbox: inbox, activity_id: activity_id} = params) do + activity = Activity.get_by_id_with_user_actor(activity_id) + actor = activity.user_actor ap_id = activity.data["id"] Logger.debug("Federating #{ap_id} to #{inbox}") @@ -155,15 +155,6 @@ defmodule Pleroma.Web.ActivityPub.Publisher do end end - def publish_one(%{actor_id: actor_id} = params) do - actor = User.get_cached_by_id(actor_id) - - params - |> Map.delete(:actor_id) - |> Map.put(:actor, actor) - |> publish_one() - end - defp signature_host(%URI{port: port, scheme: scheme, host: host}) do if port == URI.default_port(scheme) do host @@ -291,7 +282,6 @@ defmodule Pleroma.Web.ActivityPub.Publisher do __MODULE__.enqueue_one(%{ inbox: inbox, cc: cc, - actor_id: actor.id, activity_id: activity.id, unreachable_since: unreachable_since }) @@ -330,7 +320,6 @@ defmodule Pleroma.Web.ActivityPub.Publisher do __MODULE__.enqueue_one( %{ inbox: inbox, - actor_id: actor.id, activity_id: activity.id, unreachable_since: unreachable_since }, -- cgit v1.2.3 From 8893ad98997197bd89e98f7dd18825dcb1206aa4 Mon Sep 17 00:00:00 2001 From: Mark Felder Date: Mon, 29 Jul 2024 09:59:35 -0400 Subject: Fix cancelling jobs --- lib/pleroma/web/common_api.ex | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) (limited to 'lib') diff --git a/lib/pleroma/web/common_api.ex b/lib/pleroma/web/common_api.ex index 06faf845e..b90b6a6d9 100644 --- a/lib/pleroma/web/common_api.ex +++ b/lib/pleroma/web/common_api.ex @@ -714,11 +714,11 @@ defmodule Pleroma.Web.CommonAPI do end end - defp maybe_cancel_jobs(%Activity{data: %{"id" => ap_id}}) do + defp maybe_cancel_jobs(%Activity{id: activity_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) + |> where([j], j.args["params"]["activity_id"] == ^activity_id) |> Oban.cancel_all_jobs() end -- cgit v1.2.3 From b48fd89a41ad766b79a7a2336737196216cede22 Mon Sep 17 00:00:00 2001 From: Mark Felder Date: Mon, 29 Jul 2024 10:03:22 -0400 Subject: Revert unintended change to the Logger metadata tag name --- lib/pleroma/web/activity_pub/publisher.ex | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) (limited to 'lib') diff --git a/lib/pleroma/web/activity_pub/publisher.ex b/lib/pleroma/web/activity_pub/publisher.ex index 373bf6e41..e040753dc 100644 --- a/lib/pleroma/web/activity_pub/publisher.ex +++ b/lib/pleroma/web/activity_pub/publisher.ex @@ -132,7 +132,7 @@ defmodule Pleroma.Web.ActivityPub.Publisher do else {_post_result, %{status: code} = response} = e -> unless params[:unreachable_since], do: Instances.set_unreachable(inbox) - Logger.metadata(id: activity_id, inbox: inbox, status: code) + Logger.metadata(activity: activity_id, inbox: inbox, status: code) Logger.error("Publisher failed to inbox #{inbox} with status #{code}") case response do -- cgit v1.2.3