No OneTemporary
Actions

Authored By

Unknown

Size

99 KB

Referenced Files

None

Subscribers

None

View Options

	diff --git a/changelog.d/home-timeline-few-follows.change b/changelog.d/home-timeline-few-follows.change
	new file mode 100644
	index 000000000..db6038647
	--- /dev/null
	+++ b/changelog.d/home-timeline-few-follows.change
	@@ -0,0 +1 @@
	+Speed up home timelines for users who follow few accounts with per-account index scans; a post addressed to a followers collection by anyone but its owner (third parties, PeerTube channels) now only appears if it also addresses the user or a list.
	diff --git a/lib/pleroma/web/activity_pub/activity_pub.ex b/lib/pleroma/web/activity_pub/activity_pub.ex
	index b53faafb6..337765a66 100644
	--- a/lib/pleroma/web/activity_pub/activity_pub.ex
	+++ b/lib/pleroma/web/activity_pub/activity_pub.ex
	@@ -1,2145 +1,2170 @@
	# 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.ActivityPub do
	alias Pleroma.Activity
	alias Pleroma.Activity.Ir.Topics
	alias Pleroma.Config
	alias Pleroma.Constants
	alias Pleroma.Conversation
	alias Pleroma.Conversation.Participation
	alias Pleroma.Filter
	alias Pleroma.Hashtag
	alias Pleroma.Maps
	alias Pleroma.Notification
	alias Pleroma.Object
	alias Pleroma.Object.Containment
	alias Pleroma.Object.Fetcher
	alias Pleroma.Pagination
	alias Pleroma.Repo
	alias Pleroma.Upload
	alias Pleroma.User
	alias Pleroma.Web.ActivityPub.HomeTimeline
	alias Pleroma.Web.ActivityPub.MRF
	alias Pleroma.Web.ActivityPub.Transmogrifier
	alias Pleroma.Web.Streamer
	alias Pleroma.Web.WebFinger
	alias Pleroma.Workers.BackgroundWorker
	alias Pleroma.Workers.PollWorker

	import Ecto.Query
	import Pleroma.Web.ActivityPub.Utils
	import Pleroma.Web.ActivityPub.Visibility
	import Pleroma.Webhook.Notify, only: [trigger_webhooks: 2]

	require Logger
	require Pleroma.Constants

	@behaviour Pleroma.Web.ActivityPub.ActivityPub.Persisting
	@behaviour Pleroma.Web.ActivityPub.ActivityPub.Streaming

	defp get_recipients(%{"type" => "Create"} = data) do
	to = Map.get(data, "to", [])
	cc = Map.get(data, "cc", [])
	bcc = Map.get(data, "bcc", [])
	actor = Map.get(data, "actor", [])
	recipients = [to, cc, bcc, [actor]] \|> Enum.concat() \|> Enum.uniq()
	{recipients, to, cc}
	end

	defp get_recipients(data) do
	to = Map.get(data, "to", [])
	cc = Map.get(data, "cc", [])
	bcc = Map.get(data, "bcc", [])
	recipients = Enum.concat([to, cc, bcc])
	{recipients, to, cc}
	end

	defp check_actor_can_insert(%{"type" => "Delete"}), do: true
	defp check_actor_can_insert(%{"type" => "Undo"}), do: true

	defp check_actor_can_insert(%{"actor" => actor}) when is_binary(actor) do
	case User.get_cached_by_ap_id(actor) do
	%User{is_active: true} -> true
	_ -> false
	end
	end

	defp check_actor_can_insert(_), do: true

	defp check_remote_limit(%{"object" => %{"content" => content}}) when not is_nil(content) do
	limit = Config.get([:instance, :remote_limit])
	String.length(content) <= limit
	end

	defp check_remote_limit(_), do: true

	def increase_note_count_if_public(actor, object) do
	if public?(object), do: User.increase_note_count(actor), else: {:ok, actor}
	end

	def decrease_note_count_if_public(actor, object) do
	if public?(object), do: User.decrease_note_count(actor), else: {:ok, actor}
	end

	def update_last_status_at_if_public(actor, object) do
	if public?(object), do: User.update_last_status_at(actor), else: {:ok, actor}
	end

	defp increase_replies_count_if_reply(%{
	"object" => %{"inReplyTo" => reply_ap_id} = object,
	"type" => "Create"
	}) do
	if public?(object) do
	Object.increase_replies_count(reply_ap_id)
	end
	end

	defp increase_replies_count_if_reply(_create_data), do: :noop

	defp increase_quotes_count_if_quote(%{
	"object" => %{"quoteUrl" => quote_ap_id} = object,
	"type" => "Create"
	}) do
	if public?(object) do
	Object.increase_quotes_count(quote_ap_id)
	end
	end

	defp increase_quotes_count_if_quote(_create_data), do: :noop

	@object_types ~w[ChatMessage Question Answer Audio Video Image Event Article Note Page]
	@impl true
	def persist(%{"type" => type} = object, meta) when type in @object_types do
	with {:ok, object} <- Object.create(object) do
	{:ok, object, meta}
	end
	end

	@impl true
	def persist(object, meta) do
	with local <- Keyword.fetch!(meta, :local),
	{recipients, _, _} <- get_recipients(object),
	{:ok, activity} <-
	Repo.insert(%Activity{
	data: object,
	local: local,
	recipients: recipients,
	actor: object["actor"]
	}),
	# TODO: add tests for expired activities, when Note type will be supported in new pipeline
	{:ok, _} <- maybe_create_activity_expiration(activity) do
	{:ok, activity, meta}
	end
	end

	@spec insert(map(), boolean(), boolean(), boolean()) :: {:ok, Activity.t()} \| {:error, any()}
	def insert(map, local \\ true, fake \\ false, bypass_actor_check \\ false) when is_map(map) do
	with nil <- Activity.normalize(map),
	map <- lazy_put_activity_defaults(map, fake),
	{_, true} <- {:actor_check, bypass_actor_check \|\| check_actor_can_insert(map)},
	{_, true} <- {:remote_limit_pass, check_remote_limit(map)},
	{:ok, map} <- MRF.filter(map),
	{recipients, _, _} = get_recipients(map),
	{:fake, false, map, recipients} <- {:fake, fake, map, recipients},
	{:containment, :ok} <- {:containment, Containment.contain_child(map)},
	{:ok, map, object} <- insert_full_object(map),
	{:ok, activity} <- insert_activity_with_expiration(map, local, recipients) do
	# Splice in the child object if we have one.
	activity = Maps.put_if_present(activity, :object, object)

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

	# Add local posts to search index
	if local, do: Pleroma.Search.add_to_index(activity)

	{:ok, activity}
	else
	%Activity{} = activity ->
	{:ok, activity}

	{:actor_check, _} ->
	{:error, false}

	{:containment, _} = error ->
	error

	{:error, _} = error ->
	error

	{:fake, true, map, recipients} ->
	activity = %Activity{
	data: map,
	local: local,
	actor: map["actor"],
	recipients: recipients,
	id: "pleroma:fakeid"
	}

	Pleroma.Web.RichMedia.Card.get_by_activity(activity)
	{:ok, activity}

	{:remote_limit_pass, _} ->
	{:error, :remote_limit}

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

	defp insert_activity_with_expiration(data, local, recipients) do
	struct = %Activity{
	data: data,
	local: local,
	actor: data["actor"],
	recipients: recipients
	}

	with {:ok, activity} <- Repo.insert(struct) do
	maybe_create_activity_expiration(activity)
	end
	end

	def notify_and_stream(activity) do
	{:ok, notifications} = Notification.create_notifications(activity)
	Notification.stream(notifications)

	original_activity =
	case activity do
	%{data: %{"type" => "Update"}, object: %{data: %{"id" => id}}} ->
	Activity.get_create_by_object_ap_id_with_object(id)

	_ ->
	activity
	end

	conversation = create_or_bump_conversation(original_activity, original_activity.actor)
	participations = get_participations(conversation)
	stream_out(activity)
	stream_out_participations(participations)
	end

	defp maybe_create_activity_expiration(
	%{data: %{"expires_at" => %DateTime{} = expires_at}} = activity
	) do
	with {:ok, _job} <-
	Pleroma.Workers.PurgeExpiredActivity.enqueue(
	%{
	activity_id: activity.id
	},
	scheduled_at: expires_at
	) do
	{:ok, activity}
	end
	end

	defp maybe_create_activity_expiration(activity), do: {:ok, activity}

	defp create_or_bump_conversation(activity, actor) do
	with {:ok, conversation} <- Conversation.create_or_bump_for(activity),
	%User{} = user <- User.get_cached_by_ap_id(actor) do
	Participation.mark_as_read(user, conversation)
	{:ok, conversation}
	end
	end

	defp get_participations({:ok, conversation}) do
	conversation
	\|> Repo.preload(:participations, force: true)
	\|> Map.get(:participations)
	end

	defp get_participations(_), do: []

	def stream_out_participations(participations) do
	participations =
	participations
	\|> Repo.preload(:user)

	Streamer.stream("participation", participations)
	end

	@impl true
	def stream_out_participations(%Object{data: %{"context" => context}}, user) do
	with %Conversation{} = conversation <- Conversation.get_for_ap_id(context) do
	conversation = Repo.preload(conversation, :participations)

	last_activity_id =
	fetch_latest_direct_activity_id_for_context(conversation.ap_id, %{
	user: user,
	blocking_user: user
	})

	if last_activity_id do
	stream_out_participations(conversation.participations)
	end
	end
	end

	@impl true
	def stream_out_participations(_, _), do: :noop

	@impl true
	def stream_out(%Activity{data: %{"type" => data_type}} = activity)
	when data_type in ["Create", "Announce", "Delete", "Update"] do
	activity
	\|> Topics.get_activity_topics()
	\|> Streamer.stream(activity)
	end

	@impl true
	def stream_out(_activity) do
	:noop
	end

	@spec create(map(), boolean()) :: {:ok, Activity.t()} \| {:error, any()}
	def create(params, fake \\ false) do
	with {:ok, result} <- Repo.transaction(fn -> do_create(params, fake) end) do
	result
	end
	end

	defp do_create(%{to: to, actor: actor, context: context, object: object} = params, fake) do
	additional = params[:additional] \|\| %{}
	# only accept false as false value
	local = !(params[:local] == false)
	published = params[:published]
	quick_insert? = Config.get([:env]) == :benchmark

	create_data =
	make_create_data(
	%{to: to, actor: actor, published: published, context: context, object: object},
	additional
	)

	with {:ok, activity} <- insert(create_data, local, fake),
	{:fake, false, activity} <- {:fake, fake, activity},
	_ <- increase_replies_count_if_reply(create_data),
	_ <- increase_quotes_count_if_quote(create_data),
	{:quick_insert, false, activity} <- {:quick_insert, quick_insert?, activity},
	{:ok, _actor} <- increase_note_count_if_public(actor, activity),
	{:ok, _actor} <- update_last_status_at_if_public(actor, activity),
	_ <- notify_and_stream(activity),
	:ok <- maybe_schedule_poll_notifications(activity),
	:ok <- maybe_handle_group_posts(activity),
	:ok <- maybe_federate(activity) do
	{:ok, activity}
	else
	{:quick_insert, true, activity} ->
	{:ok, activity}

	{:fake, true, activity} ->
	{:ok, activity}

	{:error, message} ->
	Repo.rollback(message)
	end
	end

	defp maybe_schedule_poll_notifications(activity) do
	PollWorker.schedule_poll_end(activity)
	:ok
	end

	@spec listen(map()) :: {:ok, Activity.t()} \| {:error, any()}
	def listen(%{to: to, actor: actor, context: context, object: object} = params) do
	additional = params[:additional] \|\| %{}
	# only accept false as false value
	local = !(params[:local] == false)
	published = params[:published]

	listen_data =
	make_listen_data(
	%{to: to, actor: actor, published: published, context: context, object: object},
	additional
	)

	with {:ok, activity} <- insert(listen_data, local),
	_ <- notify_and_stream(activity),
	:ok <- maybe_federate(activity) do
	{:ok, activity}
	end
	end

	@spec unfollow(User.t(), User.t(), String.t() \| nil, boolean()) ::
	{:ok, Activity.t()} \| nil \| {:error, any()}
	def unfollow(follower, followed, activity_id \\ nil, local \\ true) do
	with {:ok, result} <-
	Repo.transaction(fn -> do_unfollow(follower, followed, activity_id, local) end) do
	result
	end
	end

	defp do_unfollow(follower, followed, activity_id, local) do
	with %Activity{} = follow_activity <- fetch_latest_follow(follower, followed),
	{:ok, follow_activity} <- update_follow_state(follow_activity, "cancelled"),
	unfollow_data <- make_unfollow_data(follower, followed, follow_activity, activity_id),
	{:ok, activity} <- insert(unfollow_data, local),
	_ <- notify_and_stream(activity),
	:ok <- maybe_federate(activity) do
	{:ok, activity}
	else
	nil -> nil
	{:error, error} -> Repo.rollback(error)
	end
	end

	@spec flag(map()) :: {:ok, Activity.t()} \| {:error, any()}
	def flag(params) do
	with {:ok, result} <- Repo.transaction(fn -> do_flag(params) end) do
	result
	end
	end

	defp do_flag(
	%{
	actor: actor,
	context: _context,
	account: account,
	statuses: statuses,
	content: content
	} = params
	) do
	# only accept false as false value
	local = !(params[:local] == false)
	forward = !(params[:forward] == false)

	additional = params[:additional] \|\| %{}

	additional =
	if forward do
	Map.merge(additional, %{"to" => [], "cc" => [account.ap_id]})
	else
	Map.merge(additional, %{"to" => [], "cc" => []})
	end

	with flag_data <- make_flag_data(params, additional),
	{:ok, activity} <- insert(flag_data, local),
	_ <- notify_and_stream(activity),
	_ <- trigger_webhooks(activity, :"report.created"),
	:ok <-
	maybe_federate(activity) do
	User.all_users_with_privilege(:reports_manage_reports)
	\|> Enum.filter(fn user -> user.ap_id != actor end)
	\|> Enum.filter(fn user -> not is_nil(user.email) end)
	\|> Enum.each(fn privileged_user ->
	privileged_user
	\|> Pleroma.Emails.AdminEmail.report(actor, account, statuses, content)
	\|> Pleroma.Emails.Mailer.deliver_async()
	end)

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

	@spec move(User.t(), User.t(), boolean()) :: {:ok, Activity.t()} \| {:error, any()}
	def move(%User{} = origin, %User{} = target, local \\ true) do
	params = %{
	"type" => "Move",
	"actor" => origin.ap_id,
	"object" => origin.ap_id,
	"target" => target.ap_id,
	"to" => [origin.follower_address]
	}

	with true <- origin.ap_id in target.also_known_as,
	{:ok, activity} <- insert(params, local),
	_ <- notify_and_stream(activity) do
	maybe_federate(activity)

	BackgroundWorker.new(%{
	"op" => "move_following",
	"origin_id" => origin.id,
	"target_id" => target.id
	})
	\|> Oban.insert()

	{:ok, activity}
	else
	false -> {:error, "Target account must have the origin in `alsoKnownAs`"}
	err -> err
	end
	end

	def fetch_activities_for_context_query(context, opts) do
	public = [Constants.as_public()]

	recipients =
	if opts[:user],
	do: [opts[:user].ap_id \| User.following(opts[:user])] ++ public,
	else: public

	from(activity in Activity)
	\|> maybe_preload_objects(opts)
	\|> maybe_preload_bookmarks(opts)
	\|> maybe_set_thread_muted_field(opts)
	\|> restrict_unauthenticated(opts[:user])
	\|> restrict_blocked(opts)
	\|> restrict_blockers_visibility(opts)
	\|> restrict_recipients(recipients, opts[:user])
	\|> restrict_filtered(opts)
	\|> where(
	[activity],
	fragment(
	"?->>'type' = ? and ?->>'context' = ?",
	activity.data,
	"Create",
	activity.data,
	^context
	)
	)
	\|> exclude_poll_votes(opts)
	\|> exclude_id(opts)
	\|> order_by([activity], desc: activity.id)
	end

	@spec fetch_activities_for_context(String.t(), keyword() \| map()) :: [Activity.t()]
	def fetch_activities_for_context(context, opts \\ %{}) do
	context
	\|> fetch_activities_for_context_query(opts)
	\|> Repo.all()
	end

	def fetch_objects_for_replies_collection(parent_ap_id, opts \\ %{}) do
	opts =
	opts
	\|> Map.put(:order_asc, true)
	\|> Map.put(:id_type, :integer)

	from(o in Object,
	where:
	fragment("?->>'inReplyTo' = ?", o.data, ^parent_ap_id) and
	fragment(
	"(?->'to' \\? ?::text OR ?->'cc' \\? ?::text)",
	o.data,
	^Pleroma.Constants.as_public(),
	o.data,
	^Pleroma.Constants.as_public()
	) and
	fragment("?->>'type' <> 'Answer'", o.data),
	select: %{id: o.id, ap_id: fragment("?->>'id'", o.data)}
	)
	\|> Pagination.fetch_paginated(opts, :keyset)
	end

	@spec fetch_latest_direct_activity_id_for_context(String.t(), keyword() \| map()) ::
	Ecto.UUID.t() \| nil
	def fetch_latest_direct_activity_id_for_context(context, opts \\ %{}) do
	context
	\|> fetch_activities_for_context_query(Map.merge(%{skip_preload: true}, opts))
	\|> restrict_visibility(%{visibility: "direct"})
	\|> limit(1)
	\|> select([a], a.id)
	\|> Repo.one()
	end

	defp fetch_paginated_optimized(query, opts, pagination) do
	# Note: tag-filtering funcs may apply "ORDER BY objects.id DESC",
	# and extra sorting on "activities.id DESC NULLS LAST" would worse the query plan
	opts = Map.put(opts, :skip_extra_order, true)

	Pagination.fetch_paginated(query, opts, pagination, pagination_binding(opts))
	end

	defp pagination_binding(%{favorited_by: _}), do: :favorited_activity
	defp pagination_binding(_), do: nil

	def fetch_home_activities(recipients, opts) do
	HomeTimeline.with_exact_queries(fn ->
	list_memberships = Pleroma.List.memberships(opts[:user])
	recipients = Enum.uniq(recipients ++ list_memberships)

	- HomeTimeline.fetch(recipients, opts, fn source, matching ->
	+ build_query = fn source, recipients, matching ->
	fetch_activities_query(recipients, opts, source, matching)
	- end)
	+ end
	+
	+ HomeTimeline.fetch(recipients, opts, build_query, fn -> rejected_actors(opts) end)
	\|> Enum.reverse()
	\|> maybe_update_cc(list_memberships, opts[:user])
	end)
	end

	+ # Authors whose posts the block and mute filters always reject.
	+ defp rejected_actors(opts) do
	+ {blocked_opts, muted_opts, _} = fetch_activities_query_ap_ids_ops(opts)
	+ blocked_actors(blocked_opts) ++ muted_actors(muted_opts)
	+ end
	+
	def fetch_activities(recipients, opts \\ %{}, pagination \\ :keyset) do
	list_memberships = Pleroma.List.memberships(opts[:user])

	fetch_activities_query(recipients ++ list_memberships, opts)
	\|> fetch_paginated_optimized(opts, pagination)
	\|> Enum.reverse()
	\|> maybe_update_cc(list_memberships, opts[:user])
	end

	@spec fetch_public_or_unlisted_activities(map(), Pagination.type()) :: [Activity.t()]
	def fetch_public_or_unlisted_activities(opts \\ %{}, pagination \\ :keyset) do
	includes_local_public = Map.get(opts, :includes_local_public, false)

	opts = Map.delete(opts, :user)

	intended_recipients =
	if includes_local_public do
	[Constants.as_public(), as_local_public()]
	else
	[Constants.as_public()]
	end

	intended_recipients
	\|> fetch_activities_query(opts)
	\|> restrict_unlisted(opts)
	\|> fetch_paginated_optimized(opts, pagination)
	end

	@spec fetch_public_activities(map(), Pagination.type()) :: [Activity.t()]
	def fetch_public_activities(opts \\ %{}, pagination \\ :keyset) do
	opts
	\|> Map.put(:restrict_unlisted, true)
	\|> fetch_public_or_unlisted_activities(pagination)
	end

	@valid_visibilities ~w[direct unlisted public private]

	defp restrict_visibility(query, %{visibility: visibility})
	when is_list(visibility) do
	if Enum.all?(visibility, &(&1 in @valid_visibilities)) do
	from(
	a in query,
	where:
	fragment(
	"activity_visibility(?, ?, ?) = ANY (?)",
	a.actor,
	a.recipients,
	a.data,
	^visibility
	)
	)
	else
	Logger.error("Could not restrict visibility to #{visibility}")
	end
	end

	defp restrict_visibility(query, %{visibility: visibility})
	when visibility in @valid_visibilities do
	from(
	a in query,
	where:
	fragment("activity_visibility(?, ?, ?) = ?", a.actor, a.recipients, a.data, ^visibility)
	)
	end

	defp restrict_visibility(_query, %{visibility: visibility})
	when visibility not in @valid_visibilities do
	Logger.error("Could not restrict visibility to #{visibility}")
	end

	defp restrict_visibility(query, _visibility), do: query

	defp exclude_visibility(query, %{exclude_visibilities: visibility})
	when is_list(visibility) do
	if Enum.all?(visibility, &(&1 in @valid_visibilities)) do
	from(
	a in query,
	where:
	not fragment(
	"activity_visibility(?, ?, ?) = ANY (?)",
	a.actor,
	a.recipients,
	a.data,
	^visibility
	)
	)
	else
	Logger.error("Could not exclude visibility to #{visibility}")
	query
	end
	end

	defp exclude_visibility(query, %{exclude_visibilities: visibility})
	when visibility in @valid_visibilities do
	from(
	a in query,
	where:
	not fragment(
	"activity_visibility(?, ?, ?) = ?",
	a.actor,
	a.recipients,
	a.data,
	^visibility
	)
	)
	end

	defp exclude_visibility(query, %{exclude_visibilities: visibility})
	when visibility not in [nil \| @valid_visibilities] do
	Logger.error("Could not exclude visibility to #{visibility}")
	query
	end

	defp exclude_visibility(query, _visibility), do: query

	defp restrict_thread_visibility(query, _, %{skip_thread_containment: true} = _),
	do: query

	defp restrict_thread_visibility(query, %{user: %User{skip_thread_containment: true}}, _),
	do: query

	defp restrict_thread_visibility(query, %{user: %User{ap_id: ap_id}}, _) do
	local_public = as_local_public()

	from(
	a in query,
	where: fragment("thread_visibility(?, (?)->>'id', ?) = true", ^ap_id, a.data, ^local_public)
	)
	end

	defp restrict_thread_visibility(query, _, _), do: query

	def fetch_user_abstract_activities(user, reading_user, params \\ %{}) do
	params =
	params
	\|> Map.put(:user, reading_user)
	\|> Map.put(:actor_id, user.ap_id)

	%{
	godmode: params[:godmode],
	reading_user: reading_user
	}
	\|> user_activities_recipients()
	\|> fetch_activities(params)
	\|> Enum.reverse()
	end

	def fetch_user_activities(user, reading_user, params \\ %{})

	def fetch_user_activities(user, reading_user, %{total: true} = params) do
	result = fetch_activities_for_user(user, reading_user, params)

	Keyword.put(result, :items, Enum.reverse(result[:items]))
	end

	def fetch_user_activities(user, reading_user, params) do
	user
	\|> fetch_activities_for_user(reading_user, params)
	\|> Enum.reverse()
	end

	defp fetch_activities_for_user(user, reading_user, params) do
	params =
	params
	\|> Map.put(:type, ["Create", "Announce"])
	\|> Map.put(:user, reading_user)
	\|> Map.put(:actor_id, user.ap_id)
	\|> Map.put(:actor_local, user.local)
	\|> Map.put(:pinned_object_ids, Map.keys(user.pinned_objects))

	params =
	if User.blocks?(reading_user, user) do
	params
	else
	params
	\|> Map.put(:blocking_user, reading_user)
	\|> Map.put(:muting_user, reading_user)
	end

	pagination_type = Map.get(params, :pagination_type) \|\| :keyset

	%{
	godmode: params[:godmode],
	reading_user: reading_user
	}
	\|> user_activities_recipients()
	\|> fetch_activities(params, pagination_type)
	end

	def fetch_statuses(reading_user, %{total: true} = params) do
	result = fetch_activities_for_reading_user(reading_user, params)
	Keyword.put(result, :items, Enum.reverse(result[:items]))
	end

	def fetch_statuses(reading_user, params) do
	reading_user
	\|> fetch_activities_for_reading_user(params)
	\|> Enum.reverse()
	end

	defp fetch_activities_for_reading_user(reading_user, params) do
	params = Map.put(params, :type, ["Create", "Announce"])

	%{
	godmode: params[:godmode],
	reading_user: reading_user
	}
	\|> user_activities_recipients()
	\|> fetch_activities(params, :offset)
	end

	defp user_activities_recipients(%{godmode: true}), do: []

	defp user_activities_recipients(%{reading_user: reading_user}) do
	if not is_nil(reading_user) and reading_user.local do
	[
	Constants.as_public(),
	as_local_public(),
	reading_user.ap_id \| User.following(reading_user)
	]
	else
	[Constants.as_public()]
	end
	end

	defp restrict_announce_object_actor(_query, %{announce_filtering_user: _, skip_preload: true}) do
	raise "Can't use the child object without preloading!"
	end

	defp restrict_announce_object_actor(query, %{announce_filtering_user: %{ap_id: actor}}) do
	from(
	[activity, object] in query,
	where:
	fragment(
	"?->>'type' != ? or ?->>'actor' != ?",
	activity.data,
	"Announce",
	object.data,
	^actor
	)
	)
	end

	defp restrict_announce_object_actor(query, _), do: query

	defp restrict_since(query, %{since_id: ""}), do: query

	defp restrict_since(query, %{favorited_by: _, since_id: _}), do: query

	defp restrict_since(query, %{since_id: since_id}) do
	from(activity in query, where: activity.id > ^since_id)
	end

	defp restrict_since(query, _), do: query

	defp restrict_embedded_tag_all(_query, %{tag_all: _tag_all, skip_preload: true}) do
	raise_on_missing_preload()
	end

	defp restrict_embedded_tag_all(query, %{tag_all: [_ \| _] = tag_all}) do
	from(
	[_activity, object] in query,
	where: fragment("(?)->'tag' \\?& (?)", object.data, ^tag_all)
	)
	end

	defp restrict_embedded_tag_all(query, %{tag_all: tag}) when is_binary(tag) do
	restrict_embedded_tag_any(query, %{tag: tag})
	end

	defp restrict_embedded_tag_all(query, _), do: query

	defp restrict_embedded_tag_any(_query, %{tag: _tag, skip_preload: true}) do
	raise_on_missing_preload()
	end

	defp restrict_embedded_tag_any(query, %{tag: [_ \| _] = tag_any}) do
	from(
	[_activity, object] in query,
	where: fragment("(?)->'tag' \\?\| (?)", object.data, ^tag_any)
	)
	end

	defp restrict_embedded_tag_any(query, %{tag: tag}) when is_binary(tag) do
	restrict_embedded_tag_any(query, %{tag: [tag]})
	end

	defp restrict_embedded_tag_any(query, _), do: query

	defp restrict_embedded_tag_reject_any(_query, %{tag_reject: _tag_reject, skip_preload: true}) do
	raise_on_missing_preload()
	end

	defp restrict_embedded_tag_reject_any(query, %{tag_reject: [_ \| _] = tag_reject}) do
	from(
	[_activity, object] in query,
	where: fragment("not (?)->'tag' \\?\| (?)", object.data, ^tag_reject)
	)
	end

	defp restrict_embedded_tag_reject_any(query, %{tag_reject: tag_reject})
	when is_binary(tag_reject) do
	restrict_embedded_tag_reject_any(query, %{tag_reject: [tag_reject]})
	end

	defp restrict_embedded_tag_reject_any(query, _), do: query

	defp object_ids_query_for_tags(tags) do
	from(hto in "hashtags_objects")
	\|> join(:inner, [hto], ht in Pleroma.Hashtag, on: hto.hashtag_id == ht.id)
	\|> where([hto, ht], ht.name in ^tags)
	\|> select([hto], hto.object_id)
	\|> distinct([hto], true)
	end

	defp restrict_hashtag_all(_query, %{tag_all: _tag, skip_preload: true}) do
	raise_on_missing_preload()
	end

	defp restrict_hashtag_all(query, %{tag_all: [single_tag]}) do
	restrict_hashtag_any(query, %{tag: single_tag})
	end

	defp restrict_hashtag_all(query, %{tag_all: [_ \| _] = tags}) do
	from(
	[_activity, object] in query,
	where:
	fragment(
	"""
	(SELECT array_agg(hashtags.name) FROM hashtags JOIN hashtags_objects
	ON hashtags_objects.hashtag_id = hashtags.id WHERE hashtags.name = ANY(?)
	AND hashtags_objects.object_id = ?) @> ?
	""",
	^tags,
	object.id,
	^tags
	)
	)
	end

	defp restrict_hashtag_all(query, %{tag_all: tag}) when is_binary(tag) do
	restrict_hashtag_all(query, %{tag_all: [tag]})
	end

	defp restrict_hashtag_all(query, _), do: query

	defp restrict_hashtag_any(_query, %{tag: _tag, skip_preload: true}) do
	raise_on_missing_preload()
	end

	defp restrict_hashtag_any(query, %{tag: [_ \| _] = tags}) do
	hashtag_ids =
	from(ht in Hashtag, where: ht.name in ^tags, select: ht.id)
	\|> Repo.all()

	# Note: NO extra ordering should be done on "activities.id desc nulls last" for optimal plan
	query =
	from(
	[_activity, object] in query,
	join: hto in "hashtags_objects",
	on: hto.object_id == object.id,
	where: hto.hashtag_id in ^hashtag_ids
	)

	if is_nil(query.distinct) do
	from([_activity, object] in query,
	distinct: [desc: object.id],
	order_by: [desc: object.id]
	)
	else
	query
	end
	end

	defp restrict_hashtag_any(query, %{tag: tag}) when is_binary(tag) do
	restrict_hashtag_any(query, %{tag: [tag]})
	end

	defp restrict_hashtag_any(query, _), do: query

	defp restrict_hashtag_reject_any(_query, %{tag_reject: _tag_reject, skip_preload: true}) do
	raise_on_missing_preload()
	end

	defp restrict_hashtag_reject_any(query, %{tag_reject: [_ \| _] = tags_reject}) do
	from(
	[_activity, object] in query,
	where: object.id not in subquery(object_ids_query_for_tags(tags_reject))
	)
	end

	defp restrict_hashtag_reject_any(query, %{tag_reject: tag_reject}) when is_binary(tag_reject) do
	restrict_hashtag_reject_any(query, %{tag_reject: [tag_reject]})
	end

	defp restrict_hashtag_reject_any(query, _), do: query

	defp raise_on_missing_preload do
	raise "Can't use the child object without preloading!"
	end

	defp restrict_recipients(query, [], _user), do: query

	defp restrict_recipients(query, recipients, nil) do
	from(activity in query, where: fragment("? && ?", ^recipients, activity.recipients))
	end

	defp restrict_recipients(query, recipients, user) do
	from(
	activity in query,
	where: fragment("? && ?", ^recipients, activity.recipients),
	or_where: activity.actor == ^user.ap_id
	)
	end

	defp restrict_recipients(query, recipients, user, :overlap) do
	restrict_recipients(query, recipients, user)
	end

	+ # Only posts addressed to the recipients, without the "OR own posts" clause.
	+ defp restrict_recipients(query, recipients, _user, :addressed) do
	+ restrict_recipients(query, recipients, nil)
	+ end
	+
	defp restrict_recipients(query, recipients, user, :exists) do
	from(activity in query,
	where:
	fragment(
	"EXISTS (SELECT 1 FROM unnest(?) AS recipient WHERE recipient = ANY(?))",
	activity.recipients,
	type(^recipients, {:array, :string})
	),
	or_where: activity.actor == ^user.ap_id
	)
	end

	# Essentially, either look for activities addressed to `recipients`, _OR_ ones
	# that reference a hashtag that the user follows
	# Firstly, two fallbacks in case there's no hashtag constraint, or the user doesn't
	# follow any
	defp restrict_recipients_or_hashtags(query, recipients, user, nil, matching) do
	restrict_recipients(query, recipients, user, matching)
	end

	defp restrict_recipients_or_hashtags(query, recipients, user, [], matching) do
	restrict_recipients(query, recipients, user, matching)
	end

	defp restrict_recipients_or_hashtags(query, recipients, _user, hashtag_ids, _matching) do
	from([activity, object] in query)
	\|> join(:left, [activity, object], hto in "hashtags_objects",
	on: hto.object_id == object.id,
	as: :hto
	)
	\|> where(
	[activity, object, hto: hto],
	(hto.hashtag_id in ^hashtag_ids and ^Constants.as_public() in activity.recipients) or
	fragment("? && ?", ^recipients, activity.recipients)
	)
	end

	defp restrict_local(query, %{local_only: true}) do
	from(activity in query, where: activity.local == true)
	end

	defp restrict_local(query, _), do: query

	defp restrict_remote(query, %{remote: true}) do
	from(activity in query, where: activity.local == false)
	end

	defp restrict_remote(query, _), do: query

	defp restrict_actor(query, %{actor_id: actor_id}) do
	from(activity in query, where: activity.actor == ^actor_id)
	end

	defp restrict_actor(query, _), do: query

	defp restrict_type(query, %{type: type}) when is_binary(type) do
	from(activity in query, where: fragment("?->>'type' = ?", activity.data, ^type))
	end

	defp restrict_type(query, %{type: type}) do
	from(activity in query, where: fragment("?->>'type' = ANY(?)", activity.data, ^type))
	end

	defp restrict_type(query, _), do: query

	defp restrict_state(query, %{state: state}) do
	from(activity in query, where: fragment("?->>'state' = ?", activity.data, ^state))
	end

	defp restrict_state(query, _), do: query

	defp restrict_report_target(query, %{report_target_id: actor_id}) do
	from(activity in query,
	where: fragment("?->'object'->>0 = ?", activity.data, ^actor_id)
	)
	end

	defp restrict_report_target(query, _), do: query

	defp restrict_assigned_account(query, %{assigned_account: assigned_account}) do
	from(activity in query,
	where: fragment("?->>'assigned_account' = ?", activity.data, ^assigned_account)
	)
	end

	defp restrict_assigned_account(query, _), do: query

	defp restrict_favorited_by(query, %{favorited_by: ap_id}) do
	newer_favorite =
	from(newer_favorite in Activity,
	where: newer_favorite.actor == ^ap_id,
	where: newer_favorite.id > parent_as(:favorited_activity).id,
	where: fragment("?->>'type' = ?", newer_favorite.data, "Like"),
	where:
	fragment(
	"associated_object_id(?) = associated_object_id(?)",
	newer_favorite.data,
	parent_as(:favorited_activity).data
	),
	select: 1
	)

	from(
	[activity, object: object] in query,
	join: favorited_activity in Activity,
	as: :favorited_activity,
	on:
	favorited_activity.actor == ^ap_id and
	fragment("?->>'type' = ?", favorited_activity.data, "Like") and
	fragment("associated_object_id(?) = (?)->>'id'", favorited_activity.data, object.data),
	where: fragment("(?)->'likes' \\? (?)", object.data, ^ap_id),
	where: not exists(subquery(newer_favorite)),
	select: %Activity{activity \| pagination_id: favorited_activity.id}
	)
	end

	defp restrict_favorited_by(query, _), do: query

	defp restrict_media(_query, %{only_media: _val, skip_preload: true}) do
	raise "Can't use the child object without preloading!"
	end

	defp restrict_media(query, %{only_media: true}) do
	from(
	[activity, object] in query,
	where: fragment("(?)->>'type' = ?", activity.data, "Create"),
	where: fragment("not (?)->'attachment' = (?)", object.data, ^[])
	)
	end

	defp restrict_media(query, _), do: query

	defp restrict_replies(query, %{exclude_replies: true}) do
	from(
	[activity, object] in query,
	where:
	fragment("?->>'inReplyTo' is null or ?->>'type' = 'Announce'", object.data, activity.data)
	)
	end

	defp restrict_replies(query, %{
	reply_filtering_user: %User{} = user,
	reply_visibility: "self"
	}) do
	from(
	[activity, object] in query,
	where:
	fragment(
	"?->>'inReplyTo' is null OR ? = ANY(?)",
	object.data,
	^user.ap_id,
	activity.recipients
	)
	)
	end

	defp restrict_replies(query, %{
	reply_filtering_user: %User{} = user,
	reply_visibility: "following"
	}) do
	from(
	[activity, object] in query,
	where:
	fragment(
	"""
	?->>'type' != 'Create' -- This isn't a Create
	OR ?->>'inReplyTo' is null -- this isn't a reply
	OR ? && array_remove(?, ?) -- The recipient is us or one of our friends,
	-- unless they are the author (because authors
	-- are also part of the recipients). This leads
	-- to a bug that self-replies by friends won't
	-- show up.
	OR ? = ? -- The actor is us
	""",
	activity.data,
	object.data,
	^[user.ap_id \| User.get_cached_user_friends_ap_ids(user)],
	activity.recipients,
	activity.actor,
	activity.actor,
	^user.ap_id
	)
	)
	end

	defp restrict_replies(query, _), do: query

	defp restrict_reblogs(query, %{exclude_reblogs: true}) do
	from(activity in query, where: fragment("?->>'type' != 'Announce'", activity.data))
	end

	defp restrict_reblogs(query, %{only_reblogs: true}) do
	from(activity in query, where: fragment("?->>'type' = 'Announce'", activity.data))
	end

	defp restrict_reblogs(query, _), do: query

	defp restrict_muted(query, %{with_muted: true}), do: query

	defp restrict_muted(query, %{muting_user: %User{} = user} = opts) do
	- mutes = opts[:muted_users_ap_ids] \|\| User.muted_users_ap_ids(user)
	+ mutes = muted_actors(opts)

	query =
	from([activity] in query,
	where: fragment("not (? = ANY(?))", activity.actor, ^mutes),
	where:
	fragment(
	"not (?->'to' \\?\| ?) or ? = ?",
	activity.data,
	^mutes,
	activity.actor,
	^user.ap_id
	)
	)

	unless opts[:skip_preload] do
	from([thread_mute: tm] in query, where: is_nil(tm.user_id))
	else
	query
	end
	end

	defp restrict_muted(query, _), do: query

	+ defp muted_actors(%{with_muted: true}), do: []
	+
	+ defp muted_actors(%{muting_user: %User{} = user} = opts),
	+ do: opts[:muted_users_ap_ids] \|\| User.muted_users_ap_ids(user)
	+
	+ defp muted_actors(_), do: []
	+
	defp restrict_blocked(query, %{blocking_user: %User{} = user} = opts) do
	- blocked_ap_ids = opts[:blocked_users_ap_ids] \|\| User.blocked_users_ap_ids(user)
	+ blocked_ap_ids = blocked_actors(opts)
	domain_blocks = user.domain_blocks \|\| []

	following_ap_ids = User.get_friends_ap_ids(user)

	query =
	if has_named_binding?(query, :object), do: query, else: Activity.with_joined_object(query)

	from(
	[activity, object: o] in query,
	# You don't block the author
	where: fragment("not (? = ANY(?))", activity.actor, ^blocked_ap_ids),

	# You don't block any recipients, and didn't author the post
	where:
	fragment(
	"((not (? && ?)) or ? = ?)",
	activity.recipients,
	^blocked_ap_ids,
	activity.actor,
	^user.ap_id
	),

	# You don't block the domain of any recipients, and didn't author the post
	where:
	fragment(
	"(recipients_contain_blocked_domains(?, ?) = false) or ? = ?",
	activity.recipients,
	^domain_blocks,
	activity.actor,
	^user.ap_id
	),

	# It's not a boost of a user you block
	where:
	fragment(
	"not (?->>'type' = 'Announce' and ?->'to' \\?\| ?)",
	activity.data,
	activity.data,
	^blocked_ap_ids
	),

	# You don't block the author's domain, and also don't follow the author
	where:
	fragment(
	"(not (split_part(?, '/', 3) = ANY(?))) or ? = ANY(?)",
	activity.actor,
	^domain_blocks,
	activity.actor,
	^following_ap_ids
	),

	# Same as above, but checks the Object
	where:
	fragment(
	"(not (split_part(?->>'actor', '/', 3) = ANY(?))) or (?->>'actor') = ANY(?)",
	o.data,
	^domain_blocks,
	o.data,
	^following_ap_ids
	)
	)
	end

	defp restrict_blocked(query, _), do: query

	+ defp blocked_actors(%{blocking_user: %User{} = user} = opts),
	+ do: opts[:blocked_users_ap_ids] \|\| User.blocked_users_ap_ids(user)
	+
	+ defp blocked_actors(_), do: []
	+
	defp restrict_blockers_visibility(query, %{blocking_user: %User{} = user}) do
	if Config.get([:activitypub, :blockers_visible]) == true do
	query
	else
	blocker_ap_ids = User.incoming_relationships_ungrouped_ap_ids(user, [:block])

	from(
	activity in query,
	# The author doesn't block you
	where: fragment("not (? = ANY(?))", activity.actor, ^blocker_ap_ids),

	# It's not a boost of a user that blocks you
	where:
	fragment(
	"not (?->>'type' = 'Announce' and ?->'to' \\?\| ?)",
	activity.data,
	activity.data,
	^blocker_ap_ids
	)
	)
	end
	end

	defp restrict_blockers_visibility(query, _), do: query

	defp restrict_unlisted(query, %{restrict_unlisted: true}) do
	from(
	activity in query,
	where:
	fragment(
	"not (coalesce(?->'cc', '{}'::jsonb) \\?\| ?)",
	activity.data,
	^[Constants.as_public()]
	)
	)
	end

	defp restrict_unlisted(query, _), do: query

	defp restrict_pinned(query, %{pinned: true, pinned_object_ids: ids}) do
	from(
	[activity, object: o] in query,
	where:
	fragment(
	"(?)->>'type' = 'Create' and associated_object_id((?)) = any (?)",
	activity.data,
	activity.data,
	^ids
	)
	)
	end

	defp restrict_pinned(query, _), do: query

	defp restrict_muted_reblogs(query, %{muting_user: %User{} = user} = opts) do
	muted_reblogs = opts[:reblog_muted_users_ap_ids] \|\| User.reblog_muted_users_ap_ids(user)

	from(
	activity in query,
	where:
	fragment(
	"not ( ?->>'type' = 'Announce' and ? = ANY(?))",
	activity.data,
	activity.actor,
	^muted_reblogs
	)
	)
	end

	defp restrict_muted_reblogs(query, _), do: query

	defp restrict_instance(query, %{instance: instance}) when is_binary(instance) do
	from(
	activity in query,
	where: fragment("split_part(actor::text, '/'::text, 3) = ?", ^instance)
	)
	end

	defp restrict_instance(query, _), do: query

	defp restrict_filtered(query, %{user: %User{} = user}) do
	case Filter.compose_regex(user) do
	nil ->
	query

	regex ->
	from([activity, object] in query,
	where:
	fragment("not(?->>'content' ~* ?)", object.data, ^regex) or
	activity.actor == ^user.ap_id
	)
	end
	end

	defp restrict_filtered(query, %{blocking_user: %User{} = user}) do
	restrict_filtered(query, %{user: user})
	end

	defp restrict_filtered(query, _), do: query

	defp restrict_unauthenticated(query, nil) do
	local = Config.restrict_unauthenticated_access?(:activities, :local)
	remote = Config.restrict_unauthenticated_access?(:activities, :remote)

	cond do
	local and remote ->
	from(activity in query, where: false)

	local ->
	from(activity in query, where: activity.local == false)

	remote ->
	from(activity in query, where: activity.local == true)

	true ->
	query
	end
	end

	defp restrict_unauthenticated(query, _), do: query

	defp restrict_activity_and_object_origin(query, local?) do
	local_object_prefix = Pleroma.Web.Endpoint.url() <> "/"

	# Keeping the correlated origin checks inside CASE prevents PostgreSQL from choosing a
	# global bitmap-and-sort plan before applying the account statuses limit.
	from([activity, object: object] in query,
	where:
	fragment(
	"CASE WHEN ? = ? AND starts_with((?->>'id'), ?) = ? THEN true ELSE false END",
	activity.local,
	^local?,
	object.data,
	^local_object_prefix,
	^local?
	)
	)
	end

	defp maybe_restrict_unauthenticated(
	query,
	%{restrict_unauthenticated: true, user: nil, actor_local: actor_local}
	)
	when is_boolean(actor_local) do
	local_restricted = Config.restrict_unauthenticated_access?(:activities, :local)
	remote_restricted = Config.restrict_unauthenticated_access?(:activities, :remote)
	actor_restricted = if actor_local, do: local_restricted, else: remote_restricted

	cond do
	actor_restricted ->
	from(activity in query, where: false)

	local_restricted ->
	restrict_activity_and_object_origin(query, false)

	remote_restricted ->
	restrict_activity_and_object_origin(query, true)

	true ->
	query
	end
	end

	defp maybe_restrict_unauthenticated(query, _), do: query

	defp restrict_quote_url(query, %{quote_url: quote_url}) do
	from([_activity, object] in query,
	where: fragment("(?)->'quoteUrl' = ?", object.data, ^quote_url)
	)
	end

	defp restrict_quote_url(query, _), do: query

	defp restrict_rule(query, %{rule_id: rule_id}) do
	from(
	activity in query,
	where: fragment("(?)->'rules' \\? (?)", activity.data, ^rule_id)
	)
	end

	defp restrict_rule(query, _), do: query

	defp exclude_poll_votes(query, %{include_poll_votes: true}), do: query

	defp exclude_poll_votes(query, _) do
	if has_named_binding?(query, :object) do
	from([activity, object: o] in query,
	where: fragment("not(?->>'type' = ?)", o.data, "Answer")
	)
	else
	query
	end
	end

	defp exclude_chat_messages(query, %{include_chat_messages: true}), do: query

	defp exclude_chat_messages(query, _) do
	if has_named_binding?(query, :object) do
	from([activity, object: o] in query,
	where: fragment("not(?->>'type' = ?)", o.data, "ChatMessage")
	)
	else
	query
	end
	end

	defp exclude_invisible_actors(query, %{type: "Flag"}), do: query
	defp exclude_invisible_actors(query, %{invisible_actors: true}), do: query

	defp exclude_invisible_actors(query, _opts) do
	query
	\|> join(:inner, [activity], u in User,
	as: :u,
	on: activity.actor == u.ap_id and u.invisible == false
	)
	end

	defp exclude_id(query, %{exclude_id: id}) when is_binary(id) do
	from(activity in query, where: activity.id != ^id)
	end

	defp exclude_id(query, _), do: query

	defp maybe_preload_objects(query, %{skip_preload: true}), do: query

	defp maybe_preload_objects(query, _) do
	query
	\|> Activity.with_preloaded_object()
	end

	defp maybe_preload_bookmarks(query, %{skip_preload: true}), do: query

	defp maybe_preload_bookmarks(query, opts) do
	query
	\|> Activity.with_preloaded_bookmark(opts[:user])
	end

	defp maybe_preload_report_notes(query, %{preload_report_notes: true}) do
	query
	\|> Activity.with_preloaded_report_notes()
	end

	defp maybe_preload_report_notes(query, _), do: query

	defp maybe_set_thread_muted_field(query, %{skip_preload: true}), do: query

	defp maybe_set_thread_muted_field(query, opts) do
	query
	\|> Activity.with_set_thread_muted_field(opts[:muting_user] \|\| opts[:user])
	end

	defp maybe_order(query, %{order: :desc}) do
	query
	\|> order_by(desc: :id)
	end

	defp maybe_order(query, %{order: :asc}) do
	query
	\|> order_by(asc: :id)
	end

	defp maybe_order(query, _), do: query

	defp normalize_fetch_activities_query_opts(opts) do
	Enum.reduce([:tag, :tag_all, :tag_reject], opts, fn key, opts ->
	case opts[key] do
	value when is_bitstring(value) ->
	Map.put(opts, key, Hashtag.normalize_name(value))

	value when is_list(value) ->
	normalized_value =
	value
	\|> Enum.map(&Hashtag.normalize_name/1)
	\|> Enum.uniq()

	Map.put(opts, key, normalized_value)

	_ ->
	opts
	end
	end)
	end

	defp fetch_activities_query_ap_ids_ops(opts) do
	source_user = opts[:muting_user]
	ap_id_relationships = if source_user, do: [:mute, :reblog_mute], else: []

	ap_id_relationships =
	if opts[:blocking_user] && opts[:blocking_user] == source_user do
	[:block \| ap_id_relationships]
	else
	ap_id_relationships
	end

	preloaded_ap_ids = User.outgoing_relationships_ap_ids(source_user, ap_id_relationships)

	restrict_blocked_opts = Map.merge(%{blocked_users_ap_ids: preloaded_ap_ids[:block]}, opts)
	restrict_muted_opts = Map.merge(%{muted_users_ap_ids: preloaded_ap_ids[:mute]}, opts)

	restrict_muted_reblogs_opts =
	Map.merge(%{reblog_muted_users_ap_ids: preloaded_ap_ids[:reblog_mute]}, opts)

	{restrict_blocked_opts, restrict_muted_opts, restrict_muted_reblogs_opts}
	end

	def fetch_activities_query(recipients, opts \\ %{}) do
	fetch_activities_query(recipients, opts, Activity, :overlap)
	end

	defp fetch_activities_query(recipients, opts, source, matching) do
	opts = normalize_fetch_activities_query_opts(opts)

	{restrict_blocked_opts, restrict_muted_opts, restrict_muted_reblogs_opts} =
	fetch_activities_query_ap_ids_ops(opts)

	config = %{
	skip_thread_containment: Config.get([:instance, :skip_thread_containment])
	}

	query =
	source
	\|> maybe_preload_objects(opts)
	\|> maybe_preload_bookmarks(opts)
	\|> maybe_preload_report_notes(opts)
	\|> maybe_set_thread_muted_field(opts)
	\|> maybe_order(opts)
	\|> maybe_restrict_unauthenticated(opts)
	\|> restrict_recipients_or_hashtags(
	recipients,
	opts[:user],
	opts[:followed_hashtags],
	matching
	)
	\|> restrict_replies(opts)
	\|> restrict_since(opts)
	\|> restrict_local(opts)
	\|> restrict_remote(opts)
	\|> restrict_actor(opts)
	\|> restrict_type(opts)
	\|> restrict_state(opts)
	\|> restrict_report_target(opts)
	\|> restrict_assigned_account(opts)
	\|> restrict_favorited_by(opts)
	\|> restrict_blocked(restrict_blocked_opts)
	\|> restrict_blockers_visibility(opts)
	\|> restrict_muted(restrict_muted_opts)
	\|> restrict_filtered(opts)
	\|> restrict_media(opts)
	\|> restrict_visibility(opts)
	\|> restrict_thread_visibility(opts, config)
	\|> restrict_reblogs(opts)
	\|> restrict_pinned(opts)
	\|> restrict_muted_reblogs(restrict_muted_reblogs_opts)
	\|> restrict_instance(opts)
	\|> restrict_announce_object_actor(opts)
	\|> restrict_filtered(opts)
	\|> restrict_rule(opts)
	\|> restrict_quote_url(opts)
	\|> maybe_restrict_deactivated_users(opts)
	\|> exclude_poll_votes(opts)
	\|> exclude_chat_messages(opts)
	\|> exclude_invisible_actors(opts)
	\|> exclude_visibility(opts)

	if Config.feature_enabled?(:improved_hashtag_timeline) do
	query
	\|> restrict_hashtag_any(opts)
	\|> restrict_hashtag_all(opts)
	\|> restrict_hashtag_reject_any(opts)
	else
	query
	\|> restrict_embedded_tag_any(opts)
	\|> restrict_embedded_tag_all(opts)
	\|> restrict_embedded_tag_reject_any(opts)
	end
	end

	@doc """
	Fetch favorites activities of user with order by sort adds to favorites
	"""
	@spec fetch_favourites(User.t(), map(), Pagination.type()) :: list(Activity.t())
	def fetch_favourites(user, params \\ %{}, pagination \\ :keyset) do
	user.ap_id
	\|> Activity.Queries.by_actor()
	\|> Activity.Queries.by_type("Like")
	\|> Activity.with_joined_object()
	\|> Object.with_joined_activity()
	\|> select([like, object, activity], %{activity \| object: object, pagination_id: like.id})
	\|> order_by([like, _, _], desc_nulls_last: like.id)
	\|> Pagination.fetch_paginated(
	Map.merge(params, %{skip_order: true}),
	pagination
	)
	end

	defp maybe_update_cc(activities, [_ \| _] = list_memberships, %User{ap_id: user_ap_id}) do
	Enum.map(activities, fn
	%{data: %{"bcc" => [_ \| _] = bcc}} = activity ->
	if Enum.any?(bcc, &(&1 in list_memberships)) do
	update_in(activity.data["cc"], &[user_ap_id \| &1])
	else
	activity
	end

	activity ->
	activity
	end)
	end

	defp maybe_update_cc(activities, _, _), do: activities

	defp fetch_activities_bounded_query(query, recipients, recipients_with_public) do
	from(activity in query,
	where:
	fragment("? && ?", activity.recipients, ^recipients) or
	(fragment("? && ?", activity.recipients, ^recipients_with_public) and
	^Constants.as_public() in activity.recipients)
	)
	end

	def fetch_activities_bounded(
	recipients,
	recipients_with_public,
	opts \\ %{},
	pagination \\ :keyset
	) do
	fetch_activities_query([], opts)
	\|> fetch_activities_bounded_query(recipients, recipients_with_public)
	\|> Pagination.fetch_paginated(opts, pagination)
	\|> Enum.reverse()
	end

	@spec upload(Upload.source(), keyword()) :: {:ok, Object.t()} \| {:error, any()}
	def upload(file, opts \\ []) do
	with {:ok, data} <- Upload.store(sanitize_upload_file(file), opts) do
	obj_data = Maps.put_if_present(data, "actor", opts[:actor])

	Repo.insert(%Object{data: obj_data})
	end
	end

	defp sanitize_upload_file(%Plug.Upload{filename: filename} = upload) when is_binary(filename) do
	%Plug.Upload{
	upload
	\| filename: Path.basename(filename)
	}
	end

	defp sanitize_upload_file(upload), do: upload

	@spec get_actor_url(any()) :: binary() \| nil
	defp get_actor_url(url) when is_binary(url), do: url
	defp get_actor_url(%{"href" => href}) when is_binary(href), do: href

	defp get_actor_url(url) when is_list(url) do
	url
	\|> List.first()
	\|> get_actor_url()
	end

	defp get_actor_url(_url), do: nil

	defp normalize_image(%{"url" => url} = data) when is_binary(url) do
	%{
	"type" => "Image",
	"url" => [%{"href" => url}]
	}
	\|> maybe_put_description(data)
	end

	defp normalize_image(%{"url" => urls}) when is_list(urls) do
	url = urls \|> List.first()

	%{"url" => url}
	\|> normalize_image()
	end

	defp normalize_image(urls) when is_list(urls), do: urls \|> List.first() \|> normalize_image()
	defp normalize_image(_), do: nil

	defp normalize_also_known_as(urls) when is_list(urls), do: urls
	defp normalize_also_known_as(url) when is_binary(url), do: [url]
	defp normalize_also_known_as(nil), do: []

	defp maybe_put_description(map, %{"name" => description}) when is_binary(description) do
	Map.put(map, "name", description)
	end

	defp maybe_put_description(map, _), do: map

	defp object_to_user_data(data, additional) do
	fields =
	data
	\|> Map.get("attachment", [])
	\|> Enum.filter(fn %{"type" => t} -> t == "PropertyValue" end)
	\|> Enum.map(fn fields -> Map.take(fields, ["name", "value"]) end)

	emojis =
	data
	\|> Map.get("tag", [])
	\|> Enum.filter(fn
	%{"type" => "Emoji"} -> true
	_ -> false
	end)
	\|> Map.new(fn %{"icon" => %{"url" => url}, "name" => name} ->
	{String.trim(name, ":"), url}
	end)

	is_locked = data["manuallyApprovesFollowers"] \|\| false
	capabilities = data["capabilities"] \|\| %{}
	accepts_chat_messages = capabilities["acceptsChatMessages"]
	data = Transmogrifier.maybe_fix_user_object(data)
	is_discoverable = data["discoverable"] \|\| false
	invisible = data["invisible"] \|\| false
	actor_type = data["type"] \|\| "Person"

	featured_address = data["featured"]
	{:ok, pinned_objects} = fetch_and_prepare_featured_from_ap_id(featured_address)

	public_key =
	if is_map(data["publicKey"]) && is_binary(data["publicKey"]["publicKeyPem"]) do
	data["publicKey"]["publicKeyPem"]
	end

	shared_inbox =
	if is_map(data["endpoints"]) && is_binary(data["endpoints"]["sharedInbox"]) do
	data["endpoints"]["sharedInbox"]
	end

	birthday =
	if is_binary(data["vcard:bday"]) do
	case Date.from_iso8601(data["vcard:bday"]) do
	{:ok, date} -> date
	{:error, _} -> nil
	end
	end

	show_birthday = !!birthday

	with {:ok, nickname} <- nickname_from_actor(data, additional) do
	{:ok,
	%{
	ap_id: data["id"],
	uri: get_actor_url(data["url"]),
	banner: normalize_image(data["image"]),
	fields: fields,
	emoji: emojis,
	is_locked: is_locked,
	is_discoverable: is_discoverable,
	invisible: invisible,
	avatar: normalize_image(data["icon"]),
	name: data["name"],
	follower_address: data["followers"],
	following_address: data["following"],
	featured_address: featured_address,
	bio: data["summary"] \|\| "",
	actor_type: actor_type,
	also_known_as: normalize_also_known_as(data["alsoKnownAs"]),
	public_key: public_key,
	inbox: data["inbox"],
	shared_inbox: shared_inbox,
	accepts_chat_messages: accepts_chat_messages,
	birthday: birthday,
	show_birthday: show_birthday,
	pinned_objects: pinned_objects,
	nickname: nickname
	}}
	end
	end

	defp nickname_from_actor(data, additional) do
	generated = generated_nickname(data)

	case additional[:nickname_from_acct] do
	^generated when is_binary(generated) ->
	{:ok, generated}

	acct when is_binary(acct) ->
	with ^acct <- webfinger_nickname(data) do
	{:ok, acct}
	else
	_ -> {:error, {:webfinger_actor_mismatch, acct, data["id"]}}
	end

	_ ->
	{:ok, generate_nickname(data)}
	end
	end

	defp generated_nickname(%{"preferredUsername" => username, "id" => ap_id})
	when is_binary(username) and is_binary(ap_id) do
	case URI.parse(ap_id) do
	%URI{host: host} when is_binary(host) -> "#{username}@#{host}"
	_ -> nil
	end
	end

	defp generated_nickname(_), do: nil

	defp webfinger_nickname(data) do
	with generated when is_binary(generated) <- generated_nickname(data),
	{:ok, %{"subject" => "acct:" <> acct, "ap_id" => ap_id}} <- WebFinger.finger(generated),
	true <- ap_id == data["id"] do
	acct
	end
	end

	defp generate_nickname(%{"preferredUsername" => username} = data) when is_binary(username) do
	generated = generated_nickname(data)

	if Config.get([WebFinger, :update_nickname_on_user_fetch]) do
	case webfinger_nickname(data) do
	acct when is_binary(acct) -> acct
	_ -> generated
	end
	else
	generated
	end
	end

	# nickname can be nil because of virtual actors
	defp generate_nickname(_), do: nil

	def fetch_follow_information_for_user(user) do
	with {:ok, following_data} <-
	Fetcher.fetch_and_contain_remote_object_from_id(user.following_address),
	{:ok, hide_follows} <- collection_private(following_data),
	{:ok, followers_data} <-
	Fetcher.fetch_and_contain_remote_object_from_id(user.follower_address),
	{:ok, hide_followers} <- collection_private(followers_data) do
	{:ok,
	%{
	hide_follows: hide_follows,
	follower_count: normalize_counter(followers_data["totalItems"]),
	following_count: normalize_counter(following_data["totalItems"]),
	hide_followers: hide_followers
	}}
	else
	{:error, _} = e -> e
	end
	end

	defp normalize_counter(counter) when is_integer(counter), do: counter
	defp normalize_counter(_), do: 0

	def maybe_update_follow_information(user_data) do
	with {:enabled, true} <- {:enabled, Config.get([:instance, :external_user_synchronization])},
	{_, true} <- {:user_type_check, user_data[:type] in ["Person", "Service"]},
	{_, true} <-
	{:collections_available,
	!!(user_data[:following_address] && user_data[:follower_address])},
	{:ok, info} <-
	fetch_follow_information_for_user(user_data) do
	info = Map.merge(user_data[:info] \|\| %{}, info)

	user_data
	\|> Map.put(:info, info)
	else
	{:user_type_check, false} ->
	user_data

	{:collections_available, false} ->
	user_data

	{:enabled, false} ->
	user_data

	e ->
	Logger.error(
	"Follower/Following counter update for #{user_data.ap_id} failed.\n" <> inspect(e)
	)

	user_data
	end
	end

	defp collection_private(%{"first" => %{"type" => type}})
	when type in ["CollectionPage", "OrderedCollectionPage"],
	do: {:ok, false}

	defp collection_private(%{"first" => first}) do
	with {:ok, %{"type" => type}} when type in ["CollectionPage", "OrderedCollectionPage"] <-
	Fetcher.fetch_and_contain_remote_object_from_id(first) do
	{:ok, false}
	else
	{:error, _} -> {:ok, true}
	end
	end

	defp collection_private(_data), do: {:ok, true}

	def user_data_from_user_object(data, additional \\ []) do
	with {:ok, data} <- MRF.filter(data),
	{:ok, data} <- object_to_user_data(data, additional) do
	{:ok, data}
	else
	{:error, _} = e -> e
	e -> {:error, e}
	end
	end

	defp fetch_and_prepare_user_from_ap_id(ap_id, additional) do
	with {:ok, data} <- Fetcher.fetch_and_contain_remote_object_from_id(ap_id),
	{:ok, data} <- user_data_from_user_object(data, additional) do
	{:ok, maybe_update_follow_information(data)}
	else
	# If this has been deleted, only log a debug and not an error
	{:error, "Object has been deleted" = e} ->
	Logger.debug("Could not decode user at fetch #{ap_id}, #{inspect(e)}")
	{:error, e}

	{:error, {:reject, reason} = e} ->
	Logger.info("Rejected user #{ap_id}: #{inspect(reason)}")
	{:error, e}

	{:error, e} ->
	Logger.error("Could not decode user at fetch #{ap_id}, #{inspect(e)}")
	{:error, e}
	end
	end

	def maybe_handle_clashing_nickname(data) do
	with nickname when is_binary(nickname) <- data[:nickname],
	%User{} = old_user <- User.get_by_nickname(nickname),
	{_, false} <- {:ap_id_comparison, data[:ap_id] == old_user.ap_id} do
	Logger.info(
	"Found an old user for #{nickname}, the old ap id is #{old_user.ap_id}, new one is #{data[:ap_id]}, renaming."
	)

	old_user
	\|> User.remote_user_changeset(%{nickname: "#{old_user.id}.#{old_user.nickname}"})
	\|> User.update_and_set_cache()
	else
	{:ap_id_comparison, true} ->
	Logger.info(
	"Found an old user for #{data[:nickname]}, but the ap id #{data[:ap_id]} is the same as the new user. Race condition? Not changing anything."
	)

	_ ->
	nil
	end
	end

	@featured_collection_types ["OrderedCollection", "Collection"]
	@featured_collection_page_types ["OrderedCollectionPage", "CollectionPage"]
	@featured_collection_item_types @featured_collection_types ++ @featured_collection_page_types

	def pin_data_from_featured_collection(%{
	"type" => type,
	"orderedItems" => objects
	})
	when type in @featured_collection_item_types do
	Map.new(objects, fn
	%{"id" => object_ap_id} -> {object_ap_id, NaiveDateTime.utc_now()}
	object_ap_id when is_binary(object_ap_id) -> {object_ap_id, NaiveDateTime.utc_now()}
	end)
	end

	def pin_data_from_featured_collection(obj) do
	Logger.error("Could not parse featured collection #{inspect(obj)}")
	%{}
	end

	def fetch_and_prepare_featured_from_ap_id(nil) do
	{:ok, %{}}
	end

	def fetch_and_prepare_featured_from_ap_id(ap_id) do
	with {:ok, data} <- Fetcher.fetch_and_contain_remote_object_from_id(ap_id) do
	{:ok, prepare_featured_collection(data)}
	else
	e ->
	Logger.error("Could not decode featured collection at fetch #{ap_id}, #{inspect(e)}")
	{:ok, %{}}
	end
	end

	defp prepare_featured_collection(%{"orderedItems" => objects} = data) when is_list(objects) do
	pin_data_from_featured_collection(data)
	end

	defp prepare_featured_collection(%{
	"type" => type,
	"first" => %{"type" => page_type} = first
	})
	when type in @featured_collection_types and page_type in @featured_collection_page_types do
	pin_data_from_featured_collection(first)
	end

	defp prepare_featured_collection(%{"type" => type, "first" => first})
	when type in @featured_collection_types and is_binary(first) do
	case Fetcher.fetch_and_contain_remote_object_from_id(first) do
	{:ok, data} ->
	pin_data_from_featured_collection(data)

	e ->
	Logger.error("Could not decode featured collection page at fetch #{first}, #{inspect(e)}")
	%{}
	end
	end

	defp prepare_featured_collection(data) do
	pin_data_from_featured_collection(data)
	end

	def enqueue_pin_fetches(%{pinned_objects: pins}) do
	# enqueue a task to fetch all pinned objects
	Enum.each(pins, fn {ap_id, _} ->
	if is_nil(Object.get_cached_by_ap_id(ap_id)) do
	Pleroma.Workers.RemoteFetcherWorker.new(%{
	"op" => "fetch_remote",
	"id" => ap_id,
	"depth" => 1
	})
	\|> Oban.insert()
	end
	end)
	end

	def enqueue_pin_fetches(_), do: nil

	def make_user_from_ap_id(ap_id, additional \\ []) do
	user = User.get_cached_by_ap_id(ap_id)

	with {:ok, data} <- fetch_and_prepare_user_from_ap_id(ap_id, additional) do
	enqueue_pin_fetches(data)

	if user do
	user
	\|> User.remote_user_changeset(data)
	\|> User.update_and_set_cache()
	else
	maybe_handle_clashing_nickname(data)

	data
	\|> User.remote_user_changeset()
	\|> Repo.insert()
	\|> User.set_cache()
	end
	end
	end

	def make_user_from_nickname(nickname) do
	with {:ok, %{"ap_id" => ap_id, "subject" => "acct:" <> acct}} when not is_nil(ap_id) <-
	WebFinger.finger(nickname) do
	make_user_from_ap_id(ap_id, nickname_from_acct: acct)
	else
	_e -> {:error, "No AP id in WebFinger"}
	end
	end

	# filter out broken threads
	defp contain_broken_threads(%Activity{} = activity, %User{} = user) do
	entire_thread_visible_for_user?(activity, user)
	end

	# do post-processing on a specific activity
	def contain_activity(%Activity{} = activity, %User{} = user) do
	contain_broken_threads(activity, user)
	end

	def fetch_direct_messages_query do
	Activity
	\|> restrict_type(%{type: "Create"})
	\|> restrict_visibility(%{visibility: "direct"})
	\|> order_by([activity], asc: activity.id)
	end

	defp maybe_restrict_deactivated_users(activity, %{type: "Flag"}), do: activity

	defp maybe_restrict_deactivated_users(activity, _opts),
	do: Activity.restrict_deactivated_users(activity)
	end
	diff --git a/lib/pleroma/web/activity_pub/home_timeline.ex b/lib/pleroma/web/activity_pub/home_timeline.ex
	index 1f9fbba79..c04e0200c 100644
	--- a/lib/pleroma/web/activity_pub/home_timeline.ex
	+++ b/lib/pleroma/web/activity_pub/home_timeline.ex
	@@ -1,163 +1,245 @@
	# Pleroma: A lightweight social networking server
	# Copyright © 2017-2026 Pleroma Authors <https://pleroma.social/>
	# SPDX-License-Identifier: AGPL-3.0-only

	defmodule Pleroma.Web.ActivityPub.HomeTimeline do
	@moduledoc false

	import Ecto.Query

	alias Pleroma.Activity
	alias Pleroma.Pagination
	alias Pleroma.Repo
	+ alias Pleroma.User

	@minimum_recipients 128
	@candidate_limit 8192
	@budget_ms 250

	# Home selection must be exact, but search still benefits from the configured
	# fuzzy GIN limit. A savepoint also restores settings inside caller transactions
	# (including SQL Sandbox), where SET LOCAL alone would leak into later queries.
	# Only read-only selection belongs here, never rendering or job insertion.
	def with_exact_queries(fun) do
	if Repo.in_transaction?() do
	exact_queries(fun)
	else
	{:ok, result} = Repo.transaction(fn -> exact_queries(fun) end)
	result
	end
	end

	defp exact_queries(fun) do
	read_only_savepoint("home_timeline", fn ->
	Repo.query!("SET LOCAL gin_fuzzy_search_limit = 0")
	fun.()
	end)
	end

	- def fetch(recipients, opts, build_query) do
	+ def fetch(recipients, opts, build_query, rejected_actors) do
	opts = Map.put(opts, :skip_extra_order, true)

	result =
	- if length(recipients) >= @minimum_recipients and opts[:followed_hashtags] in [nil, []] do
	- attempt(opts, build_query)
	- else
	- :fallback
	+ cond do
	+ opts[:followed_hashtags] not in [nil, []] -> :fallback
	+ length(recipients) >= @minimum_recipients -> attempt(recipients, opts, build_query)
	+ per_owner?(opts) -> {:ok, per_owner(recipients, opts, build_query, rejected_actors)}
	+ true -> :fallback
	end

	case result do
	{:ok, activities} ->
	activities

	:fallback ->
	- build_query.(Activity, :overlap) \|> Pagination.fetch_paginated(opts, :keyset, nil)
	+ build_query.(Activity, recipients, :overlap)
	+ \|> Pagination.fetch_paginated(opts, :keyset, nil)
	+ end
	+ end
	+
	+ defp per_owner?(%{user: %User{}, type: [_ \| _] = types}), do: Enum.all?(types, &is_binary/1)
	+ defp per_owner?(_), do: false
	+
	+ # Few recipients: walk each (owner, type) pair through the actor/type index
	+ # instead of the whole id index. Owners are the users whose followers
	+ # collection is a recipient, plus the user. Unlike the overlap query (and like
	+ # Mastodon), a post addressed to a followers collection by anyone but its
	+ # owner (third parties, PeerTube channels) only appears if it also addresses
	+ # the user or a list.
	+ defp per_owner(recipients, opts, build_query, rejected_actors) do
	+ options = Pagination.cast_params(opts)
	+
	+ users =
	+ from(u in User,
	+ where: u.follower_address in ^recipients,
	+ select: %{ap_id: u.ap_id, follower_address: u.follower_address, local: u.local}
	+ )
	+ \|> Repo.all()
	+
	+ # Skip owners whose posts the filters always reject; their whole index
	+ # slice would be scanned for nothing.
	+ rejected = rejected_actors.()
	+
	+ owners =
	+ for %{ap_id: ap_id, local: local} <- users,
	+ local or opts[:local_only] != true,
	+ ap_id not in rejected,
	+ do: ap_id
	+
	+ owners = Enum.uniq([opts[:user].ap_id \| owners])
	+ collections = Enum.map(users, & &1.follower_address)
	+
	+ case owner_ids(owners, recipients, opts, options, build_query) ++
	+ addressed_ids(recipients -- collections, options, build_query) do
	+ [] -> []
	+ ids -> reload(ids, recipients, opts, build_query)
	end
	end

	- defp attempt(opts, build_query) do
	+ defp owner_ids(owners, recipients, opts, options, build_query) do
	+ # Each pair contributes at most one page from an ordered index scan.
	+ page =
	+ build_query.(Activity, recipients, :exists)
	+ \|> where(
	+ [a],
	+ a.actor == parent_as(:owner).actor and
	+ fragment("?->>'type'", a.data) == parent_as(:owner).type
	+ )
	+ \|> page_ids(options)
	+
	+ from(
	+ o in fragment(
	+ "SELECT * FROM unnest(?::text[]) AS actor CROSS JOIN unnest(?::text[]) AS type",
	+ ^owners,
	+ ^opts[:type]
	+ ),
	+ as: :owner,
	+ inner_lateral_join: c in subquery(page),
	+ on: true,
	+ as: :candidate,
	+ select: c.id
	+ )
	+ \|> Pagination.paginate(options, :keyset, :candidate)
	+ \|> Repo.all()
	+ end
	+
	+ # Direct addressing (the user, lists, unknown collections) is rare, so GIN fits.
	+ defp addressed_ids([], _options, _build_query), do: []
	+
	+ defp addressed_ids(recipients, options, build_query) do
	+ build_query.(Activity, recipients, :addressed)
	+ \|> page_ids(options)
	+ \|> Repo.all()
	+ \|> Enum.map(& &1.id)
	+ end
	+
	+ defp page_ids(query, options) do
	+ query
	+ \|> exclude(:preload)
	+ \|> exclude(:select)
	+ \|> select([a], %{id: a.id})
	+ \|> Pagination.paginate(options)
	+ end
	+
	+ defp attempt(recipients, opts, build_query) do
	read_only_savepoint("home_timeline_fast", fn ->
	deadline = System.monotonic_time(:millisecond) + @budget_ms
	set_timeout(@budget_ms)
	options = Pagination.cast_params(opts)
	- rows = select_candidates(options, opts, build_query)
	+ rows = select_candidates(recipients, options, opts, build_query)
	ids = for {id, _count} <- rows, not is_nil(id), do: id
	{_id, count} = hd(rows)

	cond do
	length(ids) < Pagination.page_size(options) and count > @candidate_limit ->
	:fallback

	ids == [] ->
	{:ok, []}

	true ->
	- hydrate(ids, opts, build_query, deadline)
	+ hydrate(ids, recipients, opts, build_query, deadline)
	end
	end)
	rescue
	error in Postgrex.Error ->
	if error.postgres[:code] == :query_canceled do
	:fallback
	else
	reraise error, __STACKTRACE__
	end
	end

	- defp select_candidates(options, opts, build_query) do
	+ defp select_candidates(recipients, options, opts, build_query) do
	# The extra ID proves whether the requested range was exhausted. Metadata and
	# filtered IDs are selected in ONE snapshot, including when no posts match.
	range =
	Activity
	\|> candidate_order(opts)
	\|> Pagination.paginate(options)

	candidates =
	range
	\|> limit(^(@candidate_limit + 1))
	\|> select([a], %{id: a.id})

	# Both bounded scans share the same range, order and statement snapshot.
	# Keep the activity scan ordered and limited BEFORE every filter; joining
	# candidate IDs back to activities can lose that order and force a full sort.
	source = range \|> limit(^@candidate_limit) \|> subquery()

	- page =
	- build_query.(source, :exists)
	- \|> exclude(:preload)
	- \|> exclude(:select)
	- \|> select([a], %{id: a.id})
	- \|> Pagination.paginate(options)
	+ page = build_query.(source, recipients, :exists) \|> page_ids(options)

	metadata = from(c in "home_timeline_candidates", select: %{count: count(c.id)})

	from(m in subquery(metadata),
	left_join: a in subquery(page),
	on: true,
	select: {a.id, m.count}
	)
	\|> with_cte("home_timeline_candidates", as: ^candidates, materialized: true)
	\|> Repo.all()
	end

	defp candidate_order(query, %{order: direction}) when direction in [:asc, :desc] do
	order_by(query, [a], [{^direction, a.id}])
	end

	defp candidate_order(query, _), do: query

	- defp hydrate(ids, opts, build_query, deadline) do
	+ defp hydrate(ids, recipients, opts, build_query, deadline) do
	remaining = deadline - System.monotonic_time(:millisecond)

	if remaining > 0 do
	set_timeout(remaining)
	-
	- # Reload at most one page through the same filters and typed preloads. A
	- # concurrent delete/visibility change may remove a post, never add one.
	- activities =
	- build_query.(Activity, :exists)
	- \|> where([a], a.id in ^ids)
	- \|> Pagination.fetch_paginated(opts, :keyset, nil)
	-
	- {:ok, activities}
	+ {:ok, reload(ids, recipients, opts, build_query)}
	else
	:fallback
	end
	end

	+ # Reload at most one page through the same filters and typed preloads. A
	+ # concurrent delete/visibility change may remove a post, never add one.
	+ defp reload(ids, recipients, opts, build_query) do
	+ build_query.(Activity, recipients, :exists)
	+ \|> where([a], a.id in ^ids)
	+ \|> Pagination.fetch_paginated(opts, :keyset, nil)
	+ end
	+
	defp set_timeout(milliseconds) do
	Repo.query!("SELECT set_config('statement_timeout', $1, true)", ["#{milliseconds}ms"])
	end

	defp read_only_savepoint(name, fun) do
	Repo.query!("SAVEPOINT #{name}")

	try do
	fun.()
	after
	# Roll back even on success to restore the caller's settings. This also
	# recovers an aborted transaction after PostgreSQL cancels a slow attempt.
	Repo.query!("ROLLBACK TO SAVEPOINT #{name}")
	Repo.query!("RELEASE SAVEPOINT #{name}")
	end
	end
	end
	diff --git a/test/pleroma/web/activity_pub/home_timeline_test.exs b/test/pleroma/web/activity_pub/home_timeline_test.exs
	index 7af863dee..c976d3bb9 100644
	--- a/test/pleroma/web/activity_pub/home_timeline_test.exs
	+++ b/test/pleroma/web/activity_pub/home_timeline_test.exs
	@@ -1,448 +1,659 @@
	# Pleroma: A lightweight social networking server
	# Copyright © 2017-2026 Pleroma Authors <https://pleroma.social/>
	# SPDX-License-Identifier: AGPL-3.0-only

	defmodule Pleroma.Web.ActivityPub.HomeTimelineTest do
	use Pleroma.DataCase

	import Mock
	require Pleroma.Constants
	import Pleroma.Factory

	alias Pleroma.Pagination
	+ alias Pleroma.User
	+ alias Pleroma.Web.CommonAPI
	alias Pleroma.Web.ActivityPub.ActivityPub

	defp setting(name), do: Repo.query!("SELECT current_setting($1)", [name]).rows \|> hd() \|> hd()

	describe "exact home timeline queries" do
	test "uses exact GIN only for home selection and restores the caller's setting" do
	user = insert(:user)
	Repo.query!("SET LOCAL gin_fuzzy_search_limit = 321")

	with_mock Pagination, [:passthrough],
	fetch_paginated: fn _query, _opts, :keyset, nil ->
	send(self(), {:gin_limit, setting("gin_fuzzy_search_limit")})
	[]
	end do
	assert ActivityPub.fetch_home_activities([user.ap_id], %{user: user}) == []
	assert_received {:gin_limit, "0"}
	assert setting("gin_fuzzy_search_limit") == "321"

	assert ActivityPub.fetch_activities([user.ap_id], %{user: user}) == []
	assert_received {:gin_limit, "321"}
	end
	end

	test "restores settings inside an existing transaction" do
	user = insert(:user)

	assert {:ok, :ok} =
	Repo.transaction(fn ->
	Repo.query!("SET LOCAL gin_fuzzy_search_limit = 123")
	ActivityPub.fetch_home_activities([user.ap_id], %{user: user})
	assert setting("gin_fuzzy_search_limit") == "123"
	:ok
	end)
	end

	test "restores the caller's transaction after a database error" do
	user = insert(:user)
	Repo.query!("SET LOCAL gin_fuzzy_search_limit = 123")

	with_mock Pagination, [:passthrough],
	fetch_paginated: fn _query, _opts, :keyset, nil ->
	Repo.query!("SELECT 1 / 0")
	end do
	assert_raise Postgrex.Error, fn ->
	ActivityPub.fetch_home_activities([user.ap_id], %{user: user})
	end
	end

	assert setting("gin_fuzzy_search_limit") == "123"
	assert Repo.query!("SELECT 1").rows == [[1]]
	end
	end

	describe "bounded fast selection" do
	setup do
	user = insert(:user)
	recipients = [user.ap_id \| Enum.map(1..127, &"https://example.org/followers/#{&1}")]
	opts = %{user: user, type: ["Create", "Announce"], limit: 2}
	%{user: user, recipients: recipients, opts: opts}
	end

	test "gates on unique recipients and excludes followed hashtags", ctx do
	for {recipients, opts} <- [
	{Enum.take(ctx.recipients, 127), ctx.opts},
	{List.duplicate(ctx.user.ap_id, 128), ctx.opts},
	{ctx.recipients, Map.put(ctx.opts, :followed_hashtags, [123])}
	] do
	{[], queries} = queries(fn -> ActivityPub.fetch_home_activities(recipients, opts) end)
	refute fast?(queries)
	end

	{[], queries} =
	queries(fn -> ActivityPub.fetch_home_activities(ctx.recipients, ctx.opts) end)

	assert fast?(queries)
	refute fallback?(queries)
	end

	test "returns a full page with self posts, recipient posts, and preloaded objects", ctx do
	own = insert(:note_activity, user: ctx.user)

	addressed =
	insert(:note_activity,
	recipients: [ctx.user.ap_id],
	data_attrs: %{"to" => [ctx.user.ap_id]}
	)

	insert(:note_activity)

	{activities, queries} =
	queries(fn -> ActivityPub.fetch_home_activities(ctx.recipients, ctx.opts) end)

	assert Enum.map(activities, & &1.id) == [own.id, addressed.id]
	assert Enum.all?(activities, &Ecto.assoc_loaded?(&1.object))
	assert fast?(queries)
	refute fallback?(queries)
	end

	test "returns exhausted partial and empty polling pages without fallback", ctx do
	old = insert(:note_activity, user: ctx.user)
	newest = insert(:note_activity, user: ctx.user)

	for {since_id, expected} <- [{old.id, [newest.id]}, {newest.id, []}] do
	{activities, queries} =
	queries(fn ->
	ActivityPub.fetch_home_activities(
	ctx.recipients,
	Map.put(ctx.opts, :since_id, since_id)
	)
	end)

	assert Enum.map(activities, & &1.id) == expected
	assert fast?(queries)
	refute fallback?(queries)
	end
	end

	test "falls back when sparse matches lie beyond the candidate window", ctx do
	old = insert(:note_activity, user: ctx.user)
	insert_gap(8192, ctx.user)
	newest = insert(:note_activity, user: ctx.user)

	{activities, queries} =
	queries(fn -> ActivityPub.fetch_home_activities(ctx.recipients, ctx.opts) end)

	assert Enum.map(activities, & &1.id) == [old.id, newest.id]
	assert fast?(queries)
	assert fallback?(queries)
	end

	test "distinguishes exactly 8192 candidates from a truncated empty window", ctx do
	old = insert(:note_activity, user: ctx.user)
	insert_gap(8191, ctx.user)

	{activities, first_queries} =
	queries(fn -> ActivityPub.fetch_home_activities(ctx.recipients, ctx.opts) end)

	assert Enum.map(activities, & &1.id) == [old.id]
	assert fast?(first_queries)
	refute fallback?(first_queries)

	insert_gap(1, ctx.user)

	{activities, second_queries} =
	queries(fn -> ActivityPub.fetch_home_activities(ctx.recipients, ctx.opts) end)

	assert Enum.map(activities, & &1.id) == [old.id]
	assert fast?(second_queries)
	assert fallback?(second_queries)
	end

	test "applies min, max and since bounds before bounding candidates", ctx do
	oldest = insert(:note_activity, user: ctx.user)
	middle = insert(:note_activity, user: ctx.user)
	newest = insert(:note_activity, user: ctx.user)
	insert_gap(8193, ctx.user)

	for bounds <- [
	%{max_id: newest.id},
	%{min_id: oldest.id, max_id: newest.id},
	%{since_id: oldest.id, max_id: newest.id},
	%{min_id: oldest.id, limit: "1"},
	%{min_id: middle.id, order_asc: true, limit: 1}
	] do
	opts = Map.merge(ctx.opts, bounds)
	expected = ActivityPub.fetch_activities(ctx.recipients, opts) \|> Enum.map(& &1.id)

	{activities, queries} =
	queries(fn -> ActivityPub.fetch_home_activities(ctx.recipients, opts) end)

	assert Enum.map(activities, & &1.id) == expected
	assert fast?(queries)
	refute fallback?(queries)
	end
	end

	test "preserves block, mute, media, locality and visibility filters", ctx do
	author = insert(:user)
	muted = insert(:user)
	blocked = insert(:user)
	Pleroma.User.mute(ctx.user, muted)
	Pleroma.User.block(ctx.user, blocked)

	for user <- [author, muted, blocked] do
	insert(:note_activity,
	user: user,
	recipients: [ctx.user.ap_id],
	data_attrs: %{"to" => [ctx.user.ap_id]}
	)
	end

	insert(:note_activity, user: ctx.user, local: false)
	insert(:note_activity, user: ctx.user)

	media =
	insert(:note_activity,
	user: ctx.user,
	note:
	insert(:note,
	user: ctx.user,
	data: %{
	"attachment" => [%{"type" => "Image", "url" => "https://example.org/image.png"}]
	}
	)
	)

	{:ok, _bookmark} = Pleroma.Bookmark.create(ctx.user.id, media.id)

	for filters <- [
	%{},
	%{with_muted: true},
	%{local_only: true},
	%{remote: true},
	%{only_media: true},
	%{exclude_visibilities: ["public"]}
	] do
	opts =
	Map.merge(
	ctx.opts,
	Map.merge(%{blocking_user: ctx.user, muting_user: ctx.user, limit: 40}, filters)
	)

	expected = ActivityPub.fetch_activities(ctx.recipients, opts)

	{activities, queries} =
	queries(fn -> ActivityPub.fetch_home_activities(ctx.recipients, opts) end)

	assert activities == expected
	assert fast?(queries)
	refute fallback?(queries)
	end
	end

	test "preserves reply filtering and the virtual thread mute field", ctx do
	parent = insert(:note_activity, user: ctx.user)

	reply =
	insert(:note_activity,
	user: ctx.user,
	note: insert(:note, user: ctx.user, data: %{"inReplyTo" => parent.data["object"]})
	)

	{:ok, _} = Pleroma.ThreadMute.add_mute(ctx.user.id, reply.data["context"])

	opts =
	Map.merge(ctx.opts, %{
	with_muted: true,
	muting_user: ctx.user,
	reply_filtering_user: ctx.user
	})

	for filters <- [
	%{},
	%{reply_visibility: "self"},
	%{reply_visibility: "following"},
	%{exclude_replies: true}
	] do
	opts = Map.merge(opts, filters)
	expected = ActivityPub.fetch_activities(ctx.recipients, opts)

	{activities, queries} =
	queries(fn -> ActivityPub.fetch_home_activities(ctx.recipients, opts) end)

	assert activities == expected
	assert fast?(queries)
	refute fallback?(queries)
	end

	activities = ActivityPub.fetch_home_activities(ctx.recipients, opts)
	assert Enum.find(activities, &(&1.id == reply.id)).thread_muted?
	end

	test "includes list memberships in the gate and preserves list bcc postprocessing", ctx do
	owner = insert(:user)
	{:ok, list} = Pleroma.List.create(%{title: "members"}, owner)
	{:ok, list} = Pleroma.List.follow(list, ctx.user)

	activity =
	insert(:note_activity,
	user: owner,
	recipients: [list.ap_id, Pleroma.Constants.as_public()],
	data_attrs: %{"bcc" => [list.ap_id], "cc" => [], "to" => [ctx.user.ap_id]}
	)

	recipients = Enum.take(ctx.recipients, 127)

	{activities, queries} =
	queries(fn -> ActivityPub.fetch_home_activities(recipients, ctx.opts) end)

	assert [result] = activities
	assert result.id == activity.id
	assert result.data["cc"] == [ctx.user.ap_id]
	assert fast?(queries)
	refute fallback?(queries)
	end

	test "uses the configured default and maximum page sizes", ctx do
	for _ <- 1..41, do: insert(:note_activity, user: ctx.user)

	for opts <- [Map.delete(ctx.opts, :limit), Map.put(ctx.opts, :limit, "100")] do
	expected = ActivityPub.fetch_activities(ctx.recipients, opts)

	{activities, queries} =
	queries(fn -> ActivityPub.fetch_home_activities(ctx.recipients, opts) end)

	assert activities == expected
	assert fast?(queries)
	refute fallback?(queries)
	end
	end

	test "does not swallow non-cancellation errors in the fast query", ctx do
	Repo.query!("SET LOCAL gin_fuzzy_search_limit = 123")
	Repo.query!("SET LOCAL statement_timeout = '5s'")

	with_mock Repo, [:passthrough],
	all: fn query ->
	{sql, _} = Ecto.Adapters.SQL.to_sql(:all, Repo, query)

	if String.contains?(sql, "home_timeline_candidates"),
	do: Repo.query!("SELECT 1 / 0"),
	else: :meck.passthrough([query])
	end do
	error =
	assert_raise Postgrex.Error, fn ->
	ActivityPub.fetch_home_activities(ctx.recipients, ctx.opts)
	end

	assert error.postgres.code == :division_by_zero
	end

	assert setting("statement_timeout") == "5s"
	assert setting("gin_fuzzy_search_limit") == "123"
	end

	test "also recovers when bounded hydration times out", ctx do
	activity = insert(:note_activity, user: ctx.user)

	with_mock Repo, [:passthrough],
	all: fn query ->
	{sql, _} = Ecto.Adapters.SQL.to_sql(:all, Repo, query)

	if String.contains?(sql, "unnest(") and
	not String.contains?(sql, "home_timeline_candidates") do
	send(self(), :hydration_timeout)
	Repo.query!("SELECT pg_sleep(1)")
	else
	:meck.passthrough([query])
	end
	end do
	assert [result] = ActivityPub.fetch_home_activities(ctx.recipients, ctx.opts)
	assert result.id == activity.id
	assert_received :hydration_timeout
	end
	end

	test "recovers from a real statement timeout and restores both settings", ctx do
	activity = insert(:note_activity, user: ctx.user)
	Repo.query!("SET LOCAL statement_timeout = '5s'")
	Repo.query!("SET LOCAL gin_fuzzy_search_limit = 123")

	with_mock Repo, [:passthrough],
	all: fn query ->
	{sql, _} = Ecto.Adapters.SQL.to_sql(:all, Repo, query)

	if String.contains?(sql, "home_timeline_candidates") do
	assert setting("gin_fuzzy_search_limit") == "0"
	send(self(), :timed_fast_query)
	Repo.query!("SELECT pg_sleep(1)")
	else
	result = :meck.passthrough([query])

	if String.contains?(sql, " && ") do
	assert setting("gin_fuzzy_search_limit") == "0"
	assert setting("statement_timeout") == "5s"
	end

	result
	end
	end do
	assert [result] = ActivityPub.fetch_home_activities(ctx.recipients, ctx.opts)
	assert result.id == activity.id
	assert_received :timed_fast_query
	end

	assert setting("statement_timeout") == "5s"
	assert setting("gin_fuzzy_search_limit") == "123"
	assert Repo.query!("SELECT 1").rows == [[1]]
	end
	end

	+ describe "per-owner selection for few follows" do
	+ setup do
	+ user = insert(:user)
	+ followed = insert(:user)
	+ other = insert(:user)
	+ {:ok, user, followed} = User.follow(user, followed)
	+ opts = %{user: user, type: ["Create", "Announce"], limit: 20}
	+ %{user: user, followed: followed, other: other, opts: opts}
	+ end
	+
	+ test "returns followed, own, direct and list posts but not foreign DMs", ctx do
	+ stranger = insert(:user)
	+ {:ok, list} = Pleroma.List.create(%{title: "members"}, stranger)
	+ {:ok, _list} = Pleroma.List.follow(list, ctx.user)
	+
	+ expected =
	+ for visibility <- ["public", "unlisted", "private"] do
	+ {:ok, post} = CommonAPI.post(ctx.followed, %{status: "hi", visibility: visibility})
	+ post
	+ end
	+
	+ {:ok, original} = CommonAPI.post(ctx.other, %{status: "boost me"})
	+ {:ok, announce} = CommonAPI.repeat(original.id, ctx.followed)
	+ {:ok, own} = CommonAPI.post(ctx.user, %{status: "mine", visibility: "private"})
	+
	+ {:ok, dm} =
	+ CommonAPI.post(ctx.other, %{
	+ status: "@#{ctx.user.nickname} hi",
	+ visibility: "direct"
	+ })
	+
	+ list_post =
	+ insert(:note_activity,
	+ user: stranger,
	+ recipients: [list.ap_id, Pleroma.Constants.as_public()],
	+ data_attrs: %{"bcc" => [list.ap_id], "cc" => [], "to" => [ctx.user.ap_id]}
	+ )
	+
	+ {:ok, _foreign_dm} =
	+ CommonAPI.post(ctx.followed, %{
	+ status: "@#{ctx.other.nickname} secret",
	+ visibility: "direct"
	+ })
	+
	+ {:ok, _unrelated} = CommonAPI.post(ctx.other, %{status: "not followed"})
	+
	+ {activities, queries} = home(ctx.user, ctx.opts)
	+
	+ assert ids(activities) == ids(expected ++ [announce, own, dm, list_post])
	+ assert per_owner?(queries)
	+ refute fast?(queries)
	+ end
	+
	+ test "drops third-party posts addressed only to a followers collection", ctx do
	+ followers = ctx.followed.follower_address
	+
	+ _third_party =
	+ insert(:note_activity,
	+ user: ctx.other,
	+ recipients: [followers],
	+ data_attrs: %{"to" => [followers], "cc" => []}
	+ )
	+
	+ {:ok, post} = CommonAPI.post(ctx.followed, %{status: "hi"})
	+
	+ {activities, queries} = home(ctx.user, ctx.opts)
	+
	+ assert ids(activities) == [post.id]
	+ assert per_owner?(queries)
	+ end
	+
	+ test "preserves block and mute filters", ctx do
	+ muted = insert(:user)
	+ blocked = insert(:user)
	+ {:ok, user, _} = User.follow(ctx.user, muted)
	+ {:ok, user, _} = User.follow(user, blocked)
	+ {:ok, _} = User.mute(user, muted)
	+ {:ok, _} = User.block(user, blocked)
	+
	+ {:ok, visible} = CommonAPI.post(ctx.followed, %{status: "hi"})
	+ {:ok, muted_post} = CommonAPI.post(muted, %{status: "muted"})
	+ {:ok, _} = CommonAPI.post(blocked, %{status: "blocked"})
	+
	+ opts = Map.merge(ctx.opts, %{muting_user: user, blocking_user: user})
	+ {activities, queries} = home(user, opts)
	+ assert ids(activities) == [visible.id]
	+ assert per_owner?(queries)
	+
	+ {activities, _} = home(user, Map.put(opts, :with_muted, true))
	+ assert ids(activities) == [visible.id, muted_post.id]
	+ end
	+
	+ test "keyset pagination matches the exact query for interleaved owners", ctx do
	+ second = insert(:user)
	+ {:ok, user, _} = User.follow(ctx.user, second)
	+
	+ posts =
	+ for index <- 1..9 do
	+ author = Enum.at([ctx.followed, second, user], rem(index, 3))
	+ {:ok, post} = CommonAPI.post(author, %{status: "post #{index}"})
	+ post
	+ end
	+
	+ [first \| _] = posts
	+ middle = Enum.at(posts, 4)
	+ last = List.last(posts)
	+ recipients = recipients(user)
	+ opts = %{ctx.opts \| user: user}
	+
	+ for bounds <- [
	+ %{limit: 3},
	+ %{max_id: middle.id, limit: 3},
	+ %{since_id: first.id, limit: 3},
	+ %{min_id: first.id, limit: 3},
	+ %{min_id: middle.id, max_id: last.id, limit: 2},
	+ %{min_id: first.id, order_asc: true, limit: 4}
	+ ] do
	+ opts = Map.merge(opts, bounds)
	+ expected = ActivityPub.fetch_activities(recipients, opts)
	+
	+ {activities, queries} = home(user, opts)
	+
	+ assert activities == expected
	+ assert per_owner?(queries)
	+ end
	+ end
	+
	+ test "prunes muted, blocked and (for local_only) remote owners", ctx do
	+ muted = insert(:user)
	+ blocked = insert(:user)
	+ remote = insert(:user, local: false)
	+ {:ok, user, _} = User.follow(ctx.user, muted)
	+ {:ok, user, _} = User.follow(user, blocked)
	+ {:ok, user, _} = User.follow(user, remote)
	+ {:ok, _} = User.mute(user, muted)
	+ {:ok, _} = User.block(user, blocked)
	+
	+ for author <- [user, ctx.followed, muted, blocked] do
	+ {:ok, _} = CommonAPI.post(author, %{status: "hi"})
	+ end
	+
	+ insert(:note_activity, user: remote, local: false)
	+
	+ opts = Map.merge(ctx.opts, %{user: user, muting_user: user, blocking_user: user})
	+
	+ for {filters, expected_owners} <- [
	+ {%{}, [user, ctx.followed, remote]},
	+ {%{with_muted: true}, [user, ctx.followed, muted, remote]},
	+ {%{local_only: true}, [user, ctx.followed]}
	+ ] do
	+ opts = Map.merge(opts, filters)
	+
	+ {activities, owners} =
	+ owners(fn -> ActivityPub.fetch_home_activities(recipients(user), opts) end)
	+
	+ assert Enum.sort(owners) == Enum.sort(Enum.map(expected_owners, & &1.ap_id))
	+ assert activities == ActivityPub.fetch_activities(recipients(user), opts)
	+ end
	+ end
	+
	+ test "keeps the overlap query for followed hashtags", ctx do
	+ hashtag = insert(:hashtag, name: "tenshi")
	+ {:ok, tagged} = CommonAPI.post(ctx.other, %{status: "maji #tenshi"})
	+ {:ok, post} = CommonAPI.post(ctx.followed, %{status: "hi"})
	+
	+ {activities, queries} = home(ctx.user, Map.put(ctx.opts, :followed_hashtags, [hashtag.id]))
	+
	+ assert ids(activities) == [tagged.id, post.id]
	+ refute per_owner?(queries)
	+ end
	+
	+ test "keeps the overlap query without a list of types", ctx do
	+ {:ok, post} = CommonAPI.post(ctx.followed, %{status: "hi"})
	+
	+ {activities, queries} = home(ctx.user, Map.delete(ctx.opts, :type))
	+
	+ assert ids(activities) == [post.id]
	+ refute per_owner?(queries)
	+ end
	+ end
	+
	+ # The owner list is the first parameter of the per-owner query.
	+ defp owners(fun) do
	+ id = {__MODULE__, make_ref()}
	+ :ok = :telemetry.attach(id, [:pleroma, :repo, :query], &__MODULE__.record_params/4, self())
	+
	+ try do
	+ result = fun.()
	+ assert_received {:lateral_params, [owners \| _]}
	+ {result, owners}
	+ after
	+ :telemetry.detach(id)
	+ end
	+ end
	+
	+ def record_params(_event, _measurements, %{query: query, params: params}, pid) do
	+ if String.contains?(query, "LATERAL"), do: send(pid, {:lateral_params, params})
	+ end
	+
	+ defp recipients(user), do: [user.ap_id \| User.following(user)]
	+
	+ defp home(user, opts) do
	+ queries(fn -> ActivityPub.fetch_home_activities(recipients(user), opts) end)
	+ end
	+
	+ defp ids(activities), do: Enum.map(activities, & &1.id)
	+
	defp insert_gap(count, user) do
	now = NaiveDateTime.utc_now() \|> NaiveDateTime.truncate(:second)

	rows =
	for _ <- 1..count,
	do: %{
	id: FlakeId.get(),
	actor: user.ap_id,
	recipients: [],
	data: %{"type" => "Like"},
	inserted_at: now,
	updated_at: now
	}

	Repo.insert_all(Pleroma.Activity, rows)
	end

	defp queries(fun) do
	id = {__MODULE__, make_ref()}
	:ok = :telemetry.attach(id, [:pleroma, :repo, :query], &__MODULE__.record_query/4, self())

	try do
	result = fun.()
	{result, drain_queries()}
	after
	:telemetry.detach(id)
	end
	end

	def record_query(_event, _measurements, metadata, pid), do: send(pid, {:query, metadata.query})

	defp drain_queries do
	receive do
	{:query, query} -> [query \| drain_queries()]
	after
	0 -> []
	end
	end

	+ defp per_owner?(queries), do: Enum.any?(queries, &String.contains?(&1, "LATERAL"))
	+
	defp fast?(queries), do: Enum.any?(queries, &String.contains?(&1, "home_timeline_candidates"))

	defp fallback?(queries),
	do:
	Enum.any?(queries, &(String.contains?(&1, " && ") and not String.contains?(&1, "unnest(")))
	end

File Metadata

Mime Type: text/x-diff
Expires: Fri, Oct 30, 4:46 AM (1 d, 25 m)
Storage Engine: blob
Storage Format: Raw Data
Storage Handle: 1799479
Default Alt Text: (99 KB)

No OneTemporaryActions

View Options

File Metadata

Event Timeline

No OneTemporary
Actions