No OneTemporary
Actions

Size

71 KB

Referenced Files

None

Subscribers

None

View Options

	diff --git a/benchmarks/load_testing/fetcher.ex b/benchmarks/load_testing/fetcher.ex
	index a45a71d4a..0a18099d9 100644
	--- a/benchmarks/load_testing/fetcher.ex
	+++ b/benchmarks/load_testing/fetcher.ex
	@@ -1,260 +1,324 @@
	defmodule Pleroma.LoadTesting.Fetcher do
	use Pleroma.LoadTesting.Helper

	def fetch_user(user) do
	Benchee.run(%{
	"By id" => fn -> Repo.get_by(User, id: user.id) end,
	"By ap_id" => fn -> Repo.get_by(User, ap_id: user.ap_id) end,
	"By email" => fn -> Repo.get_by(User, email: user.email) end,
	"By nickname" => fn -> Repo.get_by(User, nickname: user.nickname) end
	})
	end

	+ def query_only_media_timelines(user) do
	+ home_timeline_params = %{
	+ "count" => 20,
	+ "with_muted" => "true",
	+ "type" => ["Create", "Announce"],
	+ "blocking_user" => user,
	+ "muting_user" => user,
	+ "user" => user,
	+ "only_media" => true
	+ }
	+
	+ mastodon_federated_timeline_params = %{
	+ "count" => 20,
	+ "only_media" => "true",
	+ "type" => ["Create", "Announce"],
	+ "with_muted" => "true",
	+ "blocking_user" => user,
	+ "muting_user" => user
	+ }
	+
	+ following = User.following(user)
	+
	+ Benchee.run(%{
	+ "Home timeline with only_media flag" => fn ->
	+ Pleroma.Web.ActivityPub.ActivityPub.fetch_activities(
	+ [user.ap_id \| following],
	+ home_timeline_params
	+ )
	+ end,
	+ "Home timeline with only_media not null" => fn ->
	+ Pleroma.Web.ActivityPub.ActivityPub.fetch_activities(
	+ [user.ap_id \| following],
	+ Map.put(home_timeline_params, "only_media", :is_not_null)
	+ )
	+ end,
	+ "Home timeline" => fn ->
	+ Pleroma.Web.ActivityPub.ActivityPub.fetch_activities(
	+ [user.ap_id \| following],
	+ Map.put(home_timeline_params, "only_media", "false")
	+ )
	+ end
	+ })
	+
	+ Benchee.run(%{
	+ "Public timeline with only_media flag" => fn ->
	+ Pleroma.Web.ActivityPub.ActivityPub.fetch_public_activities(
	+ mastodon_federated_timeline_params
	+ )
	+ end,
	+ "Public timeline" => fn ->
	+ Pleroma.Web.ActivityPub.ActivityPub.fetch_public_activities(
	+ Map.put(mastodon_federated_timeline_params, "only_media", "false")
	+ )
	+ end,
	+ "Public timeline with only_media not null" => fn ->
	+ Pleroma.Web.ActivityPub.ActivityPub.fetch_public_activities(
	+ Map.put(mastodon_federated_timeline_params, "only_media", :is_not_null)
	+ )
	+ end
	+ })
	+ end
	+
	def query_timelines(user) do
	home_timeline_params = %{
	"count" => 20,
	"with_muted" => true,
	"type" => ["Create", "Announce"],
	"blocking_user" => user,
	"muting_user" => user,
	"user" => user
	}

	mastodon_public_timeline_params = %{
	"count" => 20,
	"local_only" => true,
	"only_media" => "false",
	"type" => ["Create", "Announce"],
	"with_muted" => "true",
	"blocking_user" => user,
	"muting_user" => user
	}

	mastodon_federated_timeline_params = %{
	"count" => 20,
	"only_media" => "false",
	"type" => ["Create", "Announce"],
	"with_muted" => "true",
	"blocking_user" => user,
	"muting_user" => user
	}

	following = User.following(user)

	Benchee.run(%{
	"User home timeline" => fn ->
	Pleroma.Web.ActivityPub.ActivityPub.fetch_activities(
	following,
	home_timeline_params
	)
	end,
	"User mastodon public timeline" => fn ->
	Pleroma.Web.ActivityPub.ActivityPub.fetch_public_activities(
	mastodon_public_timeline_params
	)
	end,
	"User mastodon federated public timeline" => fn ->
	Pleroma.Web.ActivityPub.ActivityPub.fetch_public_activities(
	mastodon_federated_timeline_params
	)
	end
	})

	home_activities =
	Pleroma.Web.ActivityPub.ActivityPub.fetch_activities(
	following,
	home_timeline_params
	)

	public_activities =
	Pleroma.Web.ActivityPub.ActivityPub.fetch_public_activities(mastodon_public_timeline_params)

	public_federated_activities =
	Pleroma.Web.ActivityPub.ActivityPub.fetch_public_activities(
	mastodon_federated_timeline_params
	)

	Benchee.run(%{
	"Rendering home timeline" => fn ->
	Pleroma.Web.MastodonAPI.StatusView.render("index.json", %{
	activities: home_activities,
	for: user,
	as: :activity
	})
	end,
	"Rendering public timeline" => fn ->
	Pleroma.Web.MastodonAPI.StatusView.render("index.json", %{
	activities: public_activities,
	for: user,
	as: :activity
	})
	end,
	"Rendering public federated timeline" => fn ->
	Pleroma.Web.MastodonAPI.StatusView.render("index.json", %{
	activities: public_federated_activities,
	for: user,
	as: :activity
	})
	end,
	"Rendering favorites timeline" => fn ->
	conn = Phoenix.ConnTest.build_conn(:get, "http://localhost:4001/api/v1/favourites", nil)
	- Pleroma.Web.MastodonAPI.StatusController.favourites(
	- %Plug.Conn{conn \|
	- assigns: %{user: user},
	- query_params: %{"limit" => "0"},
	- body_params: %{},
	- cookies: %{},
	- params: %{},
	- path_params: %{},
	- private: %{
	- Pleroma.Web.Router => {[], %{}},
	- phoenix_router: Pleroma.Web.Router,
	- phoenix_action: :favourites,
	- phoenix_controller: Pleroma.Web.MastodonAPI.StatusController,
	- phoenix_endpoint: Pleroma.Web.Endpoint,
	- phoenix_format: "json",
	- phoenix_layout: {Pleroma.Web.LayoutView, "app.html"},
	- phoenix_recycled: true,

	- phoenix_view: Pleroma.Web.MastodonAPI.StatusView,
	- plug_session: %{"user_id" => user.id},
	- plug_session_fetch: :done,
	- plug_session_info: :write,
	- plug_skip_csrf_protection: true
	- }
	+ Pleroma.Web.MastodonAPI.StatusController.favourites(
	+ %Plug.Conn{
	+ conn
	+ \| assigns: %{user: user},
	+ query_params: %{"limit" => "0"},
	+ body_params: %{},
	+ cookies: %{},
	+ params: %{},
	+ path_params: %{},
	+ private: %{
	+ Pleroma.Web.Router => {[], %{}},
	+ phoenix_router: Pleroma.Web.Router,
	+ phoenix_action: :favourites,
	+ phoenix_controller: Pleroma.Web.MastodonAPI.StatusController,
	+ phoenix_endpoint: Pleroma.Web.Endpoint,
	+ phoenix_format: "json",
	+ phoenix_layout: {Pleroma.Web.LayoutView, "app.html"},
	+ phoenix_recycled: true,
	+ phoenix_view: Pleroma.Web.MastodonAPI.StatusView,
	+ plug_session: %{"user_id" => user.id},
	+ plug_session_fetch: :done,
	+ plug_session_info: :write,
	+ plug_skip_csrf_protection: true
	+ }
	},
	- %{})
	- end,
	+ %{}
	+ )
	+ end
	})
	end

	def query_notifications(user) do
	without_muted_params = %{"count" => "20", "with_muted" => "false"}
	with_muted_params = %{"count" => "20", "with_muted" => "true"}

	Benchee.run(%{
	"Notifications without muted" => fn ->
	Pleroma.Web.MastodonAPI.MastodonAPI.get_notifications(user, without_muted_params)
	end,
	"Notifications with muted" => fn ->
	Pleroma.Web.MastodonAPI.MastodonAPI.get_notifications(user, with_muted_params)
	end
	})

	without_muted_notifications =
	Pleroma.Web.MastodonAPI.MastodonAPI.get_notifications(user, without_muted_params)

	with_muted_notifications =
	Pleroma.Web.MastodonAPI.MastodonAPI.get_notifications(user, with_muted_params)

	Benchee.run(%{
	"Render notifications without muted" => fn ->
	Pleroma.Web.MastodonAPI.NotificationView.render("index.json", %{
	notifications: without_muted_notifications,
	for: user
	})
	end,
	"Render notifications with muted" => fn ->
	Pleroma.Web.MastodonAPI.NotificationView.render("index.json", %{
	notifications: with_muted_notifications,
	for: user
	})
	end
	})
	end

	def query_dms(user) do
	params = %{
	"count" => "20",
	"with_muted" => "true",
	"type" => "Create",
	"blocking_user" => user,
	"user" => user,
	visibility: "direct"
	}

	Benchee.run(%{
	"Direct messages with muted" => fn ->
	Pleroma.Web.ActivityPub.ActivityPub.fetch_activities_query([user.ap_id], params)
	\|> Pleroma.Pagination.fetch_paginated(params)
	end,
	"Direct messages without muted" => fn ->
	Pleroma.Web.ActivityPub.ActivityPub.fetch_activities_query([user.ap_id], params)
	\|> Pleroma.Pagination.fetch_paginated(Map.put(params, "with_muted", false))
	end
	})

	dms_with_muted =
	Pleroma.Web.ActivityPub.ActivityPub.fetch_activities_query([user.ap_id], params)
	\|> Pleroma.Pagination.fetch_paginated(params)

	dms_without_muted =
	Pleroma.Web.ActivityPub.ActivityPub.fetch_activities_query([user.ap_id], params)
	\|> Pleroma.Pagination.fetch_paginated(Map.put(params, "with_muted", false))

	Benchee.run(%{
	"Rendering dms with muted" => fn ->
	Pleroma.Web.MastodonAPI.StatusView.render("index.json", %{
	activities: dms_with_muted,
	for: user,
	as: :activity
	})
	end,
	"Rendering dms without muted" => fn ->
	Pleroma.Web.MastodonAPI.StatusView.render("index.json", %{
	activities: dms_without_muted,
	for: user,
	as: :activity
	})
	end
	})
	end

	def query_long_thread(user, activity) do
	Benchee.run(%{
	"Fetch main post" => fn ->
	Pleroma.Activity.get_by_id_with_object(activity.id)
	end,
	"Fetch context of main post" => fn ->
	Pleroma.Web.ActivityPub.ActivityPub.fetch_activities_for_context(
	activity.data["context"],
	%{
	"blocking_user" => user,
	"user" => user,
	"exclude_id" => activity.id
	}
	)
	end
	})

	activity = Pleroma.Activity.get_by_id_with_object(activity.id)

	context =
	Pleroma.Web.ActivityPub.ActivityPub.fetch_activities_for_context(
	activity.data["context"],
	%{
	"blocking_user" => user,
	"user" => user,
	"exclude_id" => activity.id
	}
	)

	Benchee.run(%{
	"Render status" => fn ->
	Pleroma.Web.MastodonAPI.StatusView.render("show.json", %{
	activity: activity,
	for: user
	})
	end,
	"Render context" => fn ->
	Pleroma.Web.MastodonAPI.StatusView.render(
	"index.json",
	for: user,
	activities: context,
	as: :activity
	)
	\|> Enum.reverse()
	end
	})
	end
	end
	diff --git a/benchmarks/load_testing/generator.ex b/benchmarks/load_testing/generator.ex
	index a957e0ffb..776a8b1d4 100644
	--- a/benchmarks/load_testing/generator.ex
	+++ b/benchmarks/load_testing/generator.ex
	@@ -1,367 +1,394 @@
	defmodule Pleroma.LoadTesting.Generator do
	use Pleroma.LoadTesting.Helper
	alias Pleroma.Web.CommonAPI

	def generate_like_activities(user, posts) do
	count_likes = Kernel.trunc(length(posts) / 4)
	IO.puts("Starting generating #{count_likes} like activities...")

	{time, _} =
	:timer.tc(fn ->
	Task.async_stream(
	- Enum.take_random(posts, count_likes),
	+ Enum.take_random(posts, count_likes),
	fn post -> {:ok, _, _} = CommonAPI.favorite(post.id, user) end,
	max_concurrency: 10,
	timeout: 30_000
	)
	\|> Stream.run()
	end)

	IO.puts("Inserting like activities take #{to_sec(time)} sec.\n")
	end

	def generate_users(opts) do
	IO.puts("Starting generating #{opts[:users_max]} users...")
	{time, _} = :timer.tc(fn -> do_generate_users(opts) end)

	IO.puts("Inserting users take #{to_sec(time)} sec.\n")
	end

	defp do_generate_users(opts) do
	max = Keyword.get(opts, :users_max)

	Task.async_stream(
	1..max,
	&generate_user_data(&1),
	max_concurrency: 10,
	timeout: 30_000
	)
	\|> Enum.to_list()
	end

	defp generate_user_data(i) do
	remote = Enum.random([true, false])

	user = %User{
	name: "Test テスト User #{i}",
	email: "user#{i}@example.com",
	nickname: "nick#{i}",
	password_hash:
	"$pbkdf2-sha512$160000$bU.OSFI7H/yqWb5DPEqyjw$uKp/2rmXw12QqnRRTqTtuk2DTwZfF8VR4MYW2xMeIlqPR/UX1nT1CEKVUx2CowFMZ5JON8aDvURrZpJjSgqXrg",
	bio: "Tester Number #{i}",
	local: remote
	}

	user_urls =
	if remote do
	base_url =
	Enum.random(["https://domain1.com", "https://domain2.com", "https://domain3.com"])

	ap_id = "#{base_url}/users/#{user.nickname}"

	%{
	ap_id: ap_id,
	follower_address: ap_id <> "/followers",
	following_address: ap_id <> "/following"
	}
	else
	%{
	ap_id: User.ap_id(user),
	follower_address: User.ap_followers(user),
	following_address: User.ap_following(user)
	}
	end

	user = Map.merge(user, user_urls)

	Repo.insert!(user)
	end

	def generate_activities(user, users) do
	- do_generate_activities(user, users)
	- end
	-
	- defp do_generate_activities(user, users) do
	IO.puts("Starting generating 20000 common activities...")

	{time, _} =
	:timer.tc(fn ->
	Task.async_stream(
	1..20_000,
	- fn _ ->
	- do_generate_activity([user \| users])
	+ fn i ->
	+ do_generate_activity([user \| users], i)
	end,
	max_concurrency: 10,
	timeout: 30_000
	)
	\|> Stream.run()
	end)

	IO.puts("Inserting common activities take #{to_sec(time)} sec.\n")
	+ end

	+ def generate_activities_with_mentions(user, users) do
	IO.puts("Starting generating 20000 activities with mentions...")

	{time, _} =
	:timer.tc(fn ->
	Task.async_stream(
	1..20_000,
	fn _ ->
	do_generate_activity_with_mention(user, users)
	end,
	max_concurrency: 10,
	timeout: 30_000
	)
	\|> Stream.run()
	end)

	IO.puts("Inserting activities with menthions take #{to_sec(time)} sec.\n")
	+ end

	+ def generate_activities_with_thread(user, users) do
	IO.puts("Starting generating 10000 activities with threads...")

	{time, _} =
	:timer.tc(fn ->
	Task.async_stream(
	1..10_000,
	fn _ ->
	do_generate_threads([user \| users])
	end,
	max_concurrency: 10,
	timeout: 30_000
	)
	\|> Stream.run()
	end)

	IO.puts("Inserting activities with threads take #{to_sec(time)} sec.\n")
	end

	- defp do_generate_activity(users) do
	+ defp do_generate_activity(users, i) when i > 19_900 and i < 19_951 and rem(i, 2) == 0 do
	+ user = Enum.random(users)
	+
	+ obj_data = %{
	+ "actor" => user.ap_id,
	+ "name" => "4467-11.jpg",
	+ "type" => "Document",
	+ "url" => [
	+ %{
	+ "href" =>
	+ "http://#{Pleroma.Web.base_url()}/media/b1b873552422a07bf53af01f3c231c841db4dfc42c35efde681abaf0f2a4eab7.jpg",
	+ "mediaType" => "image/jpeg",
	+ "type" => "Link"
	+ }
	+ ]
	+ }
	+
	+ {:ok, object} = Repo.insert(%Pleroma.Object{data: obj_data})
	+
	+ post = %{
	+ "status" => "Some status without mention with random user",
	+ "media_ids" => [object.id]
	+ }
	+
	+ CommonAPI.post(user, post)
	+ end
	+
	+ defp do_generate_activity(users, _i) do
	post = %{
	"status" => "Some status without mention with random user"
	}

	CommonAPI.post(Enum.random(users), post)
	end

	defp do_generate_activity_with_mention(user, users) do
	mentions_cnt = Enum.random([2, 3, 4, 5])
	with_user = Enum.random([true, false])
	users = Enum.shuffle(users)
	mentions_users = Enum.take(users, mentions_cnt)
	mentions_users = if with_user, do: [user \| mentions_users], else: mentions_users

	mentions_str =
	Enum.map(mentions_users, fn user -> "@" <> user.nickname end) \|> Enum.join(", ")

	post = %{
	"status" => mentions_str <> "some status with mentions random users"
	}

	CommonAPI.post(Enum.random(users), post)
	end

	defp do_generate_threads(users) do
	thread_length = Enum.random([2, 3, 4, 5])
	actor = Enum.random(users)

	post = %{
	"status" => "Start of the thread"
	}

	{:ok, activity} = CommonAPI.post(actor, post)

	Enum.each(1..thread_length, fn _ ->
	user = Enum.random(users)

	post = %{
	"status" => "@#{actor.nickname} reply to thread",
	"in_reply_to_status_id" => activity.id
	}

	CommonAPI.post(user, post)
	end)
	end

	def generate_remote_activities(user, users) do
	do_generate_remote_activities(user, users)
	end

	defp do_generate_remote_activities(user, users) do
	IO.puts("Starting generating 10000 remote activities...")

	{time, _} =
	:timer.tc(fn ->
	Task.async_stream(
	1..10_000,
	fn i ->
	do_generate_remote_activity(i, user, users)
	end,
	max_concurrency: 10,
	timeout: 30_000
	)
	\|> Stream.run()
	end)

	IO.puts("Inserting remote activities take #{to_sec(time)} sec.\n")
	end

	defp do_generate_remote_activity(i, user, users) do
	actor = Enum.random(users)
	%{host: host} = URI.parse(actor.ap_id)
	date = Date.utc_today()
	datetime = DateTime.utc_now()

	map = %{
	"actor" => actor.ap_id,
	"cc" => [actor.follower_address, user.ap_id],
	"context" => "tag:mastodon.example.org,#{date}:objectId=#{i}:objectType=Conversation",
	"id" => actor.ap_id <> "/statuses/#{i}/activity",
	"object" => %{
	"actor" => actor.ap_id,
	"atomUri" => actor.ap_id <> "/statuses/#{i}",
	"attachment" => [],
	"attributedTo" => actor.ap_id,
	"bcc" => [],
	"bto" => [],
	"cc" => [actor.follower_address, user.ap_id],
	"content" =>
	"<p><span class=\"h-card\"><a href=\"" <>
	user.ap_id <>
	"\" class=\"u-url mention\">@<span>" <> user.nickname <> "</span></a></span></p>",
	"context" => "tag:mastodon.example.org,#{date}:objectId=#{i}:objectType=Conversation",
	"conversation" =>
	"tag:mastodon.example.org,#{date}:objectId=#{i}:objectType=Conversation",
	"emoji" => %{},
	"id" => actor.ap_id <> "/statuses/#{i}",
	"inReplyTo" => nil,
	"inReplyToAtomUri" => nil,
	"published" => datetime,
	"sensitive" => true,
	"summary" => "cw",
	"tag" => [
	%{
	"href" => user.ap_id,
	"name" => "@#{user.nickname}@#{host}",
	"type" => "Mention"
	}
	],
	"to" => ["https://www.w3.org/ns/activitystreams#Public"],
	"type" => "Note",
	"url" => "http://#{host}/@#{actor.nickname}/#{i}"
	},
	"published" => datetime,
	"to" => ["https://www.w3.org/ns/activitystreams#Public"],
	"type" => "Create"
	}

	Pleroma.Web.ActivityPub.ActivityPub.insert(map, false)
	end

	def generate_dms(user, users, opts) do
	IO.puts("Starting generating #{opts[:dms_max]} DMs")
	{time, _} = :timer.tc(fn -> do_generate_dms(user, users, opts) end)
	IO.puts("Inserting dms take #{to_sec(time)} sec.\n")
	end

	defp do_generate_dms(user, users, opts) do
	Task.async_stream(
	1..opts[:dms_max],
	fn _ ->
	do_generate_dm(user, users)
	end,
	max_concurrency: 10,
	timeout: 30_000
	)
	\|> Stream.run()
	end

	defp do_generate_dm(user, users) do
	post = %{
	"status" => "@#{user.nickname} some direct message",
	"visibility" => "direct"
	}

	CommonAPI.post(Enum.random(users), post)
	end

	def generate_long_thread(user, users, opts) do
	IO.puts("Starting generating long thread with #{opts[:thread_length]} replies")
	{time, activity} = :timer.tc(fn -> do_generate_long_thread(user, users, opts) end)
	IO.puts("Inserting long thread replies take #{to_sec(time)} sec.\n")
	{:ok, activity}
	end

	defp do_generate_long_thread(user, users, opts) do
	{:ok, %{id: id} = activity} = CommonAPI.post(user, %{"status" => "Start of long thread"})

	Task.async_stream(
	1..opts[:thread_length],
	fn _ -> do_generate_thread(users, id) end,
	max_concurrency: 10,
	timeout: 30_000
	)
	\|> Stream.run()

	activity
	end

	defp do_generate_thread(users, activity_id) do
	CommonAPI.post(Enum.random(users), %{
	"status" => "reply to main post",
	"in_reply_to_status_id" => activity_id
	})
	end

	def generate_non_visible_message(user, users) do
	IO.puts("Starting generating 1000 non visible posts")

	{time, _} =
	:timer.tc(fn ->
	do_generate_non_visible_posts(user, users)
	end)

	IO.puts("Inserting non visible posts take #{to_sec(time)} sec.\n")
	end

	defp do_generate_non_visible_posts(user, users) do
	[not_friend \| users] = users

	make_friends(user, users)

	Task.async_stream(1..1000, fn _ -> do_generate_non_visible_post(not_friend, users) end,
	max_concurrency: 10,
	timeout: 30_000
	)
	\|> Stream.run()
	end

	defp make_friends(_user, []), do: nil

	defp make_friends(user, [friend \| users]) do
	{:ok, _} = User.follow(user, friend)
	{:ok, _} = User.follow(friend, user)
	make_friends(user, users)
	end

	defp do_generate_non_visible_post(not_friend, users) do
	post = %{
	"status" => "some non visible post",
	"visibility" => "private"
	}

	{:ok, activity} = CommonAPI.post(not_friend, post)

	thread_length = Enum.random([2, 3, 4, 5])

	Enum.each(1..thread_length, fn _ ->
	user = Enum.random(users)

	post = %{
	"status" => "@#{not_friend.nickname} reply to non visible post",
	"in_reply_to_status_id" => activity.id,
	"visibility" => "private"
	}

	CommonAPI.post(user, post)
	end)
	end
	end
	diff --git a/benchmarks/mix/tasks/pleroma/load_testing.ex b/benchmarks/mix/tasks/pleroma/load_testing.ex
	index 0a751adac..664adb9b6 100644
	--- a/benchmarks/mix/tasks/pleroma/load_testing.ex
	+++ b/benchmarks/mix/tasks/pleroma/load_testing.ex
	@@ -1,138 +1,146 @@
	defmodule Mix.Tasks.Pleroma.LoadTesting do
	use Mix.Task
	use Pleroma.LoadTesting.Helper
	import Mix.Pleroma
	import Pleroma.LoadTesting.Generator
	import Pleroma.LoadTesting.Fetcher

	@shortdoc "Factory for generation data"
	@moduledoc """
	Generates data like:
	- local/remote users
	- local/remote activities with notifications
	- direct messages
	- long thread
	- non visible posts

	## Generate data
	MIX_ENV=benchmark mix pleroma.load_testing --users 20000 --dms 20000 --thread_length 2000
	MIX_ENV=benchmark mix pleroma.load_testing -u 20000 -d 20000 -t 2000

	Options:
	- `--users NUMBER` - number of users to generate. Defaults to: 20000. Alias: `-u`
	- `--dms NUMBER` - number of direct messages to generate. Defaults to: 20000. Alias `-d`
	- `--thread_length` - number of messages in thread. Defaults to: 2000. ALias `-t`
	"""

	@aliases [u: :users, d: :dms, t: :thread_length]
	@switches [
	users: :integer,
	dms: :integer,
	thread_length: :integer
	]
	@users_default 20_000
	@dms_default 1_000
	@thread_length_default 2_000

	def run(args) do
	start_pleroma()
	Pleroma.Config.put([:instance, :skip_thread_containment], true)
	{opts, _} = OptionParser.parse!(args, strict: @switches, aliases: @aliases)

	users_max = Keyword.get(opts, :users, @users_default)
	dms_max = Keyword.get(opts, :dms, @dms_default)
	thread_length = Keyword.get(opts, :thread_length, @thread_length_default)

	clean_tables()

	opts =
	Keyword.put(opts, :users_max, users_max)
	\|> Keyword.put(:dms_max, dms_max)
	\|> Keyword.put(:thread_length, thread_length)

	generate_users(opts)

	# main user for queries
	IO.puts("Fetching local main user...")

	{time, user} =
	:timer.tc(fn ->
	Repo.one(
	from(u in User, where: u.local == true, order_by: fragment("RANDOM()"), limit: 1)
	)
	end)

	IO.puts("Fetching main user take #{to_sec(time)} sec.\n")

	IO.puts("Fetching local users...")

	{time, users} =
	:timer.tc(fn ->
	Repo.all(
	from(u in User,
	where: u.id != ^user.id,
	where: u.local == true,
	order_by: fragment("RANDOM()"),
	limit: 10
	)
	)
	end)

	IO.puts("Fetching local users take #{to_sec(time)} sec.\n")

	IO.puts("Fetching remote users...")

	{time, remote_users} =
	:timer.tc(fn ->
	Repo.all(
	from(u in User,
	where: u.id != ^user.id,
	where: u.local == false,
	order_by: fragment("RANDOM()"),
	limit: 10
	)
	)
	end)

	IO.puts("Fetching remote users take #{to_sec(time)} sec.\n")

	generate_activities(user, users)

	+ query_only_media_timelines(user)
	+
	+ generate_activities_with_mentions(user, users)
	+
	+ generate_activities_with_thread(user, users)
	+
	generate_remote_activities(user, remote_users)

	generate_like_activities(
	- user, Pleroma.Repo.all(Pleroma.Activity.Queries.by_type("Create"))
	+ user,
	+ Pleroma.Repo.all(Pleroma.Activity.Queries.by_type("Create"))
	)

	generate_dms(user, users, opts)

	{:ok, activity} = generate_long_thread(user, users, opts)

	generate_non_visible_message(user, users)

	IO.puts("Users in DB: #{Repo.aggregate(from(u in User), :count, :id)}")

	IO.puts("Activities in DB: #{Repo.aggregate(from(a in Pleroma.Activity), :count, :id)}")

	IO.puts("Objects in DB: #{Repo.aggregate(from(o in Pleroma.Object), :count, :id)}")

	IO.puts(
	"Notifications in DB: #{Repo.aggregate(from(n in Pleroma.Notification), :count, :id)}"
	)

	fetch_user(user)
	query_timelines(user)
	query_notifications(user)
	query_dms(user)
	query_long_thread(user, activity)
	Pleroma.Config.put([:instance, :skip_thread_containment], false)
	query_timelines(user)
	end

	defp clean_tables do
	IO.puts("Deleting old data...\n")
	+ Ecto.Adapters.SQL.query!(Repo, "TRUNCATE oban_jobs CASCADE;")
	Ecto.Adapters.SQL.query!(Repo, "TRUNCATE users CASCADE;")
	Ecto.Adapters.SQL.query!(Repo, "TRUNCATE activities CASCADE;")
	Ecto.Adapters.SQL.query!(Repo, "TRUNCATE objects CASCADE;")
	end
	end
	diff --git a/lib/pleroma/web/activity_pub/activity_pub.ex b/lib/pleroma/web/activity_pub/activity_pub.ex
	index 16e6b0057..b6b134994 100644
	--- a/lib/pleroma/web/activity_pub/activity_pub.ex
	+++ b/lib/pleroma/web/activity_pub/activity_pub.ex
	@@ -1,1422 +1,1429 @@
	# Pleroma: A lightweight social networking server
	# Copyright © 2017-2019 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.Conversation
	alias Pleroma.Conversation.Participation
	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.MRF
	alias Pleroma.Web.ActivityPub.Transmogrifier
	alias Pleroma.Web.ActivityPub.Utils
	alias Pleroma.Web.Streamer
	alias Pleroma.Web.WebFinger
	alias Pleroma.Workers.BackgroundWorker

	import Ecto.Query
	import Pleroma.Web.ActivityPub.Utils
	import Pleroma.Web.ActivityPub.Visibility

	require Logger
	require Pleroma.Constants

	# For Announce activities, we filter the recipients based on following status for any actors
	# that match actual users. See issue #164 for more information about why this is necessary.
	defp get_recipients(%{"type" => "Announce"} = data) do
	to = Map.get(data, "to", [])
	cc = Map.get(data, "cc", [])
	bcc = Map.get(data, "bcc", [])
	actor = User.get_cached_by_ap_id(data["actor"])

	recipients =
	Enum.filter(Enum.concat([to, cc, bcc]), fn recipient ->
	case User.get_cached_by_ap_id(recipient) do
	nil -> true
	user -> User.following?(user, actor)
	end
	end)

	{recipients, to, cc}
	end

	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_is_active(actor) do
	if not is_nil(actor) do
	with user <- User.get_cached_by_ap_id(actor),
	false <- user.deactivated do
	true
	else
	_e -> false
	end
	else
	true
	end
	end

	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 is_public?(object), do: User.increase_note_count(actor), else: {:ok, actor}
	end

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

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

	def increase_replies_count_if_reply(_create_data), do: :noop

	def decrease_replies_count_if_reply(%Object{
	data: %{"inReplyTo" => reply_ap_id} = object
	}) do
	if is_public?(object) do
	Object.decrease_replies_count(reply_ap_id)
	end
	end

	def decrease_replies_count_if_reply(_object), do: :noop

	def increase_poll_votes_if_vote(%{
	"object" => %{"inReplyTo" => reply_ap_id, "name" => name},
	"type" => "Create"
	}) do
	Object.increase_vote_count(reply_ap_id, name)
	end

	def increase_poll_votes_if_vote(_create_data), do: :noop

	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 <- bypass_actor_check \|\| check_actor_is_active(map["actor"]),
	{_, true} <- {:remote_limit_error, 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) do
	{:ok, activity} =
	Repo.insert(%Activity{
	data: map,
	local: local,
	actor: map["actor"],
	recipients: recipients
	})

	# Splice in the child object if we have one.
	activity =
	if not is_nil(object) do
	Map.put(activity, :object, object)
	else
	activity
	end

	BackgroundWorker.enqueue("fetch_data_for_activity", %{"activity_id" => activity.id})

	Notification.create_notifications(activity)

	conversation = create_or_bump_conversation(activity, map["actor"])
	participations = get_participations(conversation)
	stream_out(activity)
	stream_out_participations(participations)
	{:ok, activity}
	else
	%Activity{} = activity ->
	{:ok, activity}

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

	Pleroma.Web.RichMedia.Helpers.fetch_data_for_activity(activity)
	{:ok, activity}

	error ->
	{:error, error}
	end
	end

	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),
	Participation.mark_as_read(user, conversation) do
	{: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

	def stream_out_participations(%Object{data: %{"context" => context}}, user) do
	with %Conversation{} = conversation <- Conversation.get_for_ap_id(context),
	conversation = Repo.preload(conversation, :participations),
	last_activity_id =
	fetch_latest_activity_id_for_context(conversation.ap_id, %{
	"user" => user,
	"blocking_user" => user
	}) do
	if last_activity_id do
	stream_out_participations(conversation.participations)
	end
	end
	end

	def stream_out_participations(_, _), do: :noop

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

	def stream_out(_activity) do
	:noop
	end

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

	with create_data <-
	make_create_data(
	%{to: to, actor: actor, published: published, context: context, object: object},
	additional
	),
	{:ok, activity} <- insert(create_data, local, fake),
	{:fake, false, activity} <- {:fake, fake, activity},
	_ <- increase_replies_count_if_reply(create_data),
	_ <- increase_poll_votes_if_vote(create_data),
	{:quick_insert, false, activity} <- {:quick_insert, quick_insert?, activity},
	{:ok, _actor} <- increase_note_count_if_public(actor, activity),
	:ok <- maybe_federate(activity) do
	{:ok, activity}
	else
	{:quick_insert, true, activity} ->
	{:ok, activity}

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

	{:error, message} ->
	{:error, message}
	end
	end

	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]

	with listen_data <-
	make_listen_data(
	%{to: to, actor: actor, published: published, context: context, object: object},
	additional
	),
	{:ok, activity} <- insert(listen_data, local),
	:ok <- maybe_federate(activity) do
	{:ok, activity}
	else
	{:error, message} ->
	{:error, message}
	end
	end

	def accept(params) do
	accept_or_reject("Accept", params)
	end

	def reject(params) do
	accept_or_reject("Reject", params)
	end

	def accept_or_reject(type, %{to: to, actor: actor, object: object} = params) do
	local = Map.get(params, :local, true)
	activity_id = Map.get(params, :activity_id, nil)

	with data <-
	%{"to" => to, "type" => type, "actor" => actor.ap_id, "object" => object}
	\|> Utils.maybe_put("id", activity_id),
	{:ok, activity} <- insert(data, local),
	:ok <- maybe_federate(activity) do
	{:ok, activity}
	end
	end

	def update(%{to: to, cc: cc, actor: actor, object: object} = params) do
	local = !(params[:local] == false)
	activity_id = params[:activity_id]

	with data <- %{
	"to" => to,
	"cc" => cc,
	"type" => "Update",
	"actor" => actor,
	"object" => object
	},
	data <- Utils.maybe_put(data, "id", activity_id),
	{:ok, activity} <- insert(data, local),
	:ok <- maybe_federate(activity) do
	{:ok, activity}
	end
	end

	def react_with_emoji(user, object, emoji, options \\ []) do
	with local <- Keyword.get(options, :local, true),
	activity_id <- Keyword.get(options, :activity_id, nil),
	Pleroma.Emoji.is_unicode_emoji?(emoji),
	reaction_data <- make_emoji_reaction_data(user, object, emoji, activity_id),
	{:ok, activity} <- insert(reaction_data, local),
	{:ok, object} <- add_emoji_reaction_to_object(activity, object),
	:ok <- maybe_federate(activity) do
	{:ok, activity, object}
	end
	end

	def unreact_with_emoji(user, reaction_id, options \\ []) do
	with local <- Keyword.get(options, :local, true),
	activity_id <- Keyword.get(options, :activity_id, nil),
	user_ap_id <- user.ap_id,
	%Activity{actor: ^user_ap_id} = reaction_activity <- Activity.get_by_ap_id(reaction_id),
	object <- Object.normalize(reaction_activity),
	unreact_data <- make_undo_data(user, reaction_activity, activity_id),
	{:ok, activity} <- insert(unreact_data, local),
	{:ok, object} <- remove_emoji_reaction_from_object(reaction_activity, object),
	:ok <- maybe_federate(activity) do
	{:ok, activity, object}
	end
	end

	# TODO: This is weird, maybe we shouldn't check here if we can make the activity.
	def like(
	%User{ap_id: ap_id} = user,
	%Object{data: %{"id" => _}} = object,
	activity_id \\ nil,
	local \\ true
	) do
	with nil <- get_existing_like(ap_id, object),
	like_data <- make_like_data(user, object, activity_id),
	{:ok, activity} <- insert(like_data, local),
	{:ok, object} <- add_like_to_object(activity, object),
	:ok <- maybe_federate(activity) do
	{:ok, activity, object}
	else
	%Activity{} = activity -> {:ok, activity, object}
	error -> {:error, error}
	end
	end

	def unlike(%User{} = actor, %Object{} = object, activity_id \\ nil, local \\ true) do
	with %Activity{} = like_activity <- get_existing_like(actor.ap_id, object),
	unlike_data <- make_unlike_data(actor, like_activity, activity_id),
	{:ok, unlike_activity} <- insert(unlike_data, local),
	{:ok, _activity} <- Repo.delete(like_activity),
	{:ok, object} <- remove_like_from_object(like_activity, object),
	:ok <- maybe_federate(unlike_activity) do
	{:ok, unlike_activity, like_activity, object}
	else
	_e -> {:ok, object}
	end
	end

	def announce(
	%User{ap_id: _} = user,
	%Object{data: %{"id" => _}} = object,
	activity_id \\ nil,
	local \\ true,
	public \\ true
	) do
	with true <- is_announceable?(object, user, public),
	announce_data <- make_announce_data(user, object, activity_id, public),
	{:ok, activity} <- insert(announce_data, local),
	{:ok, object} <- add_announce_to_object(activity, object),
	:ok <- maybe_federate(activity) do
	{:ok, activity, object}
	else
	error -> {:error, error}
	end
	end

	def unannounce(
	%User{} = actor,
	%Object{} = object,
	activity_id \\ nil,
	local \\ true
	) do
	with %Activity{} = announce_activity <- get_existing_announce(actor.ap_id, object),
	unannounce_data <- make_unannounce_data(actor, announce_activity, activity_id),
	{:ok, unannounce_activity} <- insert(unannounce_data, local),
	:ok <- maybe_federate(unannounce_activity),
	{:ok, _activity} <- Repo.delete(announce_activity),
	{:ok, object} <- remove_announce_from_object(announce_activity, object) do
	{:ok, unannounce_activity, object}
	else
	_e -> {:ok, object}
	end
	end

	def follow(follower, followed, activity_id \\ nil, local \\ true) do
	with data <- make_follow_data(follower, followed, activity_id),
	{:ok, activity} <- insert(data, local),
	:ok <- maybe_federate(activity),
	_ <- User.set_follow_state_cache(follower.ap_id, followed.ap_id, activity.data["state"]) do
	{:ok, activity}
	end
	end

	def unfollow(follower, followed, activity_id \\ nil, local \\ true) 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),
	:ok <- maybe_federate(activity) do
	{:ok, activity}
	end
	end

	def delete(%User{ap_id: ap_id, follower_address: follower_address} = user) do
	with data <- %{
	"to" => [follower_address],
	"type" => "Delete",
	"actor" => ap_id,
	"object" => %{"type" => "Person", "id" => ap_id}
	},
	{:ok, activity} <- insert(data, true, true, true),
	:ok <- maybe_federate(activity) do
	{:ok, user}
	end
	end

	def delete(%Object{data: %{"id" => id, "actor" => actor}} = object, options \\ []) do
	local = Keyword.get(options, :local, true)
	activity_id = Keyword.get(options, :activity_id, nil)
	actor = Keyword.get(options, :actor, actor)

	user = User.get_cached_by_ap_id(actor)
	to = (object.data["to"] \|\| []) ++ (object.data["cc"] \|\| [])

	with create_activity <- Activity.get_create_by_object_ap_id(id),
	data <-
	%{
	"type" => "Delete",
	"actor" => actor,
	"object" => id,
	"to" => to,
	"deleted_activity_id" => create_activity && create_activity.id
	}
	\|> maybe_put("id", activity_id),
	{:ok, activity} <- insert(data, local, false),
	{:ok, object, _create_activity} <- Object.delete(object),
	stream_out_participations(object, user),
	_ <- decrease_replies_count_if_reply(object),
	{:ok, _actor} <- decrease_note_count_if_public(user, object),
	:ok <- maybe_federate(activity) do
	{:ok, activity}
	end
	end

	@spec block(User.t(), User.t(), String.t() \| nil, boolean) :: {:ok, Activity.t() \| nil}
	def block(blocker, blocked, activity_id \\ nil, local \\ true) do
	outgoing_blocks = Config.get([:activitypub, :outgoing_blocks])
	unfollow_blocked = Config.get([:activitypub, :unfollow_blocked])

	if unfollow_blocked do
	follow_activity = fetch_latest_follow(blocker, blocked)
	if follow_activity, do: unfollow(blocker, blocked, nil, local)
	end

	with true <- outgoing_blocks,
	block_data <- make_block_data(blocker, blocked, activity_id),
	{:ok, activity} <- insert(block_data, local),
	:ok <- maybe_federate(activity) do
	{:ok, activity}
	else
	_e -> {:ok, nil}
	end
	end

	def unblock(blocker, blocked, activity_id \\ nil, local \\ true) do
	with %Activity{} = block_activity <- fetch_latest_block(blocker, blocked),
	unblock_data <- make_unblock_data(blocker, blocked, block_activity, activity_id),
	{:ok, activity} <- insert(unblock_data, local),
	:ok <- maybe_federate(activity) do
	{:ok, activity}
	end
	end

	@spec flag(map()) :: {:ok, Activity.t()} \| any
	def 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),
	{:ok, stripped_activity} <- strip_report_status_data(activity),
	:ok <- maybe_federate(stripped_activity) do
	Enum.each(User.all_superusers(), fn superuser ->
	superuser
	\|> Pleroma.Emails.AdminEmail.report(actor, account, statuses, content)
	\|> Pleroma.Emails.Mailer.deliver_async()
	end)

	{:ok, activity}
	end
	end

	def move(%User{} = origin, %User{} = target, local \\ true) do
	params = %{
	"type" => "Move",
	"actor" => origin.ap_id,
	"object" => origin.ap_id,
	"target" => target.ap_id
	}

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

	BackgroundWorker.enqueue("move_following", %{
	"origin_id" => origin.id,
	"target_id" => target.id
	})

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

	defp fetch_activities_for_context_query(context, opts) do
	public = [Pleroma.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_blocked(opts)
	\|> restrict_recipients(recipients, opts["user"])
	\|> 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

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

	def fetch_public_activities(opts \\ %{}, pagination \\ :keyset) do
	opts = Map.drop(opts, ["user"])

	[Pleroma.Constants.as_public()]
	\|> fetch_activities_query(opts)
	\|> restrict_unlisted()
	\|> Pagination.fetch_paginated(opts, 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
	query =
	from(
	a in query,
	where:
	fragment(
	"activity_visibility(?, ?, ?) = ANY (?)",
	a.actor,
	a.recipients,
	a.data,
	^visibility
	)
	)

	query
	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 @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
	from(
	a in query,
	where: fragment("thread_visibility(?, (?)->>'id') = true", ^ap_id, a.data)
	)
	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)
	\|> Map.put("whole_db", true)

	recipients =
	user_activities_recipients(%{
	"godmode" => params["godmode"],
	"reading_user" => reading_user
	})

	fetch_activities(recipients, params)
	\|> Enum.reverse()
	end

	def fetch_user_activities(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("whole_db", true)
	\|> Map.put("pinned_activity_ids", user.pinned_activities)

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

	recipients =
	user_activities_recipients(%{
	"godmode" => params["godmode"],
	"reading_user" => reading_user
	})

	fetch_activities(recipients, params)
	\|> Enum.reverse()
	end

	def fetch_instance_activities(params) do
	params =
	params
	\|> Map.put("type", ["Create", "Announce"])
	\|> Map.put("instance", params["instance"])
	\|> Map.put("whole_db", true)

	fetch_activities([Pleroma.Constants.as_public()], params, :offset)
	\|> Enum.reverse()
	end

	defp user_activities_recipients(%{"godmode" => true}) do
	[]
	end

	defp user_activities_recipients(%{"reading_user" => reading_user}) do
	if reading_user do
	[Pleroma.Constants.as_public()] ++ [reading_user.ap_id \| User.following(reading_user)]
	else
	[Pleroma.Constants.as_public()]
	end
	end

	defp restrict_since(query, %{"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_tag_reject(_query, %{"tag_reject" => _tag_reject, "skip_preload" => true}) do
	raise "Can't use the child object without preloading!"
	end

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

	defp restrict_tag_reject(query, _), do: query

	defp restrict_tag_all(_query, %{"tag_all" => _tag_all, "skip_preload" => true}) do
	raise "Can't use the child object without preloading!"
	end

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

	defp restrict_tag_all(query, _), do: query

	defp restrict_tag(_query, %{"tag" => _tag, "skip_preload" => true}) do
	raise "Can't use the child object without preloading!"
	end

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

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

	defp restrict_tag(query, _), do: query

	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_local(query, %{"local_only" => true}) do
	from(activity in query, where: activity.local == true)
	end

	defp restrict_local(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_favorited_by(query, %{"favorited_by" => ap_id}) do
	from(
	[_activity, object] in query,
	where: fragment("(?)->'likes' \\? (?)", object.data, ^ap_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" => val}) when val == "true" or val == "1" do
	from(
	[_activity, object] in query,
	where: fragment("not (?)->'attachment' = (?)", object.data, ^[])
	)
	end

	+ defp restrict_media(query, %{"only_media" => :is_not_null}) do
	+ from(
	+ [_activity, object] in query,
	+ where: fragment("(?)->'attachment' IS NOT NULL", object.data)
	+ )
	+ end
	+
	defp restrict_media(query, _), do: query

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

	defp restrict_replies(query, _), do: query

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

	defp restrict_reblogs(query, _), do: query

	defp restrict_muted(query, %{"with_muted" => val}) when val in [true, "true", "1"], do: query

	defp restrict_muted(query, %{"muting_user" => %User{} = user} = opts) do
	mutes = opts["muted_users_ap_ids"] \|\| User.muted_users_ap_ids(user)

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

	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 restrict_blocked(query, %{"blocking_user" => %User{} = user} = opts) do
	blocked_ap_ids = opts["blocked_users_ap_ids"] \|\| User.blocked_users_ap_ids(user)
	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,
	where: fragment("not (? = ANY(?))", activity.actor, ^blocked_ap_ids),
	where: fragment("not (? && ?)", activity.recipients, ^blocked_ap_ids),
	where:
	fragment(
	"not (?->>'type' = 'Announce' and ?->'to' \\?\| ?)",
	activity.data,
	activity.data,
	^blocked_ap_ids
	),
	where:
	fragment(
	"(not (split_part(?, '/', 3) = ANY(?))) or ? = ANY(?)",
	activity.actor,
	^domain_blocks,
	activity.actor,
	^following_ap_ids
	),
	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 restrict_unlisted(query) do
	from(
	activity in query,
	where:
	fragment(
	"not (coalesce(?->'cc', '{}'::jsonb) \\?\| ?)",
	activity.data,
	^[Pleroma.Constants.as_public()]
	)
	)
	end

	defp restrict_pinned(query, %{"pinned" => "true", "pinned_activity_ids" => ids}) do
	from(activity in query, where: activity.id in ^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}) do
	users =
	from(
	u in User,
	select: u.ap_id,
	where: fragment("? LIKE ?", u.nickname, ^"%@#{instance}")
	)
	\|> Repo.all()

	from(activity in query, where: activity.actor in ^users)
	end

	defp restrict_instance(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_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 fetch_activities_query_ap_ids_ops(opts) do
	source_user = opts["muting_user"]
	ap_id_relations = if source_user, do: [:mute, :reblog_mute], else: []

	ap_id_relations =
	ap_id_relations ++
	if opts["blocking_user"] && opts["blocking_user"] == source_user do
	[:block]
	else
	[]
	end

	preloaded_ap_ids = User.outgoing_relations_ap_ids(source_user, ap_id_relations)

	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
	{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])
	}

	Activity
	\|> maybe_preload_objects(opts)
	\|> maybe_preload_bookmarks(opts)
	\|> maybe_preload_report_notes(opts)
	\|> maybe_set_thread_muted_field(opts)
	\|> maybe_order(opts)
	\|> restrict_recipients(recipients, opts["user"])
	\|> restrict_tag(opts)
	\|> restrict_tag_reject(opts)
	\|> restrict_tag_all(opts)
	\|> restrict_since(opts)
	\|> restrict_local(opts)
	\|> restrict_actor(opts)
	\|> restrict_type(opts)
	\|> restrict_state(opts)
	\|> restrict_favorited_by(opts)
	\|> restrict_blocked(restrict_blocked_opts)
	\|> restrict_muted(restrict_muted_opts)
	\|> restrict_media(opts)
	\|> restrict_visibility(opts)
	\|> restrict_thread_visibility(opts, config)
	\|> restrict_replies(opts)
	\|> restrict_reblogs(opts)
	\|> restrict_pinned(opts)
	\|> restrict_muted_reblogs(restrict_muted_reblogs_opts)
	\|> restrict_instance(opts)
	\|> Activity.restrict_deactivated_users()
	\|> exclude_poll_votes(opts)
	\|> exclude_visibility(opts)
	end

	def fetch_activities(recipients, opts \\ %{}, pagination \\ :keyset) do
	list_memberships = Pleroma.List.memberships(opts["user"])

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

	@doc """
	Fetch favorites activities of user with order by sort adds to favorites
	"""
	@spec fetch_favourites(User.t(), map(), atom()) :: 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})
	\|> order_by([like, _, _], desc: like.id)
	\|> Pagination.fetch_paginated(
	Map.merge(params, %{"skip_order" => true}),
	pagination,
	:object_activity
	)
	end

	defp maybe_update_cc(activities, list_memberships, %User{ap_id: user_ap_id})
	when is_list(list_memberships) and length(list_memberships) > 0 do
	Enum.map(activities, fn
	%{data: %{"bcc" => bcc}} = activity when is_list(bcc) and length(bcc) > 0 ->
	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

	def 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
	^Pleroma.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

	def upload(file, opts \\ []) do
	with {:ok, data} <- Upload.store(file, opts) do
	obj_data =
	if opts[:actor] do
	Map.put(data, "actor", opts[:actor])
	else
	data
	end

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

	defp object_to_user_data(data) do
	avatar =
	data["icon"]["url"] &&
	%{
	"type" => "Image",
	"url" => [%{"href" => data["icon"]["url"]}]
	}

	banner =
	data["image"]["url"] &&
	%{
	"type" => "Image",
	"url" => [%{"href" => data["image"]["url"]}]
	}

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

	locked = data["manuallyApprovesFollowers"] \|\| false
	data = Transmogrifier.maybe_fix_user_object(data)
	discoverable = data["discoverable"] \|\| false
	invisible = data["invisible"] \|\| false
	actor_type = data["type"] \|\| "Person"

	user_data = %{
	ap_id: data["id"],
	ap_enabled: true,
	source_data: data,
	banner: banner,
	fields: fields,
	locked: locked,
	discoverable: discoverable,
	invisible: invisible,
	avatar: avatar,
	name: data["name"],
	follower_address: data["followers"],
	following_address: data["following"],
	bio: data["summary"],
	actor_type: actor_type,
	also_known_as: Map.get(data, "alsoKnownAs", [])
	}

	# nickname can be nil because of virtual actors
	user_data =
	if data["preferredUsername"] do
	Map.put(
	user_data,
	:nickname,
	"#{data["preferredUsername"]}@#{URI.parse(data["id"]).host}"
	)
	else
	Map.put(user_data, :nickname, nil)
	end

	{:ok, user_data}
	end

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

	e ->
	{:error, e}
	end
	end

	defp maybe_update_follow_information(data) do
	with {:enabled, true} <-
	{:enabled, Pleroma.Config.get([:instance, :external_user_synchronization])},
	{:ok, info} <- fetch_follow_information_for_user(data) do
	info = Map.merge(data[:info] \|\| %{}, info)
	Map.put(data, :info, info)
	else
	{:enabled, false} ->
	data

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

	data
	end
	end

	defp collection_private(%{"first" => first}) do
	if is_map(first) and
	first["type"] in ["CollectionPage", "OrderedCollectionPage"] do
	{:ok, false}
	else
	with {:ok, %{"type" => type}} when type in ["CollectionPage", "OrderedCollectionPage"] <-
	Fetcher.fetch_and_contain_remote_object_from_id(first) do
	{:ok, false}
	else
	{:error, {:ok, %{status: code}}} when code in [401, 403] ->
	{:ok, true}

	{:error, _} = e ->
	e

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

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

	def user_data_from_user_object(data) do
	with {:ok, data} <- MRF.filter(data),
	{:ok, data} <- object_to_user_data(data) do
	{:ok, data}
	else
	e -> {:error, e}
	end
	end

	def fetch_and_prepare_user_from_ap_id(ap_id) do
	with {:ok, data} <- Fetcher.fetch_and_contain_remote_object_from_id(ap_id),
	{:ok, data} <- user_data_from_user_object(data),
	data <- maybe_update_follow_information(data) do
	{:ok, data}
	else
	e ->
	Logger.error("Could not decode user at fetch #{ap_id}, #{inspect(e)}")
	{:error, e}
	end
	end

	def make_user_from_ap_id(ap_id) do
	if _user = User.get_cached_by_ap_id(ap_id) do
	Transmogrifier.upgrade_user_from_ap_id(ap_id)
	else
	with {:ok, data} <- fetch_and_prepare_user_from_ap_id(ap_id) do
	User.insert_or_update_user(data)
	else
	e -> {:error, e}
	end
	end
	end

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

	# filter out broken threads
	def 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
	end

File Metadata

Mime Type: text/x-diff
Expires: Sun, Oct 11, 5:38 AM (1 d, 13 h)
Storage Engine: blob
Storage Format: Raw Data
Storage Handle: 1785621
Default Alt Text: (71 KB)

No OneTemporaryActions

View Options

File Metadata

Event Timeline

No OneTemporary
Actions