No OneTemporary
Actions

Authored By

Unknown

Size

41 KB

Referenced Files

None

Subscribers

None

View Options

	diff --git a/changelog.d/oban-cancel-badreq.change b/changelog.d/oban-cancel-badreq.change
	new file mode 100644
	index 000000000..c7951735c
	--- /dev/null
	+++ b/changelog.d/oban-cancel-badreq.change
	@@ -0,0 +1 @@
	+Publisher jobs will not retry if the error received is a 400
	diff --git a/changelog.d/oban-cancel-poll-result.change b/changelog.d/oban-cancel-poll-result.change
	new file mode 100644
	index 000000000..b51c460a7
	--- /dev/null
	+++ b/changelog.d/oban-cancel-poll-result.change
	@@ -0,0 +1 @@
	+PollWorker jobs will not retry if the activity no longer exists.
	diff --git a/changelog.d/oban-cancel-receiverworker.change b/changelog.d/oban-cancel-receiverworker.change
	new file mode 100644
	index 000000000..70ad22d60
	--- /dev/null
	+++ b/changelog.d/oban-cancel-receiverworker.change
	@@ -0,0 +1 @@
	+Improved detecting unrecoverable errors for incoming federation jobs
	diff --git a/changelog.d/oban-timeouts.skip b/changelog.d/oban-timeouts.skip
	new file mode 100644
	index 000000000..e69de29bb
	diff --git a/lib/pleroma/web/activity_pub/publisher.ex b/lib/pleroma/web/activity_pub/publisher.ex
	index 90c4274f2..c8bdf2250 100644
	--- a/lib/pleroma/web/activity_pub/publisher.ex
	+++ b/lib/pleroma/web/activity_pub/publisher.ex
	@@ -1,357 +1,358 @@
	# Pleroma: A lightweight social networking server
	# Copyright © 2017-2022 Pleroma Authors <https://pleroma.social/>
	# SPDX-License-Identifier: AGPL-3.0-only

	defmodule Pleroma.Web.ActivityPub.Publisher do
	alias Pleroma.Activity
	alias Pleroma.Config
	alias Pleroma.Delivery
	alias Pleroma.HTTP
	alias Pleroma.Instances
	alias Pleroma.Object
	alias Pleroma.Repo
	alias Pleroma.User
	alias Pleroma.Web.ActivityPub.Relay
	alias Pleroma.Web.ActivityPub.Transmogrifier
	alias Pleroma.Workers.PublisherWorker

	require Pleroma.Constants

	import Pleroma.Web.ActivityPub.Visibility

	require Logger

	@moduledoc """
	ActivityPub outgoing federation module.
	"""

	@doc """
	Enqueue publishing a single activity.
	"""
	@spec enqueue_one(map(), Keyword.t()) :: {:ok, %Oban.Job{}}
	def enqueue_one(%{} = params, worker_args \\ []) do
	PublisherWorker.enqueue(
	"publish_one",
	%{"params" => params},
	worker_args
	)
	end

	@doc """
	Gathers a set of remote users given an IR envelope.
	"""
	def remote_users(%User{id: user_id}, %{data: %{"to" => to} = data}) do
	cc = Map.get(data, "cc", [])

	bcc =
	data
	\|> Map.get("bcc", [])
	\|> Enum.reduce([], fn ap_id, bcc ->
	case Pleroma.List.get_by_ap_id(ap_id) do
	%Pleroma.List{user_id: ^user_id} = list ->
	{:ok, following} = Pleroma.List.get_following(list)
	bcc ++ Enum.map(following, & &1.ap_id)

	_ ->
	bcc
	end
	end)

	[to, cc, bcc]
	\|> Enum.concat()
	\|> Enum.map(&User.get_cached_by_ap_id/1)
	\|> Enum.filter(fn user -> user && !user.local end)
	end

	@doc """
	Determine if an activity can be represented by running it through Transmogrifier.
	"""
	def representable?(%Activity{} = activity) do
	with {:ok, _data} <- Transmogrifier.prepare_outgoing(activity.data) do
	true
	else
	_e ->
	false
	end
	end

	@doc """
	Publish a single message to a peer. Takes a struct with the following
	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
	"""
	def publish_one(%{inbox: inbox, json: json, actor: %User{} = actor, id: id} = params) do
	Logger.debug("Federating #{id} to #{inbox}")
	uri = %{path: path} = URI.parse(inbox)
	digest = "SHA-256=" <> (:crypto.hash(:sha256, json) \|> Base.encode64())

	date = Pleroma.Signature.signed_date()

	signature =
	Pleroma.Signature.sign(actor, %{
	"(request-target)": "post #{path}",
	host: signature_host(uri),
	"content-length": byte_size(json),
	digest: digest,
	date: date
	})

	with {:ok, %{status: code}} = result when code in 200..299 <-
	HTTP.post(
	inbox,
	json,
	[
	{"Content-Type", "application/activity+json"},
	{"Date", date},
	{"signature", signature},
	{"digest", digest}
	]
	) do
	if not Map.has_key?(params, :unreachable_since) \|\| params[:unreachable_since] do
	Instances.set_reachable(inbox)
	end

	result
	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.error("Publisher failed to inbox #{inbox} with status #{code}")

	case response do
	+ %{status: 400} -> {:cancel, :bad_request}
	%{status: 403} -> {:cancel, :forbidden}
	%{status: 404} -> {:cancel, :not_found}
	%{status: 410} -> {:cancel, :not_found}
	_ -> {:error, e}
	end

	{:error, :pool_full} ->
	Logger.debug("Publisher snoozing worker job due to full connection pool")
	{:snooze, 30}

	e ->
	unless params[:unreachable_since], do: Instances.set_unreachable(inbox)
	Logger.metadata(activity: id, inbox: inbox)
	Logger.error("Publisher failed to inbox #{inbox} #{inspect(e)}")
	{:error, e}
	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
	else
	"#{host}:#{port}"
	end
	end

	def should_federate?(nil, _), do: false
	def should_federate?(_, true), do: true

	def should_federate?(inbox, _) do
	%{host: host} = URI.parse(inbox)

	quarantined_instances =
	Config.get([:instance, :quarantined_instances], [])
	\|> Pleroma.Web.ActivityPub.MRF.instance_list_from_tuples()
	\|> Pleroma.Web.ActivityPub.MRF.subdomains_regex()

	!Pleroma.Web.ActivityPub.MRF.subdomain_match?(quarantined_instances, host)
	end

	@spec recipients(User.t(), Activity.t()) :: [[User.t()]]
	defp recipients(actor, activity) do
	followers =
	if actor.follower_address in activity.recipients do
	User.get_external_followers(actor)
	else
	[]
	end

	fetchers =
	with %Activity{data: %{"type" => "Delete"}} <- activity,
	%Object{id: object_id} <- Object.normalize(activity, fetch: false),
	fetchers <- User.get_delivered_users_by_object_id(object_id),
	_ <- Delivery.delete_all_by_object_id(object_id) do
	fetchers
	else
	_ ->
	[]
	end

	mentioned = remote_users(actor, activity)
	non_mentioned = (followers ++ fetchers) -- mentioned

	[mentioned, non_mentioned]
	end

	defp get_cc_ap_ids(ap_id, recipients) do
	host = Map.get(URI.parse(ap_id), :host)

	recipients
	\|> Enum.filter(fn %User{ap_id: ap_id} -> Map.get(URI.parse(ap_id), :host) == host end)
	\|> Enum.map(& &1.ap_id)
	end

	defp maybe_use_sharedinbox(%User{shared_inbox: nil, inbox: inbox}), do: inbox
	defp maybe_use_sharedinbox(%User{shared_inbox: shared_inbox}), do: shared_inbox

	@doc """
	Determine a user inbox to use based on heuristics. These heuristics
	are based on an approximation of the ``sharedInbox`` rules in the
	[ActivityPub specification][ap-sharedinbox].

	Please do not edit this function (or its children) without reading
	the spec, as editing the code is likely to introduce some breakage
	without some familiarity.

	[ap-sharedinbox]: https://www.w3.org/TR/activitypub/#shared-inbox-delivery
	"""
	def determine_inbox(
	%Activity{data: activity_data},
	%User{inbox: inbox} = user
	) do
	to = activity_data["to"] \|\| []
	cc = activity_data["cc"] \|\| []
	type = activity_data["type"]

	cond do
	type == "Delete" ->
	maybe_use_sharedinbox(user)

	Pleroma.Constants.as_public() in to \|\| Pleroma.Constants.as_public() in cc ->
	maybe_use_sharedinbox(user)

	length(to) + length(cc) > 1 ->
	maybe_use_sharedinbox(user)

	true ->
	inbox
	end
	end

	@doc """
	Publishes an activity with BCC to all relevant peers.
	"""

	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)

	inboxes =
	[priority_recipients, recipients]
	\|> Enum.map(fn recipients ->
	recipients
	\|> Enum.map(fn %User{} = user ->
	determine_inbox(activity, user)
	end)
	\|> Enum.uniq()
	\|> 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)

	# Get all the recipients on the same host and add them to cc. Otherwise, a remote
	# 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,
	actor_id: actor.id,
	id: activity.data["id"],
	unreachable_since: unreachable_since
	})
	end)
	end)
	end)
	end

	# Publishes an activity to all relevant peers.
	def publish(%User{} = actor, %Activity{} = activity) do
	public = public?(activity)

	if public && Config.get([:instance, :allow_relay]) do
	Logger.debug(fn -> "Relaying #{activity.data["id"]} out" end)
	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 ->
	recipients
	\|> Enum.map(fn %User{} = user ->
	determine_inbox(activity, user)
	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} ->
	__MODULE__.enqueue_one(
	%{
	inbox: inbox,
	json: json,
	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
	[
	%{"rel" => "self", "type" => "application/activity+json", "href" => user.ap_id},
	%{
	"rel" => "self",
	"type" => "application/ld+json; profile=\"https://www.w3.org/ns/activitystreams\"",
	"href" => user.ap_id
	},
	%{
	"rel" => "http://ostatus.org/schema/1.0/subscribe",
	"template" => "#{Pleroma.Web.Endpoint.url()}/ostatus_subscribe?acct={uri}"
	}
	]
	end

	def gather_nodeinfo_protocol_names, do: ["activitypub"]
	end
	diff --git a/lib/pleroma/web/activity_pub/side_effects.ex b/lib/pleroma/web/activity_pub/side_effects.ex
	index 2a141b0f5..cc1c7a0af 100644
	--- a/lib/pleroma/web/activity_pub/side_effects.ex
	+++ b/lib/pleroma/web/activity_pub/side_effects.ex
	@@ -1,631 +1,631 @@
	# Pleroma: A lightweight social networking server
	# Copyright © 2017-2022 Pleroma Authors <https://pleroma.social/>
	# SPDX-License-Identifier: AGPL-3.0-only

	defmodule Pleroma.Web.ActivityPub.SideEffects do
	@moduledoc """
	This module looks at an inserted object and executes the side effects that it
	implies. For example, a `Like` activity will increase the like count on the
	liked object, a `Follow` activity will add the user to the follower
	collection, and so on.
	"""
	alias Pleroma.Activity
	alias Pleroma.Chat
	alias Pleroma.Chat.MessageReference
	alias Pleroma.FollowingRelationship
	alias Pleroma.Notification
	alias Pleroma.Object
	alias Pleroma.Repo
	alias Pleroma.User
	alias Pleroma.Web.ActivityPub.ActivityPub
	alias Pleroma.Web.ActivityPub.Builder
	alias Pleroma.Web.ActivityPub.Pipeline
	alias Pleroma.Web.ActivityPub.Utils
	alias Pleroma.Web.Streamer
	alias Pleroma.Workers.PollWorker

	require Pleroma.Constants
	require Logger

	@cachex Pleroma.Config.get([:cachex, :provider], Cachex)
	@logger Pleroma.Config.get([:side_effects, :logger], Logger)

	@behaviour Pleroma.Web.ActivityPub.SideEffects.Handling

	defp ap_streamer, do: Pleroma.Config.get([:side_effects, :ap_streamer], ActivityPub)

	@impl true
	def handle(object, meta \\ [])

	# Task this handles
	# - Follows
	# - Sends a notification
	@impl true
	def handle(
	%{
	data: %{
	"actor" => actor,
	"type" => "Accept",
	"object" => follow_activity_id
	}
	} = object,
	meta
	) do
	with %Activity{actor: follower_id} = follow_activity <-
	Activity.get_by_ap_id(follow_activity_id),
	%User{} = followed <- User.get_cached_by_ap_id(actor),
	%User{} = follower <- User.get_cached_by_ap_id(follower_id),
	{:ok, follow_activity} <- Utils.update_follow_state_for_all(follow_activity, "accept"),
	{:ok, _follower, followed} <-
	FollowingRelationship.update(follower, followed, :follow_accept) do
	Notification.update_notification_type(followed, follow_activity)
	end

	{:ok, object, meta}
	end

	# Task this handles
	# - Rejects all existing follow activities for this person
	# - Updates the follow state
	# - Dismisses notification
	@impl true
	def handle(
	%{
	data: %{
	"actor" => actor,
	"type" => "Reject",
	"object" => follow_activity_id
	}
	} = object,
	meta
	) do
	with %Activity{actor: follower_id} = follow_activity <-
	Activity.get_by_ap_id(follow_activity_id),
	%User{} = followed <- User.get_cached_by_ap_id(actor),
	%User{} = follower <- User.get_cached_by_ap_id(follower_id),
	{:ok, _follow_activity} <- Utils.update_follow_state_for_all(follow_activity, "reject") do
	FollowingRelationship.update(follower, followed, :follow_reject)
	Notification.dismiss(follow_activity)
	end

	{:ok, object, meta}
	end

	# Tasks this handle
	# - Follows if possible
	# - Sends a notification
	# - Generates accept or reject if appropriate
	@impl true
	def handle(
	%{
	data: %{
	"id" => follow_id,
	"type" => "Follow",
	"object" => followed_user,
	"actor" => following_user
	}
	} = object,
	meta
	) do
	with %User{} = follower <- User.get_cached_by_ap_id(following_user),
	%User{} = followed <- User.get_cached_by_ap_id(followed_user),
	{_, {:ok, _, _}, _, _} <-
	{:following, User.follow(follower, followed, :follow_pending), follower, followed} do
	if followed.local && !followed.is_locked do
	{:ok, accept_data, _} = Builder.accept(followed, object)
	{:ok, _activity, _} = Pipeline.common_pipeline(accept_data, local: true)
	end
	else
	{:following, {:error, _}, _follower, followed} ->
	{:ok, reject_data, _} = Builder.reject(followed, object)
	{:ok, _activity, _} = Pipeline.common_pipeline(reject_data, local: true)

	_ ->
	nil
	end

	{:ok, notifications} = Notification.create_notifications(object)

	meta =
	meta
	\|> add_notifications(notifications)

	updated_object = Activity.get_by_ap_id(follow_id)

	{:ok, updated_object, meta}
	end

	# Tasks this handles:
	# - Unfollow and block
	@impl true
	def handle(
	%{data: %{"type" => "Block", "object" => blocked_user, "actor" => blocking_user}} =
	object,
	meta
	) do
	with %User{} = blocker <- User.get_cached_by_ap_id(blocking_user),
	%User{} = blocked <- User.get_cached_by_ap_id(blocked_user) do
	User.block(blocker, blocked)
	end

	{:ok, object, meta}
	end

	# Tasks this handles:
	# - Update the user
	# - Update a non-user object (Note, Question, etc.)
	#
	# For a local user, we also get a changeset with the full information, so we
	# can update non-federating, non-activitypub settings as well.
	@impl true
	def handle(%{data: %{"type" => "Update", "object" => updated_object}} = object, meta) do
	updated_object_id = updated_object["id"]

	with {_, true} <- {:has_id, is_binary(updated_object_id)},
	%{"type" => type} <- updated_object,
	{_, is_user} <- {:is_user, type in Pleroma.Constants.actor_types()} do
	if is_user do
	handle_update_user(object, meta)
	else
	handle_update_object(object, meta)
	end
	else
	_ ->
	{:ok, object, meta}
	end
	end

	# Tasks this handles:
	# - Add like to object
	# - Set up notification
	@impl true
	def handle(%{data: %{"type" => "Like"}} = object, meta) do
	liked_object = Object.get_by_ap_id(object.data["object"])
	Utils.add_like_to_object(object, liked_object)

	{:ok, notifications} = Notification.create_notifications(object)

	meta =
	meta
	\|> add_notifications(notifications)

	{:ok, object, meta}
	end

	# Tasks this handles
	# - Actually create object
	# - Rollback if we couldn't create it
	# - Increase the user note count
	# - Increase the reply count
	# - Increase replies count
	# - Set up ActivityExpiration
	# - Set up notifications
	# - Index incoming posts for search (if needed)
	@impl true
	def handle(%{data: %{"type" => "Create"}} = activity, meta) do
	with {:ok, object, meta} <- handle_object_creation(meta[:object_data], activity, meta),
	%User{} = user <- User.get_cached_by_ap_id(activity.data["actor"]) do
	{:ok, notifications} = Notification.create_notifications(activity)
	{:ok, _user} = ActivityPub.increase_note_count_if_public(user, object)
	{:ok, _user} = ActivityPub.update_last_status_at_if_public(user, object)

	if in_reply_to = object.data["type"] != "Answer" && object.data["inReplyTo"] do
	Object.increase_replies_count(in_reply_to)
	end

	if quote_url = object.data["quoteUrl"] do
	Object.increase_quotes_count(quote_url)
	end

	reply_depth = (meta[:depth] \|\| 0) + 1

	# FIXME: Force inReplyTo to replies
	if Pleroma.Web.Federator.allowed_thread_distance?(reply_depth) and
	object.data["replies"] != nil do
	for reply_id <- object.data["replies"] do
	Pleroma.Workers.RemoteFetcherWorker.enqueue("fetch_remote", %{
	"id" => reply_id,
	"depth" => reply_depth
	})
	end
	end

	Pleroma.Web.RichMedia.Card.get_by_activity(activity)

	Pleroma.Search.add_to_index(Map.put(activity, :object, object))

	Utils.maybe_handle_group_posts(activity)

	meta =
	meta
	\|> add_notifications(notifications)

	ap_streamer().stream_out(activity)

	{:ok, activity, meta}
	else
	e -> Repo.rollback(e)
	end
	end

	# Tasks this handles:
	# - Add announce to object
	# - Set up notification
	# - Stream out the announce
	@impl true
	def handle(%{data: %{"type" => "Announce"}} = object, meta) do
	announced_object = Object.get_by_ap_id(object.data["object"])
	user = User.get_cached_by_ap_id(object.data["actor"])

	Utils.add_announce_to_object(object, announced_object)

	{:ok, notifications} = Notification.create_notifications(object)

	if !User.internal?(user), do: ap_streamer().stream_out(object)

	meta =
	meta
	\|> add_notifications(notifications)

	{:ok, object, meta}
	end

	@impl true
	def handle(%{data: %{"type" => "Undo", "object" => undone_object}} = object, meta) do
	with undone_object <- Activity.get_by_ap_id(undone_object),
	:ok <- handle_undoing(undone_object) do
	{:ok, object, meta}
	end
	end

	# Tasks this handles:
	# - Add reaction to object
	# - Set up notification
	@impl true
	def handle(%{data: %{"type" => "EmojiReact"}} = object, meta) do
	reacted_object = Object.get_by_ap_id(object.data["object"])
	Utils.add_emoji_reaction_to_object(object, reacted_object)

	{:ok, notifications} = Notification.create_notifications(object)

	meta =
	meta
	\|> add_notifications(notifications)

	{:ok, object, meta}
	end

	# Tasks this handles:
	# - Delete and unpins the create activity
	# - Replace object with Tombstone
	# - Reduce the user note count
	# - Reduce the reply count
	# - Stream out the activity
	# - Removes posts from search index (if needed)
	@impl true
	def handle(%{data: %{"type" => "Delete", "object" => deleted_object}} = object, meta) do
	deleted_object =
	Object.normalize(deleted_object, fetch: false) \|\|
	User.get_cached_by_ap_id(deleted_object)

	result =
	case deleted_object do
	%Object{} ->
	with {_, {:ok, deleted_object, _activity}} <- {:object, Object.delete(deleted_object)},
	{_, actor} when is_binary(actor) <- {:actor, deleted_object.data["actor"]},
	{_, %User{} = user} <- {:user, User.get_cached_by_ap_id(actor)} do
	User.remove_pinned_object_id(user, deleted_object.data["id"])

	{:ok, user} = ActivityPub.decrease_note_count_if_public(user, deleted_object)

	if in_reply_to = deleted_object.data["inReplyTo"] do
	Object.decrease_replies_count(in_reply_to)
	end

	if quote_url = deleted_object.data["quoteUrl"] do
	Object.decrease_quotes_count(quote_url)
	end

	MessageReference.delete_for_object(deleted_object)

	ap_streamer().stream_out(object)
	ap_streamer().stream_out_participations(deleted_object, user)
	:ok
	else
	{:actor, _} ->
	@logger.error("The object doesn't have an actor: #{inspect(deleted_object)}")
	:no_object_actor

	{:user, _} ->
	@logger.error(
	"The object's actor could not be resolved to a user: #{inspect(deleted_object)}"
	)

	:no_object_user

	{:object, _} ->
	@logger.error("The object could not be deleted: #{inspect(deleted_object)}")
	{:error, object}
	end

	%User{} ->
	with {:ok, _} <- User.delete(deleted_object) do
	:ok
	end
	end

	if result == :ok do
	# Only remove from index when deleting actual objects, not users or anything else
	with %Pleroma.Object{} <- deleted_object do
	Pleroma.Search.remove_from_index(deleted_object)
	end

	{:ok, object, meta}
	else
	{:error, result}
	end
	end

	# Tasks this handles:
	# - adds pin to user
	# - removes expiration job for pinned activity, if was set for expiration
	@impl true
	def handle(%{data: %{"type" => "Add"} = data} = object, meta) do
	with %User{} = user <- User.get_cached_by_ap_id(data["actor"]),
	{:ok, _user} <- User.add_pinned_object_id(user, data["object"]) do
	# if pinned activity was scheduled for deletion, we remove job
	if expiration = Pleroma.Workers.PurgeExpiredActivity.get_expiration(meta[:activity_id]) do
	Oban.cancel_job(expiration.id)
	end

	{:ok, object, meta}
	else
	nil ->
	{:error, :user_not_found}

	{:error, changeset} ->
	if changeset.errors[:pinned_objects] do
	{:error, :pinned_statuses_limit_reached}
	else
	changeset.errors
	end
	end
	end

	# Tasks this handles:
	# - removes pin from user
	# - removes corresponding Add activity
	# - if activity had expiration, recreates activity expiration job
	@impl true
	def handle(%{data: %{"type" => "Remove"} = data} = object, meta) do
	with %User{} = user <- User.get_cached_by_ap_id(data["actor"]),
	{:ok, _user} <- User.remove_pinned_object_id(user, data["object"]) do
	data["object"]
	\|> Activity.add_by_params_query(user.ap_id, user.featured_address)
	\|> Repo.delete_all()

	# if pinned activity was scheduled for deletion, we reschedule it for deletion
	if meta[:expires_at] do
	# MRF.ActivityExpirationPolicy used UTC timestamps for expires_at in original implementation
	{:ok, expires_at} =
	Pleroma.EctoType.ActivityPub.ObjectValidators.DateTime.cast(meta[:expires_at])

	Pleroma.Workers.PurgeExpiredActivity.enqueue(%{
	activity_id: meta[:activity_id],
	expires_at: expires_at
	})
	end

	{:ok, object, meta}
	else
	nil -> {:error, :user_not_found}
	error -> error
	end
	end

	# Nothing to do
	@impl true
	def handle(object, meta) do
	{:ok, object, meta}
	end

	defp handle_update_user(
	%{data: %{"type" => "Update", "object" => updated_object}} = object,
	meta
	) do
	if changeset = Keyword.get(meta, :user_update_changeset) do
	changeset
	\|> User.update_and_set_cache()
	else
	{:ok, new_user_data} = ActivityPub.user_data_from_user_object(updated_object)

	User.get_by_ap_id(updated_object["id"])
	\|> User.remote_user_changeset(new_user_data)
	\|> User.update_and_set_cache()
	end

	{:ok, object, meta}
	end

	defp handle_update_object(
	%{data: %{"type" => "Update", "object" => updated_object}} = object,
	meta
	) do
	orig_object_ap_id = updated_object["id"]
	orig_object = Object.get_by_ap_id(orig_object_ap_id)
	- orig_object_data = orig_object.data
	+ orig_object_data = Map.get(orig_object, :data)

	updated_object =
	if meta[:local] do
	# If this is a local Update, we don't process it by transmogrifier,
	# so we use the embedded object as-is.
	updated_object
	else
	meta[:object_data]
	end

	if orig_object_data["type"] in Pleroma.Constants.updatable_object_types() do
	{:ok, _, updated} =
	Object.Updater.do_update_and_invalidate_cache(orig_object, updated_object)

	if updated do
	object
	\|> Activity.normalize()
	\|> ActivityPub.notify_and_stream()
	end
	end

	{:ok, object, meta}
	end

	def handle_object_creation(%{"type" => "ChatMessage"} = object, _activity, meta) do
	with {:ok, object, meta} <- Pipeline.common_pipeline(object, meta) do
	actor = User.get_cached_by_ap_id(object.data["actor"])
	recipient = User.get_cached_by_ap_id(hd(object.data["to"]))

	streamables =
	[[actor, recipient], [recipient, actor]]
	\|> Enum.uniq()
	\|> Enum.map(fn [user, other_user] ->
	if user.local do
	{:ok, chat} = Chat.bump_or_create(user.id, other_user.ap_id)
	{:ok, cm_ref} = MessageReference.create(chat, object, user.ap_id != actor.ap_id)

	@cachex.put(
	:chat_message_id_idempotency_key_cache,
	cm_ref.id,
	meta[:idempotency_key]
	)

	{
	["user", "user:pleroma_chat"],
	{user, %{cm_ref \| chat: chat, object: object}}
	}
	end
	end)
	\|> Enum.filter(& &1)

	meta =
	meta
	\|> add_streamables(streamables)

	{:ok, object, meta}
	end
	end

	def handle_object_creation(%{"type" => "Question"} = object, activity, meta) do
	with {:ok, object, meta} <- Pipeline.common_pipeline(object, meta) do
	PollWorker.schedule_poll_end(activity)
	{:ok, object, meta}
	end
	end

	def handle_object_creation(%{"type" => "Answer"} = object_map, _activity, meta) do
	with {:ok, object, meta} <- Pipeline.common_pipeline(object_map, meta) do
	Object.increase_vote_count(
	object.data["inReplyTo"],
	object.data["name"],
	object.data["actor"]
	)

	{:ok, object, meta}
	end
	end

	def handle_object_creation(%{"type" => objtype} = object, _activity, meta)
	when objtype in ~w[Audio Video Image Event Article Note Page] do
	with {:ok, object, meta} <- Pipeline.common_pipeline(object, meta) do
	{:ok, object, meta}
	end
	end

	# Nothing to do
	def handle_object_creation(object, _activity, meta) do
	{:ok, object, meta}
	end

	defp undo_like(nil, object), do: delete_object(object)

	defp undo_like(%Object{} = liked_object, object) do
	with {:ok, _} <- Utils.remove_like_from_object(object, liked_object) do
	delete_object(object)
	end
	end

	def handle_undoing(%{data: %{"type" => "Like"}} = object) do
	object.data["object"]
	\|> Object.get_by_ap_id()
	\|> undo_like(object)
	end

	def handle_undoing(%{data: %{"type" => "EmojiReact"}} = object) do
	with %Object{} = reacted_object <- Object.get_by_ap_id(object.data["object"]),
	{:ok, _} <- Utils.remove_emoji_reaction_from_object(object, reacted_object),
	{:ok, _} <- Repo.delete(object) do
	:ok
	end
	end

	def handle_undoing(%{data: %{"type" => "Announce"}} = object) do
	with %Object{} = liked_object <- Object.get_by_ap_id(object.data["object"]),
	{:ok, _} <- Utils.remove_announce_from_object(object, liked_object),
	{:ok, _} <- Repo.delete(object) do
	:ok
	end
	end

	def handle_undoing(
	%{data: %{"type" => "Block", "actor" => blocker, "object" => blocked}} = object
	) do
	with %User{} = blocker <- User.get_cached_by_ap_id(blocker),
	%User{} = blocked <- User.get_cached_by_ap_id(blocked),
	{:ok, _} <- User.unblock(blocker, blocked),
	{:ok, _} <- Repo.delete(object) do
	:ok
	end
	end

	def handle_undoing(object), do: {:error, ["don't know how to handle", object]}

	@spec delete_object(Activity.t()) :: :ok \| {:error, Ecto.Changeset.t()}
	defp delete_object(object) do
	with {:ok, _} <- Repo.delete(object), do: :ok
	end

	defp stream_notifications(meta) do
	Keyword.get(meta, :notifications, [])
	\|> Notification.stream()

	meta
	end

	defp send_streamables(meta) do
	Keyword.get(meta, :streamables, [])
	\|> Enum.each(fn {topics, items} ->
	Streamer.stream(topics, items)
	end)

	meta
	end

	defp add_streamables(meta, streamables) do
	existing = Keyword.get(meta, :streamables, [])

	meta
	\|> Keyword.put(:streamables, streamables ++ existing)
	end

	defp add_notifications(meta, notifications) do
	existing = Keyword.get(meta, :notifications, [])

	meta
	\|> Keyword.put(:notifications, notifications ++ existing)
	end

	@impl true
	def handle_after_transaction(meta) do
	meta
	\|> stream_notifications()
	\|> send_streamables()
	end
	end
	diff --git a/lib/pleroma/workers/background_worker.ex b/lib/pleroma/workers/background_worker.ex
	index d74cc08f1..870aef3c6 100644
	--- a/lib/pleroma/workers/background_worker.ex
	+++ b/lib/pleroma/workers/background_worker.ex
	@@ -1,44 +1,44 @@
	# Pleroma: A lightweight social networking server
	# Copyright © 2017-2022 Pleroma Authors <https://pleroma.social/>
	# SPDX-License-Identifier: AGPL-3.0-only

	defmodule Pleroma.Workers.BackgroundWorker do
	alias Pleroma.User

	use Pleroma.Workers.WorkerHelper, queue: "background"

	@impl Oban.Worker

	def perform(%Job{args: %{"op" => "user_activation", "user_id" => user_id, "status" => status}}) do
	user = User.get_cached_by_id(user_id)
	User.perform(:set_activation_async, user, status)
	end

	def perform(%Job{args: %{"op" => "force_password_reset", "user_id" => user_id}}) do
	user = User.get_cached_by_id(user_id)
	User.perform(:force_password_reset, user)
	end

	def perform(%Job{args: %{"op" => op, "user_id" => user_id, "identifiers" => identifiers}})
	when op in ["blocks_import", "follow_import", "mutes_import"] do
	user = User.get_cached_by_id(user_id)
	{:ok, User.Import.perform(String.to_existing_atom(op), user, identifiers)}
	end

	def perform(%Job{
	args: %{"op" => "move_following", "origin_id" => origin_id, "target_id" => target_id}
	}) do
	origin = User.get_cached_by_id(origin_id)
	target = User.get_cached_by_id(target_id)

	Pleroma.FollowingRelationship.move_following(origin, target)
	end

	def perform(%Job{args: %{"op" => "verify_fields_links", "user_id" => user_id}}) do
	user = User.get_by_id(user_id)
	User.perform(:verify_fields_links, user)
	end

	@impl Oban.Worker
	- def timeout(_job), do: :timer.seconds(5)
	+ def timeout(_job), do: :timer.seconds(15)
	end
	diff --git a/lib/pleroma/workers/poll_worker.ex b/lib/pleroma/workers/poll_worker.ex
	index 3fcac9bc3..af8997e70 100644
	--- a/lib/pleroma/workers/poll_worker.ex
	+++ b/lib/pleroma/workers/poll_worker.ex
	@@ -1,49 +1,52 @@
	# Pleroma: A lightweight social networking server
	# Copyright © 2017-2022 Pleroma Authors <https://pleroma.social/>
	# SPDX-License-Identifier: AGPL-3.0-only

	defmodule Pleroma.Workers.PollWorker do
	@moduledoc """
	Generates notifications when a poll ends.
	"""
	use Pleroma.Workers.WorkerHelper, queue: "background"

	alias Pleroma.Activity
	alias Pleroma.Notification
	alias Pleroma.Object

	@impl Oban.Worker
	def perform(%Job{args: %{"op" => "poll_end", "activity_id" => activity_id}}) do
	with %Activity{} = activity <- find_poll_activity(activity_id),
	{:ok, notifications} <- Notification.create_poll_notifications(activity) do
	Notification.stream(notifications)
	+ else
	+ {:error, :poll_activity_not_found} = e -> {:cancel, e}
	+ e -> {:error, e}
	end
	end

	@impl Oban.Worker
	def timeout(_job), do: :timer.seconds(5)

	defp find_poll_activity(activity_id) do
	with nil <- Activity.get_by_id(activity_id) do
	{:error, :poll_activity_not_found}
	end
	end

	def schedule_poll_end(%Activity{data: %{"type" => "Create"}, id: activity_id} = activity) do
	with %Object{data: %{"type" => "Question", "closed" => closed}} when is_binary(closed) <-
	Object.normalize(activity),
	{:ok, end_time} <- NaiveDateTime.from_iso8601(closed),
	:gt <- NaiveDateTime.compare(end_time, NaiveDateTime.utc_now()) do
	%{
	op: "poll_end",
	activity_id: activity_id
	}
	\|> new(scheduled_at: end_time)
	\|> Oban.insert()
	else
	_ -> {:error, activity}
	end
	end

	def schedule_poll_end(activity), do: {:error, activity}
	end
	diff --git a/lib/pleroma/workers/receiver_worker.ex b/lib/pleroma/workers/receiver_worker.ex
	index 8b2052c23..d01813c25 100644
	--- a/lib/pleroma/workers/receiver_worker.ex
	+++ b/lib/pleroma/workers/receiver_worker.ex
	@@ -1,59 +1,61 @@
	# Pleroma: A lightweight social networking server
	# Copyright © 2017-2022 Pleroma Authors <https://pleroma.social/>
	# SPDX-License-Identifier: AGPL-3.0-only

	defmodule Pleroma.Workers.ReceiverWorker do
	alias Pleroma.Signature
	alias Pleroma.User
	alias Pleroma.Web.Federator

	use Pleroma.Workers.WorkerHelper, queue: "federator_incoming"

	@impl Oban.Worker

	def perform(%Job{
	args: %{"op" => "incoming_ap_doc", "req_headers" => req_headers, "params" => params}
	}) do
	# Oban's serialization converts our tuple headers to lists.
	# Revert it for the signature validation.
	req_headers = Enum.into(req_headers, [], &List.to_tuple(&1))

	conn_data = %{params: params, req_headers: req_headers}

	with {:ok, %User{} = _actor} <- User.get_or_fetch_by_ap_id(conn_data.params["actor"]),
	{:ok, _public_key} <- Signature.refetch_public_key(conn_data),
	{:signature, true} <- {:signature, HTTPSignatures.validate_conn(conn_data)},
	{:ok, res} <- Federator.perform(:incoming_ap_doc, params) do
	{:ok, res}
	else
	e -> process_errors(e)
	end
	end

	def perform(%Job{args: %{"op" => "incoming_ap_doc", "params" => params}}) do
	with {:ok, res} <- Federator.perform(:incoming_ap_doc, params) do
	{:ok, res}
	else
	e -> process_errors(e)
	end
	end

	@impl Oban.Worker
	def timeout(%_{args: %{"timeout" => timeout}}), do: timeout

	def timeout(_job), do: :timer.seconds(5)

	defp process_errors(errors) do
	case errors do
	{:error, :origin_containment_failed} -> {:cancel, :origin_containment_failed}
	{:error, :already_present} -> {:cancel, :already_present}
	- {:error, {:validate_object, reason}} -> {:cancel, reason}
	- {:error, {:error, {:validate, reason}}} -> {:cancel, reason}
	- {:error, {:reject, reason}} -> {:cancel, reason}
	+ {:error, {:validate_object, _} = reason} -> {:cancel, reason}
	+ {:error, {:error, {:validate, {:error, _changeset} = reason}}} -> {:cancel, reason}
	+ {:error, {:reject, _} = reason} -> {:cancel, reason}
	{:signature, false} -> {:cancel, :invalid_signature}
	- {:error, {:error, reason = "Object has been deleted"}} -> {:cancel, reason}
	+ {:error, "Object has been deleted"} = reason -> {:cancel, reason}
	+ {:error, {:side_effects, {:error, :no_object_actor}} = reason} -> {:cancel, reason}
	+ {:error, :not_found} = reason -> {:cancel, reason}
	{:error, _} = e -> e
	e -> {:error, e}
	end
	end
	end
	diff --git a/lib/pleroma/workers/remote_fetcher_worker.ex b/lib/pleroma/workers/remote_fetcher_worker.ex
	index 1c8874b59..60096e14b 100644
	--- a/lib/pleroma/workers/remote_fetcher_worker.ex
	+++ b/lib/pleroma/workers/remote_fetcher_worker.ex
	@@ -1,35 +1,35 @@
	# Pleroma: A lightweight social networking server
	# Copyright © 2017-2022 Pleroma Authors <https://pleroma.social/>
	# SPDX-License-Identifier: AGPL-3.0-only

	defmodule Pleroma.Workers.RemoteFetcherWorker do
	alias Pleroma.Object.Fetcher

	use Pleroma.Workers.WorkerHelper, queue: "background"

	@impl Oban.Worker
	def perform(%Job{args: %{"op" => "fetch_remote", "id" => id} = args}) do
	case Fetcher.fetch_object_from_id(id, depth: args["depth"]) do
	{:ok, _object} ->
	:ok

	{:reject, reason} ->
	{:cancel, reason}

	{:error, :forbidden} ->
	{:cancel, :forbidden}

	{:error, :not_found} ->
	{:cancel, :not_found}

	{:error, :allowed_depth} ->
	{:cancel, :allowed_depth}

	{:error, _} = e ->
	e
	end
	end

	@impl Oban.Worker
	- def timeout(_job), do: :timer.seconds(10)
	+ def timeout(_job), do: :timer.seconds(15)
	end
	diff --git a/lib/pleroma/workers/user_refresh_worker.ex b/lib/pleroma/workers/user_refresh_worker.ex
	index 0c04fb237..fb90e9c9c 100644
	--- a/lib/pleroma/workers/user_refresh_worker.ex
	+++ b/lib/pleroma/workers/user_refresh_worker.ex
	@@ -1,17 +1,17 @@
	# Pleroma: A lightweight social networking server
	# Copyright © 2017-2022 Pleroma Authors <https://pleroma.social/>
	# SPDX-License-Identifier: AGPL-3.0-only

	defmodule Pleroma.Workers.UserRefreshWorker do
	use Oban.Worker, queue: :background, max_attempts: 1, unique: [period: 300]

	alias Pleroma.User

	@impl true
	def perform(%Job{args: %{"ap_id" => ap_id}}) do
	User.fetch_by_ap_id(ap_id)
	end

	@impl Oban.Worker
	- def timeout(_job), do: :timer.seconds(5)
	+ def timeout(_job), do: :timer.seconds(15)
	end
	diff --git a/test/pleroma/workers/receiver_worker_test.exs b/test/pleroma/workers/receiver_worker_test.exs
	index b9b6d6af2..2b8bd3c40 100644
	--- a/test/pleroma/workers/receiver_worker_test.exs
	+++ b/test/pleroma/workers/receiver_worker_test.exs
	@@ -1,52 +1,52 @@
	# Pleroma: A lightweight social networking server
	# Copyright © 2017-2022 Pleroma Authors <https://pleroma.social/>
	# SPDX-License-Identifier: AGPL-3.0-only

	defmodule Pleroma.Workers.ReceiverWorkerTest do
	use Pleroma.DataCase
	use Oban.Testing, repo: Pleroma.Repo

	import Mock
	import Pleroma.Factory

	alias Pleroma.Workers.ReceiverWorker

	test "it does not retry MRF reject" do
	params = insert(:note).data

	with_mock Pleroma.Web.ActivityPub.Transmogrifier,
	handle_incoming: fn _ -> {:reject, "MRF"} end do
	- assert {:cancel, "MRF"} =
	+ assert {:cancel, {:reject, "MRF"}} =
	ReceiverWorker.perform(%Oban.Job{
	args: %{"op" => "incoming_ap_doc", "params" => params}
	})
	end
	end

	test "it does not retry ObjectValidator reject" do
	params =
	insert(:note_activity).data
	\|> Map.put("id", Pleroma.Web.ActivityPub.Utils.generate_activity_id())
	\|> Map.put("object", %{
	"type" => "Note",
	"id" => Pleroma.Web.ActivityPub.Utils.generate_object_id()
	})

	with_mock Pleroma.Web.ActivityPub.ObjectValidator, [:passthrough],
	validate: fn _, _ -> {:error, %Ecto.Changeset{}} end do
	assert {:cancel, {:error, %Ecto.Changeset{}}} =
	ReceiverWorker.perform(%Oban.Job{
	args: %{"op" => "incoming_ap_doc", "params" => params}
	})
	end
	end

	test "it does not retry duplicates" do
	params = insert(:note_activity).data

	assert {:cancel, :already_present} =
	ReceiverWorker.perform(%Oban.Job{
	args: %{"op" => "incoming_ap_doc", "params" => params}
	})
	end
	end

File Metadata

Mime Type: text/x-diff
Expires: Thu, Oct 29, 7:51 PM (1 d, 22 h)
Storage Engine: blob
Storage Format: Raw Data
Storage Handle: 1798597
Default Alt Text: (41 KB)

No OneTemporaryActions

View Options

File Metadata

Event Timeline

No OneTemporary
Actions