No OneTemporary
Actions

Size

151 KB

Referenced Files

None

Subscribers

None

View Options

	diff --git a/lib/pleroma/application.ex b/lib/pleroma/application.ex
	index 13be77fec..a00bc0624 100644
	--- a/lib/pleroma/application.ex
	+++ b/lib/pleroma/application.ex
	@@ -1,236 +1,243 @@
	# Pleroma: A lightweight social networking server
	# Copyright © 2017-2020 Pleroma Authors <https://pleroma.social/>
	# SPDX-License-Identifier: AGPL-3.0-only

	defmodule Pleroma.Application do
	use Application

	import Cachex.Spec

	alias Pleroma.Config

	require Logger

	@name Mix.Project.config()[:name]
	@version Mix.Project.config()[:version]
	@repository Mix.Project.config()[:source_url]
	@env Mix.env()

	def name, do: @name
	def version, do: @version
	def named_version, do: @name <> " " <> @version
	def repository, do: @repository

	def user_agent do
	case Config.get([:http, :user_agent], :default) do
	:default ->
	info = "#{Pleroma.Web.base_url()} <#{Config.get([:instance, :email], "")}>"
	named_version() <> "; " <> info

	custom ->
	custom
	end
	end

	# See http://elixir-lang.org/docs/stable/elixir/Application.html
	# for more information on OTP Applications
	def start(_type, _args) do
	Pleroma.Config.Holder.save_default()
	Pleroma.HTML.compile_scrubbers()
	Config.DeprecationWarnings.warn()
	Pleroma.Plugs.HTTPSecurityPlug.warn_if_disabled()
	Pleroma.Repo.check_migrations_applied!()
	setup_instrumenters()
	load_custom_modules()

	adapter = Application.get_env(:tesla, :adapter)

	if adapter == Tesla.Adapter.Gun do
	if version = Pleroma.OTPVersion.version() do
	[major, minor] =
	version
	\|> String.split(".")
	\|> Enum.map(&String.to_integer/1)
	\|> Enum.take(2)

	if (major == 22 and minor < 2) or major < 22 do
	raise "
	!!!OTP VERSION WARNING!!!
	You are using gun adapter with OTP version #{version}, which doesn't support correct handling of unordered certificates chains.
	"
	end
	else
	raise "
	!!!OTP VERSION WARNING!!!
	To support correct handling of unordered certificates chains - OTP version must be > 22.2.
	"
	end
	end

	# Define workers and child supervisors to be supervised
	children =
	[
	Pleroma.Repo,
	Config.TransferTask,
	Pleroma.Emoji,
	Pleroma.Plugs.RateLimiter.Supervisor
	] ++
	cachex_children() ++
	http_children(adapter, @env) ++
	[
	Pleroma.Stats,
	Pleroma.JobQueueMonitor,
	{Oban, Config.get(Oban)}
	] ++
	task_children(@env) ++
	streamer_child(@env) ++
	chat_child(@env, chat_enabled?()) ++
	[
	Pleroma.Web.Endpoint,
	Pleroma.Gopher.Server
	]

	# See http://elixir-lang.org/docs/stable/elixir/Supervisor.html
	# for other strategies and supported options
	opts = [strategy: :one_for_one, name: Pleroma.Supervisor]
	Supervisor.start_link(children, opts)
	end

	def load_custom_modules do
	dir = Config.get([:modules, :runtime_dir])

	if dir && File.exists?(dir) do
	dir
	\|> Pleroma.Utils.compile_dir()
	\|> case do
	{:error, _errors, _warnings} ->
	raise "Invalid custom modules"

	{:ok, modules, _warnings} ->
	if @env != :test do
	Enum.each(modules, fn mod ->
	Logger.info("Custom module loaded: #{inspect(mod)}")
	end)
	end

	:ok
	end
	end
	end

	defp setup_instrumenters do
	require Prometheus.Registry

	if Application.get_env(:prometheus, Pleroma.Repo.Instrumenter) do
	:ok =
	:telemetry.attach(
	"prometheus-ecto",
	[:pleroma, :repo, :query],
	&Pleroma.Repo.Instrumenter.handle_event/4,
	%{}
	)

	Pleroma.Repo.Instrumenter.setup()
	end

	Pleroma.Web.Endpoint.MetricsExporter.setup()
	Pleroma.Web.Endpoint.PipelineInstrumenter.setup()
	Pleroma.Web.Endpoint.Instrumenter.setup()
	end

	defp cachex_children do
	[
	build_cachex("used_captcha", ttl_interval: seconds_valid_interval()),
	build_cachex("user", default_ttl: 25_000, ttl_interval: 1000, limit: 2500),
	build_cachex("object", default_ttl: 25_000, ttl_interval: 1000, limit: 2500),
	build_cachex("rich_media", default_ttl: :timer.minutes(120), limit: 5000),
	build_cachex("scrubber", limit: 2500),
	build_cachex("idempotency", expiration: idempotency_expiration(), limit: 2500),
	build_cachex("web_resp", limit: 2500),
	build_cachex("emoji_packs", expiration: emoji_packs_expiration(), limit: 10),
	build_cachex("failed_proxy_url", limit: 2500)
	]
	end

	defp emoji_packs_expiration,
	do: expiration(default: :timer.seconds(5 * 60), interval: :timer.seconds(60))

	defp idempotency_expiration,
	do: expiration(default: :timer.seconds(6 * 60 * 60), interval: :timer.seconds(60))

	defp seconds_valid_interval,
	do: :timer.seconds(Config.get!([Pleroma.Captcha, :seconds_valid]))

	defp build_cachex(type, opts),
	do: %{
	id: String.to_atom("cachex_" <> type),
	start: {Cachex, :start_link, [String.to_atom(type <> "_cache"), opts]},
	type: :worker
	}

	defp chat_enabled?, do: Config.get([:chat, :enabled])

	defp streamer_child(env) when env in [:test, :benchmark], do: []

	defp streamer_child(_) do
	- [{Registry, [name: Pleroma.Web.Streamer.registry(), partitions: System.schedulers_online()]}]
	+ [
	+ {Registry,
	+ [
	+ name: Pleroma.Web.Streamer.registry(),
	+ keys: :duplicate,
	+ partitions: System.schedulers_online()
	+ ]}
	+ ]
	end

	defp chat_child(_env, true) do
	[Pleroma.Web.ChatChannel.ChatChannelState]
	end

	defp chat_child(_, _), do: []

	defp task_children(:test) do
	[
	%{
	id: :web_push_init,
	start: {Task, :start_link, [&Pleroma.Web.Push.init/0]},
	restart: :temporary
	}
	]
	end

	defp task_children(_) do
	[
	%{
	id: :web_push_init,
	start: {Task, :start_link, [&Pleroma.Web.Push.init/0]},
	restart: :temporary
	},
	%{
	id: :internal_fetch_init,
	start: {Task, :start_link, [&Pleroma.Web.ActivityPub.InternalFetchActor.init/0]},
	restart: :temporary
	}
	]
	end

	# start hackney and gun pools in tests
	defp http_children(_, :test) do
	hackney_options = Config.get([:hackney_pools, :federation])
	hackney_pool = :hackney_pool.child_spec(:federation, hackney_options)
	[hackney_pool, Pleroma.Pool.Supervisor]
	end

	defp http_children(Tesla.Adapter.Hackney, _) do
	pools = [:federation, :media]

	pools =
	if Config.get([Pleroma.Upload, :proxy_remote]) do
	[:upload \| pools]
	else
	pools
	end

	for pool <- pools do
	options = Config.get([:hackney_pools, pool])
	:hackney_pool.child_spec(pool, options)
	end
	end

	defp http_children(Tesla.Adapter.Gun, _), do: [Pleroma.Pool.Supervisor]

	defp http_children(_, _), do: []
	end
	diff --git a/lib/pleroma/web/activity_pub/activity_pub.ex b/lib/pleroma/web/activity_pub/activity_pub.ex
	index 099df5879..8baaf97ac 100644
	--- a/lib/pleroma/web/activity_pub/activity_pub.ex
	+++ b/lib/pleroma/web/activity_pub/activity_pub.ex
	@@ -1,1592 +1,1610 @@
	# Pleroma: A lightweight social networking server
	# Copyright © 2017-2020 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.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",
	"actor" => actor
	}) do
	Object.increase_vote_count(reply_ap_id, name, actor)
	end

	def increase_poll_votes_if_vote(_create_data), do: :noop

	@spec persist(map(), keyword()) :: {:ok, Activity.t() \| Object.t()}
	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"]
	}) 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 <- 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

	+ def notify_and_stream(activity) do
	+ Notification.create_notifications(activity)
	+
	+ conversation = create_or_bump_conversation(activity, activity.actor)
	+ participations = get_participations(conversation)
	+ stream_out(activity)
	+ stream_out_participations(participations)
	+ 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

	@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

	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),
	+ _ <- notify_and_stream(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

	@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]

	with listen_data <-
	make_listen_data(
	%{to: to, actor: actor, published: published, context: context, object: object},
	additional
	),
	{:ok, activity} <- insert(listen_data, local),
	+ _ <- notify_and_stream(activity),
	:ok <- maybe_federate(activity) do
	{:ok, activity}
	end
	end

	@spec accept(map()) :: {:ok, Activity.t()} \| {:error, any()}
	def accept(params) do
	accept_or_reject("Accept", params)
	end

	@spec reject(map()) :: {:ok, Activity.t()} \| {:error, any()}
	def reject(params) do
	accept_or_reject("Reject", params)
	end

	@spec accept_or_reject(String.t(), map()) :: {:ok, Activity.t()} \| {:error, any()}
	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),
	+ _ <- notify_and_stream(activity),
	:ok <- maybe_federate(activity) do
	{:ok, activity}
	end
	end

	@spec update(map()) :: {:ok, Activity.t()} \| {:error, any()}
	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),
	+ _ <- notify_and_stream(activity),
	:ok <- maybe_federate(activity) do
	{:ok, activity}
	end
	end

	@spec react_with_emoji(User.t(), Object.t(), String.t(), keyword()) ::
	{:ok, Activity.t(), Object.t()} \| {:error, any()}
	def react_with_emoji(user, object, emoji, options \\ []) do
	with {:ok, result} <-
	Repo.transaction(fn -> do_react_with_emoji(user, object, emoji, options) end) do
	result
	end
	end

	defp do_react_with_emoji(user, object, emoji, options) do
	with local <- Keyword.get(options, :local, true),
	activity_id <- Keyword.get(options, :activity_id, nil),
	true <- 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),
	+ _ <- notify_and_stream(activity),
	:ok <- maybe_federate(activity) do
	{:ok, activity, object}
	else
	false -> {:error, false}
	{:error, error} -> Repo.rollback(error)
	end
	end

	@spec unreact_with_emoji(User.t(), String.t(), keyword()) ::
	{:ok, Activity.t(), Object.t()} \| {:error, any()}
	def unreact_with_emoji(user, reaction_id, options \\ []) do
	with {:ok, result} <-
	Repo.transaction(fn -> do_unreact_with_emoji(user, reaction_id, options) end) do
	result
	end
	end

	defp do_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),
	+ _ <- notify_and_stream(activity),
	:ok <- maybe_federate(activity) do
	{:ok, activity, object}
	else
	{:error, error} -> Repo.rollback(error)
	end
	end

	@spec unlike(User.t(), Object.t(), String.t() \| nil, boolean()) ::
	{:ok, Activity.t(), Activity.t(), Object.t()} \| {:ok, Object.t()} \| {:error, any()}
	def unlike(%User{} = actor, %Object{} = object, activity_id \\ nil, local \\ true) do
	with {:ok, result} <-
	Repo.transaction(fn -> do_unlike(actor, object, activity_id, local) end) do
	result
	end
	end

	defp do_unlike(actor, object, activity_id, local) 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),
	+ _ <- notify_and_stream(unlike_activity),
	:ok <- maybe_federate(unlike_activity) do
	{:ok, unlike_activity, like_activity, object}
	else
	nil -> {:ok, object}
	{:error, error} -> Repo.rollback(error)
	end
	end

	@spec announce(User.t(), Object.t(), String.t() \| nil, boolean(), boolean()) ::
	{:ok, Activity.t(), Object.t()} \| {:error, any()}
	def announce(
	%User{ap_id: _} = user,
	%Object{data: %{"id" => _}} = object,
	activity_id \\ nil,
	local \\ true,
	public \\ true
	) do
	with {:ok, result} <-
	Repo.transaction(fn -> do_announce(user, object, activity_id, local, public) end) do
	result
	end
	end

	defp do_announce(user, object, activity_id, local, public) do
	with true <- is_announceable?(object, user, public),
	object <- Object.get_by_id(object.id),
	announce_data <- make_announce_data(user, object, activity_id, public),
	{:ok, activity} <- insert(announce_data, local),
	{:ok, object} <- add_announce_to_object(activity, object),
	+ _ <- notify_and_stream(activity),
	:ok <- maybe_federate(activity) do
	{:ok, activity, object}
	else
	false -> {:error, false}
	{:error, error} -> Repo.rollback(error)
	end
	end

	@spec unannounce(User.t(), Object.t(), String.t() \| nil, boolean()) ::
	{:ok, Activity.t(), Object.t()} \| {:ok, Object.t()} \| {:error, any()}
	def unannounce(
	%User{} = actor,
	%Object{} = object,
	activity_id \\ nil,
	local \\ true
	) do
	with {:ok, result} <-
	Repo.transaction(fn -> do_unannounce(actor, object, activity_id, local) end) do
	result
	end
	end

	defp do_unannounce(actor, object, activity_id, local) 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),
	+ _ <- notify_and_stream(unannounce_activity),
	: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
	nil -> {:ok, object}
	{:error, error} -> Repo.rollback(error)
	end
	end

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

	defp do_follow(follower, followed, activity_id, local) do
	with data <- make_follow_data(follower, followed, activity_id),
	{:ok, activity} <- insert(data, local),
	+ _ <- notify_and_stream(activity),
	:ok <- maybe_federate(activity) do
	{:ok, activity}
	else
	{:error, error} -> Repo.rollback(error)
	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 block(User.t(), User.t(), String.t() \| nil, boolean()) ::
	{:ok, Activity.t()} \| {:error, any()}
	def block(blocker, blocked, activity_id \\ nil, local \\ true) do
	with {:ok, result} <-
	Repo.transaction(fn -> do_block(blocker, blocked, activity_id, local) end) do
	result
	end
	end

	defp do_block(blocker, blocked, activity_id, local) 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),
	+ _ <- notify_and_stream(activity),
	:ok <- maybe_federate(activity) do
	{:ok, activity}
	else
	{:error, error} -> Repo.rollback(error)
	end
	end

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

	defp do_unblock(blocker, blocked, activity_id, local) 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),
	+ _ <- 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(
	%{
	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),
	+ _ <- notify_and_stream(activity),
	:ok <- maybe_federate(stripped_activity) do
	User.all_superusers()
	\|> Enum.filter(fn user -> not is_nil(user.email) end)
	\|> Enum.each(fn superuser ->
	superuser
	\|> Pleroma.Emails.AdminEmail.report(actor, account, statuses, content)
	\|> Pleroma.Emails.Mailer.deliver_async()
	end)

	{:ok, activity}
	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
	}

	with true <- origin.ap_id in target.also_known_as,
	- {:ok, activity} <- insert(params, local) do
	+ {:ok, activity} <- insert(params, local),
	+ _ <- notify_and_stream(activity) 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

	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_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

	@spec fetch_public_activities(map(), Pagination.type()) :: [Activity.t()]
	def fetch_public_activities(opts \\ %{}, pagination \\ :keyset) do
	opts = Map.drop(opts, ["user"])

	[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 [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
	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)

	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("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_statuses(reading_user, params) do
	params =
	params
	\|> Map.put("type", ["Create", "Announce"])

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

	fetch_activities(recipients, 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
	[Constants.as_public()] ++ [reading_user.ap_id \| User.following(reading_user)]
	else
	[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 in [true, "true", "1"] do
	from(
	[_activity, object] in query,
	where: fragment("not (?)->'attachment' = (?)", object.data, ^[])
	)
	end

	defp restrict_media(query, _), do: query

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

	defp restrict_replies(query, %{
	"reply_filtering_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,
	"reply_visibility" => "following"
	}) do
	from(
	[activity, object] in query,
	where:
	fragment(
	"?->>'inReplyTo' is null OR ? && array_remove(?, ?) OR ? = ?",
	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" => val}) when val in [true, "true", "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,
	^[Constants.as_public()]
	)
	)
	end

	# TODO: when all endpoints migrated to OpenAPI compare `pinned` with `true` (boolean) only,
	# the same for `restrict_media/2`, `restrict_replies/2`, 'restrict_reblogs/2'
	# and `restrict_muted/2`

	defp restrict_pinned(query, %{"pinned" => pinned, "pinned_activity_ids" => ids})
	when pinned in [true, "true", "1"] 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_relationships = if source_user, do: [:mute, :reblog_mute], else: []

	ap_id_relationships =
	ap_id_relationships ++
	if opts["blocking_user"] && opts["blocking_user"] == source_user do
	[:block]
	else
	[]
	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
	{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_replies(opts)
	\|> 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_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(), 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})
	\|> 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
	^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(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

	@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 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)

	emojis =
	data
	\|> Map.get("tag", [])
	\|> Enum.filter(fn
	%{"type" => "Emoji"} -> true
	_ -> false
	end)
	\|> Enum.reduce(%{}, fn %{"icon" => %{"url" => url}, "name" => name}, acc ->
	Map.put(acc, String.trim(name, ":"), url)
	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"

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

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

	user_data = %{
	ap_id: data["id"],
	uri: get_actor_url(data["url"]),
	ap_enabled: true,
	banner: banner,
	fields: fields,
	emoji: emojis,
	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", []),
	public_key: public_key,
	inbox: data["inbox"],
	shared_inbox: shared_inbox
	}

	# 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),
	{: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
	e -> {:error, 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, %{status: code}}} when code in [401, 403] -> {:ok, true}
	{:error, _} = e -> e
	e -> {:error, e}
	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
	{:error, "Object has been deleted"} = e ->
	Logger.debug("Could not decode user at fetch #{ap_id}, #{inspect(e)}")
	{:error, e}

	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
	user = User.get_cached_by_ap_id(ap_id)

	if user && !User.ap_enabled?(user) do
	Transmogrifier.upgrade_user_from_ap_id(ap_id)
	else
	with {:ok, data} <- fetch_and_prepare_user_from_ap_id(ap_id) do
	if user do
	user
	\|> User.remote_user_changeset(data)
	\|> User.update_and_set_cache()
	else
	data
	\|> User.remote_user_changeset()
	\|> Repo.insert()
	\|> User.set_cache()
	end
	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
	diff --git a/lib/pleroma/web/mastodon_api/websocket_handler.ex b/lib/pleroma/web/mastodon_api/websocket_handler.ex
	index 6f6aa8164..6ef3fe2dd 100644
	--- a/lib/pleroma/web/mastodon_api/websocket_handler.ex
	+++ b/lib/pleroma/web/mastodon_api/websocket_handler.ex
	@@ -1,175 +1,152 @@
	# Pleroma: A lightweight social networking server
	# Copyright © 2017-2020 Pleroma Authors <https://pleroma.social/>
	# SPDX-License-Identifier: AGPL-3.0-only

	defmodule Pleroma.Web.MastodonAPI.WebsocketHandler do
	require Logger

	alias Pleroma.Repo
	alias Pleroma.User
	alias Pleroma.Web.OAuth.Token
	alias Pleroma.Web.Streamer

	@behaviour :cowboy_websocket

	# Cowboy timeout period.
	- @timeout :timer.seconds(60)
	- # Keepalive timer.
	- @keepalive_interval :timer.seconds(30)
	+ @timeout :timer.seconds(30)
	# Hibernate every X messages
	@hibernate_every 100

	@streams [
	"public",
	"public:local",
	"public:media",
	"public:local:media",
	"user",
	"user:notification",
	"direct",
	"list",
	"hashtag"
	]
	@anonymous_streams ["public", "public:local", "hashtag"]

	def init(%{qs: qs} = req, state) do
	with params <- :cow_qs.parse_qs(qs),
	sec_websocket <- :cowboy_req.header("sec-websocket-protocol", req, nil),
	access_token <- List.keyfind(params, "access_token", 0),
	{_, stream} <- List.keyfind(params, "stream", 0),
	{:ok, user} <- allow_request(stream, [access_token, sec_websocket]),
	topic when is_binary(topic) <- expand_topic(stream, params) do
	req =
	if sec_websocket do
	:cowboy_req.set_resp_header("sec-websocket-protocol", sec_websocket, req)
	else
	req
	end

	- {:ok, timer} = keepalive_timer()
	-
	- {:cowboy_websocket, req, %{user: user, topic: topic, count: 0, timer: timer},
	- %{idle_timeout: @timeout}}
	+ {:cowboy_websocket, req, %{user: user, topic: topic, count: 0}, %{idle_timeout: @timeout}}
	else
	{:error, code} ->
	Logger.debug("#{__MODULE__} denied connection: #{inspect(code)} - #{inspect(req)}")
	{:ok, req} = :cowboy_req.reply(code, req)
	{:ok, req, state}

	error ->
	Logger.debug("#{__MODULE__} denied connection: #{inspect(error)} - #{inspect(req)}")
	{:ok, req} = :cowboy_req.reply(400, req)
	{:ok, req, state}
	end
	end

	def websocket_init(state) do
	- send(self(), :subscribe)
	- {:ok, state}
	- end
	-
	- # We never receive messages.
	- def websocket_handle(_frame, state) do
	- {:ok, state}
	- end
	-
	- def websocket_info(:subscribe, state) do
	Logger.debug(
	"#{__MODULE__} accepted websocket connection for user #{
	(state.user \|\| %{id: "anonymous"}).id
	}, topic #{state.topic}"
	)

	- Streamer.add_socket(state.topic, state.assigns.user)
	+ Streamer.add_socket(state.topic, state.user)
	{:ok, state}
	end

	- def websocket_info(:keepalive, state) do
	- {:ok, timer} = keepalive_timer()
	- {:reply, {:text, ""}, %{state \| timer: timer, count: 0}, :hibernate}
	+ # We never receive messages.
	+ def websocket_handle(_frame, state) do
	+ {:ok, state}
	end

	def websocket_info({:render_with_user, view, template, item}, state) do
	user = %User{} = User.get_cached_by_ap_id(state.user.ap_id)

	- unless Pleroma.Streamer.filtered_by_user?(user, item) do
	+ unless Streamer.filtered_by_user?(user, item) do
	websocket_info({:text, view.render(template, user, item)}, %{state \| user: user})
	else
	{:ok, state}
	end
	end

	def websocket_info({:text, message}, state) do
	- # Cancel the current keep-alive timer, and re-queue one.
	- Process.cancel_timer(state.timer, async: true)
	- {:ok, timer} = keepalive_timer()
	# If the websocket processed X messages, force an hibernate/GC.
	# We don't hibernate at every message to balance CPU usage/latency with RAM usage.
	if state.count > @hibernate_every do
	- {:reply, {:text, message}, %{state \| count: 0, timer: timer}, :hibernate}
	+ {:reply, {:text, message}, %{state \| count: 0}, :hibernate}
	else
	- {:reply, {:text, message}, %{state \| count: state.count + 1, timer: timer}}
	+ {:reply, {:text, message}, %{state \| count: state.count + 1}}
	end
	end

	def terminate(reason, _req, state) do
	Logger.debug(
	"#{__MODULE__} terminating websocket connection for user #{
	(state.user \|\| %{id: "anonymous"}).id
	}, topic #{state.topic \|\| "?"}: #{inspect(reason)}"
	)

	- if state.timer, do: Process.cancel_timer(state.timer)
	Streamer.remove_socket(state.topic)
	:ok
	end

	# Public streams without authentication.
	defp allow_request(stream, [nil, nil]) when stream in @anonymous_streams do
	{:ok, nil}
	end

	# Authenticated streams.
	defp allow_request(stream, [access_token, sec_websocket]) when stream in @streams do
	token =
	with {"access_token", token} <- access_token do
	token
	else
	_ -> sec_websocket
	end

	with true <- is_bitstring(token),
	%Token{user_id: user_id} <- Repo.get_by(Token, token: token),
	user = %User{} <- User.get_cached_by_id(user_id) do
	{:ok, user}
	else
	_ -> {:error, 403}
	end
	end

	# Not authenticated.
	defp allow_request(stream, _) when stream in @streams, do: {:error, 403}

	# No matching stream.
	defp allow_request(_, _), do: {:error, 404}

	defp expand_topic("hashtag", params) do
	case List.keyfind(params, "tag", 0) do
	{_, tag} -> "hashtag:#{tag}"
	_ -> nil
	end
	end

	defp expand_topic("list", params) do
	case List.keyfind(params, "list", 0) do
	{_, list} -> "list:#{list}"
	_ -> nil
	end
	end

	defp expand_topic(topic, _), do: topic
	-
	- defp keepalive_timer() do
	- Process.send_after(self(), :keepalive, @keepalive_interval)
	- end
	end
	diff --git a/lib/pleroma/web/streamer/streamer.ex b/lib/pleroma/web/streamer/streamer.ex
	index 4a06068db..5ad4aa936 100644
	--- a/lib/pleroma/web/streamer/streamer.ex
	+++ b/lib/pleroma/web/streamer/streamer.ex
	@@ -1,242 +1,243 @@
	# Pleroma: A lightweight social networking server
	# Copyright © 2017-2020 Pleroma Authors <https://pleroma.social/>
	# SPDX-License-Identifier: AGPL-3.0-only

	defmodule Pleroma.Web.Streamer do
	require Logger

	alias Pleroma.Activity
	alias Pleroma.Config
	alias Pleroma.Conversation.Participation
	alias Pleroma.Notification
	alias Pleroma.Object
	alias Pleroma.User
	alias Pleroma.Web.ActivityPub.ActivityPub
	alias Pleroma.Web.ActivityPub.Visibility
	alias Pleroma.Web.CommonAPI
	- alias Pleroma.Web.Streamer.State
	- alias Pleroma.Web.Streamer.StreamerSocket
	alias Pleroma.Web.StreamerView

	- @timeout 60_000
	@mix_env Mix.env()
	@registry Pleroma.Web.StreamerRegistry

	- def registry(), do: @registry
	+ def registry, do: @registry

	- def add_socket(topic, user = %User{}) do
	- Registry.register(@registry, user_topic(topic, user), true)
	+ def add_socket(topic, %User{} = user) do
	+ if should_env_send?(), do: Registry.register(@registry, user_topic(topic, user), true)
	end

	def add_socket(topic, _) do
	- Registry.register(@registry, topic, false)
	+ if should_env_send?(), do: Registry.register(@registry, topic, false)
	end

	def remove_socket(topic) do
	- Registry.unregister(topic)
	+ if should_env_send?(), do: Registry.unregister(@registry, topic)
	end

	def stream(topics, item) when is_list(topics) do
	if should_env_send?() do
	Enum.each(topics, fn t ->
	spawn(fn -> do_stream(t, item) end)
	end)
	end

	:ok
	end

	def stream(topic, items) when is_list(items) do
	if should_env_send?() do
	Enum.each(items, fn i ->
	spawn(fn -> do_stream(topic, i) end)
	end)

	:ok
	end
	end

	def stream(topic, item) do
	if should_env_send?() do
	spawn(fn -> do_stream(topic, item) end)
	end

	:ok
	end

	def filtered_by_user?(%User{} = user, %Activity{} = item) do
	%{block: blocked_ap_ids, mute: muted_ap_ids, reblog_mute: reblog_muted_ap_ids} =
	User.outgoing_relationships_ap_ids(user, [:block, :mute, :reblog_mute])

	recipient_blocks = MapSet.new(blocked_ap_ids ++ muted_ap_ids)
	recipients = MapSet.new(item.recipients)
	domain_blocks = Pleroma.Web.ActivityPub.MRF.subdomains_regex(user.domain_blocks)

	with parent <- Object.normalize(item) \|\| item,
	true <-
	Enum.all?([blocked_ap_ids, muted_ap_ids], &(item.actor not in &1)),
	true <- item.data["type"] != "Announce" \|\| item.actor not in reblog_muted_ap_ids,
	true <- Enum.all?([blocked_ap_ids, muted_ap_ids], &(parent.data["actor"] not in &1)),
	true <- MapSet.disjoint?(recipients, recipient_blocks),
	%{host: item_host} <- URI.parse(item.actor),
	%{host: parent_host} <- URI.parse(parent.data["actor"]),
	false <- Pleroma.Web.ActivityPub.MRF.subdomain_match?(domain_blocks, item_host),
	false <- Pleroma.Web.ActivityPub.MRF.subdomain_match?(domain_blocks, parent_host),
	true <- thread_containment(item, user),
	false <- CommonAPI.thread_muted?(user, item) do
	false
	else
	_ -> true
	end
	end

	def filtered_by_user?(%User{} = user, %Notification{activity: activity}) do
	filtered_by_user?(user, activity)
	end

	defp do_stream("direct", item) do
	recipient_topics =
	User.get_recipients_from_activity(item)
	\|> Enum.map(fn %{id: id} -> "direct:#{id}" end)

	Enum.each(recipient_topics, fn user_topic ->
	Logger.debug("Trying to push direct message to #{user_topic}\n\n")
	push_to_socket(user_topic, item)
	end)
	end

	defp do_stream("participation", participation) do
	user_topic = "direct:#{participation.user_id}"
	Logger.debug("Trying to push a conversation participation to #{user_topic}\n\n")

	push_to_socket(user_topic, participation)
	end

	defp do_stream("list", item) do
	# filter the recipient list if the activity is not public, see #270.
	recipient_lists =
	case Visibility.is_public?(item) do
	true ->
	Pleroma.List.get_lists_from_activity(item)

	_ ->
	Pleroma.List.get_lists_from_activity(item)
	\|> Enum.filter(fn list ->
	owner = User.get_cached_by_id(list.user_id)

	Visibility.visible_for_user?(item, owner)
	end)
	end

	recipient_topics =
	recipient_lists
	\|> Enum.map(fn %{id: id} -> "list:#{id}" end)

	Enum.each(recipient_topics, fn list_topic ->
	Logger.debug("Trying to push message to #{list_topic}\n\n")
	push_to_socket(list_topic, item)
	end)
	end

	defp do_stream(topic, %Notification{} = item)
	when topic in ["user", "user:notification"] do
	Registry.dispatch(@registry, "#{topic}:#{item.user_id}", fn list ->
	Enum.each(list, fn {pid, _auth} ->
	send(pid, {:render_with_user, StreamerView, "notification.json", item})
	end)
	end)
	end

	defp do_stream("user", item) do
	Logger.debug("Trying to push to users")

	recipient_topics =
	User.get_recipients_from_activity(item)
	\|> Enum.map(fn %{id: id} -> "user:#{id}" end)

	Enum.each(recipient_topics, fn topic ->
	push_to_socket(topic, item)
	end)
	end

	defp do_stream(topic, item) do
	Logger.debug("Trying to push to #{topic}")
	Logger.debug("Pushing item to #{topic}")
	push_to_socket(topic, item)
	end

	defp push_to_socket(topic, %Participation{} = participation) do
	rendered = StreamerView.render("conversation.json", participation)

	Registry.dispatch(@registry, topic, fn list ->
	Enum.each(list, fn {pid, _} ->
	send(pid, {:text, rendered})
	end)
	end)
	end

	defp push_to_socket(topic, %Activity{
	data: %{"type" => "Delete", "deleted_activity_id" => deleted_activity_id}
	}) do
	rendered = Jason.encode!(%{event: "delete", payload: to_string(deleted_activity_id)})

	Registry.dispatch(@registry, topic, fn list ->
	Enum.each(list, fn {pid, _} ->
	send(pid, {:text, rendered})
	end)
	end)
	end

	defp push_to_socket(_topic, %Activity{data: %{"type" => "Delete"}}), do: :noop

	defp push_to_socket(topic, item) do
	anon_render = StreamerView.render("update.json", item)

	Registry.dispatch(@registry, topic, fn list ->
	Enum.each(list, fn {pid, auth?} ->
	if auth? do
	send(pid, {:render_with_user, StreamerView, "update.json", item})
	else
	send(pid, {:text, anon_render})
	end
	end)
	end)
	end

	defp thread_containment(_activity, %User{skip_thread_containment: true}), do: true

	defp thread_containment(activity, user) do
	if Config.get([:instance, :skip_thread_containment]) do
	true
	else
	ActivityPub.contain_activity(activity, user)
	end
	end

	- defp should_env_send? do
	- handle_should_send(@mix_env)
	- end
	-
	- defp handle_should_send(:test) do
	- case Process.whereis(@registry) do
	- nil ->
	- false
	+ # In test environement, only return true if the registry is started.
	+ # In benchmark environment, returns false.
	+ # In any other environment, always returns true.
	+ cond do
	+ @mix_env == :test ->
	+ def should_env_send? do
	+ case Process.whereis(@registry) do
	+ nil ->
	+ false

	- pid ->
	- Process.alive?(pid)
	- end
	- end
	+ pid ->
	+ Process.alive?(pid)
	+ end
	+ end

	- defp handle_should_send(:benchmark), do: false
	+ @mix_env == :benchmark ->
	+ def should_env_send?, do: false

	- defp handle_should_send(_), do: true
	+ true ->
	+ def should_env_send?, do: true
	+ end

	defp user_topic(topic, user)
	when topic in ~w[user user:notification direct] do
	"#{topic}:#{user.id}"
	end

	defp user_topic(topic, _) do
	topic
	end
	end
	diff --git a/test/integration/mastodon_websocket_test.exs b/test/integration/mastodon_websocket_test.exs
	index bd229c55f..109c7b4cb 100644
	--- a/test/integration/mastodon_websocket_test.exs
	+++ b/test/integration/mastodon_websocket_test.exs
	@@ -1,131 +1,128 @@
	# Pleroma: A lightweight social networking server
	# Copyright © 2017-2020 Pleroma Authors <https://pleroma.social/>
	# SPDX-License-Identifier: AGPL-3.0-only

	defmodule Pleroma.Integration.MastodonWebsocketTest do
	use Pleroma.DataCase

	import ExUnit.CaptureLog
	import Pleroma.Factory

	alias Pleroma.Integration.WebsocketClient
	alias Pleroma.Web.CommonAPI
	alias Pleroma.Web.OAuth

	+ @moduletag needs_streamer: true, capture_log: true
	+
	@path Pleroma.Web.Endpoint.url()
	\|> URI.parse()
	\|> Map.put(:scheme, "ws")
	\|> Map.put(:path, "/api/v1/streaming")
	\|> URI.to_string()

	- setup_all do
	- start_supervised(Pleroma.Web.Streamer.supervisor())
	- :ok
	- end
	-
	def start_socket(qs \\ nil, headers \\ []) do
	path =
	case qs do
	nil -> @path
	qs -> @path <> qs
	end

	WebsocketClient.start_link(self(), path, headers)
	end

	test "refuses invalid requests" do
	capture_log(fn ->
	assert {:error, {400, _}} = start_socket()
	assert {:error, {404, _}} = start_socket("?stream=ncjdk")
	Process.sleep(30)
	end)
	end

	test "requires authentication and a valid token for protected streams" do
	capture_log(fn ->
	assert {:error, {403, _}} = start_socket("?stream=user&access_token=aaaaaaaaaaaa")
	assert {:error, {403, _}} = start_socket("?stream=user")
	Process.sleep(30)
	end)
	end

	test "allows public streams without authentication" do
	assert {:ok, _} = start_socket("?stream=public")
	assert {:ok, _} = start_socket("?stream=public:local")
	assert {:ok, _} = start_socket("?stream=hashtag&tag=lain")
	end

	test "receives well formatted events" do
	user = insert(:user)
	{:ok, _} = start_socket("?stream=public")
	{:ok, activity} = CommonAPI.post(user, %{"status" => "nice echo chamber"})

	assert_receive {:text, raw_json}, 1_000
	assert {:ok, json} = Jason.decode(raw_json)

	assert "update" == json["event"]
	assert json["payload"]
	assert {:ok, json} = Jason.decode(json["payload"])

	view_json =
	Pleroma.Web.MastodonAPI.StatusView.render("show.json", activity: activity, for: nil)
	\|> Jason.encode!()
	\|> Jason.decode!()

	assert json == view_json
	end

	describe "with a valid user token" do
	setup do
	{:ok, app} =
	Pleroma.Repo.insert(
	OAuth.App.register_changeset(%OAuth.App{}, %{
	client_name: "client",
	scopes: ["scope"],
	redirect_uris: "url"
	})
	)

	user = insert(:user)

	{:ok, auth} = OAuth.Authorization.create_authorization(app, user)

	{:ok, token} = OAuth.Token.exchange_token(app, auth)

	%{user: user, token: token}
	end

	test "accepts valid tokens", state do
	assert {:ok, _} = start_socket("?stream=user&access_token=#{state.token.token}")
	end

	test "accepts the 'user' stream", %{token: token} = _state do
	assert {:ok, _} = start_socket("?stream=user&access_token=#{token.token}")

	assert capture_log(fn ->
	assert {:error, {403, "Forbidden"}} = start_socket("?stream=user")
	Process.sleep(30)
	end) =~ ":badarg"
	end

	test "accepts the 'user:notification' stream", %{token: token} = _state do
	assert {:ok, _} = start_socket("?stream=user:notification&access_token=#{token.token}")

	assert capture_log(fn ->
	assert {:error, {403, "Forbidden"}} = start_socket("?stream=user:notification")
	Process.sleep(30)
	end) =~ ":badarg"
	end

	test "accepts valid token on Sec-WebSocket-Protocol header", %{token: token} do
	assert {:ok, _} = start_socket("?stream=user", [{"Sec-WebSocket-Protocol", token.token}])

	assert capture_log(fn ->
	assert {:error, {403, "Forbidden"}} =
	start_socket("?stream=user", [{"Sec-WebSocket-Protocol", "I am a friend"}])

	Process.sleep(30)
	end) =~ ":badarg"
	end
	end
	end
	diff --git a/test/notification_test.exs b/test/notification_test.exs
	index 601a6c0ca..5c85f3368 100644
	--- a/test/notification_test.exs
	+++ b/test/notification_test.exs
	@@ -1,1030 +1,1034 @@
	# Pleroma: A lightweight social networking server
	# Copyright © 2017-2020 Pleroma Authors <https://pleroma.social/>
	# SPDX-License-Identifier: AGPL-3.0-only

	defmodule Pleroma.NotificationTest do
	use Pleroma.DataCase

	import Pleroma.Factory
	import Mock

	alias Pleroma.FollowingRelationship
	alias Pleroma.Notification
	alias Pleroma.Tests.ObanHelpers
	alias Pleroma.User
	alias Pleroma.Web.ActivityPub.Transmogrifier
	alias Pleroma.Web.CommonAPI
	alias Pleroma.Web.MastodonAPI.NotificationView
	alias Pleroma.Web.Push
	alias Pleroma.Web.Streamer

	describe "create_notifications" do
	test "creates a notification for an emoji reaction" do
	user = insert(:user)
	other_user = insert(:user)

	{:ok, activity} = CommonAPI.post(user, %{"status" => "yeah"})
	{:ok, activity, _object} = CommonAPI.react_with_emoji(activity.id, other_user, "☕")

	{:ok, [notification]} = Notification.create_notifications(activity)

	assert notification.user_id == user.id
	end

	test "notifies someone when they are directly addressed" do
	user = insert(:user)
	other_user = insert(:user)
	third_user = insert(:user)

	{:ok, activity} =
	CommonAPI.post(user, %{
	"status" => "hey @#{other_user.nickname} and @#{third_user.nickname}"
	})

	{:ok, [notification, other_notification]} = Notification.create_notifications(activity)

	notified_ids = Enum.sort([notification.user_id, other_notification.user_id])
	assert notified_ids == [other_user.id, third_user.id]
	assert notification.activity_id == activity.id
	assert other_notification.activity_id == activity.id
	end

	test "it creates a notification for subscribed users" do
	user = insert(:user)
	subscriber = insert(:user)

	User.subscribe(subscriber, user)

	{:ok, status} = CommonAPI.post(user, %{"status" => "Akariiiin"})
	{:ok, [notification]} = Notification.create_notifications(status)

	assert notification.user_id == subscriber.id
	end

	test "does not create a notification for subscribed users if status is a reply" do
	user = insert(:user)
	other_user = insert(:user)
	subscriber = insert(:user)

	User.subscribe(subscriber, other_user)

	{:ok, activity} = CommonAPI.post(user, %{"status" => "test post"})

	{:ok, _reply_activity} =
	CommonAPI.post(other_user, %{
	"status" => "test reply",
	"in_reply_to_status_id" => activity.id
	})

	user_notifications = Notification.for_user(user)
	assert length(user_notifications) == 1

	subscriber_notifications = Notification.for_user(subscriber)
	assert Enum.empty?(subscriber_notifications)
	end
	end

	describe "CommonApi.post/2 notification-related functionality" do
	test_with_mock "creates but does NOT send notification to blocker user",
	Push,
	[:passthrough],
	[] do
	user = insert(:user)
	blocker = insert(:user)
	{:ok, _user_relationship} = User.block(blocker, user)

	{:ok, _activity} = CommonAPI.post(user, %{"status" => "hey @#{blocker.nickname}!"})

	blocker_id = blocker.id
	assert [%Notification{user_id: ^blocker_id}] = Repo.all(Notification)
	refute called(Push.send(:_))
	end

	test_with_mock "creates but does NOT send notification to notification-muter user",
	Push,
	[:passthrough],
	[] do
	user = insert(:user)
	muter = insert(:user)
	{:ok, _user_relationships} = User.mute(muter, user)

	{:ok, _activity} = CommonAPI.post(user, %{"status" => "hey @#{muter.nickname}!"})

	muter_id = muter.id
	assert [%Notification{user_id: ^muter_id}] = Repo.all(Notification)
	refute called(Push.send(:_))
	end

	test_with_mock "creates but does NOT send notification to thread-muter user",
	Push,
	[:passthrough],
	[] do
	user = insert(:user)
	thread_muter = insert(:user)

	{:ok, activity} = CommonAPI.post(user, %{"status" => "hey @#{thread_muter.nickname}!"})

	{:ok, _} = CommonAPI.add_mute(thread_muter, activity)

	{:ok, _same_context_activity} =
	CommonAPI.post(user, %{
	"status" => "hey-hey-hey @#{thread_muter.nickname}!",
	"in_reply_to_status_id" => activity.id
	})

	[pre_mute_notification, post_mute_notification] =
	Repo.all(from(n in Notification, where: n.user_id == ^thread_muter.id, order_by: n.id))

	pre_mute_notification_id = pre_mute_notification.id
	post_mute_notification_id = post_mute_notification.id

	assert called(
	Push.send(
	:meck.is(fn
	%Notification{id: ^pre_mute_notification_id} -> true
	_ -> false
	end)
	)
	)

	refute called(
	Push.send(
	:meck.is(fn
	%Notification{id: ^post_mute_notification_id} -> true
	_ -> false
	end)
	)
	)
	end
	end

	describe "create_notification" do
	@tag needs_streamer: true
	test "it creates a notification for user and send to the 'user' and the 'user:notification' stream" do
	user = insert(:user)
	- task = Task.async(fn -> assert_receive {:text, _}, 4_000 end)
	- task_user_notification = Task.async(fn -> assert_receive {:text, _}, 4_000 end)
	- Streamer.add_socket("user", %{transport_pid: task.pid, assigns: %{user: user}})

	- Streamer.add_socket(
	- "user:notification",
	- %{transport_pid: task_user_notification.pid, assigns: %{user: user}}
	- )
	+ task =
	+ Task.async(fn ->
	+ Streamer.add_socket("user", user)
	+ assert_receive {:render_with_user, _, _, _}, 4_000
	+ end)
	+
	+ task_user_notification =
	+ Task.async(fn ->
	+ Streamer.add_socket("user:notification", user)
	+ assert_receive {:render_with_user, _, _, _}, 4_000
	+ end)

	activity = insert(:note_activity)

	notify = Notification.create_notification(activity, user)
	assert notify.user_id == user.id
	Task.await(task)
	Task.await(task_user_notification)
	end

	test "it creates a notification for user if the user blocks the activity author" do
	activity = insert(:note_activity)
	author = User.get_cached_by_ap_id(activity.data["actor"])
	user = insert(:user)
	{:ok, _user_relationship} = User.block(user, author)

	assert Notification.create_notification(activity, user)
	end

	test "it creates a notification for the user if the user mutes the activity author" do
	muter = insert(:user)
	muted = insert(:user)
	{:ok, _} = User.mute(muter, muted)
	muter = Repo.get(User, muter.id)
	{:ok, activity} = CommonAPI.post(muted, %{"status" => "Hi @#{muter.nickname}"})

	assert Notification.create_notification(activity, muter)
	end

	test "notification created if user is muted without notifications" do
	muter = insert(:user)
	muted = insert(:user)

	{:ok, _user_relationships} = User.mute(muter, muted, false)

	{:ok, activity} = CommonAPI.post(muted, %{"status" => "Hi @#{muter.nickname}"})

	assert Notification.create_notification(activity, muter)
	end

	test "it creates a notification for an activity from a muted thread" do
	muter = insert(:user)
	other_user = insert(:user)
	{:ok, activity} = CommonAPI.post(muter, %{"status" => "hey"})
	CommonAPI.add_mute(muter, activity)

	{:ok, activity} =
	CommonAPI.post(other_user, %{
	"status" => "Hi @#{muter.nickname}",
	"in_reply_to_status_id" => activity.id
	})

	assert Notification.create_notification(activity, muter)
	end

	test "it disables notifications from followers" do
	follower = insert(:user)

	followed =
	insert(:user, notification_settings: %Pleroma.User.NotificationSetting{followers: false})

	User.follow(follower, followed)
	{:ok, activity} = CommonAPI.post(follower, %{"status" => "hey @#{followed.nickname}"})
	refute Notification.create_notification(activity, followed)
	end

	test "it disables notifications from non-followers" do
	follower = insert(:user)

	followed =
	insert(:user,
	notification_settings: %Pleroma.User.NotificationSetting{non_followers: false}
	)

	{:ok, activity} = CommonAPI.post(follower, %{"status" => "hey @#{followed.nickname}"})
	refute Notification.create_notification(activity, followed)
	end

	test "it disables notifications from people the user follows" do
	follower =
	insert(:user, notification_settings: %Pleroma.User.NotificationSetting{follows: false})

	followed = insert(:user)
	User.follow(follower, followed)
	follower = Repo.get(User, follower.id)
	{:ok, activity} = CommonAPI.post(followed, %{"status" => "hey @#{follower.nickname}"})
	refute Notification.create_notification(activity, follower)
	end

	test "it disables notifications from people the user does not follow" do
	follower =
	insert(:user, notification_settings: %Pleroma.User.NotificationSetting{non_follows: false})

	followed = insert(:user)
	{:ok, activity} = CommonAPI.post(followed, %{"status" => "hey @#{follower.nickname}"})
	refute Notification.create_notification(activity, follower)
	end

	test "it doesn't create a notification for user if he is the activity author" do
	activity = insert(:note_activity)
	author = User.get_cached_by_ap_id(activity.data["actor"])

	refute Notification.create_notification(activity, author)
	end

	test "it doesn't create duplicate notifications for follow+subscribed users" do
	user = insert(:user)
	subscriber = insert(:user)

	{:ok, _, _, _} = CommonAPI.follow(subscriber, user)
	User.subscribe(subscriber, user)
	{:ok, status} = CommonAPI.post(user, %{"status" => "Akariiiin"})
	{:ok, [_notif]} = Notification.create_notifications(status)
	end

	test "it doesn't create subscription notifications if the recipient cannot see the status" do
	user = insert(:user)
	subscriber = insert(:user)

	User.subscribe(subscriber, user)

	{:ok, status} = CommonAPI.post(user, %{"status" => "inwisible", "visibility" => "direct"})

	assert {:ok, []} == Notification.create_notifications(status)
	end
	end

	describe "follow / follow_request notifications" do
	test "it creates `follow` notification for approved Follow activity" do
	user = insert(:user)
	followed_user = insert(:user, locked: false)

	{:ok, _, _, _activity} = CommonAPI.follow(user, followed_user)
	assert FollowingRelationship.following?(user, followed_user)
	assert [notification] = Notification.for_user(followed_user)

	assert %{type: "follow"} =
	NotificationView.render("show.json", %{
	notification: notification,
	for: followed_user
	})
	end

	test "it creates `follow_request` notification for pending Follow activity" do
	user = insert(:user)
	followed_user = insert(:user, locked: true)

	{:ok, _, _, _activity} = CommonAPI.follow(user, followed_user)
	refute FollowingRelationship.following?(user, followed_user)
	assert [notification] = Notification.for_user(followed_user)

	render_opts = %{notification: notification, for: followed_user}
	assert %{type: "follow_request"} = NotificationView.render("show.json", render_opts)

	# After request is accepted, the same notification is rendered with type "follow":
	assert {:ok, _} = CommonAPI.accept_follow_request(user, followed_user)

	notification_id = notification.id
	assert [%{id: ^notification_id}] = Notification.for_user(followed_user)
	assert %{type: "follow"} = NotificationView.render("show.json", render_opts)
	end

	test "it doesn't create a notification for follow-unfollow-follow chains" do
	user = insert(:user)
	followed_user = insert(:user, locked: false)

	{:ok, _, _, _activity} = CommonAPI.follow(user, followed_user)
	assert FollowingRelationship.following?(user, followed_user)
	assert [notification] = Notification.for_user(followed_user)

	CommonAPI.unfollow(user, followed_user)
	{:ok, _, _, _activity_dupe} = CommonAPI.follow(user, followed_user)

	notification_id = notification.id
	assert [%{id: ^notification_id}] = Notification.for_user(followed_user)
	end

	test "dismisses the notification on follow request rejection" do
	user = insert(:user, locked: true)
	follower = insert(:user)
	{:ok, _, _, _follow_activity} = CommonAPI.follow(follower, user)
	assert [notification] = Notification.for_user(user)
	{:ok, _follower} = CommonAPI.reject_follow_request(follower, user)
	assert [] = Notification.for_user(user)
	end
	end

	describe "get notification" do
	test "it gets a notification that belongs to the user" do
	user = insert(:user)
	other_user = insert(:user)

	{:ok, activity} = CommonAPI.post(user, %{"status" => "hey @#{other_user.nickname}"})

	{:ok, [notification]} = Notification.create_notifications(activity)
	{:ok, notification} = Notification.get(other_user, notification.id)

	assert notification.user_id == other_user.id
	end

	test "it returns error if the notification doesn't belong to the user" do
	user = insert(:user)
	other_user = insert(:user)

	{:ok, activity} = CommonAPI.post(user, %{"status" => "hey @#{other_user.nickname}"})

	{:ok, [notification]} = Notification.create_notifications(activity)
	{:error, _notification} = Notification.get(user, notification.id)
	end
	end

	describe "dismiss notification" do
	test "it dismisses a notification that belongs to the user" do
	user = insert(:user)
	other_user = insert(:user)

	{:ok, activity} = CommonAPI.post(user, %{"status" => "hey @#{other_user.nickname}"})

	{:ok, [notification]} = Notification.create_notifications(activity)
	{:ok, notification} = Notification.dismiss(other_user, notification.id)

	assert notification.user_id == other_user.id
	end

	test "it returns error if the notification doesn't belong to the user" do
	user = insert(:user)
	other_user = insert(:user)

	{:ok, activity} = CommonAPI.post(user, %{"status" => "hey @#{other_user.nickname}"})

	{:ok, [notification]} = Notification.create_notifications(activity)
	{:error, _notification} = Notification.dismiss(user, notification.id)
	end
	end

	describe "clear notification" do
	test "it clears all notifications belonging to the user" do
	user = insert(:user)
	other_user = insert(:user)
	third_user = insert(:user)

	{:ok, activity} =
	CommonAPI.post(user, %{
	"status" => "hey @#{other_user.nickname} and @#{third_user.nickname} !"
	})

	{:ok, _notifs} = Notification.create_notifications(activity)

	{:ok, activity} =
	CommonAPI.post(user, %{
	"status" => "hey again @#{other_user.nickname} and @#{third_user.nickname} !"
	})

	{:ok, _notifs} = Notification.create_notifications(activity)
	Notification.clear(other_user)

	assert Notification.for_user(other_user) == []
	assert Notification.for_user(third_user) != []
	end
	end

	describe "set_read_up_to()" do
	test "it sets all notifications as read up to a specified notification ID" do
	user = insert(:user)
	other_user = insert(:user)

	{:ok, _activity} =
	CommonAPI.post(user, %{
	"status" => "hey @#{other_user.nickname}!"
	})

	{:ok, _activity} =
	CommonAPI.post(user, %{
	"status" => "hey again @#{other_user.nickname}!"
	})

	[n2, n1] = notifs = Notification.for_user(other_user)
	assert length(notifs) == 2

	assert n2.id > n1.id

	{:ok, _activity} =
	CommonAPI.post(user, %{
	"status" => "hey yet again @#{other_user.nickname}!"
	})

	Notification.set_read_up_to(other_user, n2.id)

	[n3, n2, n1] = Notification.for_user(other_user)

	assert n1.seen == true
	assert n2.seen == true
	assert n3.seen == false
	end
	end

	describe "for_user_since/2" do
	defp days_ago(days) do
	NaiveDateTime.add(
	NaiveDateTime.truncate(NaiveDateTime.utc_now(), :second),
	-days * 60 * 60 * 24,
	:second
	)
	end

	test "Returns recent notifications" do
	user1 = insert(:user)
	user2 = insert(:user)

	Enum.each(0..10, fn i ->
	{:ok, _activity} =
	CommonAPI.post(user1, %{
	"status" => "hey ##{i} @#{user2.nickname}!"
	})
	end)

	{old, new} = Enum.split(Notification.for_user(user2), 5)

	Enum.each(old, fn notification ->
	notification
	\|> cast(%{updated_at: days_ago(10)}, [:updated_at])
	\|> Pleroma.Repo.update!()
	end)

	recent_notifications_ids =
	user2
	\|> Notification.for_user_since(
	NaiveDateTime.add(NaiveDateTime.utc_now(), -5 * 86_400, :second)
	)
	\|> Enum.map(& &1.id)

	Enum.each(old, fn %{id: id} ->
	refute id in recent_notifications_ids
	end)

	Enum.each(new, fn %{id: id} ->
	assert id in recent_notifications_ids
	end)
	end
	end

	describe "notification target determination / get_notified_from_activity/2" do
	test "it sends notifications to addressed users in new messages" do
	user = insert(:user)
	other_user = insert(:user)

	{:ok, activity} =
	CommonAPI.post(user, %{
	"status" => "hey @#{other_user.nickname}!"
	})

	{enabled_receivers, _disabled_receivers} = Notification.get_notified_from_activity(activity)

	assert other_user in enabled_receivers
	end

	test "it sends notifications to mentioned users in new messages" do
	user = insert(:user)
	other_user = insert(:user)

	create_activity = %{
	"@context" => "https://www.w3.org/ns/activitystreams",
	"type" => "Create",
	"to" => ["https://www.w3.org/ns/activitystreams#Public"],
	"actor" => user.ap_id,
	"object" => %{
	"type" => "Note",
	"content" => "message with a Mention tag, but no explicit tagging",
	"tag" => [
	%{
	"type" => "Mention",
	"href" => other_user.ap_id,
	"name" => other_user.nickname
	}
	],
	"attributedTo" => user.ap_id
	}
	}

	{:ok, activity} = Transmogrifier.handle_incoming(create_activity)

	{enabled_receivers, _disabled_receivers} = Notification.get_notified_from_activity(activity)

	assert other_user in enabled_receivers
	end

	test "it does not send notifications to users who are only cc in new messages" do
	user = insert(:user)
	other_user = insert(:user)

	create_activity = %{
	"@context" => "https://www.w3.org/ns/activitystreams",
	"type" => "Create",
	"to" => ["https://www.w3.org/ns/activitystreams#Public"],
	"cc" => [other_user.ap_id],
	"actor" => user.ap_id,
	"object" => %{
	"type" => "Note",
	"content" => "hi everyone",
	"attributedTo" => user.ap_id
	}
	}

	{:ok, activity} = Transmogrifier.handle_incoming(create_activity)

	{enabled_receivers, _disabled_receivers} = Notification.get_notified_from_activity(activity)

	assert other_user not in enabled_receivers
	end

	test "it does not send notification to mentioned users in likes" do
	user = insert(:user)
	other_user = insert(:user)
	third_user = insert(:user)

	{:ok, activity_one} =
	CommonAPI.post(user, %{
	"status" => "hey @#{other_user.nickname}!"
	})

	{:ok, activity_two} = CommonAPI.favorite(third_user, activity_one.id)

	{enabled_receivers, _disabled_receivers} =
	Notification.get_notified_from_activity(activity_two)

	assert other_user not in enabled_receivers
	end

	test "it does not send notification to mentioned users in announces" do
	user = insert(:user)
	other_user = insert(:user)
	third_user = insert(:user)

	{:ok, activity_one} =
	CommonAPI.post(user, %{
	"status" => "hey @#{other_user.nickname}!"
	})

	{:ok, activity_two, _} = CommonAPI.repeat(activity_one.id, third_user)

	{enabled_receivers, _disabled_receivers} =
	Notification.get_notified_from_activity(activity_two)

	assert other_user not in enabled_receivers
	end

	test "it returns blocking recipient in disabled recipients list" do
	user = insert(:user)
	other_user = insert(:user)
	{:ok, _user_relationship} = User.block(other_user, user)

	{:ok, activity} = CommonAPI.post(user, %{"status" => "hey @#{other_user.nickname}!"})

	{enabled_receivers, disabled_receivers} = Notification.get_notified_from_activity(activity)

	assert [] == enabled_receivers
	assert [other_user] == disabled_receivers
	end

	test "it returns notification-muting recipient in disabled recipients list" do
	user = insert(:user)
	other_user = insert(:user)
	{:ok, _user_relationships} = User.mute(other_user, user)

	{:ok, activity} = CommonAPI.post(user, %{"status" => "hey @#{other_user.nickname}!"})

	{enabled_receivers, disabled_receivers} = Notification.get_notified_from_activity(activity)

	assert [] == enabled_receivers
	assert [other_user] == disabled_receivers
	end

	test "it returns thread-muting recipient in disabled recipients list" do
	user = insert(:user)
	other_user = insert(:user)

	{:ok, activity} = CommonAPI.post(user, %{"status" => "hey @#{other_user.nickname}!"})

	{:ok, _} = CommonAPI.add_mute(other_user, activity)

	{:ok, same_context_activity} =
	CommonAPI.post(user, %{
	"status" => "hey-hey-hey @#{other_user.nickname}!",
	"in_reply_to_status_id" => activity.id
	})

	{enabled_receivers, disabled_receivers} =
	Notification.get_notified_from_activity(same_context_activity)

	assert [other_user] == disabled_receivers
	refute other_user in enabled_receivers
	end

	test "it returns non-following domain-blocking recipient in disabled recipients list" do
	blocked_domain = "blocked.domain"
	user = insert(:user, %{ap_id: "https://#{blocked_domain}/@actor"})
	other_user = insert(:user)

	{:ok, other_user} = User.block_domain(other_user, blocked_domain)

	{:ok, activity} = CommonAPI.post(user, %{"status" => "hey @#{other_user.nickname}!"})

	{enabled_receivers, disabled_receivers} = Notification.get_notified_from_activity(activity)

	assert [] == enabled_receivers
	assert [other_user] == disabled_receivers
	end

	test "it returns following domain-blocking recipient in enabled recipients list" do
	blocked_domain = "blocked.domain"
	user = insert(:user, %{ap_id: "https://#{blocked_domain}/@actor"})
	other_user = insert(:user)

	{:ok, other_user} = User.block_domain(other_user, blocked_domain)
	{:ok, other_user} = User.follow(other_user, user)

	{:ok, activity} = CommonAPI.post(user, %{"status" => "hey @#{other_user.nickname}!"})

	{enabled_receivers, disabled_receivers} = Notification.get_notified_from_activity(activity)

	assert [other_user] == enabled_receivers
	assert [] == disabled_receivers
	end
	end

	describe "notification lifecycle" do
	test "liking an activity results in 1 notification, then 0 if the activity is deleted" do
	user = insert(:user)
	other_user = insert(:user)

	{:ok, activity} = CommonAPI.post(user, %{"status" => "test post"})

	assert Enum.empty?(Notification.for_user(user))

	{:ok, _} = CommonAPI.favorite(other_user, activity.id)

	assert length(Notification.for_user(user)) == 1

	{:ok, _} = CommonAPI.delete(activity.id, user)

	assert Enum.empty?(Notification.for_user(user))
	end

	test "liking an activity results in 1 notification, then 0 if the activity is unliked" do
	user = insert(:user)
	other_user = insert(:user)

	{:ok, activity} = CommonAPI.post(user, %{"status" => "test post"})

	assert Enum.empty?(Notification.for_user(user))

	{:ok, _} = CommonAPI.favorite(other_user, activity.id)

	assert length(Notification.for_user(user)) == 1

	{:ok, _, _, _} = CommonAPI.unfavorite(activity.id, other_user)

	assert Enum.empty?(Notification.for_user(user))
	end

	test "repeating an activity results in 1 notification, then 0 if the activity is deleted" do
	user = insert(:user)
	other_user = insert(:user)

	{:ok, activity} = CommonAPI.post(user, %{"status" => "test post"})

	assert Enum.empty?(Notification.for_user(user))

	{:ok, _, _} = CommonAPI.repeat(activity.id, other_user)

	assert length(Notification.for_user(user)) == 1

	{:ok, _} = CommonAPI.delete(activity.id, user)

	assert Enum.empty?(Notification.for_user(user))
	end

	test "repeating an activity results in 1 notification, then 0 if the activity is unrepeated" do
	user = insert(:user)
	other_user = insert(:user)

	{:ok, activity} = CommonAPI.post(user, %{"status" => "test post"})

	assert Enum.empty?(Notification.for_user(user))

	{:ok, _, _} = CommonAPI.repeat(activity.id, other_user)

	assert length(Notification.for_user(user)) == 1

	{:ok, _, _} = CommonAPI.unrepeat(activity.id, other_user)

	assert Enum.empty?(Notification.for_user(user))
	end

	test "liking an activity which is already deleted does not generate a notification" do
	user = insert(:user)
	other_user = insert(:user)

	{:ok, activity} = CommonAPI.post(user, %{"status" => "test post"})

	assert Enum.empty?(Notification.for_user(user))

	{:ok, _deletion_activity} = CommonAPI.delete(activity.id, user)

	assert Enum.empty?(Notification.for_user(user))

	{:error, :not_found} = CommonAPI.favorite(other_user, activity.id)

	assert Enum.empty?(Notification.for_user(user))
	end

	test "repeating an activity which is already deleted does not generate a notification" do
	user = insert(:user)
	other_user = insert(:user)

	{:ok, activity} = CommonAPI.post(user, %{"status" => "test post"})

	assert Enum.empty?(Notification.for_user(user))

	{:ok, _deletion_activity} = CommonAPI.delete(activity.id, user)

	assert Enum.empty?(Notification.for_user(user))

	{:error, _} = CommonAPI.repeat(activity.id, other_user)

	assert Enum.empty?(Notification.for_user(user))
	end

	test "replying to a deleted post without tagging does not generate a notification" do
	user = insert(:user)
	other_user = insert(:user)

	{:ok, activity} = CommonAPI.post(user, %{"status" => "test post"})
	{:ok, _deletion_activity} = CommonAPI.delete(activity.id, user)

	{:ok, _reply_activity} =
	CommonAPI.post(other_user, %{
	"status" => "test reply",
	"in_reply_to_status_id" => activity.id
	})

	assert Enum.empty?(Notification.for_user(user))
	end

	test "notifications are deleted if a local user is deleted" do
	user = insert(:user)
	other_user = insert(:user)

	{:ok, _activity} =
	CommonAPI.post(user, %{"status" => "hi @#{other_user.nickname}", "visibility" => "direct"})

	refute Enum.empty?(Notification.for_user(other_user))

	{:ok, job} = User.delete(user)
	ObanHelpers.perform(job)

	assert Enum.empty?(Notification.for_user(other_user))
	end

	test "notifications are deleted if a remote user is deleted" do
	remote_user = insert(:user)
	local_user = insert(:user)

	dm_message = %{
	"@context" => "https://www.w3.org/ns/activitystreams",
	"type" => "Create",
	"actor" => remote_user.ap_id,
	"id" => remote_user.ap_id <> "/activities/test",
	"to" => [local_user.ap_id],
	"cc" => [],
	"object" => %{
	"type" => "Note",
	"content" => "Hello!",
	"tag" => [
	%{
	"type" => "Mention",
	"href" => local_user.ap_id,
	"name" => "@#{local_user.nickname}"
	}
	],
	"to" => [local_user.ap_id],
	"cc" => [],
	"attributedTo" => remote_user.ap_id
	}
	}

	{:ok, _dm_activity} = Transmogrifier.handle_incoming(dm_message)

	refute Enum.empty?(Notification.for_user(local_user))

	delete_user_message = %{
	"@context" => "https://www.w3.org/ns/activitystreams",
	"id" => remote_user.ap_id <> "/activities/delete",
	"actor" => remote_user.ap_id,
	"type" => "Delete",
	"object" => remote_user.ap_id
	}

	remote_user_url = remote_user.ap_id

	Tesla.Mock.mock(fn
	%{method: :get, url: ^remote_user_url} ->
	%Tesla.Env{status: 404, body: ""}
	end)

	{:ok, _delete_activity} = Transmogrifier.handle_incoming(delete_user_message)
	ObanHelpers.perform_all()

	assert Enum.empty?(Notification.for_user(local_user))
	end

	@tag capture_log: true
	test "move activity generates a notification" do
	%{ap_id: old_ap_id} = old_user = insert(:user)
	%{ap_id: new_ap_id} = new_user = insert(:user, also_known_as: [old_ap_id])
	follower = insert(:user)
	other_follower = insert(:user, %{allow_following_move: false})

	User.follow(follower, old_user)
	User.follow(other_follower, old_user)

	old_user_url = old_user.ap_id

	body =
	File.read!("test/fixtures/users_mock/localhost.json")
	\|> String.replace("{{nickname}}", old_user.nickname)
	\|> Jason.encode!()

	Tesla.Mock.mock(fn
	%{method: :get, url: ^old_user_url} ->
	%Tesla.Env{status: 200, body: body}
	end)

	Pleroma.Web.ActivityPub.ActivityPub.move(old_user, new_user)
	ObanHelpers.perform_all()

	assert [
	%{
	activity: %{
	data: %{"type" => "Move", "actor" => ^old_ap_id, "target" => ^new_ap_id}
	}
	}
	] = Notification.for_user(follower)

	assert [
	%{
	activity: %{
	data: %{"type" => "Move", "actor" => ^old_ap_id, "target" => ^new_ap_id}
	}
	}
	] = Notification.for_user(other_follower)
	end
	end

	describe "for_user" do
	test "it returns notifications for muted user without notifications" do
	user = insert(:user)
	muted = insert(:user)
	{:ok, _user_relationships} = User.mute(user, muted, false)

	{:ok, _activity} = CommonAPI.post(muted, %{"status" => "hey @#{user.nickname}"})

	assert length(Notification.for_user(user)) == 1
	end

	test "it doesn't return notifications for muted user with notifications" do
	user = insert(:user)
	muted = insert(:user)
	{:ok, _user_relationships} = User.mute(user, muted)

	{:ok, _activity} = CommonAPI.post(muted, %{"status" => "hey @#{user.nickname}"})

	assert Notification.for_user(user) == []
	end

	test "it doesn't return notifications for blocked user" do
	user = insert(:user)
	blocked = insert(:user)
	{:ok, _user_relationship} = User.block(user, blocked)

	{:ok, _activity} = CommonAPI.post(blocked, %{"status" => "hey @#{user.nickname}"})

	assert Notification.for_user(user) == []
	end

	test "it doesn't return notifications for domain-blocked non-followed user" do
	user = insert(:user)
	blocked = insert(:user, ap_id: "http://some-domain.com")
	{:ok, user} = User.block_domain(user, "some-domain.com")

	{:ok, _activity} = CommonAPI.post(blocked, %{"status" => "hey @#{user.nickname}"})

	assert Notification.for_user(user) == []
	end

	test "it returns notifications for domain-blocked but followed user" do
	user = insert(:user)
	blocked = insert(:user, ap_id: "http://some-domain.com")

	{:ok, user} = User.block_domain(user, "some-domain.com")
	{:ok, _} = User.follow(user, blocked)

	{:ok, _activity} = CommonAPI.post(blocked, %{"status" => "hey @#{user.nickname}"})

	assert length(Notification.for_user(user)) == 1
	end

	test "it doesn't return notifications for muted thread" do
	user = insert(:user)
	another_user = insert(:user)

	{:ok, activity} = CommonAPI.post(another_user, %{"status" => "hey @#{user.nickname}"})

	{:ok, _} = Pleroma.ThreadMute.add_mute(user.id, activity.data["context"])
	assert Notification.for_user(user) == []
	end

	test "it returns notifications from a muted user when with_muted is set" do
	user = insert(:user)
	muted = insert(:user)
	{:ok, _user_relationships} = User.mute(user, muted)

	{:ok, _activity} = CommonAPI.post(muted, %{"status" => "hey @#{user.nickname}"})

	assert length(Notification.for_user(user, %{with_muted: true})) == 1
	end

	test "it doesn't return notifications from a blocked user when with_muted is set" do
	user = insert(:user)
	blocked = insert(:user)
	{:ok, _user_relationship} = User.block(user, blocked)

	{:ok, _activity} = CommonAPI.post(blocked, %{"status" => "hey @#{user.nickname}"})

	assert Enum.empty?(Notification.for_user(user, %{with_muted: true}))
	end

	test "when with_muted is set, " <>
	"it doesn't return notifications from a domain-blocked non-followed user" do
	user = insert(:user)
	blocked = insert(:user, ap_id: "http://some-domain.com")
	{:ok, user} = User.block_domain(user, "some-domain.com")

	{:ok, _activity} = CommonAPI.post(blocked, %{"status" => "hey @#{user.nickname}"})

	assert Enum.empty?(Notification.for_user(user, %{with_muted: true}))
	end

	test "it returns notifications from muted threads when with_muted is set" do
	user = insert(:user)
	another_user = insert(:user)

	{:ok, activity} = CommonAPI.post(another_user, %{"status" => "hey @#{user.nickname}"})

	{:ok, _} = Pleroma.ThreadMute.add_mute(user.id, activity.data["context"])
	assert length(Notification.for_user(user, %{with_muted: true})) == 1
	end
	end
	end
	diff --git a/test/support/builders/activity_builder.ex b/test/support/builders/activity_builder.ex
	index 6e5a8e059..7c4950bfa 100644
	--- a/test/support/builders/activity_builder.ex
	+++ b/test/support/builders/activity_builder.ex
	@@ -1,49 +1,57 @@
	defmodule Pleroma.Builders.ActivityBuilder do
	alias Pleroma.Web.ActivityPub.ActivityPub

	def build(data \\ %{}, opts \\ %{}) do
	user = opts[:user] \|\| Pleroma.Factory.insert(:user)

	activity = %{
	"id" => Pleroma.Web.ActivityPub.Utils.generate_object_id(),
	"actor" => user.ap_id,
	"to" => ["https://www.w3.org/ns/activitystreams#Public"],
	"type" => "Create",
	"object" => %{
	"type" => "Note",
	"content" => "test",
	"to" => ["https://www.w3.org/ns/activitystreams#Public"]
	}
	}

	Map.merge(activity, data)
	end

	def insert(data \\ %{}, opts \\ %{}) do
	activity = build(data, opts)
	- ActivityPub.insert(activity)
	+
	+ case ActivityPub.insert(activity) do
	+ ok = {:ok, activity} ->
	+ ActivityPub.notify_and_stream(activity)
	+ ok
	+
	+ error ->
	+ error
	+ end
	end

	def insert_list(times, data \\ %{}, opts \\ %{}) do
	Enum.map(1..times, fn _n ->
	{:ok, activity} = insert(data, opts)
	activity
	end)
	end

	def public_and_non_public do
	user = Pleroma.Factory.insert(:user)

	public = build(%{"id" => 1}, %{user: user})
	non_public = build(%{"id" => 2, "to" => [user.follower_address]}, %{user: user})

	{:ok, public} = ActivityPub.insert(public)
	{:ok, non_public} = ActivityPub.insert(non_public)

	%{
	public: public,
	non_public: non_public,
	user: user
	}
	end
	end
	diff --git a/test/support/conn_case.ex b/test/support/conn_case.ex
	index 91c03b1a8..b23918dd1 100644
	--- a/test/support/conn_case.ex
	+++ b/test/support/conn_case.ex
	@@ -1,147 +1,151 @@
	# Pleroma: A lightweight social networking server
	# Copyright © 2017-2020 Pleroma Authors <https://pleroma.social/>
	# SPDX-License-Identifier: AGPL-3.0-only

	defmodule Pleroma.Web.ConnCase do
	@moduledoc """
	This module defines the test case to be used by
	tests that require setting up a connection.

	Such tests rely on `Phoenix.ConnTest` and also
	import other functionality to make it easier
	to build common datastructures and query the data layer.

	Finally, if the test case interacts with the database,
	it cannot be async. For this reason, every test runs
	inside a transaction which is reset at the beginning
	of the test unless the test case is marked as async.
	"""

	use ExUnit.CaseTemplate

	using do
	quote do
	# Import conveniences for testing with connections
	use Phoenix.ConnTest
	use Pleroma.Tests.Helpers
	import Pleroma.Web.Router.Helpers

	alias Pleroma.Config

	# The default endpoint for testing
	@endpoint Pleroma.Web.Endpoint

	# Sets up OAuth access with specified scopes
	defp oauth_access(scopes, opts \\ []) do
	user =
	Keyword.get_lazy(opts, :user, fn ->
	Pleroma.Factory.insert(:user)
	end)

	token =
	Keyword.get_lazy(opts, :oauth_token, fn ->
	Pleroma.Factory.insert(:oauth_token, user: user, scopes: scopes)
	end)

	conn =
	build_conn()
	\|> assign(:user, user)
	\|> assign(:token, token)

	%{user: user, token: token, conn: conn}
	end

	defp request_content_type(%{conn: conn}) do
	conn = put_req_header(conn, "content-type", "multipart/form-data")
	[conn: conn]
	end

	defp json_response_and_validate_schema(
	%{
	private: %{
	open_api_spex: %{operation_id: op_id, operation_lookup: lookup, spec: spec}
	}
	} = conn,
	status
	) do
	content_type =
	conn
	\|> Plug.Conn.get_resp_header("content-type")
	\|> List.first()
	\|> String.split(";")
	\|> List.first()

	status = Plug.Conn.Status.code(status)

	unless lookup[op_id].responses[status] do
	err = "Response schema not found for #{status} #{conn.method} #{conn.request_path}"
	flunk(err)
	end

	schema = lookup[op_id].responses[status].content[content_type].schema
	json = json_response(conn, status)

	case OpenApiSpex.cast_value(json, schema, spec) do
	{:ok, _data} ->
	json

	{:error, errors} ->
	errors =
	Enum.map(errors, fn error ->
	message = OpenApiSpex.Cast.Error.message(error)
	path = OpenApiSpex.Cast.Error.path_to_string(error)
	"#{message} at #{path}"
	end)

	flunk(
	"Response does not conform to schema of #{op_id} operation: #{
	Enum.join(errors, "\n")
	}\n#{inspect(json)}"
	)
	end
	end

	defp json_response_and_validate_schema(conn, _status) do
	flunk("Response schema not found for #{conn.method} #{conn.request_path} #{conn.status}")
	end

	defp ensure_federating_or_authenticated(conn, url, user) do
	initial_setting = Config.get([:instance, :federating])
	on_exit(fn -> Config.put([:instance, :federating], initial_setting) end)

	Config.put([:instance, :federating], false)

	conn
	\|> get(url)
	\|> response(403)

	conn
	\|> assign(:user, user)
	\|> get(url)
	\|> response(200)

	Config.put([:instance, :federating], true)

	conn
	\|> get(url)
	\|> response(200)
	end
	end
	end

	setup tags do
	Cachex.clear(:user_cache)
	Cachex.clear(:object_cache)
	:ok = Ecto.Adapters.SQL.Sandbox.checkout(Pleroma.Repo)

	unless tags[:async] do
	Ecto.Adapters.SQL.Sandbox.mode(Pleroma.Repo, {:shared, self()})
	end

	if tags[:needs_streamer] do
	- start_supervised(Pleroma.Web.Streamer.supervisor())
	+ start_supervised(%{
	+ id: Pleroma.Web.Streamer.registry(),
	+ start:
	+ {Registry, :start_link, [[keys: :duplicate, name: Pleroma.Web.Streamer.registry()]]}
	+ })
	end

	{:ok, conn: Phoenix.ConnTest.build_conn()}
	end
	end
	diff --git a/test/support/data_case.ex b/test/support/data_case.ex
	index 1669f2520..ba8848952 100644
	--- a/test/support/data_case.ex
	+++ b/test/support/data_case.ex
	@@ -1,80 +1,84 @@
	# Pleroma: A lightweight social networking server
	# Copyright © 2017-2020 Pleroma Authors <https://pleroma.social/>
	# SPDX-License-Identifier: AGPL-3.0-only

	defmodule Pleroma.DataCase do
	@moduledoc """
	This module defines the setup for tests requiring
	access to the application's data layer.

	You may define functions here to be used as helpers in
	your tests.

	Finally, if the test case interacts with the database,
	it cannot be async. For this reason, every test runs
	inside a transaction which is reset at the beginning
	of the test unless the test case is marked as async.
	"""

	use ExUnit.CaseTemplate

	using do
	quote do
	alias Pleroma.Repo

	import Ecto
	import Ecto.Changeset
	import Ecto.Query
	import Pleroma.DataCase
	use Pleroma.Tests.Helpers
	end
	end

	setup tags do
	Cachex.clear(:user_cache)
	Cachex.clear(:object_cache)
	:ok = Ecto.Adapters.SQL.Sandbox.checkout(Pleroma.Repo)

	unless tags[:async] do
	Ecto.Adapters.SQL.Sandbox.mode(Pleroma.Repo, {:shared, self()})
	end

	if tags[:needs_streamer] do
	- start_supervised(Pleroma.Web.Streamer.supervisor())
	+ start_supervised(%{
	+ id: Pleroma.Web.Streamer.registry(),
	+ start:
	+ {Registry, :start_link, [[keys: :duplicate, name: Pleroma.Web.Streamer.registry()]]}
	+ })
	end

	:ok
	end

	def ensure_local_uploader(context) do
	test_uploader = Map.get(context, :uploader, Pleroma.Uploaders.Local)
	uploader = Pleroma.Config.get([Pleroma.Upload, :uploader])
	filters = Pleroma.Config.get([Pleroma.Upload, :filters])

	Pleroma.Config.put([Pleroma.Upload, :uploader], test_uploader)
	Pleroma.Config.put([Pleroma.Upload, :filters], [])

	on_exit(fn ->
	Pleroma.Config.put([Pleroma.Upload, :uploader], uploader)
	Pleroma.Config.put([Pleroma.Upload, :filters], filters)
	end)

	:ok
	end

	@doc """
	A helper that transform changeset errors to a map of messages.

	changeset = Accounts.create_user(%{password: "short"})
	assert "password is too short" in errors_on(changeset).password
	assert %{password: ["password is too short"]} = errors_on(changeset)

	"""
	def errors_on(changeset) do
	Ecto.Changeset.traverse_errors(changeset, fn {message, opts} ->
	Enum.reduce(opts, message, fn {key, value}, acc ->
	String.replace(acc, "%{#{key}}", to_string(value))
	end)
	end)
	end
	end
	diff --git a/test/web/streamer/ping_test.exs b/test/web/streamer/ping_test.exs
	deleted file mode 100644
	index 5df6c1cc3..000000000
	--- a/test/web/streamer/ping_test.exs
	+++ /dev/null
	@@ -1,36 +0,0 @@
	-# Pleroma: A lightweight social networking server
	-# Copyright © 2017-2020 Pleroma Authors <https://pleroma.social/>
	-# SPDX-License-Identifier: AGPL-3.0-only
	-
	-defmodule Pleroma.Web.PingTest do
	- use Pleroma.DataCase
	-
	- import Pleroma.Factory
	- alias Pleroma.Web.Streamer
	-
	- setup do
	- start_supervised({Streamer.supervisor(), [ping_interval: 30]})
	-
	- :ok
	- end
	-
	- describe "sockets" do
	- setup do
	- user = insert(:user)
	- {:ok, %{user: user}}
	- end
	-
	- test "it sends pings", %{user: user} do
	- task =
	- Task.async(fn ->
	- assert_receive {:text, received_event}, 40
	- assert_receive {:text, received_event}, 40
	- assert_receive {:text, received_event}, 40
	- end)
	-
	- Streamer.add_socket("public", %{transport_pid: task.pid, assigns: %{user: user}})
	-
	- Task.await(task)
	- end
	- end
	-end
	diff --git a/test/web/streamer/state_test.exs b/test/web/streamer/state_test.exs
	deleted file mode 100644
	index a755e75c0..000000000
	--- a/test/web/streamer/state_test.exs
	+++ /dev/null
	@@ -1,54 +0,0 @@
	-# Pleroma: A lightweight social networking server
	-# Copyright © 2017-2020 Pleroma Authors <https://pleroma.social/>
	-# SPDX-License-Identifier: AGPL-3.0-only
	-
	-defmodule Pleroma.Web.StateTest do
	- use Pleroma.DataCase
	-
	- import Pleroma.Factory
	- alias Pleroma.Web.Streamer
	- alias Pleroma.Web.Streamer.StreamerSocket
	-
	- @moduletag needs_streamer: true
	-
	- describe "sockets" do
	- setup do
	- user = insert(:user)
	- user2 = insert(:user)
	- {:ok, %{user: user, user2: user2}}
	- end
	-
	- test "it can add a socket", %{user: user} do
	- Streamer.add_socket("public", %{transport_pid: 1, assigns: %{user: user}})
	-
	- assert(%{"public" => [%StreamerSocket{transport_pid: 1}]} = Streamer.get_sockets())
	- end
	-
	- test "it can add multiple sockets per user", %{user: user} do
	- Streamer.add_socket("public", %{transport_pid: 1, assigns: %{user: user}})
	- Streamer.add_socket("public", %{transport_pid: 2, assigns: %{user: user}})
	-
	- assert(
	- %{
	- "public" => [
	- %StreamerSocket{transport_pid: 2},
	- %StreamerSocket{transport_pid: 1}
	- ]
	- } = Streamer.get_sockets()
	- )
	- end
	-
	- test "it will not add a duplicate socket", %{user: user} do
	- Streamer.add_socket("activity", %{transport_pid: 1, assigns: %{user: user}})
	- Streamer.add_socket("activity", %{transport_pid: 1, assigns: %{user: user}})
	-
	- assert(
	- %{
	- "activity" => [
	- %StreamerSocket{transport_pid: 1}
	- ]
	- } = Streamer.get_sockets()
	- )
	- end
	- end
	-end
	diff --git a/test/web/streamer/streamer_test.exs b/test/web/streamer/streamer_test.exs
	index 3c0f240f5..ee530f4e9 100644
	--- a/test/web/streamer/streamer_test.exs
	+++ b/test/web/streamer/streamer_test.exs
	@@ -1,670 +1,498 @@
	# Pleroma: A lightweight social networking server
	# Copyright © 2017-2020 Pleroma Authors <https://pleroma.social/>
	# SPDX-License-Identifier: AGPL-3.0-only

	defmodule Pleroma.Web.StreamerTest do
	use Pleroma.DataCase

	import Pleroma.Factory

	alias Pleroma.Conversation.Participation
	alias Pleroma.List
	alias Pleroma.User
	alias Pleroma.Web.CommonAPI
	alias Pleroma.Web.Streamer
	- alias Pleroma.Web.Streamer.StreamerSocket
	- alias Pleroma.Web.Streamer.Worker

	@moduletag needs_streamer: true, capture_log: true

	- @streamer_timeout 150
	- @streamer_start_wait 10
	setup do: clear_config([:instance, :skip_thread_containment])

	describe "user streams" do
	setup do
	user = insert(:user)
	notify = insert(:notification, user: user, activity: build(:note_activity))
	{:ok, %{user: user, notify: notify}}
	end

	test "it streams the user's post in the 'user' stream", %{user: user} do
	- task =
	- Task.async(fn ->
	- assert_receive {:text, _}, @streamer_timeout
	- end)
	-
	- Streamer.add_socket(
	- "user",
	- %{transport_pid: task.pid, assigns: %{user: user}}
	- )
	-
	+ Streamer.add_socket("user", user)
	{:ok, activity} = CommonAPI.post(user, %{"status" => "hey"})
	-
	- Streamer.stream("user", activity)
	- Task.await(task)
	+ assert_receive {:render_with_user, _, _, ^activity}
	+ refute Streamer.filtered_by_user?(user, activity)
	end

	test "it streams boosts of the user in the 'user' stream", %{user: user} do
	- task =
	- Task.async(fn ->
	- assert_receive {:text, _}, @streamer_timeout
	- end)
	-
	- Streamer.add_socket(
	- "user",
	- %{transport_pid: task.pid, assigns: %{user: user}}
	- )
	+ Streamer.add_socket("user", user)

	other_user = insert(:user)
	{:ok, activity} = CommonAPI.post(other_user, %{"status" => "hey"})
	{:ok, announce, _} = CommonAPI.repeat(activity.id, user)

	- Streamer.stream("user", announce)
	- Task.await(task)
	+ assert_receive {:render_with_user, Pleroma.Web.StreamerView, "update.json", ^announce}
	+ refute Streamer.filtered_by_user?(user, announce)
	end

	test "it sends notify to in the 'user' stream", %{user: user, notify: notify} do
	- task =
	- Task.async(fn ->
	- assert_receive {:text, _}, @streamer_timeout
	- end)
	-
	- Streamer.add_socket(
	- "user",
	- %{transport_pid: task.pid, assigns: %{user: user}}
	- )
	-
	+ Streamer.add_socket("user", user)
	Streamer.stream("user", notify)
	- Task.await(task)
	+ assert_receive {:render_with_user, _, _, ^notify}
	+ refute Streamer.filtered_by_user?(user, notify)
	end

	test "it sends notify to in the 'user:notification' stream", %{user: user, notify: notify} do
	- task =
	- Task.async(fn ->
	- assert_receive {:text, _}, @streamer_timeout
	- end)
	-
	- Streamer.add_socket(
	- "user:notification",
	- %{transport_pid: task.pid, assigns: %{user: user}}
	- )
	-
	+ Streamer.add_socket("user:notification", user)
	Streamer.stream("user:notification", notify)
	- Task.await(task)
	+ assert_receive {:render_with_user, _, _, ^notify}
	+ refute Streamer.filtered_by_user?(user, notify)
	end

	test "it doesn't send notify to the 'user:notification' stream when a user is blocked", %{
	user: user
	} do
	blocked = insert(:user)
	{:ok, _user_relationship} = User.block(user, blocked)

	- task = Task.async(fn -> refute_receive {:text, _}, @streamer_timeout end)
	-
	- Streamer.add_socket(
	- "user:notification",
	- %{transport_pid: task.pid, assigns: %{user: user}}
	- )
	+ Streamer.add_socket("user:notification", user)

	{:ok, activity} = CommonAPI.post(user, %{"status" => ":("})
	- {:ok, notif} = CommonAPI.favorite(blocked, activity.id)
	+ {:ok, _} = CommonAPI.favorite(blocked, activity.id)

	- Streamer.stream("user:notification", notif)
	- Task.await(task)
	+ refute_receive _
	end

	test "it doesn't send notify to the 'user:notification' stream when a thread is muted", %{
	user: user
	} do
	user2 = insert(:user)

	- task = Task.async(fn -> refute_receive {:text, _}, @streamer_timeout end)
	+ {:ok, activity} = CommonAPI.post(user, %{"status" => "super hot take"})
	+ {:ok, _} = CommonAPI.add_mute(user, activity)

	- Streamer.add_socket(
	- "user:notification",
	- %{transport_pid: task.pid, assigns: %{user: user}}
	- )
	+ Streamer.add_socket("user:notification", user)

	- {:ok, activity} = CommonAPI.post(user, %{"status" => "super hot take"})
	- {:ok, activity} = CommonAPI.add_mute(user, activity)
	- {:ok, notif} = CommonAPI.favorite(user2, activity.id)
	+ {:ok, favorite_activity} = CommonAPI.favorite(user2, activity.id)

	- Streamer.stream("user:notification", notif)
	- Task.await(task)
	+ refute_receive _
	+ assert Streamer.filtered_by_user?(user, favorite_activity)
	end

	- test "it doesn't send notify to the 'user:notification' stream' when a domain is blocked", %{
	+ test "it sends favorite to 'user:notification' stream'", %{
	user: user
	} do
	user2 = insert(:user, %{ap_id: "https://hecking-lewd-place.com/user/meanie"})

	- task = Task.async(fn -> refute_receive {:text, _}, @streamer_timeout end)
	+ {:ok, activity} = CommonAPI.post(user, %{"status" => "super hot take"})
	+ Streamer.add_socket("user:notification", user)
	+ {:ok, favorite_activity} = CommonAPI.favorite(user2, activity.id)
	+
	+ assert_receive {:render_with_user, _, "notification.json", notif}
	+ assert notif.activity.id == favorite_activity.id
	+ refute Streamer.filtered_by_user?(user, notif)
	+ end

	- Streamer.add_socket(
	- "user:notification",
	- %{transport_pid: task.pid, assigns: %{user: user}}
	- )
	+ test "it doesn't send the 'user:notification' stream' when a domain is blocked", %{
	+ user: user
	+ } do
	+ user2 = insert(:user, %{ap_id: "https://hecking-lewd-place.com/user/meanie"})

	{:ok, user} = User.block_domain(user, "hecking-lewd-place.com")
	{:ok, activity} = CommonAPI.post(user, %{"status" => "super hot take"})
	- {:ok, notif} = CommonAPI.favorite(user2, activity.id)
	+ Streamer.add_socket("user:notification", user)
	+ {:ok, favorite_activity} = CommonAPI.favorite(user2, activity.id)

	- Streamer.stream("user:notification", notif)
	- Task.await(task)
	+ refute_receive _
	+ assert Streamer.filtered_by_user?(user, favorite_activity)
	end

	test "it sends follow activities to the 'user:notification' stream", %{
	user: user
	} do
	user_url = user.ap_id
	+ user2 = insert(:user)

	body =
	File.read!("test/fixtures/users_mock/localhost.json")
	\|> String.replace("{{nickname}}", user.nickname)
	\|> Jason.encode!()

	Tesla.Mock.mock_global(fn
	%{method: :get, url: ^user_url} ->
	%Tesla.Env{status: 200, body: body}
	end)

	- user2 = insert(:user)
	- task = Task.async(fn -> assert_receive {:text, _}, @streamer_timeout end)
	-
	- Process.sleep(@streamer_start_wait)
	-
	- Streamer.add_socket(
	- "user:notification",
	- %{transport_pid: task.pid, assigns: %{user: user}}
	- )
	+ Streamer.add_socket("user:notification", user)
	+ {:ok, _follower, _followed, follow_activity} = CommonAPI.follow(user2, user)

	- {:ok, _follower, _followed, _activity} = CommonAPI.follow(user2, user)
	-
	- # We don't directly pipe the notification to the streamer as it's already
	- # generated as a side effect of CommonAPI.follow().
	- Task.await(task)
	+ assert_receive {:render_with_user, _, "notification.json", notif}
	+ assert notif.activity.id == follow_activity.id
	+ refute Streamer.filtered_by_user?(user, notif)
	end
	end

	- test "it sends to public" do
	+ test "it sends to public authenticated" do
	user = insert(:user)
	other_user = insert(:user)

	- task =
	- Task.async(fn ->
	- assert_receive {:text, _}, @streamer_timeout
	- end)
	-
	- fake_socket = %StreamerSocket{
	- transport_pid: task.pid,
	- user: user
	- }
	-
	- {:ok, activity} = CommonAPI.post(other_user, %{"status" => "Test"})
	+ Streamer.add_socket("public", other_user)

	- topics = %{
	- "public" => [fake_socket]
	- }
	-
	- Worker.push_to_socket(topics, "public", activity)
	-
	- Task.await(task)
	+ {:ok, activity} = CommonAPI.post(user, %{"status" => "Test"})
	+ assert_receive {:render_with_user, _, _, ^activity}
	+ refute Streamer.filtered_by_user?(user, activity)
	end

	test "works for deletions" do
	user = insert(:user)
	other_user = insert(:user)
	{:ok, activity} = CommonAPI.post(other_user, %{"status" => "Test"})

	- task =
	- Task.async(fn ->
	- expected_event =
	- %{
	- "event" => "delete",
	- "payload" => activity.id
	- }
	- \|> Jason.encode!()
	-
	- assert_receive {:text, received_event}, @streamer_timeout
	- assert received_event == expected_event
	- end)
	+ Streamer.add_socket("public", user)

	- fake_socket = %StreamerSocket{
	- transport_pid: task.pid,
	- user: user
	- }
	+ {:ok, _} = CommonAPI.delete(activity.id, other_user)
	+ activity_id = activity.id
	+ assert_receive {:text, event}
	+ assert %{"event" => "delete", "payload" => ^activity_id} = Jason.decode!(event)
	+ end

	- {:ok, activity} = CommonAPI.delete(activity.id, other_user)
	+ test "it sends to public unauthenticated" do
	+ user = insert(:user)

	- topics = %{
	- "public" => [fake_socket]
	- }
	+ Streamer.add_socket("public", nil)

	- Worker.push_to_socket(topics, "public", activity)
	+ {:ok, activity} = CommonAPI.post(user, %{"status" => "Test"})
	+ activity_id = activity.id
	+ assert_receive {:text, event}
	+ assert %{"event" => "update", "payload" => payload} = Jason.decode!(event)
	+ assert %{"id" => ^activity_id} = Jason.decode!(payload)

	- Task.await(task)
	+ {:ok, _} = CommonAPI.delete(activity.id, user)
	+ assert_receive {:text, event}
	+ assert %{"event" => "delete", "payload" => ^activity_id} = Jason.decode!(event)
	end

	describe "thread_containment" do
	- test "it doesn't send to user if recipients invalid and thread containment is enabled" do
	+ test "it filters to user if recipients invalid and thread containment is enabled" do
	Pleroma.Config.put([:instance, :skip_thread_containment], false)
	author = insert(:user)
	user = insert(:user)
	User.follow(user, author, :follow_accept)

	activity =
	insert(:note_activity,
	note:
	insert(:note,
	user: author,
	data: %{"to" => ["TEST-FFF"]}
	)
	)

	- task = Task.async(fn -> refute_receive {:text, _}, 1_000 end)
	- fake_socket = %StreamerSocket{transport_pid: task.pid, user: user}
	- topics = %{"public" => [fake_socket]}
	- Worker.push_to_socket(topics, "public", activity)
	-
	- Task.await(task)
	+ Streamer.add_socket("public", user)
	+ Streamer.stream("public", activity)
	+ assert_receive {:render_with_user, _, _, ^activity}
	+ assert Streamer.filtered_by_user?(user, activity)
	end

	test "it sends message if recipients invalid and thread containment is disabled" do
	Pleroma.Config.put([:instance, :skip_thread_containment], true)
	author = insert(:user)
	user = insert(:user)
	User.follow(user, author, :follow_accept)

	activity =
	insert(:note_activity,
	note:
	insert(:note,
	user: author,
	data: %{"to" => ["TEST-FFF"]}
	)
	)

	- task = Task.async(fn -> assert_receive {:text, _}, 1_000 end)
	- fake_socket = %StreamerSocket{transport_pid: task.pid, user: user}
	- topics = %{"public" => [fake_socket]}
	- Worker.push_to_socket(topics, "public", activity)
	+ Streamer.add_socket("public", user)
	+ Streamer.stream("public", activity)

	- Task.await(task)
	+ assert_receive {:render_with_user, _, _, ^activity}
	+ refute Streamer.filtered_by_user?(user, activity)
	end

	test "it sends message if recipients invalid and thread containment is enabled but user's thread containment is disabled" do
	Pleroma.Config.put([:instance, :skip_thread_containment], false)
	author = insert(:user)
	user = insert(:user, skip_thread_containment: true)
	User.follow(user, author, :follow_accept)

	activity =
	insert(:note_activity,
	note:
	insert(:note,
	user: author,
	data: %{"to" => ["TEST-FFF"]}
	)
	)

	- task = Task.async(fn -> assert_receive {:text, _}, 1_000 end)
	- fake_socket = %StreamerSocket{transport_pid: task.pid, user: user}
	- topics = %{"public" => [fake_socket]}
	- Worker.push_to_socket(topics, "public", activity)
	+ Streamer.add_socket("public", user)
	+ Streamer.stream("public", activity)

	- Task.await(task)
	+ assert_receive {:render_with_user, _, _, ^activity}
	+ refute Streamer.filtered_by_user?(user, activity)
	end
	end

	describe "blocks" do
	- test "it doesn't send messages involving blocked users" do
	+ test "it filters messages involving blocked users" do
	user = insert(:user)
	blocked_user = insert(:user)
	{:ok, _user_relationship} = User.block(user, blocked_user)

	+ Streamer.add_socket("public", user)
	{:ok, activity} = CommonAPI.post(blocked_user, %{"status" => "Test"})
	-
	- task =
	- Task.async(fn ->
	- refute_receive {:text, _}, 1_000
	- end)
	-
	- fake_socket = %StreamerSocket{
	- transport_pid: task.pid,
	- user: user
	- }
	-
	- topics = %{
	- "public" => [fake_socket]
	- }
	-
	- Worker.push_to_socket(topics, "public", activity)
	-
	- Task.await(task)
	+ assert_receive {:render_with_user, _, _, ^activity}
	+ assert Streamer.filtered_by_user?(user, activity)
	end

	- test "it doesn't send messages transitively involving blocked users" do
	+ test "it filters messages transitively involving blocked users" do
	blocker = insert(:user)
	blockee = insert(:user)
	friend = insert(:user)

	- task =
	- Task.async(fn ->
	- refute_receive {:text, _}, 1_000
	- end)
	-
	- fake_socket = %StreamerSocket{
	- transport_pid: task.pid,
	- user: blocker
	- }
	-
	- topics = %{
	- "public" => [fake_socket]
	- }
	+ Streamer.add_socket("public", blocker)

	{:ok, _user_relationship} = User.block(blocker, blockee)

	{:ok, activity_one} = CommonAPI.post(friend, %{"status" => "hey! @#{blockee.nickname}"})

	- Worker.push_to_socket(topics, "public", activity_one)
	+ assert_receive {:render_with_user, _, _, ^activity_one}
	+ assert Streamer.filtered_by_user?(blocker, activity_one)

	{:ok, activity_two} = CommonAPI.post(blockee, %{"status" => "hey! @#{friend.nickname}"})

	- Worker.push_to_socket(topics, "public", activity_two)
	+ assert_receive {:render_with_user, _, _, ^activity_two}
	+ assert Streamer.filtered_by_user?(blocker, activity_two)

	{:ok, activity_three} = CommonAPI.post(blockee, %{"status" => "hey! @#{blocker.nickname}"})

	- Worker.push_to_socket(topics, "public", activity_three)
	-
	- Task.await(task)
	+ assert_receive {:render_with_user, _, _, ^activity_three}
	+ assert Streamer.filtered_by_user?(blocker, activity_three)
	end
	end

	- test "it doesn't send unwanted DMs to list" do
	- user_a = insert(:user)
	- user_b = insert(:user)
	- user_c = insert(:user)
	-
	- {:ok, user_a} = User.follow(user_a, user_b)
	-
	- {:ok, list} = List.create("Test", user_a)
	- {:ok, list} = List.follow(list, user_b)
	-
	- {:ok, activity} =
	- CommonAPI.post(user_b, %{
	- "status" => "@#{user_c.nickname} Test",
	- "visibility" => "direct"
	- })
	-
	- task =
	- Task.async(fn ->
	- refute_receive {:text, _}, 1_000
	- end)
	-
	- fake_socket = %StreamerSocket{
	- transport_pid: task.pid,
	- user: user_a
	- }
	-
	- topics = %{
	- "list:#{list.id}" => [fake_socket]
	- }
	-
	- Worker.handle_call({:stream, "list", activity}, self(), topics)
	-
	- Task.await(task)
	- end
	-
	- test "it doesn't send unwanted private posts to list" do
	- user_a = insert(:user)
	- user_b = insert(:user)
	+ describe "lists" do
	+ test "it doesn't send unwanted DMs to list" do
	+ user_a = insert(:user)
	+ user_b = insert(:user)
	+ user_c = insert(:user)

	- {:ok, list} = List.create("Test", user_a)
	- {:ok, list} = List.follow(list, user_b)
	+ {:ok, user_a} = User.follow(user_a, user_b)

	- {:ok, activity} =
	- CommonAPI.post(user_b, %{
	- "status" => "Test",
	- "visibility" => "private"
	- })
	+ {:ok, list} = List.create("Test", user_a)
	+ {:ok, list} = List.follow(list, user_b)

	- task =
	- Task.async(fn ->
	- refute_receive {:text, _}, 1_000
	- end)
	+ Streamer.add_socket("list:#{list.id}", user_a)

	- fake_socket = %StreamerSocket{
	- transport_pid: task.pid,
	- user: user_a
	- }
	+ {:ok, _activity} =
	+ CommonAPI.post(user_b, %{
	+ "status" => "@#{user_c.nickname} Test",
	+ "visibility" => "direct"
	+ })

	- topics = %{
	- "list:#{list.id}" => [fake_socket]
	- }
	+ refute_receive _
	+ end

	- Worker.handle_call({:stream, "list", activity}, self(), topics)
	+ test "it doesn't send unwanted private posts to list" do
	+ user_a = insert(:user)
	+ user_b = insert(:user)

	- Task.await(task)
	- end
	+ {:ok, list} = List.create("Test", user_a)
	+ {:ok, list} = List.follow(list, user_b)

	- test "it sends wanted private posts to list" do
	- user_a = insert(:user)
	- user_b = insert(:user)
	+ Streamer.add_socket("list:#{list.id}", user_a)

	- {:ok, user_a} = User.follow(user_a, user_b)
	+ {:ok, _activity} =
	+ CommonAPI.post(user_b, %{
	+ "status" => "Test",
	+ "visibility" => "private"
	+ })

	- {:ok, list} = List.create("Test", user_a)
	- {:ok, list} = List.follow(list, user_b)
	+ refute_receive _
	+ end

	- {:ok, activity} =
	- CommonAPI.post(user_b, %{
	- "status" => "Test",
	- "visibility" => "private"
	- })
	+ test "it sends wanted private posts to list" do
	+ user_a = insert(:user)
	+ user_b = insert(:user)

	- task =
	- Task.async(fn ->
	- assert_receive {:text, _}, 1_000
	- end)
	+ {:ok, user_a} = User.follow(user_a, user_b)

	- fake_socket = %StreamerSocket{
	- transport_pid: task.pid,
	- user: user_a
	- }
	+ {:ok, list} = List.create("Test", user_a)
	+ {:ok, list} = List.follow(list, user_b)

	- Streamer.add_socket(
	- "list:#{list.id}",
	- fake_socket
	- )
	+ Streamer.add_socket("list:#{list.id}", user_a)

	- Worker.handle_call({:stream, "list", activity}, self(), %{})
	+ {:ok, activity} =
	+ CommonAPI.post(user_b, %{
	+ "status" => "Test",
	+ "visibility" => "private"
	+ })

	- Task.await(task)
	+ assert_receive {:render_with_user, _, _, ^activity}
	+ refute Streamer.filtered_by_user?(user_a, activity)
	+ end
	end

	- test "it doesn't send muted reblogs" do
	- user1 = insert(:user)
	- user2 = insert(:user)
	- user3 = insert(:user)
	- CommonAPI.hide_reblogs(user1, user2)
	-
	- {:ok, create_activity} = CommonAPI.post(user3, %{"status" => "I'm kawen"})
	- {:ok, announce_activity, _} = CommonAPI.repeat(create_activity.id, user2)
	-
	- task =
	- Task.async(fn ->
	- refute_receive {:text, _}, 1_000
	- end)
	-
	- fake_socket = %StreamerSocket{
	- transport_pid: task.pid,
	- user: user1
	- }
	-
	- topics = %{
	- "public" => [fake_socket]
	- }
	-
	- Worker.push_to_socket(topics, "public", announce_activity)
	+ describe "muted reblogs" do
	+ test "it filters muted reblogs" do
	+ user1 = insert(:user)
	+ user2 = insert(:user)
	+ user3 = insert(:user)
	+ CommonAPI.follow(user1, user2)
	+ CommonAPI.hide_reblogs(user1, user2)

	- Task.await(task)
	- end
	+ {:ok, create_activity} = CommonAPI.post(user3, %{"status" => "I'm kawen"})

	- test "it does send non-reblog notification for reblog-muted actors" do
	- user1 = insert(:user)
	- user2 = insert(:user)
	- user3 = insert(:user)
	- CommonAPI.hide_reblogs(user1, user2)
	+ Streamer.add_socket("user", user1)
	+ {:ok, announce_activity, _} = CommonAPI.repeat(create_activity.id, user2)
	+ assert_receive {:render_with_user, _, _, ^announce_activity}
	+ assert Streamer.filtered_by_user?(user1, announce_activity)
	+ end

	- {:ok, create_activity} = CommonAPI.post(user3, %{"status" => "I'm kawen"})
	- {:ok, favorite_activity} = CommonAPI.favorite(user2, create_activity.id)
	+ test "it filters reblog notification for reblog-muted actors" do
	+ user1 = insert(:user)
	+ user2 = insert(:user)
	+ CommonAPI.follow(user1, user2)
	+ CommonAPI.hide_reblogs(user1, user2)

	- task =
	- Task.async(fn ->
	- assert_receive {:text, _}, 1_000
	- end)
	+ {:ok, create_activity} = CommonAPI.post(user1, %{"status" => "I'm kawen"})
	+ Streamer.add_socket("user", user1)
	+ {:ok, _favorite_activity, _} = CommonAPI.repeat(create_activity.id, user2)

	- fake_socket = %StreamerSocket{
	- transport_pid: task.pid,
	- user: user1
	- }
	+ assert_receive {:render_with_user, _, "notification.json", notif}
	+ assert Streamer.filtered_by_user?(user1, notif)
	+ end

	- topics = %{
	- "public" => [fake_socket]
	- }
	+ test "it send non-reblog notification for reblog-muted actors" do
	+ user1 = insert(:user)
	+ user2 = insert(:user)
	+ CommonAPI.follow(user1, user2)
	+ CommonAPI.hide_reblogs(user1, user2)

	- Worker.push_to_socket(topics, "public", favorite_activity)
	+ {:ok, create_activity} = CommonAPI.post(user1, %{"status" => "I'm kawen"})
	+ Streamer.add_socket("user", user1)
	+ {:ok, _favorite_activity} = CommonAPI.favorite(user2, create_activity.id)

	- Task.await(task)
	+ assert_receive {:render_with_user, _, "notification.json", notif}
	+ refute Streamer.filtered_by_user?(user1, notif)
	+ end
	end

	- test "it doesn't send posts from muted threads" do
	+ test "it filters posts from muted threads" do
	user = insert(:user)
	user2 = insert(:user)
	+ Streamer.add_socket("user", user2)
	{:ok, user2, user, _activity} = CommonAPI.follow(user2, user)
	-
	{:ok, activity} = CommonAPI.post(user, %{"status" => "super hot take"})
	-
	- {:ok, activity} = CommonAPI.add_mute(user2, activity)
	-
	- task = Task.async(fn -> refute_receive {:text, _}, @streamer_timeout end)
	-
	- Streamer.add_socket(
	- "user",
	- %{transport_pid: task.pid, assigns: %{user: user2}}
	- )
	-
	- Streamer.stream("user", activity)
	- Task.await(task)
	+ {:ok, _} = CommonAPI.add_mute(user2, activity)
	+ assert_receive {:render_with_user, _, _, ^activity}
	+ assert Streamer.filtered_by_user?(user2, activity)
	end

	describe "direct streams" do
	setup do
	:ok
	end

	test "it sends conversation update to the 'direct' stream", %{} do
	user = insert(:user)
	another_user = insert(:user)

	- task =
	- Task.async(fn ->
	- assert_receive {:text, received_event}, @streamer_timeout
	-
	- assert %{"event" => "conversation", "payload" => received_payload} =
	- Jason.decode!(received_event)
	-
	- assert %{"last_status" => last_status} = Jason.decode!(received_payload)
	- [participation] = Participation.for_user(user)
	- assert last_status["pleroma"]["direct_conversation_id"] == participation.id
	- end)
	-
	- Streamer.add_socket(
	- "direct",
	- %{transport_pid: task.pid, assigns: %{user: user}}
	- )
	+ Streamer.add_socket("direct", user)

	{:ok, _create_activity} =
	CommonAPI.post(another_user, %{
	"status" => "hey @#{user.nickname}",
	"visibility" => "direct"
	})

	- Task.await(task)
	+ assert_receive {:text, received_event}
	+
	+ assert %{"event" => "conversation", "payload" => received_payload} =
	+ Jason.decode!(received_event)
	+
	+ assert %{"last_status" => last_status} = Jason.decode!(received_payload)
	+ [participation] = Participation.for_user(user)
	+ assert last_status["pleroma"]["direct_conversation_id"] == participation.id
	end

	test "it doesn't send conversation update to the 'direct' stream when the last message in the conversation is deleted" do
	user = insert(:user)
	another_user = insert(:user)

	+ Streamer.add_socket("direct", user)
	+
	{:ok, create_activity} =
	CommonAPI.post(another_user, %{
	"status" => "hi @#{user.nickname}",
	"visibility" => "direct"
	})

	- task =
	- Task.async(fn ->
	- assert_receive {:text, received_event}, @streamer_timeout
	- assert %{"event" => "delete", "payload" => _} = Jason.decode!(received_event)
	+ create_activity_id = create_activity.id
	+ assert_receive {:render_with_user, _, _, ^create_activity}
	+ assert_receive {:text, received_conversation1}
	+ assert %{"event" => "conversation", "payload" => _} = Jason.decode!(received_conversation1)

	- refute_receive {:text, _}, @streamer_timeout
	- end)
	+ {:ok, _} = CommonAPI.delete(create_activity_id, another_user)

	- Process.sleep(@streamer_start_wait)
	+ assert_receive {:text, received_event}

	- Streamer.add_socket(
	- "direct",
	- %{transport_pid: task.pid, assigns: %{user: user}}
	- )
	+ assert %{"event" => "delete", "payload" => ^create_activity_id} =
	+ Jason.decode!(received_event)

	- {:ok, _} = CommonAPI.delete(create_activity.id, another_user)
	-
	- Task.await(task)
	+ refute_receive _
	end

	test "it sends conversation update to the 'direct' stream when a message is deleted" do
	user = insert(:user)
	another_user = insert(:user)
	+ Streamer.add_socket("direct", user)

	{:ok, create_activity} =
	CommonAPI.post(another_user, %{
	"status" => "hi @#{user.nickname}",
	"visibility" => "direct"
	})

	{:ok, create_activity2} =
	CommonAPI.post(another_user, %{
	- "status" => "hi @#{user.nickname}",
	+ "status" => "hi @#{user.nickname} 2",
	"in_reply_to_status_id" => create_activity.id,
	"visibility" => "direct"
	})

	- task =
	- Task.async(fn ->
	- assert_receive {:text, received_event}, @streamer_timeout
	- assert %{"event" => "delete", "payload" => _} = Jason.decode!(received_event)
	-
	- assert_receive {:text, received_event}, @streamer_timeout
	+ assert_receive {:render_with_user, _, _, ^create_activity}
	+ assert_receive {:render_with_user, _, _, ^create_activity2}
	+ assert_receive {:text, received_conversation1}
	+ assert %{"event" => "conversation", "payload" => _} = Jason.decode!(received_conversation1)
	+ assert_receive {:text, received_conversation1}
	+ assert %{"event" => "conversation", "payload" => _} = Jason.decode!(received_conversation1)

	- assert %{"event" => "conversation", "payload" => received_payload} =
	- Jason.decode!(received_event)
	-
	- assert %{"last_status" => last_status} = Jason.decode!(received_payload)
	- assert last_status["id"] == to_string(create_activity.id)
	- end)
	+ {:ok, _} = CommonAPI.delete(create_activity2.id, another_user)

	- Process.sleep(@streamer_start_wait)
	+ assert_receive {:text, received_event}
	+ assert %{"event" => "delete", "payload" => _} = Jason.decode!(received_event)

	- Streamer.add_socket(
	- "direct",
	- %{transport_pid: task.pid, assigns: %{user: user}}
	- )
	+ assert_receive {:text, received_event}

	- {:ok, _} = CommonAPI.delete(create_activity2.id, another_user)
	+ assert %{"event" => "conversation", "payload" => received_payload} =
	+ Jason.decode!(received_event)

	- Task.await(task)
	+ assert %{"last_status" => last_status} = Jason.decode!(received_payload)
	+ assert last_status["id"] == to_string(create_activity.id)
	end
	end
	end

File Metadata

Mime Type: text/x-diff
Expires: Fri, Aug 28, 7:48 PM (20 h, 27 m)
Storage Engine: blob
Storage Format: Raw Data
Storage Handle: 1736621
Default Alt Text: (151 KB)

No OneTemporaryActions

View Options

File Metadata

Event Timeline

No OneTemporary
Actions