No OneTemporary
Actions

Authored By

Unknown

Size

33 KB

Referenced Files

None

Subscribers

None

View Options

	diff --git a/lib/pleroma/web/activity_pub/home_timeline.ex b/lib/pleroma/web/activity_pub/home_timeline.ex
	index 8864662b6..7df47bf7a 100644
	--- a/lib/pleroma/web/activity_pub/home_timeline.ex
	+++ b/lib/pleroma/web/activity_pub/home_timeline.ex
	@@ -1,265 +1,267 @@
	# 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, rejected_actors) do
	opts = Map.put(opts, :skip_extra_order, true)

	result =
	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, 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)

	owner_ids = owner_ids(owners, recipients, opts, options, build_query)
	bound = page_bound(owner_ids, options)

	case owner_ids ++ addressed_ids(recipients -- collections, options, build_query, bound) do
	[] -> []
	ids -> reload(ids, recipients, opts, build_query)
	end
	end

	# A full page of owner posts ends where the page ends: a post addressed to
	# the user beyond its last id cannot make the page.
	+ defp page_bound([], _options), do: nil
	+
	defp page_bound(ids, options) do
	if length(ids) >= Pagination.page_size(options) do
	{if(ascending?(options), do: :below, else: :above), List.last(ids)}
	end
	end

	# the order Pagination gives a keyset page
	defp ascending?(%{min_id: _}), do: true
	defp ascending?(%{max_id: _}), do: false
	defp ascending?(options), do: options[:order_asc] == true

	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, :owner)
	\|> 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, _bound), do: []

	defp addressed_ids(recipients, options, build_query, bound) do
	build_query.(Activity, recipients, :addressed)
	\|> within(bound)
	\|> page_ids(options)
	\|> Repo.all()
	\|> Enum.map(& &1.id)
	end

	defp within(query, nil), do: query
	defp within(query, {:above, id}), do: where(query, [a], a.id > ^id)
	defp within(query, {:below, id}), do: where(query, [a], a.id < ^id)

	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(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, 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(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, 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, recipients, opts, build_query, deadline) do
	remaining = deadline - System.monotonic_time(:millisecond)

	if remaining > 0 do
	set_timeout(remaining)
	{: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 173ac8351..e2b84b92f 100644
	--- a/test/pleroma/web/activity_pub/home_timeline_test.exs
	+++ b/test/pleroma/web/activity_pub/home_timeline_test.exs
	@@ -1,705 +1,706 @@
	# 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

	# A type list next to the owner's type makes PostgreSQL < 17 read every
	# (owner, type) row and sort them instead of stopping after one page.
	test "leaves the type to the owner in the per-owner query", ctx do
	{:ok, post} = CommonAPI.post(ctx.followed, %{status: "hi"})

	{activities, queries} = home(ctx.user, ctx.opts)

	assert ids(activities) == [post.id]
	lateral = Enum.find(queries, &String.contains?(&1, "LATERAL"))
	refute lateral =~ ~r/->>'type' = ANY\(/
	end

	test "searches direct addressing only within a full page of owner posts", ctx do
	{:ok, old_dm} =
	CommonAPI.post(ctx.other, %{status: "@#{ctx.user.nickname} old", visibility: "direct"})

	posts =
	for index <- 1..4 do
	{:ok, post} = CommonAPI.post(ctx.followed, %{status: "post #{index}"})
	post
	end

	{:ok, new_dm} =
	CommonAPI.post(ctx.other, %{status: "@#{ctx.user.nickname} new", visibility: "direct"})

	recipients = recipients(ctx.user)

	for bounds <- [
	+ %{limit: 0},
	%{limit: 2},
	%{limit: 6},
	%{max_id: new_dm.id, limit: 2},
	%{min_id: old_dm.id, limit: 2},
	%{min_id: hd(posts).id, limit: 2},
	%{order_asc: true, limit: 2}
	] do
	opts = Map.merge(ctx.opts, bounds)
	{activities, _queries} = home(ctx.user, opts)
	assert activities == ActivityPub.fetch_activities(recipients, opts)
	end

	# a full page of owner posts bounds the search for direct addressing
	{_activities, queries} = home(ctx.user, Map.put(ctx.opts, :limit, 2))
	addressed = Enum.find(queries, &(&1 =~ ~r/&& a0\."recipients"/ and &1 =~ "LIMIT"))
	assert addressed =~ ~r/a0\."id" > \$\d+/
	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: Mon, Nov 2, 1:45 AM (13 h, 33 m)
Storage Engine: blob
Storage Format: Raw Data
Storage Handle: 1803494
Default Alt Text: (33 KB)

No OneTemporaryActions

View Options

File Metadata

Event Timeline

No OneTemporary
Actions