No OneTemporary
Actions

Size

138 KB

Referenced Files

None

Subscribers

None

View Options

	diff --git a/lib/pleroma/application.ex b/lib/pleroma/application.ex
	index 4999911e7..88a4bc65d 100644
	--- a/lib/pleroma/application.ex
	+++ b/lib/pleroma/application.ex
	@@ -1,180 +1,180 @@
	# Pleroma: A lightweight social networking server
	# Copyright © 2017-2019 Pleroma Authors <https://pleroma.social/>
	# SPDX-License-Identifier: AGPL-3.0-only

	defmodule Pleroma.Application do
	import Cachex.Spec
	use Application

	@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
	info = "#{Pleroma.Web.base_url()} <#{Pleroma.Config.get([:instance, :email], "")}>"
	named_version() <> "; " <> info
	end

	# See http://elixir-lang.org/docs/stable/elixir/Application.html
	# for more information on OTP Applications
	def start(_type, _args) do
	Pleroma.Config.DeprecationWarnings.warn()
	setup_instrumenters()

	# Define workers and child supervisors to be supervised
	children =
	[
	Pleroma.Repo,
	Pleroma.Config.TransferTask,
	Pleroma.Emoji,
	Pleroma.Captcha,
	Pleroma.FlakeId,
	Pleroma.ScheduledActivityWorker,
	Pleroma.ActivityExpirationWorker
	] ++
	cachex_children() ++
	hackney_pool_children() ++
	[
	Pleroma.Web.Federator.RetryQueue,
	Pleroma.Stats,
	%{
	id: :web_push_init,
	start: {Task, :start_link, [&Pleroma.Web.Push.init/0]},
	restart: :temporary
	},
	%{
	id: :federator_init,
	start: {Task, :start_link, [&Pleroma.Web.Federator.init/0]},
	restart: :temporary
	},
	%{
	id: :internal_fetch_init,
	start: {Task, :start_link, [&Pleroma.Web.ActivityPub.InternalFetchActor.init/0]},
	restart: :temporary
	}
	] ++
	oauth_cleanup_child(oauth_cleanup_enabled?()) ++
	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]
	result = Supervisor.start_link(children, opts)
	:ok = after_supervisor_start()
	result
	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

	def enabled_hackney_pools do
	[:media] ++
	if Application.get_env(:tesla, :adapter) == Tesla.Adapter.Hackney do
	[:federation]
	else
	[]
	end ++
	if Pleroma.Config.get([Pleroma.Upload, :proxy_remote]) do
	[:upload]
	else
	[]
	end
	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)
	]
	end

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

	defp seconds_valid_interval,
	do: :timer.seconds(Pleroma.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: Pleroma.Config.get([:chat, :enabled])

	defp oauth_cleanup_enabled?,
	do: Pleroma.Config.get([:oauth2, :clean_expired_tokens], false)

	defp streamer_child(:test), do: []

	defp streamer_child(_) do
	- [PleromaWeb.Streamer]
	+ [PleromaWeb.Streamer.supervisor()]
	end

	defp oauth_cleanup_child(true),
	do: [Pleroma.Web.OAuth.Token.CleanWorker]

	defp oauth_cleanup_child(_), do: []

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

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

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

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

	defp after_supervisor_start do
	with digest_config <- Application.get_env(:pleroma, :email_notifications)[:digest],
	true <- digest_config[:active] do
	PleromaJobQueue.schedule(
	digest_config[:schedule],
	:digest_emails,
	Pleroma.DigestEmailWorker
	)
	end

	:ok
	end
	end
	diff --git a/lib/pleroma/notification.ex b/lib/pleroma/notification.ex
	index 4f16962bd..046214a51 100644
	--- a/lib/pleroma/notification.ex
	+++ b/lib/pleroma/notification.ex
	@@ -1,304 +1,306 @@
	# Pleroma: A lightweight social networking server
	# Copyright © 2017-2019 Pleroma Authors <https://pleroma.social/>
	# SPDX-License-Identifier: AGPL-3.0-only

	defmodule Pleroma.Notification do
	use Ecto.Schema

	alias Pleroma.Activity
	alias Pleroma.Notification
	alias Pleroma.Object
	alias Pleroma.Pagination
	alias Pleroma.Repo
	alias Pleroma.User
	alias Pleroma.Web.CommonAPI.Utils
	alias Pleroma.Web.Push
	alias PleromaWeb.Streamer

	import Ecto.Query
	import Ecto.Changeset

	@type t :: %__MODULE__{}

	schema "notifications" do
	field(:seen, :boolean, default: false)
	belongs_to(:user, User, type: Pleroma.FlakeId)
	belongs_to(:activity, Activity, type: Pleroma.FlakeId)

	timestamps()
	end

	def changeset(%Notification{} = notification, attrs) do
	notification
	\|> cast(attrs, [:seen])
	end

	def for_user_query(user, opts \\ []) do
	query =
	Notification
	\|> where(user_id: ^user.id)
	\|> where(
	[n, a],
	fragment(
	"? not in (SELECT ap_id FROM users WHERE info->'deactivated' @> 'true')",
	a.actor
	)
	)
	\|> join(:inner, [n], activity in assoc(n, :activity))
	\|> join(:left, [n, a], object in Object,
	on:
	fragment(
	"(?->>'id') = COALESCE((? -> 'object'::text) ->> 'id'::text)",
	object.data,
	a.data
	)
	)
	\|> preload([n, a, o], activity: {a, object: o})

	if opts[:with_muted] do
	query
	else
	where(query, [n, a], a.actor not in ^user.info.muted_notifications)
	\|> where([n, a], a.actor not in ^user.info.blocks)
	\|> where(
	[n, a],
	fragment("substring(? from '.://([^/])')", a.actor) not in ^user.info.domain_blocks
	)
	\|> join(:left, [n, a], tm in Pleroma.ThreadMute,
	on: tm.user_id == ^user.id and tm.context == fragment("?->>'context'", a.data)
	)
	\|> where([n, a, o, tm], is_nil(tm.user_id))
	end
	end

	def for_user(user, opts \\ %{}) do
	user
	\|> for_user_query(opts)
	\|> Pagination.fetch_paginated(opts)
	end

	@doc """
	Returns notifications for user received since given date.

	## Examples

	iex> Pleroma.Notification.for_user_since(%Pleroma.User{}, ~N[2019-04-13 11:22:33])
	[%Pleroma.Notification{}, %Pleroma.Notification{}]

	iex> Pleroma.Notification.for_user_since(%Pleroma.User{}, ~N[2019-04-15 11:22:33])
	[]
	"""
	@spec for_user_since(Pleroma.User.t(), NaiveDateTime.t()) :: [t()]
	def for_user_since(user, date) do
	from(n in for_user_query(user),
	where: n.updated_at > ^date
	)
	\|> Repo.all()
	end

	def set_read_up_to(%{id: user_id} = _user, id) do
	query =
	from(
	n in Notification,
	where: n.user_id == ^user_id,
	where: n.id <= ^id,
	where: n.seen == false,
	update: [
	set: [
	seen: true,
	updated_at: ^NaiveDateTime.utc_now()
	]
	],
	# Ideally we would preload object and activities here
	# but Ecto does not support preloads in update_all
	select: n.id
	)

	{_, notification_ids} = Repo.update_all(query, [])

	Notification
	\|> where([n], n.id in ^notification_ids)
	\|> join(:inner, [n], activity in assoc(n, :activity))
	\|> join(:left, [n, a], object in Object,
	on:
	fragment(
	"(?->>'id') = COALESCE((? -> 'object'::text) ->> 'id'::text)",
	object.data,
	a.data
	)
	)
	\|> preload([n, a, o], activity: {a, object: o})
	\|> Repo.all()
	end

	def read_one(%User{} = user, notification_id) do
	with {:ok, %Notification{} = notification} <- get(user, notification_id) do
	notification
	\|> changeset(%{seen: true})
	\|> Repo.update()
	end
	end

	def get(%{id: user_id} = _user, id) do
	query =
	from(
	n in Notification,
	where: n.id == ^id,
	join: activity in assoc(n, :activity),
	preload: [activity: activity]
	)

	notification = Repo.one(query)

	case notification do
	%{user_id: ^user_id} ->
	{:ok, notification}

	_ ->
	{:error, "Cannot get notification"}
	end
	end

	def clear(user) do
	from(n in Notification, where: n.user_id == ^user.id)
	\|> Repo.delete_all()
	end

	def destroy_multiple(%{id: user_id} = _user, ids) do
	from(n in Notification,
	where: n.id in ^ids,
	where: n.user_id == ^user_id
	)
	\|> Repo.delete_all()
	end

	def dismiss(%{id: user_id} = _user, id) do
	notification = Repo.get(Notification, id)

	case notification do
	%{user_id: ^user_id} ->
	Repo.delete(notification)

	_ ->
	{:error, "Cannot dismiss notification"}
	end
	end

	def create_notifications(%Activity{data: %{"to" => _, "type" => "Create"}} = activity) do
	object = Object.normalize(activity)

	unless object && object.data["type"] == "Answer" do
	users = get_notified_from_activity(activity)
	notifications = Enum.map(users, fn user -> create_notification(activity, user) end)
	{:ok, notifications}
	else
	{:ok, []}
	end
	end

	def create_notifications(%Activity{data: %{"to" => _, "type" => type}} = activity)
	when type in ["Like", "Announce", "Follow"] do
	users = get_notified_from_activity(activity)
	notifications = Enum.map(users, fn user -> create_notification(activity, user) end)
	{:ok, notifications}
	end

	def create_notifications(_), do: {:ok, []}

	# TODO move to sql, too.
	def create_notification(%Activity{} = activity, %User{} = user) do
	unless skip?(activity, user) do
	notification = %Notification{user_id: user.id, activity: activity}
	{:ok, notification} = Repo.insert(notification)
	- Streamer.stream("user", notification)
	- Streamer.stream("user:notification", notification)
	+
	+ ["user", "user:notification"]
	+ \|> Streamer.stream(notification)
	+
	Push.send(notification)
	notification
	end
	end

	def get_notified_from_activity(activity, local_only \\ true)

	def get_notified_from_activity(
	%Activity{data: %{"to" => _, "type" => type} = _data} = activity,
	local_only
	)
	when type in ["Create", "Like", "Announce", "Follow"] do
	recipients =
	[]
	\|> Utils.maybe_notify_to_recipients(activity)
	\|> Utils.maybe_notify_mentioned_recipients(activity)
	\|> Utils.maybe_notify_subscribers(activity)
	\|> Enum.uniq()

	User.get_users_from_set(recipients, local_only)
	end

	def get_notified_from_activity(_, _local_only), do: []

	@spec skip?(Activity.t(), User.t()) :: boolean()
	def skip?(activity, user) do
	[
	:self,
	:followers,
	:follows,
	:non_followers,
	:non_follows,
	:recently_followed
	]
	\|> Enum.any?(&skip?(&1, activity, user))
	end

	@spec skip?(atom(), Activity.t(), User.t()) :: boolean()
	def skip?(:self, activity, user) do
	activity.data["actor"] == user.ap_id
	end

	def skip?(
	:followers,
	activity,
	%{info: %{notification_settings: %{"followers" => false}}} = user
	) do
	actor = activity.data["actor"]
	follower = User.get_cached_by_ap_id(actor)
	User.following?(follower, user)
	end

	def skip?(
	:non_followers,
	activity,
	%{info: %{notification_settings: %{"non_followers" => false}}} = user
	) do
	actor = activity.data["actor"]
	follower = User.get_cached_by_ap_id(actor)
	!User.following?(follower, user)
	end

	def skip?(:follows, activity, %{info: %{notification_settings: %{"follows" => false}}} = user) do
	actor = activity.data["actor"]
	followed = User.get_cached_by_ap_id(actor)
	User.following?(user, followed)
	end

	def skip?(
	:non_follows,
	activity,
	%{info: %{notification_settings: %{"non_follows" => false}}} = user
	) do
	actor = activity.data["actor"]
	followed = User.get_cached_by_ap_id(actor)
	!User.following?(user, followed)
	end

	def skip?(:recently_followed, %{data: %{"type" => "Follow"}} = activity, user) do
	actor = activity.data["actor"]

	Notification.for_user(user)
	\|> Enum.any?(fn
	%{activity: %{data: %{"type" => "Follow", "actor" => ^actor}}} -> true
	_ -> false
	end)
	end

	def skip?(_, _, _), do: false
	end
	diff --git a/lib/pleroma/web/activity_pub/activity_pub.ex b/lib/pleroma/web/activity_pub/activity_pub.ex
	index 9e3ccfd38..edd91e46e 100644
	--- a/lib/pleroma/web/activity_pub/activity_pub.ex
	+++ b/lib/pleroma/web/activity_pub/activity_pub.ex
	@@ -1,1156 +1,1155 @@
	# Pleroma: A lightweight social networking server
	# Copyright © 2017-2019 Pleroma Authors <https://pleroma.social/>
	# SPDX-License-Identifier: AGPL-3.0-only

	defmodule Pleroma.Web.ActivityPub.ActivityPub do
	alias Pleroma.Activity
	alias Pleroma.Config
	alias Pleroma.Conversation
	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.Topics
	alias Pleroma.Web.ActivityPub.Transmogrifier
	alias Pleroma.Web.WebFinger
	alias PleromaWeb.Streamer

	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.info.deactivated do
	true
	else
	_e -> false
	end
	else
	true
	end
	end

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

	defp check_remote_limit(_), do: true

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

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

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

	def increase_replies_count_if_reply(_create_data), do: :noop

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

	def decrease_replies_count_if_reply(_object), do: :noop

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

	def increase_poll_votes_if_vote(_create_data), do: :noop

	def insert(map, local \\ true, fake \\ false, bypass_actor_check \\ false) when is_map(map) do
	with nil <- Activity.normalize(map),
	map <- lazy_put_activity_defaults(map, fake),
	true <- bypass_actor_check \|\| check_actor_is_active(map["actor"]),
	{_, true} <- {:remote_limit_error, check_remote_limit(map)},
	{:ok, map} <- MRF.filter(map),
	{recipients, _, _} = get_recipients(map),
	{:fake, false, map, recipients} <- {:fake, fake, map, recipients},
	:ok <- 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

	PleromaJobQueue.enqueue(:background, Pleroma.Web.RichMedia.Helpers, [:fetch, activity])

	Notification.create_notifications(activity)

	participations =
	activity
	\|> Conversation.create_or_bump_for()
	\|> get_participations()

	stream_out(activity)
	stream_out_participations(participations)
	{:ok, activity}
	else
	%Activity{} = activity ->
	{:ok, activity}

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

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

	error ->
	{:error, error}
	end
	end

	defp get_participations({:ok, %{participations: participations}}), do: participations
	defp get_participations(_), do: []

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

	Streamer.stream("participation", participations)
	end

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

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

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

	def stream_out(_activity) do
	:noop
	end

	def create(%{to: to, actor: actor, context: context, object: object} = params, fake \\ false) do
	additional = params[:additional] \|\| %{}
	# only accept false as false value
	local = !(params[:local] == false)
	published = params[:published]

	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),
	# Changing note count prior to enqueuing federation task in order to avoid
	# race conditions on updating user.info
	{:ok, _actor} <- increase_note_count_if_public(actor, activity),
	:ok <- maybe_federate(activity) do
	{:ok, activity}
	else
	{:fake, true, activity} ->
	{:ok, activity}

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

	def accept(%{to: to, actor: actor, object: object} = params) do
	# only accept false as false value
	local = !(params[:local] == false)

	with data <- %{"to" => to, "type" => "Accept", "actor" => actor.ap_id, "object" => object},
	{:ok, activity} <- insert(data, local),
	:ok <- maybe_federate(activity) do
	{:ok, activity}
	end
	end

	def reject(%{to: to, actor: actor, object: object} = params) do
	# only accept false as false value
	local = !(params[:local] == false)

	with data <- %{"to" => to, "type" => "Reject", "actor" => actor.ap_id, "object" => object},
	{:ok, activity} <- insert(data, local),
	:ok <- maybe_federate(activity) do
	{:ok, activity}
	end
	end

	def update(%{to: to, cc: cc, actor: actor, object: object} = params) do
	# only accept false as false value
	local = !(params[:local] == false)

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

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

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

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

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

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

	def unfollow(follower, followed, activity_id \\ nil, local \\ true) do
	with %Activity{} = follow_activity <- fetch_latest_follow(follower, followed),
	{:ok, follow_activity} <- update_follow_state(follow_activity, "cancelled"),
	unfollow_data <- make_unfollow_data(follower, followed, follow_activity, activity_id),
	{:ok, activity} <- insert(unfollow_data, local),
	:ok <- maybe_federate(activity) do
	{:ok, activity}
	end
	end

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

	def delete(%Object{data: %{"id" => id, "actor" => actor}} = object, local \\ true) do
	user = User.get_cached_by_ap_id(actor)
	to = (object.data["to"] \|\| []) ++ (object.data["cc"] \|\| [])

	with {:ok, object, activity} <- Object.delete(object),
	data <- %{
	"type" => "Delete",
	"actor" => actor,
	"object" => id,
	"to" => to,
	"deleted_activity_id" => activity && activity.id
	},
	{:ok, activity} <- insert(data, local, false),
	stream_out_participations(object, user),
	_ <- decrease_replies_count_if_reply(object),
	# Changing note count prior to enqueuing federation task in order to avoid
	# race conditions on updating user.info
	{:ok, _actor} <- decrease_note_count_if_public(user, object),
	:ok <- maybe_federate(activity) do
	{:ok, activity}
	end
	end

	def block(blocker, blocked, activity_id \\ nil, local \\ true) do
	outgoing_blocks = Config.get([:activitypub, :outgoing_blocks])
	unfollow_blocked = Config.get([:activitypub, :unfollow_blocked])

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

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

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

	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] \|\| %{}

	params = %{
	actor: actor,
	context: context,
	account: account,
	statuses: statuses,
	content: content
	}

	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 <- maybe_federate(activity) do
	Enum.each(User.all_superusers(), fn superuser ->
	superuser
	\|> Pleroma.Emails.AdminEmail.report(actor, account, statuses, content)
	\|> Pleroma.Emails.Mailer.deliver_async()
	end)

	{:ok, activity}
	end
	end

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

	recipients =
	if opts["user"], do: [opts["user"].ap_id \| opts["user"].following] ++ 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()) ::
	Pleroma.FlakeId.t() \| nil
	def fetch_latest_activity_id_for_context(context, opts \\ %{}) do
	context
	\|> fetch_activities_for_context_query(Map.merge(%{"skip_preload" => true}, opts))
	\|> limit(1)
	\|> select([a], a.id)
	\|> Repo.one()
	end

	def fetch_public_activities(opts \\ %{}) do
	q = fetch_activities_query([Pleroma.Constants.as_public()], opts)

	q
	\|> restrict_unlisted()
	\|> Pagination.fetch_paginated(opts)
	\|> Enum.reverse()
	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 restrict_thread_visibility(query, _, %{skip_thread_containment: true} = _),
	do: query

	defp restrict_thread_visibility(
	query,
	%{"user" => %User{info: %{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_activities(user, reading_user, params \\ %{}) do
	params =
	params
	\|> Map.put("type", ["Create", "Announce"])
	\|> Map.put("user", reading_user)
	\|> Map.put("actor_id", user.ap_id)
	\|> Map.put("whole_db", true)
	\|> Map.put("pinned_activity_ids", user.info.pinned_activities)

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

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

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

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

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

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

	defp restrict_since(query, _), do: query

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

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

	defp restrict_tag_reject(query, _), do: query

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

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

	defp restrict_tag_all(query, _), do: query

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

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

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

	defp restrict_tag(query, _), do: query

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

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

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

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

	defp restrict_local(query, _), do: query

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

	defp restrict_actor(query, _), do: query

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

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

	defp restrict_type(query, _), do: query

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

	defp restrict_state(query, _), do: query

	defp restrict_favorited_by(query, %{"favorited_by" => ap_id}) do
	from(
	[_activity, object] in query,
	where: fragment("(?)->'likes' \\? (?)", object.data, ^ap_id)
	)
	end

	defp restrict_favorited_by(query, _), do: query

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

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

	defp restrict_media(query, _), do: query

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

	defp restrict_replies(query, _), do: query

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

	defp restrict_reblogs(query, _), do: query

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

	defp restrict_muted(query, %{"muting_user" => %User{info: info}} = opts) do
	mutes = info.mutes

	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{info: info}}) do
	blocks = info.blocks \|\| []
	domain_blocks = info.domain_blocks \|\| []

	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, ^blocks),
	where: fragment("not (? && ?)", activity.recipients, ^blocks),
	where:
	fragment(
	"not (?->>'type' = 'Announce' and ?->'to' \\?\| ?)",
	activity.data,
	activity.data,
	^blocks
	),
	where: fragment("not (split_part(?, '/', 3) = ANY(?))", activity.actor, ^domain_blocks),
	where: fragment("not (split_part(?->>'actor', '/', 3) = ANY(?))", o.data, ^domain_blocks)
	)
	end

	defp restrict_blocked(query, _), do: query

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

	defp restrict_pinned(query, %{"pinned" => "true", "pinned_activity_ids" => ids}) do
	from(activity in query, where: activity.id in ^ids)
	end

	defp restrict_pinned(query, _), do: query

	defp restrict_muted_reblogs(query, %{"muting_user" => %User{info: info}}) do
	muted_reblogs = info.muted_reblogs \|\| []

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

	def fetch_activities_query(recipients, opts \\ %{}) do
	config = %{
	skip_thread_containment: Config.get([:instance, :skip_thread_containment])
	}

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

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

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

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

	activity ->
	activity
	end)
	end

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

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

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

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

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

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

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

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

	locked = data["manuallyApprovesFollowers"] \|\| false
	data = Transmogrifier.maybe_fix_user_object(data)

	user_data = %{
	ap_id: data["id"],
	info: %{
	ap_enabled: true,
	source_data: data,
	banner: banner,
	fields: fields,
	locked: locked
	},
	avatar: avatar,
	name: data["name"],
	follower_address: data["followers"],
	following_address: data["following"],
	bio: data["summary"]
	}

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

	{:ok, user_data}
	end

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

	e ->
	{:error, e}
	end
	end

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

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

	data
	end
	end

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

	{:error, _} = e ->
	e

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

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

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

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

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

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

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

	def fetch_direct_messages_query do
	Activity
	\|> restrict_type(%{"type" => "Create"})
	\|> restrict_visibility(%{visibility: "direct"})
	\|> order_by([activity], asc: activity.id)
	end
	end
	diff --git a/lib/pleroma/web/activity_pub/topics.ex b/lib/pleroma/web/activity_pub/topics.ex
	index a016014b1..ed7e77c1a 100644
	--- a/lib/pleroma/web/activity_pub/topics.ex
	+++ b/lib/pleroma/web/activity_pub/topics.ex
	@@ -1,70 +1,68 @@
	# Pleroma: A lightweight social networking server
	# Copyright © 2017-2019 Pleroma Authors <https://pleroma.social/>
	# SPDX-License-Identifier: AGPL-3.0-only

	defmodule Pleroma.Web.ActivityPub.Topics do
	-
	alias Pleroma.Object
	alias Pleroma.Web.ActivityPub.Visibility

	def get_activity_topics(activity) do
	activity
	- \|> Object.normalize
	+ \|> Object.normalize()
	\|> generate_topics(activity)
	- \|> List.flatten
	+ \|> List.flatten()
	end

	defp generate_topics(%{data: %{type: "Answer"}}, _) do
	[]
	end

	defp generate_topics(object, activity) do
	["user", "list"] ++ visibility_tags(object, activity)
	end

	defp visibility_tags(object, activity) do
	case Visibility.get_visibility(activity) do
	"public" ->
	if activity.local do
	["public", "public:local"]
	else
	["public"]
	end
	\|> item_creation_tags(object, activity)

	"direct" ->
	["direct"]

	_ ->
	[]
	end
	end

	defp item_creation_tags(tags, %{data: %{type: "Create"}} = object, activity) do
	tags ++ hashtags_to_topics(object) ++ attachment_topics(object, activity)
	end

	defp item_creation_tags(tags, _, _) do
	tags
	end

	defp hashtags_to_topics(obj) do
	obj.data
	\|> Map.get("tag", [])
	\|> Enum.filter(fn tag -> is_bitstring(tag) end)
	\|> Enum.map(fn tag -> "hashtag:" <> tag end)
	end

	defp attachment_topics(%{data: %{"attachment" => []}}, _activity) do
	[]
	end

	defp attachment_topics(_object, activity) do
	if activity.local do
	["public:media", "public:local:media"]
	else
	["public:media"]
	end
	end
	-
	end
	diff --git a/lib/pleroma_web/streamer/state.ex b/lib/pleroma_web/streamer/state.ex
	index 03479c877..2c9284245 100644
	--- a/lib/pleroma_web/streamer/state.ex
	+++ b/lib/pleroma_web/streamer/state.ex
	@@ -1,90 +1,90 @@
	defmodule PleromaWeb.Streamer.State do
	use GenServer
	require Logger

	alias PleromaWeb.Streamer.StreamerSocket

	def start_link(_) do
	GenServer.start_link(__MODULE__, %{sockets: %{}, streams: []}, name: __MODULE__)
	end

	def add_socket(topic, socket) do
	GenServer.call(__MODULE__, %{action: :add, socket: socket, topic: topic})
	end

	def remove_socket(topic, socket) do
	GenServer.call(__MODULE__, %{action: :remove, socket: socket, topic: topic})
	end

	- def get_sockets() do
	+ def get_sockets do
	%{sockets: stream_sockets} = GenServer.call(__MODULE__, :get_state)
	stream_sockets
	end

	def add_stream_item(topic, item) do
	GenServer.call(__MODULE__, %{action: :add, item: item, topic: topic})
	end

	def init(init_arg) do
	{:ok, init_arg}
	end

	def handle_call(:get_state, _from, state) do
	{:reply, state, state}
	end

	def handle_call(
	%{action: :add, topic: topic, socket: socket},
	_from,
	%{sockets: sockets} = state
	) do
	internal_topic = internal_topic(topic, socket)
	stream_socket = StreamerSocket.from_socket(socket)

	sockets_for_topic =
	sockets
	\|> Map.get(internal_topic, [])
	\|> List.insert_at(0, stream_socket)
	\|> Enum.uniq()

	state = Kernel.put_in(state, [:sockets, internal_topic], sockets_for_topic)
	Logger.debug("Got new conn for #{topic}")
	{:reply, state, state}
	end

	def handle_call(
	%{action: :add, topic: topic, item: _item},
	_from,
	%{streams: _streams} = state
	) do
	state = Map.put(state, :streams, [])
	Logger.debug("Got new conn for #{topic}")
	{:reply, state, state}
	end

	def handle_call(
	%{action: :remove, topic: topic, socket: socket},
	_from,
	%{sockets: sockets} = state
	) do
	internal_topic = internal_topic(topic, socket)
	stream_socket = PleromaWeb.Streamer.StreamerSocket.from_socket(socket)

	sockets_for_topic =
	sockets
	\|> Map.get(internal_topic, [])
	\|> List.delete(stream_socket)

	state = Kernel.put_in(state, [:sockets, internal_topic], sockets_for_topic)
	{:reply, state, state}
	end

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

	defp internal_topic(topic, _) do
	topic
	end
	end
	diff --git a/lib/pleroma_web/streamer/streamer.ex b/lib/pleroma_web/streamer/streamer.ex
	index 6a65f80a0..0b0446a76 100644
	--- a/lib/pleroma_web/streamer/streamer.ex
	+++ b/lib/pleroma_web/streamer/streamer.ex
	@@ -1,299 +1,40 @@
	# Pleroma: A lightweight social networking server
	# Copyright © 2017-2019 Pleroma Authors <https://pleroma.social/>
	# SPDX-License-Identifier: AGPL-3.0-only

	defmodule PleromaWeb.Streamer do
	use GenServer
	- 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.MastodonAPI.NotificationView
	+
	alias PleromaWeb.Streamer.State
	- alias PleromaWeb.Streamer.StreamerSocket
	+ alias PleromaWeb.Streamer.Worker

	def start_link(_) do
	GenServer.start_link(__MODULE__, %{}, name: __MODULE__)
	end

	def add_socket(topic, socket) do
	State.add_socket(topic, socket)
	end

	def remove_socket(topic, socket) do
	State.remove_socket(topic, socket)
	end

	- def get_sockets() do
	+ def get_sockets do
	State.get_sockets()
	end

	- def stream(topics, item) when is_list(topics) do
	- Enum.each(topics, fn t ->
	- GenServer.cast(__MODULE__, %{action: :stream, topic: t, item: item})
	- end)
	- end
	-
	- def stream(topic, items) when is_list(items) do
	- Enum.each(items, fn i ->
	- GenServer.cast(__MODULE__, %{action: :stream, topic: topic, item: i})
	- end)
	+ def direct_push(topics, topic, activity) do
	+ Worker.push_to_socket(topics, topic, activity)
	end

	- def stream(topic, item) do
	- GenServer.cast(__MODULE__, %{action: :stream, topic: topic, item: item})
	+ def stream(topics, items) do
	+ GenServer.cast(Worker, %{action: :stream, topic: topics, item: items})
	end

	def supervisor, do: PleromaWeb.Streamer.Supervisor

	def init(args) do
	{:ok, args}
	end
	-
	- def handle_cast(%{action: :stream, topic: "direct", item: item}, state) 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(State.get_sockets(), user_topic, item)
	- end)
	-
	- {:noreply, state}
	- end
	-
	- def handle_cast(%{action: :stream, topic: "participation", item: participation}, state) do
	- user_topic = "direct:#{participation.user_id}"
	- Logger.debug("Trying to push a conversation participation to #{user_topic}\n\n")
	-
	- push_to_socket(State.get_sockets(), user_topic, participation)
	-
	- {:noreply, state}
	- end
	-
	- def handle_cast(%{action: :stream, topic: "list", item: item}, state) 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(State.get_sockets(), list_topic, item)
	- end)
	-
	- {:noreply, state}
	- end
	-
	- def handle_cast(
	- %{action: :stream, topic: topic, item: %Notification{} = item},
	- state
	- )
	- when topic in ["user", "user:notification"] do
	- State.get_sockets()
	- \|> Map.get("#{topic}:#{item.user_id}", [])
	- \|> Enum.each(fn %StreamerSocket{transport_pid: transport_pid, user: socket_user} ->
	- with %User{} = user <- User.get_cached_by_ap_id(socket_user.ap_id),
	- true <- should_send?(user, item) do
	- send(transport_pid, {:text, represent_notification(socket_user, item)})
	- end
	- end)
	-
	- {:noreply, state}
	- end
	-
	- def handle_cast(%{action: :stream, topic: "user", item: item}, state) 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(State.get_sockets(), topic, item)
	- end)
	-
	- {:noreply, state}
	- end
	-
	- def handle_cast(%{action: :stream, topic: topic, item: item}, state) do
	- Logger.debug("Trying to push to #{topic}")
	- Logger.debug("Pushing item to #{topic}")
	- push_to_socket(State.get_sockets(), topic, item)
	- {:noreply, state}
	- end
	-
	- def handle_cast(m, state) do
	- Logger.info("Unknown: #{inspect(m)}, #{inspect(state)}")
	- {:noreply, state}
	- end
	-
	- defp represent_update(%Activity{} = activity, %User{} = user) do
	- %{
	- event: "update",
	- payload:
	- Pleroma.Web.MastodonAPI.StatusView.render(
	- "status.json",
	- activity: activity,
	- for: user
	- )
	- \|> Jason.encode!()
	- }
	- \|> Jason.encode!()
	- end
	-
	- defp represent_update(%Activity{} = activity) do
	- %{
	- event: "update",
	- payload:
	- Pleroma.Web.MastodonAPI.StatusView.render(
	- "status.json",
	- activity: activity
	- )
	- \|> Jason.encode!()
	- }
	- \|> Jason.encode!()
	- end
	-
	- def represent_conversation(%Participation{} = participation) do
	- %{
	- event: "conversation",
	- payload:
	- Pleroma.Web.MastodonAPI.ConversationView.render("participation.json", %{
	- participation: participation,
	- for: participation.user
	- })
	- \|> Jason.encode!()
	- }
	- \|> Jason.encode!()
	- end
	-
	- @spec represent_notification(User.t(), Notification.t()) :: binary()
	- defp represent_notification(%User{} = user, %Notification{} = notify) do
	- %{
	- event: "notification",
	- payload:
	- NotificationView.render(
	- "show.json",
	- %{notification: notify, for: user}
	- )
	- \|> Jason.encode!()
	- }
	- \|> Jason.encode!()
	- end
	-
	- defp should_send?(%User{} = user, %Activity{} = item) do
	- blocks = user.info.blocks \|\| []
	- mutes = user.info.mutes \|\| []
	- reblog_mutes = user.info.muted_reblogs \|\| []
	- domain_blocks = Pleroma.Web.ActivityPub.MRF.subdomains_regex(user.info.domain_blocks)
	-
	- with parent when not is_nil(parent) <- Object.normalize(item),
	- true <- Enum.all?([blocks, mutes, reblog_mutes], &(item.actor not in &1)),
	- true <- Enum.all?([blocks, mutes], &(parent.data["actor"] not in &1)),
	- %{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
	- true
	- else
	- _ -> false
	- end
	- end
	-
	- defp should_send?(%User{} = user, %Notification{activity: activity}) do
	- should_send?(user, activity)
	- end
	-
	- def push_to_socket(topics, topic, %Activity{data: %{"type" => "Announce"}} = item) do
	- Enum.each(topics[topic] \|\| [], fn %StreamerSocket{
	- transport_pid: transport_pid,
	- user: socket_user
	- } ->
	- # Get the current user so we have up-to-date blocks etc.
	- if socket_user do
	- user = User.get_cached_by_ap_id(socket_user.ap_id)
	-
	- if should_send?(user, item) do
	- send(transport_pid, {:text, represent_update(item, user)})
	- end
	- else
	- send(transport_pid, {:text, represent_update(item)})
	- end
	- end)
	- end
	-
	- def push_to_socket(topics, topic, %Participation{} = participation) do
	- Enum.each(topics[topic] \|\| [], fn %StreamerSocket{transport_pid: transport_pid} ->
	- send(transport_pid, {:text, represent_conversation(participation)})
	- end)
	- end
	-
	- def push_to_socket(topics, topic, %Activity{
	- data: %{"type" => "Delete", "deleted_activity_id" => deleted_activity_id}
	- }) do
	- Enum.each(topics[topic] \|\| [], fn %StreamerSocket{transport_pid: transport_pid} ->
	- send(
	- transport_pid,
	- {:text, %{event: "delete", payload: to_string(deleted_activity_id)} \|> Jason.encode!()}
	- )
	- end)
	- end
	-
	- def push_to_socket(_topics, _topic, %Activity{data: %{"type" => "Delete"}}), do: :noop
	-
	- def push_to_socket(topics, topic, item) do
	- Enum.each(topics[topic] \|\| [], fn %StreamerSocket{
	- transport_pid: transport_pid,
	- user: socket_user
	- } ->
	- # Get the current user so we have up-to-date blocks etc.
	- if socket_user do
	- user = User.get_cached_by_ap_id(socket_user.ap_id)
	- blocks = user.info.blocks \|\| []
	- mutes = user.info.mutes \|\| []
	-
	- with true <- Enum.all?([blocks, mutes], &(item.actor not in &1)),
	- true <- thread_containment(item, user) do
	- send(transport_pid, {:text, represent_update(item, user)})
	- end
	- else
	- send(transport_pid, {:text, represent_update(item)})
	- end
	- end)
	- end
	-
	- @spec thread_containment(Activity.t(), User.t()) :: boolean()
	- defp thread_containment(_activity, %User{info: %{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
	end
	diff --git a/lib/pleroma_web/streamer/supervisor.ex b/lib/pleroma_web/streamer/supervisor.ex
	index d2439021e..368cccac1 100644
	--- a/lib/pleroma_web/streamer/supervisor.ex
	+++ b/lib/pleroma_web/streamer/supervisor.ex
	@@ -1,18 +1,30 @@
	defmodule PleromaWeb.Streamer.Supervisor do
	use Supervisor

	def start_link(opts) do
	Supervisor.start_link(__MODULE__, opts, name: __MODULE__)
	end

	def init(args) do
	children = [
	{PleromaWeb.Streamer.State, args},
	{PleromaWeb.Streamer.Ping, args},
	- {PleromaWeb.Streamer, args}
	+ {PleromaWeb.Streamer.Worker, args},
	+ {PleromaWeb.Streamer, args},
	+ # :poolboy.child_spec(:streamer_worker, poolboy_config())
	]

	opts = [strategy: :one_for_one, name: PleromaWeb.Streamer.Supervisor]
	Supervisor.init(children, opts)
	end
	+
	+
	+# defp poolboy_config do
	+# [
	+# {:name, {:local, :streamer_worker}},
	+# {:worker_module, PleromaWeb.Streamer.Worker},
	+# {:size, 10},
	+# {:max_overflow, 2}
	+# ]
	+# end
	end
	diff --git a/lib/pleroma_web/streamer/streamer.ex b/lib/pleroma_web/streamer/worker.ex
	similarity index 64%
	copy from lib/pleroma_web/streamer/streamer.ex
	copy to lib/pleroma_web/streamer/worker.ex
	index 6a65f80a0..fd3fc1ca9 100644
	--- a/lib/pleroma_web/streamer/streamer.ex
	+++ b/lib/pleroma_web/streamer/worker.ex
	@@ -1,299 +1,220 @@
	-# Pleroma: A lightweight social networking server
	-# Copyright © 2017-2019 Pleroma Authors <https://pleroma.social/>
	-# SPDX-License-Identifier: AGPL-3.0-only
	-
	-defmodule PleromaWeb.Streamer do
	+defmodule PleromaWeb.Streamer.Worker do
	use GenServer
	+
	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.MastodonAPI.NotificationView
	alias PleromaWeb.Streamer.State
	alias PleromaWeb.Streamer.StreamerSocket
	+ alias PleromaWeb.StreamerView

	def start_link(_) do
	GenServer.start_link(__MODULE__, %{}, name: __MODULE__)
	end

	- def add_socket(topic, socket) do
	- State.add_socket(topic, socket)
	- end
	-
	- def remove_socket(topic, socket) do
	- State.remove_socket(topic, socket)
	+ def init(init_arg) do
	+ {:ok, init_arg}
	end

	- def get_sockets() do
	- State.get_sockets()
	- end
	-
	- def stream(topics, item) when is_list(topics) do
	+ def handle_cast(%{action: :stream, topic: topics, item: item}, state) when is_list(topics) do
	Enum.each(topics, fn t ->
	- GenServer.cast(__MODULE__, %{action: :stream, topic: t, item: item})
	+ do_stream(%{topic: t, item: item})
	end)
	+
	+ {:noreply, state}
	end

	- def stream(topic, items) when is_list(items) do
	+ def handle_cast(%{action: :stream, topic: topic, item: items}, state) when is_list(items) do
	Enum.each(items, fn i ->
	- GenServer.cast(__MODULE__, %{action: :stream, topic: topic, item: i})
	+ do_stream(%{topic: topic, item: i})
	end)
	- end

	- def stream(topic, item) do
	- GenServer.cast(__MODULE__, %{action: :stream, topic: topic, item: item})
	+ {:noreply, state}
	end

	- def supervisor, do: PleromaWeb.Streamer.Supervisor
	+ def handle_cast(%{action: :stream, topic: topic, item: item}, state) do
	+ do_stream(%{topic: topic, item: item})

	- def init(args) do
	- {:ok, args}
	+ {:noreply, state}
	end

	- def handle_cast(%{action: :stream, topic: "direct", item: item}, state) do
	+ defp do_stream(%{topic: "direct", item: 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(State.get_sockets(), user_topic, item)
	end)
	-
	- {:noreply, state}
	end

	- def handle_cast(%{action: :stream, topic: "participation", item: participation}, state) do
	+ defp do_stream(%{topic: "participation", item: participation}) do
	user_topic = "direct:#{participation.user_id}"
	Logger.debug("Trying to push a conversation participation to #{user_topic}\n\n")

	push_to_socket(State.get_sockets(), user_topic, participation)
	-
	- {:noreply, state}
	end

	- def handle_cast(%{action: :stream, topic: "list", item: item}, state) do
	+ defp do_stream(%{topic: "list", item: 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(State.get_sockets(), list_topic, item)
	end)
	-
	- {:noreply, state}
	end

	- def handle_cast(
	- %{action: :stream, topic: topic, item: %Notification{} = item},
	- state
	- )
	- when topic in ["user", "user:notification"] do
	+ defp do_stream(%{topic: topic, item: %Notification{} = item})
	+ when topic in ["user", "user:notification"] do
	State.get_sockets()
	\|> Map.get("#{topic}:#{item.user_id}", [])
	\|> Enum.each(fn %StreamerSocket{transport_pid: transport_pid, user: socket_user} ->
	with %User{} = user <- User.get_cached_by_ap_id(socket_user.ap_id),
	true <- should_send?(user, item) do
	- send(transport_pid, {:text, represent_notification(socket_user, item)})
	+ send(transport_pid, {:text, StreamerView.render("notification.json", socket_user, item)})
	end
	end)
	-
	- {:noreply, state}
	end

	- def handle_cast(%{action: :stream, topic: "user", item: item}, state) do
	+ defp do_stream(%{topic: "user", item: 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(State.get_sockets(), topic, item)
	end)
	-
	- {:noreply, state}
	end

	- def handle_cast(%{action: :stream, topic: topic, item: item}, state) do
	+ defp do_stream(%{topic: topic, item: item}) do
	Logger.debug("Trying to push to #{topic}")
	Logger.debug("Pushing item to #{topic}")
	push_to_socket(State.get_sockets(), topic, item)
	- {:noreply, state}
	- end
	-
	- def handle_cast(m, state) do
	- Logger.info("Unknown: #{inspect(m)}, #{inspect(state)}")
	- {:noreply, state}
	- end
	-
	- defp represent_update(%Activity{} = activity, %User{} = user) do
	- %{
	- event: "update",
	- payload:
	- Pleroma.Web.MastodonAPI.StatusView.render(
	- "status.json",
	- activity: activity,
	- for: user
	- )
	- \|> Jason.encode!()
	- }
	- \|> Jason.encode!()
	- end
	-
	- defp represent_update(%Activity{} = activity) do
	- %{
	- event: "update",
	- payload:
	- Pleroma.Web.MastodonAPI.StatusView.render(
	- "status.json",
	- activity: activity
	- )
	- \|> Jason.encode!()
	- }
	- \|> Jason.encode!()
	- end
	-
	- def represent_conversation(%Participation{} = participation) do
	- %{
	- event: "conversation",
	- payload:
	- Pleroma.Web.MastodonAPI.ConversationView.render("participation.json", %{
	- participation: participation,
	- for: participation.user
	- })
	- \|> Jason.encode!()
	- }
	- \|> Jason.encode!()
	end

	- @spec represent_notification(User.t(), Notification.t()) :: binary()
	- defp represent_notification(%User{} = user, %Notification{} = notify) do
	- %{
	- event: "notification",
	- payload:
	- NotificationView.render(
	- "show.json",
	- %{notification: notify, for: user}
	- )
	- \|> Jason.encode!()
	- }
	- \|> Jason.encode!()
	+ defp do_stream(m) do
	+ Logger.info("Unknown: #{inspect(m)}")
	end

	defp should_send?(%User{} = user, %Activity{} = item) do
	blocks = user.info.blocks \|\| []
	mutes = user.info.mutes \|\| []
	reblog_mutes = user.info.muted_reblogs \|\| []
	domain_blocks = Pleroma.Web.ActivityPub.MRF.subdomains_regex(user.info.domain_blocks)

	with parent when not is_nil(parent) <- Object.normalize(item),
	true <- Enum.all?([blocks, mutes, reblog_mutes], &(item.actor not in &1)),
	true <- Enum.all?([blocks, mutes], &(parent.data["actor"] not in &1)),
	%{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
	true
	else
	_ -> false
	end
	end

	defp should_send?(%User{} = user, %Notification{activity: activity}) do
	should_send?(user, activity)
	end

	def push_to_socket(topics, topic, %Activity{data: %{"type" => "Announce"}} = item) do
	Enum.each(topics[topic] \|\| [], fn %StreamerSocket{
	transport_pid: transport_pid,
	user: socket_user
	} ->
	# Get the current user so we have up-to-date blocks etc.
	if socket_user do
	user = User.get_cached_by_ap_id(socket_user.ap_id)

	if should_send?(user, item) do
	- send(transport_pid, {:text, represent_update(item, user)})
	+ send(transport_pid, {:text, StreamerView.render("update.json", item, user)})
	end
	else
	- send(transport_pid, {:text, represent_update(item)})
	+ send(transport_pid, {:text, StreamerView.render("update.json", item)})
	end
	end)
	end

	def push_to_socket(topics, topic, %Participation{} = participation) do
	Enum.each(topics[topic] \|\| [], fn %StreamerSocket{transport_pid: transport_pid} ->
	- send(transport_pid, {:text, represent_conversation(participation)})
	+ send(transport_pid, {:text, StreamerView.render("conversation.json", participation)})
	end)
	end

	def push_to_socket(topics, topic, %Activity{
	data: %{"type" => "Delete", "deleted_activity_id" => deleted_activity_id}
	}) do
	Enum.each(topics[topic] \|\| [], fn %StreamerSocket{transport_pid: transport_pid} ->
	send(
	transport_pid,
	{:text, %{event: "delete", payload: to_string(deleted_activity_id)} \|> Jason.encode!()}
	)
	end)
	end

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

	def push_to_socket(topics, topic, item) do
	Enum.each(topics[topic] \|\| [], fn %StreamerSocket{
	transport_pid: transport_pid,
	user: socket_user
	} ->
	# Get the current user so we have up-to-date blocks etc.
	if socket_user do
	user = User.get_cached_by_ap_id(socket_user.ap_id)
	blocks = user.info.blocks \|\| []
	mutes = user.info.mutes \|\| []

	with true <- Enum.all?([blocks, mutes], &(item.actor not in &1)),
	true <- thread_containment(item, user) do
	- send(transport_pid, {:text, represent_update(item, user)})
	+ send(transport_pid, {:text, StreamerView.render("update.json", item, user)})
	end
	else
	- send(transport_pid, {:text, represent_update(item)})
	+ send(transport_pid, {:text, StreamerView.render("update.json", item)})
	end
	end)
	end

	@spec thread_containment(Activity.t(), User.t()) :: boolean()
	defp thread_containment(_activity, %User{info: %{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
	end
	diff --git a/lib/pleroma_web/views/streamer_view.ex b/lib/pleroma_web/views/streamer_view.ex
	new file mode 100644
	index 000000000..c3feacf5c
	--- /dev/null
	+++ b/lib/pleroma_web/views/streamer_view.ex
	@@ -0,0 +1,67 @@
	+# Pleroma: A lightweight social networking server
	+# Copyright © 2017-2019 Pleroma Authors <https://pleroma.social/>
	+# SPDX-License-Identifier: AGPL-3.0-only
	+
	+defmodule PleromaWeb.StreamerView do
	+ use Pleroma.Web, :view
	+
	+ alias Pleroma.Activity
	+ alias Pleroma.Conversation.Participation
	+ alias Pleroma.Notification
	+ alias Pleroma.Web.MastodonAPI.NotificationView
	+ alias Pleroma.User
	+
	+ def render("update.json", %Activity{} = activity, %User{} = user) do
	+ %{
	+ event: "update",
	+ payload:
	+ Pleroma.Web.MastodonAPI.StatusView.render(
	+ "status.json",
	+ activity: activity,
	+ for: user
	+ )
	+ \|> Jason.encode!()
	+ }
	+ \|> Jason.encode!()
	+ end
	+
	+ def render("notification.json", %User{} = user, %Notification{} = notify) do
	+ %{
	+ event: "notification",
	+ payload:
	+ NotificationView.render(
	+ "show.json",
	+ %{notification: notify, for: user}
	+ )
	+ \|> Jason.encode!()
	+ }
	+ \|> Jason.encode!()
	+ end
	+
	+ def render("update.json", %Activity{} = activity) do
	+ %{
	+ event: "update",
	+ payload:
	+ Pleroma.Web.MastodonAPI.StatusView.render(
	+ "status.json",
	+ activity: activity
	+ )
	+ \|> Jason.encode!()
	+ }
	+ \|> Jason.encode!()
	+ end
	+
	+ def render("conversation.json", %Participation{} = participation) do
	+ %{
	+ event: "conversation",
	+ payload:
	+ Pleroma.Web.MastodonAPI.ConversationView.render("participation.json", %{
	+ participation: participation,
	+ for: participation.user
	+ })
	+ \|> Jason.encode!()
	+ }
	+ \|> Jason.encode!()
	+ end
	+
	+end
	diff --git a/mix.lock b/mix.lock
	index 2639e96e9..8ca34144b 100644
	--- a/mix.lock
	+++ b/mix.lock
	@@ -1,96 +1,97 @@
	%{
	"accept": {:hex, :accept, "0.3.5", "b33b127abca7cc948bbe6caa4c263369abf1347cfa9d8e699c6d214660f10cd1", [:rebar3], [], "hexpm"},
	"auto_linker": {:git, "https://git.pleroma.social/pleroma/auto_linker.git", "95e8188490e97505c56636c1379ffdf036c1fdde", [ref: "95e8188490e97505c56636c1379ffdf036c1fdde"]},
	"base64url": {:hex, :base64url, "0.0.1", "36a90125f5948e3afd7be97662a1504b934dd5dac78451ca6e9abf85a10286be", [:rebar], [], "hexpm"},
	"bbcode": {:hex, :bbcode, "0.1.1", "0023e2c7814119b2e620b7add67182e3f6019f92bfec9a22da7e99821aceba70", [:mix], [{:nimble_parsec, "~> 0.5", [hex: :nimble_parsec, repo: "hexpm", optional: false]}], "hexpm"},
	"benchee": {:hex, :benchee, "1.0.1", "66b211f9bfd84bd97e6d1beaddf8fc2312aaabe192f776e8931cb0c16f53a521", [:mix], [{:deep_merge, "~> 1.0", [hex: :deep_merge, repo: "hexpm", optional: false]}], "hexpm"},
	"bunt": {:hex, :bunt, "0.2.0", "951c6e801e8b1d2cbe58ebbd3e616a869061ddadcc4863d0a2182541acae9a38", [:mix], [], "hexpm"},
	"cachex": {:hex, :cachex, "3.0.3", "4e2d3e05814a5738f5ff3903151d5c25636d72a3527251b753f501ad9c657967", [:mix], [{:eternal, "~> 1.2", [hex: :eternal, repo: "hexpm", optional: false]}, {:unsafe, "~> 1.0", [hex: :unsafe, repo: "hexpm", optional: false]}], "hexpm"},
	"calendar": {:hex, :calendar, "0.17.6", "ec291cb2e4ba499c2e8c0ef5f4ace974e2f9d02ae9e807e711a9b0c7850b9aee", [:mix], [{:tzdata, "~> 0.5.20 or ~> 0.1.201603 or ~> 1.0", [hex: :tzdata, repo: "hexpm", optional: false]}], "hexpm"},
	"certifi": {:hex, :certifi, "2.5.1", "867ce347f7c7d78563450a18a6a28a8090331e77fa02380b4a21962a65d36ee5", [:rebar3], [{:parse_trans, "~>3.3", [hex: :parse_trans, repo: "hexpm", optional: false]}], "hexpm"},
	"combine": {:hex, :combine, "0.10.0", "eff8224eeb56498a2af13011d142c5e7997a80c8f5b97c499f84c841032e429f", [:mix], [], "hexpm"},
	"comeonin": {:hex, :comeonin, "4.1.2", "3eb5620fd8e35508991664b4c2b04dd41e52f1620b36957be837c1d7784b7592", [:mix], [{:argon2_elixir, "~> 1.2", [hex: :argon2_elixir, repo: "hexpm", optional: true]}, {:bcrypt_elixir, "~> 0.12.1 or ~> 1.0", [hex: :bcrypt_elixir, repo: "hexpm", optional: true]}, {:pbkdf2_elixir, "~> 0.12", [hex: :pbkdf2_elixir, repo: "hexpm", optional: true]}], "hexpm"},
	"connection": {:hex, :connection, "1.0.4", "a1cae72211f0eef17705aaededacac3eb30e6625b04a6117c1b2db6ace7d5976", [:mix], [], "hexpm"},
	"cors_plug": {:hex, :cors_plug, "1.5.2", "72df63c87e4f94112f458ce9d25800900cc88608c1078f0e4faddf20933eda6e", [:mix], [{:plug, "~> 1.3 or ~> 1.4 or ~> 1.5", [hex: :plug, repo: "hexpm", optional: false]}], "hexpm"},
	"cowboy": {:hex, :cowboy, "2.6.3", "99aa50e94e685557cad82e704457336a453d4abcb77839ad22dbe71f311fcc06", [:rebar3], [{:cowlib, "~> 2.7.3", [hex: :cowlib, repo: "hexpm", optional: false]}, {:ranch, "~> 1.7.1", [hex: :ranch, repo: "hexpm", optional: false]}], "hexpm"},
	"cowlib": {:hex, :cowlib, "2.7.3", "a7ffcd0917e6d50b4d5fb28e9e2085a0ceb3c97dea310505f7460ff5ed764ce9", [:rebar3], [], "hexpm"},
	"credo": {:hex, :credo, "0.9.3", "76fa3e9e497ab282e0cf64b98a624aa11da702854c52c82db1bf24e54ab7c97a", [:mix], [{:bunt, "~> 0.2.0", [hex: :bunt, repo: "hexpm", optional: false]}, {:poison, ">= 0.0.0", [hex: :poison, repo: "hexpm", optional: false]}], "hexpm"},
	"crontab": {:hex, :crontab, "1.1.7", "b9219f0bdc8678b94143655a8f229716c5810c0636a4489f98c0956137e53985", [:mix], [{:ecto, "~> 1.0 or ~> 2.0 or ~> 3.0", [hex: :ecto, repo: "hexpm", optional: true]}], "hexpm"},
	"crypt": {:git, "https://github.com/msantos/crypt", "1f2b58927ab57e72910191a7ebaeff984382a1d3", [ref: "1f2b58927ab57e72910191a7ebaeff984382a1d3"]},
	"db_connection": {:hex, :db_connection, "2.0.6", "bde2f85d047969c5b5800cb8f4b3ed6316c8cb11487afedac4aa5f93fd39abfa", [:mix], [{:connection, "~> 1.0.2", [hex: :connection, repo: "hexpm", optional: false]}], "hexpm"},
	"decimal": {:hex, :decimal, "1.8.0", "ca462e0d885f09a1c5a342dbd7c1dcf27ea63548c65a65e67334f4b61803822e", [:mix], [], "hexpm"},
	"deep_merge": {:hex, :deep_merge, "1.0.0", "b4aa1a0d1acac393bdf38b2291af38cb1d4a52806cf7a4906f718e1feb5ee961", [:mix], [], "hexpm"},
	"earmark": {:hex, :earmark, "1.3.2", "b840562ea3d67795ffbb5bd88940b1bed0ed9fa32834915125ea7d02e35888a5", [:mix], [], "hexpm"},
	"ecto": {:hex, :ecto, "3.1.4", "69d852da7a9f04ede725855a35ede48d158ca11a404fe94f8b2fb3b2162cd3c9", [:mix], [{:decimal, "~> 1.6", [hex: :decimal, repo: "hexpm", optional: false]}, {:jason, "~> 1.0", [hex: :jason, repo: "hexpm", optional: true]}], "hexpm"},
	"ecto_sql": {:hex, :ecto_sql, "3.1.3", "2c536139190492d9de33c5fefac7323c5eaaa82e1b9bf93482a14649042f7cd9", [:mix], [{:db_connection, "~> 2.0", [hex: :db_connection, repo: "hexpm", optional: false]}, {:ecto, "~> 3.1.0", [hex: :ecto, repo: "hexpm", optional: false]}, {:mariaex, "~> 0.9.1", [hex: :mariaex, repo: "hexpm", optional: true]}, {:myxql, "~> 0.2.0", [hex: :myxql, repo: "hexpm", optional: true]}, {:postgrex, "~> 0.14.0", [hex: :postgrex, repo: "hexpm", optional: true]}, {:telemetry, "~> 0.4.0", [hex: :telemetry, repo: "hexpm", optional: false]}], "hexpm"},
	"esshd": {:hex, :esshd, "0.1.0", "6f93a2062adb43637edad0ea7357db2702a4b80dd9683482fe00f5134e97f4c1", [:mix], [], "hexpm"},
	"eternal": {:hex, :eternal, "1.2.0", "e2a6b6ce3b8c248f7dc31451aefca57e3bdf0e48d73ae5043229380a67614c41", [:mix], [], "hexpm"},
	"ex2ms": {:hex, :ex2ms, "1.5.0", "19e27f9212be9a96093fed8cdfbef0a2b56c21237196d26760f11dfcfae58e97", [:mix], [], "hexpm"},
	"ex_aws": {:hex, :ex_aws, "2.1.0", "b92651527d6c09c479f9013caa9c7331f19cba38a650590d82ebf2c6c16a1d8a", [:mix], [{:configparser_ex, "~> 2.0", [hex: :configparser_ex, repo: "hexpm", optional: true]}, {:hackney, "1.6.3 or 1.6.5 or 1.7.1 or 1.8.6 or ~> 1.9", [hex: :hackney, repo: "hexpm", optional: true]}, {:jsx, "~> 2.8", [hex: :jsx, repo: "hexpm", optional: true]}, {:poison, ">= 1.2.0", [hex: :poison, repo: "hexpm", optional: true]}, {:sweet_xml, "~> 0.6", [hex: :sweet_xml, repo: "hexpm", optional: true]}, {:xml_builder, "~> 0.1.0", [hex: :xml_builder, repo: "hexpm", optional: true]}], "hexpm"},
	"ex_aws_s3": {:hex, :ex_aws_s3, "2.0.1", "9e09366e77f25d3d88c5393824e613344631be8db0d1839faca49686e99b6704", [:mix], [{:ex_aws, "~> 2.0", [hex: :ex_aws, repo: "hexpm", optional: false]}, {:sweet_xml, ">= 0.0.0", [hex: :sweet_xml, repo: "hexpm", optional: true]}], "hexpm"},
	"ex_const": {:hex, :ex_const, "0.2.4", "d06e540c9d834865b012a17407761455efa71d0ce91e5831e86881b9c9d82448", [:mix], [], "hexpm"},
	"ex_doc": {:hex, :ex_doc, "0.20.2", "1bd0dfb0304bade58beb77f20f21ee3558cc3c753743ae0ddbb0fd7ba2912331", [:mix], [{:earmark, "~> 1.3", [hex: :earmark, repo: "hexpm", optional: false]}, {:makeup_elixir, "~> 0.10", [hex: :makeup_elixir, repo: "hexpm", optional: false]}], "hexpm"},
	"ex_machina": {:hex, :ex_machina, "2.3.0", "92a5ad0a8b10ea6314b876a99c8c9e3f25f4dde71a2a835845b136b9adaf199a", [:mix], [{:ecto, "~> 2.2 or ~> 3.0", [hex: :ecto, repo: "hexpm", optional: true]}, {:ecto_sql, "~> 3.0", [hex: :ecto_sql, repo: "hexpm", optional: true]}], "hexpm"},
	"ex_rated": {:hex, :ex_rated, "1.3.3", "30ecbdabe91f7eaa9d37fa4e81c85ba420f371babeb9d1910adbcd79ec798d27", [:mix], [{:ex2ms, "~> 1.5", [hex: :ex2ms, repo: "hexpm", optional: false]}], "hexpm"},
	"ex_syslogger": {:git, "https://github.com/slashmili/ex_syslogger.git", "f3963399047af17e038897c69e20d552e6899e1d", [tag: "1.4.0"]},
	"excoveralls": {:hex, :excoveralls, "0.11.1", "dd677fbdd49114fdbdbf445540ec735808250d56b011077798316505064edb2c", [:mix], [{:hackney, "~> 1.0", [hex: :hackney, repo: "hexpm", optional: false]}, {:jason, "~> 1.0", [hex: :jason, repo: "hexpm", optional: false]}], "hexpm"},
	"floki": {:hex, :floki, "0.20.4", "be42ac911fece24b4c72f3b5846774b6e61b83fe685c2fc9d62093277fb3bc86", [:mix], [{:html_entities, "~> 0.4.0", [hex: :html_entities, repo: "hexpm", optional: false]}, {:mochiweb, "~> 2.15", [hex: :mochiweb, repo: "hexpm", optional: false]}], "hexpm"},
	"gen_smtp": {:hex, :gen_smtp, "0.14.0", "39846a03522456077c6429b4badfd1d55e5e7d0fdfb65e935b7c5e38549d9202", [:rebar3], [], "hexpm"},
	"gettext": {:hex, :gettext, "0.17.0", "abe21542c831887a2b16f4c94556db9c421ab301aee417b7c4fbde7fbdbe01ec", [:mix], [], "hexpm"},
	"hackney": {:hex, :hackney, "1.15.1", "9f8f471c844b8ce395f7b6d8398139e26ddca9ebc171a8b91342ee15a19963f4", [:rebar3], [{:certifi, "2.5.1", [hex: :certifi, repo: "hexpm", optional: false]}, {:idna, "6.0.0", [hex: :idna, repo: "hexpm", optional: false]}, {:metrics, "1.0.1", [hex: :metrics, repo: "hexpm", optional: false]}, {:mimerl, "~>1.1", [hex: :mimerl, repo: "hexpm", optional: false]}, {:ssl_verify_fun, "1.1.4", [hex: :ssl_verify_fun, repo: "hexpm", optional: false]}], "hexpm"},
	"html_entities": {:hex, :html_entities, "0.4.0", "f2fee876858cf6aaa9db608820a3209e45a087c5177332799592142b50e89a6b", [:mix], [], "hexpm"},
	"html_sanitize_ex": {:hex, :html_sanitize_ex, "1.3.0", "f005ad692b717691203f940c686208aa3d8ffd9dd4bb3699240096a51fa9564e", [:mix], [{:mochiweb, "~> 2.15", [hex: :mochiweb, repo: "hexpm", optional: false]}], "hexpm"},
	"http_signatures": {:git, "https://git.pleroma.social/pleroma/http_signatures.git", "293d77bb6f4a67ac8bde1428735c3b42f22cbb30", [ref: "293d77bb6f4a67ac8bde1428735c3b42f22cbb30"]},
	"httpoison": {:hex, :httpoison, "1.2.0", "2702ed3da5fd7a8130fc34b11965c8cfa21ade2f232c00b42d96d4967c39a3a3", [:mix], [{:hackney, "~> 1.8", [hex: :hackney, repo: "hexpm", optional: false]}], "hexpm"},
	"idna": {:hex, :idna, "6.0.0", "689c46cbcdf3524c44d5f3dde8001f364cd7608a99556d8fbd8239a5798d4c10", [:rebar3], [{:unicode_util_compat, "0.4.1", [hex: :unicode_util_compat, repo: "hexpm", optional: false]}], "hexpm"},
	"jason": {:hex, :jason, "1.1.2", "b03dedea67a99223a2eaf9f1264ce37154564de899fd3d8b9a21b1a6fd64afe7", [:mix], [{:decimal, "~> 1.0", [hex: :decimal, repo: "hexpm", optional: true]}], "hexpm"},
	"joken": {:hex, :joken, "2.0.1", "ec9ab31bf660f343380da033b3316855197c8d4c6ef597fa3fcb451b326beb14", [:mix], [{:jose, "~> 1.9", [hex: :jose, repo: "hexpm", optional: false]}], "hexpm"},
	"jose": {:hex, :jose, "1.9.0", "4167c5f6d06ffaebffd15cdb8da61a108445ef5e85ab8f5a7ad926fdf3ada154", [:mix, :rebar3], [{:base64url, "~> 0.0.1", [hex: :base64url, repo: "hexpm", optional: false]}], "hexpm"},
	"makeup": {:hex, :makeup, "0.8.0", "9cf32aea71c7fe0a4b2e9246c2c4978f9070257e5c9ce6d4a28ec450a839b55f", [:mix], [{:nimble_parsec, "~> 0.5.0", [hex: :nimble_parsec, repo: "hexpm", optional: false]}], "hexpm"},
	"makeup_elixir": {:hex, :makeup_elixir, "0.13.0", "be7a477997dcac2e48a9d695ec730b2d22418292675c75aa2d34ba0909dcdeda", [:mix], [{:makeup, "~> 0.8", [hex: :makeup, repo: "hexpm", optional: false]}], "hexpm"},
	"meck": {:hex, :meck, "0.8.13", "ffedb39f99b0b99703b8601c6f17c7f76313ee12de6b646e671e3188401f7866", [:rebar3], [], "hexpm"},
	"metrics": {:hex, :metrics, "1.0.1", "25f094dea2cda98213cecc3aeff09e940299d950904393b2a29d191c346a8486", [:rebar3], [], "hexpm"},
	"mime": {:hex, :mime, "1.3.1", "30ce04ab3175b6ad0bdce0035cba77bba68b813d523d1aac73d9781b4d193cf8", [:mix], [], "hexpm"},
	"mimerl": {:hex, :mimerl, "1.2.0", "67e2d3f571088d5cfd3e550c383094b47159f3eee8ffa08e64106cdf5e981be3", [:rebar3], [], "hexpm"},
	"mochiweb": {:hex, :mochiweb, "2.18.0", "eb55f1db3e6e960fac4e6db4e2db9ec3602cc9f30b86cd1481d56545c3145d2e", [:rebar3], [], "hexpm"},
	"mock": {:hex, :mock, "0.3.3", "42a433794b1291a9cf1525c6d26b38e039e0d3a360732b5e467bfc77ef26c914", [:mix], [{:meck, "~> 0.8.13", [hex: :meck, repo: "hexpm", optional: false]}], "hexpm"},
	"mogrify": {:hex, :mogrify, "0.6.1", "de1b527514f2d95a7bbe9642eb556061afb337e220cf97adbf3a4e6438ed70af", [:mix], [], "hexpm"},
	"mox": {:hex, :mox, "0.5.1", "f86bb36026aac1e6f924a4b6d024b05e9adbed5c63e8daa069bd66fb3292165b", [:mix], [], "hexpm"},
	"nimble_parsec": {:hex, :nimble_parsec, "0.5.0", "90e2eca3d0266e5c53f8fbe0079694740b9c91b6747f2b7e3c5d21966bba8300", [:mix], [], "hexpm"},
	"parse_trans": {:hex, :parse_trans, "3.3.0", "09765507a3c7590a784615cfd421d101aec25098d50b89d7aa1d66646bc571c1", [:rebar3], [], "hexpm"},
	"pbkdf2_elixir": {:hex, :pbkdf2_elixir, "0.12.3", "6706a148809a29c306062862c803406e88f048277f6e85b68faf73291e820b84", [:mix], [], "hexpm"},
	"phoenix": {:hex, :phoenix, "1.4.9", "746d098e10741c334d88143d3c94cab1756435f94387a63441792e66ec0ee974", [:mix], [{:jason, "~> 1.0", [hex: :jason, repo: "hexpm", optional: true]}, {:phoenix_pubsub, "~> 1.1", [hex: :phoenix_pubsub, repo: "hexpm", optional: false]}, {:plug, "~> 1.8.1 or ~> 1.9", [hex: :plug, repo: "hexpm", optional: false]}, {:plug_cowboy, "~> 1.0 or ~> 2.0", [hex: :plug_cowboy, repo: "hexpm", optional: true]}, {:telemetry, "~> 0.4", [hex: :telemetry, repo: "hexpm", optional: false]}], "hexpm"},
	"phoenix_ecto": {:hex, :phoenix_ecto, "4.0.0", "c43117a136e7399ea04ecaac73f8f23ee0ffe3e07acfcb8062fe5f4c9f0f6531", [:mix], [{:ecto, "~> 3.0", [hex: :ecto, repo: "hexpm", optional: false]}, {:phoenix_html, "~> 2.9", [hex: :phoenix_html, repo: "hexpm", optional: true]}, {:plug, "~> 1.0", [hex: :plug, repo: "hexpm", optional: false]}], "hexpm"},
	"phoenix_html": {:hex, :phoenix_html, "2.13.1", "fa8f034b5328e2dfa0e4131b5569379003f34bc1fafdaa84985b0b9d2f12e68b", [:mix], [{:plug, "~> 1.5", [hex: :plug, repo: "hexpm", optional: false]}], "hexpm"},
	"phoenix_pubsub": {:hex, :phoenix_pubsub, "1.1.2", "496c303bdf1b2e98a9d26e89af5bba3ab487ba3a3735f74bf1f4064d2a845a3e", [:mix], [], "hexpm"},
	"phoenix_swoosh": {:hex, :phoenix_swoosh, "0.2.0", "a7e0b32077cd6d2323ae15198839b05d9caddfa20663fd85787479e81f89520e", [:mix], [{:phoenix, "~> 1.0", [hex: :phoenix, repo: "hexpm", optional: false]}, {:phoenix_html, "~> 2.2", [hex: :phoenix_html, repo: "hexpm", optional: false]}, {:swoosh, "~> 0.1", [hex: :swoosh, repo: "hexpm", optional: false]}], "hexpm"},
	"pleroma_job_queue": {:hex, :pleroma_job_queue, "0.3.0", "b84538d621f0c3d6fcc1cff9d5648d3faaf873b8b21b94e6503428a07a48ec47", [:mix], [{:crontab, "~> 1.1", [hex: :crontab, repo: "hexpm", optional: false]}], "hexpm"},
	"plug": {:hex, :plug, "1.8.2", "0bcce1daa420f189a6491f3940cc77ea7fb1919761175c9c3b59800d897440fc", [:mix], [{:mime, "~> 1.0", [hex: :mime, repo: "hexpm", optional: false]}, {:plug_crypto, "~> 1.0", [hex: :plug_crypto, repo: "hexpm", optional: false]}, {:telemetry, "~> 0.4", [hex: :telemetry, repo: "hexpm", optional: true]}], "hexpm"},
	"plug_cowboy": {:hex, :plug_cowboy, "2.1.0", "b75768153c3a8a9e8039d4b25bb9b14efbc58e9c4a6e6a270abff1cd30cbe320", [:mix], [{:cowboy, "~> 2.5", [hex: :cowboy, repo: "hexpm", optional: false]}, {:plug, "~> 1.7", [hex: :plug, repo: "hexpm", optional: false]}], "hexpm"},
	"plug_crypto": {:hex, :plug_crypto, "1.0.0", "18e49317d3fa343f24620ed22795ec29d4a5e602d52d1513ccea0b07d8ea7d4d", [:mix], [], "hexpm"},
	"plug_static_index_html": {:hex, :plug_static_index_html, "1.0.0", "840123d4d3975585133485ea86af73cb2600afd7f2a976f9f5fd8b3808e636a0", [:mix], [{:plug, "~> 1.0", [hex: :plug, repo: "hexpm", optional: false]}], "hexpm"},
	"poison": {:hex, :poison, "3.1.0", "d9eb636610e096f86f25d9a46f35a9facac35609a7591b3be3326e99a0484665", [:mix], [], "hexpm"},
	+ "poolboy": {:hex, :poolboy, "1.5.2", "392b007a1693a64540cead79830443abf5762f5d30cf50bc95cb2c1aaafa006b", [:rebar3], [], "hexpm"},
	"postgrex": {:hex, :postgrex, "0.14.3", "5754dee2fdf6e9e508cbf49ab138df964278700b764177e8f3871e658b345a1e", [:mix], [{:connection, "~> 1.0", [hex: :connection, repo: "hexpm", optional: false]}, {:db_connection, "~> 2.0", [hex: :db_connection, repo: "hexpm", optional: false]}, {:decimal, "~> 1.5", [hex: :decimal, repo: "hexpm", optional: false]}, {:jason, "~> 1.0", [hex: :jason, repo: "hexpm", optional: true]}], "hexpm"},
	"prometheus": {:hex, :prometheus, "4.4.1", "1e96073b3ed7788053768fea779cbc896ddc3bdd9ba60687f2ad50b252ac87d6", [:mix, :rebar3], [], "hexpm"},
	"prometheus_ecto": {:hex, :prometheus_ecto, "1.4.1", "6c768ea9654de871e5b32fab2eac348467b3021604ebebbcbd8bcbe806a65ed5", [:mix], [{:ecto, "~> 2.0 or ~> 3.0", [hex: :ecto, repo: "hexpm", optional: false]}, {:prometheus_ex, "~> 1.1 or ~> 2.0 or ~> 3.0", [hex: :prometheus_ex, repo: "hexpm", optional: false]}], "hexpm"},
	"prometheus_ex": {:hex, :prometheus_ex, "3.0.5", "fa58cfd983487fc5ead331e9a3e0aa622c67232b3ec71710ced122c4c453a02f", [:mix], [{:prometheus, "~> 4.0", [hex: :prometheus, repo: "hexpm", optional: false]}], "hexpm"},
	"prometheus_phoenix": {:hex, :prometheus_phoenix, "1.3.0", "c4b527e0b3a9ef1af26bdcfbfad3998f37795b9185d475ca610fe4388fdd3bb5", [:mix], [{:phoenix, "~> 1.4", [hex: :phoenix, repo: "hexpm", optional: false]}, {:prometheus_ex, "~> 1.3 or ~> 2.0 or ~> 3.0", [hex: :prometheus_ex, repo: "hexpm", optional: false]}], "hexpm"},
	"prometheus_plugs": {:hex, :prometheus_plugs, "1.1.5", "25933d48f8af3a5941dd7b621c889749894d8a1082a6ff7c67cc99dec26377c5", [:mix], [{:accept, "~> 0.1", [hex: :accept, repo: "hexpm", optional: false]}, {:plug, "~> 1.0", [hex: :plug, repo: "hexpm", optional: false]}, {:prometheus_ex, "~> 1.1 or ~> 2.0 or ~> 3.0", [hex: :prometheus_ex, repo: "hexpm", optional: false]}, {:prometheus_process_collector, "~> 1.1", [hex: :prometheus_process_collector, repo: "hexpm", optional: true]}], "hexpm"},
	"quack": {:hex, :quack, "0.1.1", "cca7b4da1a233757fdb44b3334fce80c94785b3ad5a602053b7a002b5a8967bf", [:mix], [{:poison, ">= 1.0.0", [hex: :poison, repo: "hexpm", optional: false]}, {:tesla, "~> 1.2.0", [hex: :tesla, repo: "hexpm", optional: false]}], "hexpm"},
	"ranch": {:hex, :ranch, "1.7.1", "6b1fab51b49196860b733a49c07604465a47bdb78aa10c1c16a3d199f7f8c881", [:rebar3], [], "hexpm"},
	"recon": {:git, "https://github.com/ferd/recon.git", "75d70c7c08926d2f24f1ee6de14ee50fe8a52763", [tag: "2.4.0"]},
	"ssl_verify_fun": {:hex, :ssl_verify_fun, "1.1.4", "f0eafff810d2041e93f915ef59899c923f4568f4585904d010387ed74988e77b", [:make, :mix, :rebar3], [], "hexpm"},
	"sweet_xml": {:hex, :sweet_xml, "0.6.6", "fc3e91ec5dd7c787b6195757fbcf0abc670cee1e4172687b45183032221b66b8", [:mix], [], "hexpm"},
	"swoosh": {:hex, :swoosh, "0.23.2", "7dda95ff0bf54a2298328d6899c74dae1223777b43563ccebebb4b5d2b61df38", [:mix], [{:cowboy, "~> 1.0.1 or ~> 1.1 or ~> 2.4", [hex: :cowboy, repo: "hexpm", optional: true]}, {:gen_smtp, "~> 0.13", [hex: :gen_smtp, repo: "hexpm", optional: true]}, {:hackney, "~> 1.9", [hex: :hackney, repo: "hexpm", optional: false]}, {:jason, "~> 1.0", [hex: :jason, repo: "hexpm", optional: false]}, {:mail, "~> 0.2", [hex: :mail, repo: "hexpm", optional: true]}, {:mime, "~> 1.1", [hex: :mime, repo: "hexpm", optional: false]}, {:plug_cowboy, ">= 1.0.0", [hex: :plug_cowboy, repo: "hexpm", optional: true]}], "hexpm"},
	"syslog": {:git, "https://github.com/Vagabond/erlang-syslog.git", "4a6c6f2c996483e86c1320e9553f91d337bcb6aa", [tag: "1.0.5"]},
	"telemetry": {:hex, :telemetry, "0.4.0", "8339bee3fa8b91cb84d14c2935f8ecf399ccd87301ad6da6b71c09553834b2ab", [:rebar3], [], "hexpm"},
	"tesla": {:hex, :tesla, "1.2.1", "864783cc27f71dd8c8969163704752476cec0f3a51eb3b06393b3971dc9733ff", [:mix], [{:exjsx, ">= 3.0.0", [hex: :exjsx, repo: "hexpm", optional: true]}, {:fuse, "~> 2.4", [hex: :fuse, repo: "hexpm", optional: true]}, {:hackney, "~> 1.6", [hex: :hackney, repo: "hexpm", optional: true]}, {:ibrowse, "~> 4.4.0", [hex: :ibrowse, repo: "hexpm", optional: true]}, {:jason, ">= 1.0.0", [hex: :jason, repo: "hexpm", optional: true]}, {:mime, "~> 1.0", [hex: :mime, repo: "hexpm", optional: false]}, {:poison, ">= 1.0.0", [hex: :poison, repo: "hexpm", optional: true]}], "hexpm"},
	"timex": {:hex, :timex, "3.6.1", "efdf56d0e67a6b956cc57774353b0329c8ab7726766a11547e529357ffdc1d56", [:mix], [{:combine, "~> 0.10", [hex: :combine, repo: "hexpm", optional: false]}, {:gettext, "~> 0.10", [hex: :gettext, repo: "hexpm", optional: false]}, {:tzdata, "~> 0.1.8 or ~> 0.5 or ~> 1.0.0", [hex: :tzdata, repo: "hexpm", optional: false]}], "hexpm"},
	"trailing_format_plug": {:hex, :trailing_format_plug, "0.0.7", "64b877f912cf7273bed03379936df39894149e35137ac9509117e59866e10e45", [:mix], [{:plug, "> 0.12.0", [hex: :plug, repo: "hexpm", optional: false]}], "hexpm"},
	"tzdata": {:hex, :tzdata, "0.5.21", "8cbf3607fcce69636c672d5be2bbb08687fe26639a62bdcc283d267277db7cf0", [:mix], [{:hackney, "~> 1.0", [hex: :hackney, repo: "hexpm", optional: false]}], "hexpm"},
	"ueberauth": {:hex, :ueberauth, "0.6.1", "9e90d3337dddf38b1ca2753aca9b1e53d8a52b890191cdc55240247c89230412", [:mix], [{:plug, "~> 1.5", [hex: :plug, repo: "hexpm", optional: false]}], "hexpm"},
	"unicode_util_compat": {:hex, :unicode_util_compat, "0.4.1", "d869e4c68901dd9531385bb0c8c40444ebf624e60b6962d95952775cac5e90cd", [:rebar3], [], "hexpm"},
	"unsafe": {:hex, :unsafe, "1.0.1", "a27e1874f72ee49312e0a9ec2e0b27924214a05e3ddac90e91727bc76f8613d8", [:mix], [], "hexpm"},
	"web_push_encryption": {:hex, :web_push_encryption, "0.2.1", "d42cecf73420d9dc0053ba3299cc8c8d6ff2be2487d67ca2a57265868e4d9a98", [:mix], [{:httpoison, "~> 1.0", [hex: :httpoison, repo: "hexpm", optional: false]}, {:jose, "~> 1.8", [hex: :jose, repo: "hexpm", optional: false]}, {:poison, "~> 3.0", [hex: :poison, repo: "hexpm", optional: false]}], "hexpm"},
	"websocket_client": {:git, "https://github.com/jeremyong/websocket_client.git", "9a6f65d05ebf2725d62fb19262b21f1805a59fbf", []},
	}
	diff --git a/test/notification_test.exs b/test/notification_test.exs
	index d2c24f8f5..6a1c60929 100644
	--- a/test/notification_test.exs
	+++ b/test/notification_test.exs
	@@ -1,728 +1,728 @@
	# Pleroma: A lightweight social networking server
	# Copyright © 2017-2018 Pleroma Authors <https://pleroma.social/>
	# SPDX-License-Identifier: AGPL-3.0-only

	defmodule Pleroma.NotificationTest do
	use Pleroma.DataCase

	import Pleroma.Factory

	alias Pleroma.Notification
	alias Pleroma.User
	+ alias PleromaWeb.Streamer
	alias Pleroma.Web.ActivityPub.Transmogrifier
	alias Pleroma.Web.CommonAPI
	- alias PleromaWeb.Streamer

	describe "create_notifications" do
	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 "create_notification" do
	setup do
	start_supervised(Streamer.supervisor())

	:ok
	end

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

	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} = User.block(user, author)

	assert Notification.create_notification(activity, user)
	end

	test "it creates a notificatin 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, muter} = 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, info: %{notification_settings: %{"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, info: %{notification_settings: %{"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, info: %{notification_settings: %{"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, info: %{notification_settings: %{"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 a notification for follow-unfollow-follow chains" do
	user = insert(:user)
	followed_user = insert(:user)
	{:ok, _, _, activity} = CommonAPI.follow(user, followed_user)
	Notification.create_notification(activity, followed_user)
	CommonAPI.unfollow(user, followed_user)
	{:ok, _, _, activity_dupe} = CommonAPI.follow(user, followed_user)
	refute Notification.create_notification(activity_dupe, followed_user)
	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 "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" 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}!"
	})

	assert other_user in Notification.get_notified_from_activity(activity)
	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)

	assert other_user in Notification.get_notified_from_activity(activity)
	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)

	assert other_user not in Notification.get_notified_from_activity(activity)
	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(activity_one.id, third_user)

	assert other_user not in Notification.get_notified_from_activity(activity_two)
	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)

	assert other_user not in Notification.get_notified_from_activity(activity_two)
	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(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 "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(activity.id, other_user)

	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, _} = CommonAPI.favorite(activity.id, other_user)

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

	User.delete(user)

	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
	}

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

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

	describe "for_user" do
	test "it returns notifications for muted user without notifications" do
	user = insert(:user)
	muted = insert(:user)
	{:ok, user} = 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} = 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} = User.block(user, blocked)

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

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

	test "it doesn't return notificatitons for blocked domain" 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 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 for muted user with notifications and with_muted parameter" do
	user = insert(:user)
	muted = insert(:user)
	{:ok, user} = 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 returns notifications for blocked user and with_muted parameter" do
	user = insert(:user)
	blocked = insert(:user)
	{:ok, user} = User.block(user, blocked)

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

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

	test "it returns notificatitons for blocked domain and with_muted parameter" 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 length(Notification.for_user(user, %{with_muted: true})) == 1
	end

	test "it returns notifications for muted thread with_muted parameter" 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/pleroma_web/streamer/streamer_test.exs b/test/pleroma_web/streamer/streamer_test.exs
	index 183bfba84..1b559f339 100644
	--- a/test/pleroma_web/streamer/streamer_test.exs
	+++ b/test/pleroma_web/streamer/streamer_test.exs
	@@ -1,500 +1,523 @@
	# Pleroma: A lightweight social networking server
	# Copyright © 2017-2018 Pleroma Authors <https://pleroma.social/>
	# SPDX-License-Identifier: AGPL-3.0-only

	defmodule PleromaWeb.StreamerTest do
	use Pleroma.DataCase

	alias Pleroma.List
	alias Pleroma.User
	alias Pleroma.Web.CommonAPI
	alias PleromaWeb.Streamer
	import Pleroma.Factory
	alias PleromaWeb.Streamer.StreamerSocket
	+ alias PleromaWeb.Streamer.Worker

	setup do
	skip_thread_containment = Pleroma.Config.get([:instance, :skip_thread_containment])

	on_exit(fn ->
	Pleroma.Config.put([:instance, :skip_thread_containment], skip_thread_containment)
	end)

	start_supervised(Streamer.supervisor())

	:ok
	end

	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 sends notify to in the 'user' stream", %{user: user, notify: notify} do
	task =
	Task.async(fn ->
	assert_receive {:text, _}, 4_000
	end)

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

	Streamer.stream("user", notify)
	Task.await(task)
	end

	test "it sends notify to in the 'user:notification' stream", %{user: user, notify: notify} do
	task =
	Task.async(fn ->
	assert_receive {:text, _}, 4_000
	end)

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

	Streamer.stream("user:notification", notify)
	Task.await(task)
	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} = User.block(user, blocked)

	task = Task.async(fn -> refute_receive {:text, _}, 4_000 end)

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

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

	Streamer.stream("user:notification", notif)
	Task.await(task)
	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, _}, 4_000 end)

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

	{:ok, activity} = CommonAPI.post(user, %{"status" => "super hot take"})
	{:ok, activity} = CommonAPI.add_mute(user, activity)
	{:ok, notif, _} = CommonAPI.favorite(activity.id, user2)
	Streamer.stream("user:notification", notif)
	Task.await(task)
	end

	test "it doesn't send notify to the 'user:notification' stream' when a domain is blocked", %{
	user: user
	} do
	user2 = insert(:user, %{ap_id: "https://hecking-lewd-place.com/user/meanie"})
	task = Task.async(fn -> refute_receive {:text, _}, 4_000 end)

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

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

	Streamer.stream("user:notification", notif)
	Task.await(task)
	end
	end

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

	task =
	Task.async(fn ->
	assert_receive {:text, _}, 4_000
	end)

	fake_socket = %StreamerSocket{
	transport_pid: task.pid,
	user: user
	}

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

	topics = %{
	"public" => [fake_socket]
	}

	- Streamer.push_to_socket(topics, "public", activity)
	+ Streamer.direct_push(topics, "public", activity)

	Task.await(task)

	task =
	Task.async(fn ->
	expected_event =
	%{
	"event" => "delete",
	"payload" => activity.id
	}
	\|> Jason.encode!()

	assert_receive {:text, received_event}, 4_000
	assert received_event == expected_event
	end)

	fake_socket = %StreamerSocket{
	transport_pid: task.pid,
	user: user
	}

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

	topics = %{
	"public" => [fake_socket]
	}

	- Streamer.push_to_socket(topics, "public", activity)
	+ Streamer.direct_push(topics, "public", activity)

	Task.await(task)
	end

	describe "thread_containment" do
	test "it doesn't send 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, following: [author.ap_id])

	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]}
	- Streamer.push_to_socket(topics, "public", activity)
	+ Streamer.direct_push(topics, "public", activity)

	Task.await(task)
	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, following: [author.ap_id])

	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]}
	- Streamer.push_to_socket(topics, "public", activity)
	+ Streamer.direct_push(topics, "public", activity)

	Task.await(task)
	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, following: [author.ap_id], info: %{skip_thread_containment: true})

	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]}
	- Streamer.push_to_socket(topics, "public", activity)
	+ Streamer.direct_push(topics, "public", activity)

	Task.await(task)
	end
	end

	test "it doesn't send to blocked users" do
	user = insert(:user)
	blocked_user = insert(:user)
	{:ok, user} = User.block(user, blocked_user)

	task =
	Task.async(fn ->
	refute_receive {:text, _}, 1_000
	end)

	fake_socket = %StreamerSocket{
	transport_pid: task.pid,
	user: user
	}

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

	topics = %{
	"public" => [fake_socket]
	}

	- Streamer.push_to_socket(topics, "public", activity)
	+ Streamer.direct_push(topics, "public", activity)

	Task.await(task)
	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)

	task =
	Task.async(fn ->
	refute_receive {:text, _}, 1_000
	end)

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

	- Streamer.handle_cast(%{action: :stream, topic: "list", item: activity}, topics)
	+ Worker.handle_cast(%{action: :stream, topic: "list", item: activity}, topics)

	Task.await(task)
	end

	test "it doesn't send unwanted private posts to list" do
	user_a = insert(:user)
	user_b = insert(:user)

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

	task =
	Task.async(fn ->
	refute_receive {:text, _}, 1_000
	end)

	fake_socket = %StreamerSocket{
	transport_pid: task.pid,
	user: user_a
	}

	{:ok, activity} =
	CommonAPI.post(user_b, %{
	"status" => "Test",
	"visibility" => "private"
	})

	topics = %{
	"list:#{list.id}" => [fake_socket]
	}

	- Streamer.handle_cast(%{action: :stream, topic: "list", item: activity}, topics)
	+ Worker.handle_cast(%{action: :stream, topic: "list", item: activity}, topics)

	Task.await(task)
	end

	test "it sends wanted private posts to list" do
	user_a = insert(:user)
	user_b = 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)

	task =
	Task.async(fn ->
	assert_receive {:text, _}, 1_000
	end)

	fake_socket = %StreamerSocket{
	transport_pid: task.pid,
	user: user_a
	}

	{:ok, activity} =
	CommonAPI.post(user_b, %{
	"status" => "Test",
	"visibility" => "private"
	})

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

	- Streamer.handle_cast(%{action: :stream, topic: "list", item: activity}, %{})
	+ Worker.handle_cast(%{action: :stream, topic: "list", item: activity}, %{})

	Task.await(task)
	end

	test "it doesn't send muted reblogs" do
	user1 = insert(:user)
	user2 = insert(:user)
	user3 = insert(:user)
	CommonAPI.hide_reblogs(user1, user2)

	task =
	Task.async(fn ->
	refute_receive {:text, _}, 1_000
	end)

	fake_socket = %StreamerSocket{
	transport_pid: task.pid,
	user: user1
	}

	{:ok, create_activity} = CommonAPI.post(user3, %{"status" => "I'm kawen"})
	{:ok, announce_activity, _} = CommonAPI.repeat(create_activity.id, user2)

	topics = %{
	"public" => [fake_socket]
	}

	- Streamer.push_to_socket(topics, "public", announce_activity)
	+ Streamer.direct_push(topics, "public", announce_activity)

	Task.await(task)
	end

	+ test "it doesn't send posts from muted threads" do
	+ user = insert(:user)
	+ user2 = insert(:user)
	+ {: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, _}, 4_000 end)
	+
	+ Process.sleep(4000)
	+
	+ Streamer.add_socket(
	+ "user",
	+ %{transport_pid: task.pid, assigns: %{user: user2}}
	+ )
	+
	+ Streamer.stream("user", activity)
	+ Task.await(task)
	+ 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}, 4_000
	end)

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

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

	Task.await(task)
	end

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

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

	task =
	Task.async(fn ->
	assert_receive {:text, received_event}, 4_000
	assert_receive {:text, received_event}, 4_000
	assert_receive {:text, received_event}, 4_000
	assert %{"event" => "delete", "payload" => _} = Jason.decode!(received_event)

	refute_receive {:text, _}, 4_000
	end)

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

	{:ok, _} = CommonAPI.delete(create_activity.id, another_user)

	Task.await(task)
	end

	test "it sends conversation update to the 'direct' stream when a message is deleted" do
	user = insert(:user)
	another_user = insert(: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}",
	"in_reply_to_status_id" => create_activity.id,
	"visibility" => "direct"
	})

	task =
	Task.async(fn ->
	assert_receive {:text, received_event}, 4_000
	assert_receive {:text, received_event}, 4_000
	assert_receive {:text, received_event}, 4_000
	assert %{"event" => "delete", "payload" => _} = Jason.decode!(received_event)

	assert_receive {:text, received_event}, 4_000

	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)

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

	{:ok, _} = CommonAPI.delete(create_activity2.id, another_user)

	Task.await(task)
	end
	end
	end

File Metadata

Mime Type: text/x-diff
Expires: Wed, Oct 14, 2:57 AM (1 d, 21 h)
Storage Engine: blob
Storage Format: Raw Data
Storage Handle: 1786987
Default Alt Text: (138 KB)

No OneTemporaryActions

View Options

File Metadata

Event Timeline

No OneTemporary
Actions