No OneTemporary
Actions

Authored By

Unknown

Size

82 KB

Referenced Files

None

Subscribers

None

View Options

	diff --git a/changelog.d/stream-marker-updates.add b/changelog.d/stream-marker-updates.add
	new file mode 100644
	index 000000000..e9fda3e59
	--- /dev/null
	+++ b/changelog.d/stream-marker-updates.add
	@@ -0,0 +1 @@
	+Stream marker updates
	diff --git a/lib/pleroma/notification.ex b/lib/pleroma/notification.ex
	index 75f4ba503..59b32bbf0 100644
	--- a/lib/pleroma/notification.ex
	+++ b/lib/pleroma/notification.ex
	@@ -1,768 +1,773 @@
	# Pleroma: A lightweight social networking server
	# Copyright © 2017-2022 Pleroma Authors <https://pleroma.social/>
	# SPDX-License-Identifier: AGPL-3.0-only

	defmodule Pleroma.Notification do
	use Ecto.Schema

	alias Ecto.Multi
	alias Pleroma.Activity
	alias Pleroma.FollowingRelationship
	alias Pleroma.Marker
	alias Pleroma.Notification
	alias Pleroma.Object
	alias Pleroma.Pagination
	alias Pleroma.Repo
	alias Pleroma.ThreadMute
	alias Pleroma.User
	alias Pleroma.Web.CommonAPI
	alias Pleroma.Web.CommonAPI.Utils
	alias Pleroma.Web.Push
	alias Pleroma.Web.Streamer

	import Ecto.Query
	import Ecto.Changeset

	require Logger

	@type t :: %__MODULE__{}

	@include_muted_option :with_muted

	schema "notifications" do
	field(:seen, :boolean, default: false)
	# This is an enum type in the database. If you add a new notification type,
	# remember to add a migration to add it to the `notifications_type` enum
	# as well.
	field(:type, :string)
	belongs_to(:user, User, type: FlakeId.Ecto.CompatType)
	belongs_to(:activity, Activity, type: FlakeId.Ecto.CompatType)

	timestamps()
	end

	def update_notification_type(user, activity) do
	with %__MODULE__{} = notification <-
	Repo.get_by(__MODULE__, user_id: user.id, activity_id: activity.id) do
	type =
	activity
	\|> type_from_activity()

	notification
	\|> changeset(%{type: type})
	\|> Repo.update()
	end
	end

	@spec unread_notifications_count(User.t()) :: integer()
	def unread_notifications_count(%User{id: user_id}) do
	from(q in __MODULE__,
	where: q.user_id == ^user_id and q.seen == false
	)
	\|> Repo.aggregate(:count, :id)
	end

	@notification_types ~w{
	favourite
	follow
	follow_request
	mention
	move
	pleroma:chat_mention
	pleroma:emoji_reaction
	pleroma:report
	reblog
	poll
	status
	}

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

	@spec last_read_query(User.t()) :: Ecto.Queryable.t()
	def last_read_query(user) do
	from(q in Pleroma.Notification,
	where: q.user_id == ^user.id,
	where: q.seen == true,
	select: type(q.id, :string),
	limit: 1,
	order_by: fragment("? desc nulls last", q.id)
	)
	end

	defp for_user_query_ap_id_opts(user, opts) do
	ap_id_relationships =
	[:block] ++
	if opts[@include_muted_option], do: [], else: [:notification_mute]

	preloaded_ap_ids = User.outgoing_relationships_ap_ids(user, ap_id_relationships)

	exclude_blocked_opts = Map.merge(%{blocked_users_ap_ids: preloaded_ap_ids[:block]}, opts)

	exclude_notification_muted_opts =
	Map.merge(%{notification_muted_users_ap_ids: preloaded_ap_ids[:notification_mute]}, opts)

	{exclude_blocked_opts, exclude_notification_muted_opts}
	end

	def for_user_query(user, opts \\ %{}) do
	{exclude_blocked_opts, exclude_notification_muted_opts} =
	for_user_query_ap_id_opts(user, opts)

	Notification
	\|> where(user_id: ^user.id)
	\|> join(:inner, [n], activity in assoc(n, :activity))
	\|> join(:left, [n, a], object in Object,
	on:
	fragment(
	"(?->>'id') = associated_object_id(?)",
	object.data,
	a.data
	)
	)
	\|> join(:inner, [_n, a], u in User, on: u.ap_id == a.actor, as: :user_actor)
	\|> preload([n, a, o], activity: {a, object: o})
	\|> where([user_actor: user_actor], user_actor.is_active)
	\|> exclude_notification_muted(user, exclude_notification_muted_opts)
	\|> exclude_blocked(user, exclude_blocked_opts)
	\|> exclude_blockers(user)
	\|> exclude_filtered(user)
	\|> exclude_visibility(opts)
	end

	# Excludes blocked users and non-followed domain-blocked users
	defp exclude_blocked(query, user, opts) do
	blocked_ap_ids = opts[:blocked_users_ap_ids] \|\| User.blocked_users_ap_ids(user)

	query
	\|> where([..., user_actor: user_actor], user_actor.ap_id not in ^blocked_ap_ids)
	\|> FollowingRelationship.keep_following_or_not_domain_blocked(user)
	end

	defp exclude_blockers(query, user) do
	if Pleroma.Config.get([:activitypub, :blockers_visible]) == true do
	query
	else
	blocker_ap_ids = User.incoming_relationships_ungrouped_ap_ids(user, [:block])

	query
	\|> where([..., user_actor: user_actor], user_actor.ap_id not in ^blocker_ap_ids)
	end
	end

	defp exclude_notification_muted(query, _, %{@include_muted_option => true}) do
	query
	end

	defp exclude_notification_muted(query, user, opts) do
	notification_muted_ap_ids =
	opts[:notification_muted_users_ap_ids] \|\| User.notification_muted_users_ap_ids(user)

	query
	\|> where([..., user_actor: user_actor], user_actor.ap_id not in ^notification_muted_ap_ids)
	\|> join(:left, [n, a], tm in ThreadMute,
	on: tm.user_id == ^user.id and tm.context == fragment("?->>'context'", a.data),
	as: :thread_mute
	)
	\|> where([thread_mute: thread_mute], is_nil(thread_mute.user_id))
	end

	defp exclude_filtered(query, user) do
	case Pleroma.Filter.compose_regex(user) do
	nil ->
	query

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

	@valid_visibilities ~w[direct unlisted public private]

	defp exclude_visibility(query, %{exclude_visibilities: visibility})
	when is_list(visibility) do
	if Enum.all?(visibility, &(&1 in @valid_visibilities)) do
	query
	\|> join(:left, [n, a], mutated_activity in Pleroma.Activity,
	on:
	fragment(
	"associated_object_id(?)",
	a.data
	) ==
	fragment(
	"associated_object_id(?)",
	mutated_activity.data
	) and
	fragment("(?->>'type' = 'Like' or ?->>'type' = 'Announce')", a.data, a.data) and
	fragment("?->>'type'", mutated_activity.data) == "Create",
	as: :mutated_activity
	)
	\|> where(
	[n, a, mutated_activity: mutated_activity],
	not fragment(
	"""
	CASE WHEN (?->>'type') = 'Like' or (?->>'type') = 'Announce'
	THEN (activity_visibility(?, ?, ?) = ANY (?))
	ELSE (activity_visibility(?, ?, ?) = ANY (?)) END
	""",
	a.data,
	a.data,
	mutated_activity.actor,
	mutated_activity.recipients,
	mutated_activity.data,
	^visibility,
	a.actor,
	a.recipients,
	a.data,
	^visibility
	)
	)
	else
	Logger.error("Could not exclude visibility to #{visibility}")
	query
	end
	end

	defp exclude_visibility(query, %{exclude_visibilities: visibility})
	when visibility in @valid_visibilities do
	exclude_visibility(query, [visibility])
	end

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

	defp exclude_visibility(query, _visibility), do: query

	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,
	# Ideally we would preload object and activities here
	# but Ecto does not support preloads in update_all
	select: n.id
	)

	- Multi.new()
	- \|> Multi.update_all(:ids, query, set: [seen: true, updated_at: NaiveDateTime.utc_now()])
	- \|> Marker.multi_set_last_read_id(user, "notifications")
	- \|> Repo.transaction()
	+ {:ok, %{marker: marker}} =
	+ Multi.new()
	+ \|> Multi.update_all(:ids, query, set: [seen: true, updated_at: NaiveDateTime.utc_now()])
	+ \|> Marker.multi_set_last_read_id(user, "notifications")
	+ \|> Repo.transaction()
	+
	+ Streamer.stream(["user", "user:notification"], marker)
	+
	+ {:ok, %{marker: marker}}
	end

	@spec read_one(User.t(), String.t()) ::
	{:ok, Notification.t()} \| {:error, Ecto.Changeset.t()} \| nil
	def read_one(%User{} = user, notification_id) do
	with {:ok, %Notification{} = notification} <- get(user, notification_id) do
	Multi.new()
	\|> Multi.update(:update, changeset(notification, %{seen: true}))
	\|> Marker.multi_set_last_read_id(user, "notifications")
	\|> Repo.transaction()
	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(%Pleroma.Activity{} = activity) do
	Notification
	\|> where([n], n.activity_id == ^activity.id)
	\|> Repo.delete_all()
	\|> case do
	{_, notifications} -> {:ok, notifications}
	_ -> {:error, "Cannot dismiss notification"}
	end
	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

	@spec create_notifications(Activity.t()) :: {:ok, [Notification.t()] \| []}
	def create_notifications(activity)

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

	if object && object.data["type"] == "Answer" do
	{:ok, []}
	else
	do_create_notifications(activity)
	end
	end

	def create_notifications(%Activity{data: %{"type" => type}} = activity)
	when type in ["Follow", "Like", "Announce", "Move", "EmojiReact", "Flag", "Update"] do
	do_create_notifications(activity)
	end

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

	defp do_create_notifications(%Activity{} = activity) do
	enabled_receivers = get_notified_from_activity(activity)

	enabled_subscribers = get_notified_subscribers_from_activity(activity)

	notifications =
	(Enum.map(enabled_receivers, fn user ->
	create_notification(activity, user)
	end) ++
	Enum.map(enabled_subscribers -- enabled_receivers, fn user ->
	create_notification(activity, user, type: "status")
	end))
	\|> Enum.reject(&is_nil/1)

	{:ok, notifications}
	end

	defp type_from_activity(%{data: %{"type" => type}} = activity) do
	case type do
	"Follow" ->
	if Activity.follow_accepted?(activity) do
	"follow"
	else
	"follow_request"
	end

	"Announce" ->
	"reblog"

	"Like" ->
	"favourite"

	"Move" ->
	"move"

	"EmojiReact" ->
	"pleroma:emoji_reaction"

	"Flag" ->
	"pleroma:report"

	# Compatibility with old reactions
	"EmojiReaction" ->
	"pleroma:emoji_reaction"

	"Create" ->
	activity
	\|> type_from_activity_object()

	"Update" ->
	"update"

	t ->
	raise "No notification type for activity type #{t}"
	end
	end

	defp type_from_activity_object(%{data: %{"type" => "Create", "object" => %{}}}), do: "mention"

	defp type_from_activity_object(%{data: %{"type" => "Create"}} = activity) do
	object = Object.get_by_ap_id(activity.data["object"])

	case object && object.data["type"] do
	"ChatMessage" -> "pleroma:chat_mention"
	_ -> "mention"
	end
	end

	# TODO move to sql, too.
	def create_notification(%Activity{} = activity, %User{} = user, opts \\ []) do
	type = Keyword.get(opts, :type, type_from_activity(activity))

	unless skip?(activity, user, opts) do
	{:ok, %{notification: notification}} =
	Multi.new()
	\|> Multi.insert(:notification, %Notification{
	user_id: user.id,
	activity: activity,
	seen: mark_as_read?(activity, user),
	type: type
	})
	\|> Marker.multi_set_last_read_id(user, "notifications")
	\|> Repo.transaction()

	notification
	end
	end

	def create_poll_notifications(%Activity{} = activity) do
	with %Object{data: %{"type" => "Question", "actor" => actor} = data} <-
	Object.normalize(activity) do
	voters =
	case data do
	%{"voters" => voters} when is_list(voters) -> voters
	_ -> []
	end

	notifications =
	Enum.reduce([actor \| voters], [], fn ap_id, acc ->
	with %User{local: true} = user <- User.get_by_ap_id(ap_id) do
	[create_notification(activity, user, type: "poll") \| acc]
	else
	_ -> acc
	end
	end)

	{:ok, notifications}
	end
	end

	@doc """
	Returns a tuple with 2 elements:
	{notification-enabled receivers, currently disabled receivers (blocking / [thread] muting)}

	NOTE: might be called for FAKE Activities, see ActivityPub.Utils.get_notified_from_object/1
	"""
	@spec get_notified_from_activity(Activity.t(), boolean()) :: list(User.t())
	def get_notified_from_activity(activity, local_only \\ true)

	def get_notified_from_activity(%Activity{data: %{"type" => type}} = activity, local_only)
	when type in [
	"Create",
	"Like",
	"Announce",
	"Follow",
	"Move",
	"EmojiReact",
	"Flag",
	"Update"
	] do
	potential_receiver_ap_ids = get_potential_receiver_ap_ids(activity)

	potential_receivers =
	User.get_users_from_set(potential_receiver_ap_ids, local_only: local_only)

	notification_enabled_ap_ids =
	potential_receiver_ap_ids
	\|> exclude_domain_blocker_ap_ids(activity, potential_receivers)
	\|> exclude_relationship_restricted_ap_ids(activity)
	\|> exclude_thread_muter_ap_ids(activity)

	Enum.filter(potential_receivers, fn u -> u.ap_id in notification_enabled_ap_ids end)
	end

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

	def get_notified_subscribers_from_activity(activity, local_only \\ true)

	def get_notified_subscribers_from_activity(
	%Activity{data: %{"type" => "Create"}} = activity,
	local_only
	) do
	notification_enabled_ap_ids =
	[]
	\|> Utils.maybe_notify_subscribers(activity)

	potential_receivers =
	User.get_users_from_set(notification_enabled_ap_ids, local_only: local_only)

	Enum.filter(potential_receivers, fn u -> u.ap_id in notification_enabled_ap_ids end)
	end

	def get_notified_subscribers_from_activity(_, _), do: []

	# For some activities, only notify the author of the object
	def get_potential_receiver_ap_ids(%{data: %{"type" => type, "object" => object_id}})
	when type in ~w{Like Announce EmojiReact} do
	case Object.get_cached_by_ap_id(object_id) do
	%Object{data: %{"actor" => actor}} ->
	[actor]

	_ ->
	[]
	end
	end

	def get_potential_receiver_ap_ids(%{data: %{"type" => "Follow", "object" => object_id}}) do
	[object_id]
	end

	def get_potential_receiver_ap_ids(%{data: %{"type" => "Flag", "actor" => actor}}) do
	(User.all_users_with_privilege(:reports_manage_reports)
	\|> Enum.map(fn user -> user.ap_id end)) --
	[actor]
	end

	# Update activity: notify all who repeated this
	def get_potential_receiver_ap_ids(%{data: %{"type" => "Update", "actor" => actor}} = activity) do
	with %Object{data: %{"id" => object_id}} <- Object.normalize(activity, fetch: false) do
	repeaters =
	Activity.Queries.by_type("Announce")
	\|> Activity.Queries.by_object_id(object_id)
	\|> Activity.with_joined_user_actor()
	\|> where([a, u], u.local)
	\|> select([a, u], u.ap_id)
	\|> Repo.all()

	repeaters -- [actor]
	end
	end

	def get_potential_receiver_ap_ids(activity) do
	[]
	\|> Utils.maybe_notify_to_recipients(activity)
	\|> Utils.maybe_notify_mentioned_recipients(activity)
	\|> Utils.maybe_notify_followers(activity)
	\|> Enum.uniq()
	end

	@doc "Filters out AP IDs domain-blocking and not following the activity's actor"
	def exclude_domain_blocker_ap_ids(ap_ids, activity, preloaded_users \\ [])

	def exclude_domain_blocker_ap_ids([], _activity, _preloaded_users), do: []

	def exclude_domain_blocker_ap_ids(ap_ids, %Activity{} = activity, preloaded_users) do
	activity_actor_domain = activity.actor && URI.parse(activity.actor).host

	users =
	ap_ids
	\|> Enum.map(fn ap_id ->
	Enum.find(preloaded_users, &(&1.ap_id == ap_id)) \|\|
	User.get_cached_by_ap_id(ap_id)
	end)
	\|> Enum.filter(& &1)

	domain_blocker_ap_ids = for u <- users, activity_actor_domain in u.domain_blocks, do: u.ap_id

	domain_blocker_follower_ap_ids =
	if Enum.any?(domain_blocker_ap_ids) do
	activity
	\|> Activity.user_actor()
	\|> FollowingRelationship.followers_ap_ids(domain_blocker_ap_ids)
	else
	[]
	end

	ap_ids
	\|> Kernel.--(domain_blocker_ap_ids)
	\|> Kernel.++(domain_blocker_follower_ap_ids)
	end

	@doc "Filters out AP IDs of users basing on their relationships with activity actor user"
	def exclude_relationship_restricted_ap_ids([], _activity), do: []

	def exclude_relationship_restricted_ap_ids(ap_ids, %Activity{} = activity) do
	relationship_restricted_ap_ids =
	activity
	\|> Activity.user_actor()
	\|> User.incoming_relationships_ungrouped_ap_ids([
	:block,
	:notification_mute
	])

	Enum.uniq(ap_ids) -- relationship_restricted_ap_ids
	end

	@doc "Filters out AP IDs of users who mute activity thread"
	def exclude_thread_muter_ap_ids([], _activity), do: []

	def exclude_thread_muter_ap_ids(ap_ids, %Activity{} = activity) do
	thread_muter_ap_ids = ThreadMute.muter_ap_ids(activity.data["context"])

	Enum.uniq(ap_ids) -- thread_muter_ap_ids
	end

	def skip?(activity, user, opts \\ [])

	@spec skip?(Activity.t(), User.t(), Keyword.t()) :: boolean()
	def skip?(%Activity{} = activity, %User{} = user, opts) do
	[
	:self,
	:internal,
	:invisible,
	:block_from_strangers,
	:recently_followed,
	:filtered
	]
	\|> Enum.find(&skip?(&1, activity, user, opts))
	end

	def skip?(_activity, _user, _opts), do: false

	@spec skip?(atom(), Activity.t(), User.t(), Keyword.t()) :: boolean()
	def skip?(:self, %Activity{} = activity, %User{} = user, opts) do
	cond do
	opts[:type] == "poll" -> false
	activity.data["actor"] == user.ap_id -> true
	true -> false
	end
	end

	def skip?(:internal, %Activity{} = activity, _user, _opts) do
	actor = activity.data["actor"]
	user = User.get_cached_by_ap_id(actor)
	User.internal?(user)
	end

	def skip?(:invisible, %Activity{} = activity, _user, _opts) do
	actor = activity.data["actor"]
	user = User.get_cached_by_ap_id(actor)
	User.invisible?(user)
	end

	def skip?(
	:block_from_strangers,
	%Activity{} = activity,
	%User{notification_settings: %{block_from_strangers: true}} = user,
	opts
	) do
	actor = activity.data["actor"]
	follower = User.get_cached_by_ap_id(actor)

	cond do
	opts[:type] == "poll" -> false
	user.ap_id == actor -> false
	!User.following?(user, follower) -> true
	true -> false
	end
	end

	# To do: consider defining recency in hours and checking FollowingRelationship with a single SQL
	def skip?(
	:recently_followed,
	%Activity{data: %{"type" => "Follow"}} = activity,
	%User{} = user,
	_opts
	) do
	actor = activity.data["actor"]

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

	def skip?(:filtered, %{data: %{"type" => type}}, _user, _opts) when type in ["Follow", "Move"],
	do: false

	def skip?(:filtered, activity, user, _opts) do
	object = Object.normalize(activity, fetch: false)

	cond do
	is_nil(object) ->
	false

	object.data["actor"] == user.ap_id ->
	false

	not is_nil(regex = Pleroma.Filter.compose_regex(user, :re)) ->
	Regex.match?(regex, object.data["content"])

	true ->
	false
	end
	end

	def skip?(_type, _activity, _user, _opts), do: false

	def mark_as_read?(activity, target_user) do
	user = Activity.user_actor(activity)
	User.mutes_user?(target_user, user) \|\| CommonAPI.thread_muted?(activity, target_user)
	end

	def for_user_and_activity(user, activity) do
	from(n in __MODULE__,
	where: n.user_id == ^user.id,
	where: n.activity_id == ^activity.id
	)
	\|> Repo.one()
	end

	@spec mark_context_as_read(User.t(), String.t()) :: {integer(), nil \| [term()]}
	def mark_context_as_read(%User{id: id}, context) do
	from(
	n in Notification,
	join: a in assoc(n, :activity),
	where: n.user_id == ^id,
	where: n.seen == false,
	where: fragment("?->>'context'", a.data) == ^context
	)
	\|> Repo.update_all(set: [seen: true])
	end

	@doc "Streams a list of notifications over websockets and web push"
	@spec stream(list(Notification.t())) :: :ok
	def stream(notifications) do
	Enum.each(notifications, fn notification ->
	Streamer.stream(["user", "user:notification"], notification)
	Push.send(notification)
	end)
	end
	end
	diff --git a/lib/pleroma/web/streamer.ex b/lib/pleroma/web/streamer.ex
	index cc149e04c..aba42ee78 100644
	--- a/lib/pleroma/web/streamer.ex
	+++ b/lib/pleroma/web/streamer.ex
	@@ -1,431 +1,442 @@
	# Pleroma: A lightweight social networking server
	# Copyright © 2017-2022 Pleroma Authors <https://pleroma.social/>
	# SPDX-License-Identifier: AGPL-3.0-only

	defmodule Pleroma.Web.Streamer do
	require Logger
	require Pleroma.Constants

	alias Pleroma.Activity
	alias Pleroma.Chat.MessageReference
	alias Pleroma.Config
	alias Pleroma.Conversation.Participation
	+ alias Pleroma.Marker
	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.OAuth.Token
	alias Pleroma.Web.Plugs.OAuthScopesPlug
	alias Pleroma.Web.StreamerView
	require Pleroma.Constants

	@registry Pleroma.Web.StreamerRegistry

	def registry, do: @registry

	@public_streams Pleroma.Constants.public_streams()
	@local_streams ["public:local", "public:local:media"]
	@user_streams ["user", "user:notification", "direct", "user:pleroma_chat"]

	@doc "Expands and authorizes a stream, and registers the process for streaming."
	@spec get_topic_and_add_socket(
	stream :: String.t(),
	User.t() \| nil,
	Token.t() \| nil,
	map() \| nil
	) ::
	{:ok, topic :: String.t()} \| {:error, :bad_topic} \| {:error, :unauthorized}
	def get_topic_and_add_socket(stream, user, oauth_token, params \\ %{}) do
	with {:ok, topic} <- get_topic(stream, user, oauth_token, params) do
	add_socket(topic, oauth_token)
	end
	end

	defp can_access_stream(user, oauth_token, kind) do
	with {_, true} <- {:restrict?, Config.restrict_unauthenticated_access?(:timelines, kind)},
	{_, %User{id: user_id}, %Token{user_id: user_id}} <- {:user, user, oauth_token},
	{_, true} <-
	{:scopes,
	OAuthScopesPlug.filter_descendants(["read:statuses"], oauth_token.scopes) != []} do
	true
	else
	{:restrict?, _} ->
	true

	_ ->
	false
	end
	end

	@doc "Expand and authorizes a stream"
	@spec get_topic(stream :: String.t() \| nil, User.t() \| nil, Token.t() \| nil, map()) ::
	{:ok, topic :: String.t() \| nil} \| {:error, :bad_topic}
	def get_topic(stream, user, oauth_token, params \\ %{})

	def get_topic(nil = _stream, _user, _oauth_token, _params) do
	{:ok, nil}
	end

	# Allow all public steams if the instance allows unauthenticated access.
	# Otherwise, only allow users with valid oauth tokens.
	def get_topic(stream, user, oauth_token, _params) when stream in @public_streams do
	kind = if stream in @local_streams, do: :local, else: :federated

	if can_access_stream(user, oauth_token, kind) do
	{:ok, stream}
	else
	{:error, :unauthorized}
	end
	end

	# Allow all hashtags streams.
	def get_topic("hashtag", _user, _oauth_token, %{"tag" => tag} = _params) do
	{:ok, "hashtag:" <> tag}
	end

	# Allow remote instance streams.
	def get_topic("public:remote", user, oauth_token, %{"instance" => instance} = _params) do
	if can_access_stream(user, oauth_token, :federated) do
	{:ok, "public:remote:" <> instance}
	else
	{:error, :unauthorized}
	end
	end

	def get_topic("public:remote:media", user, oauth_token, %{"instance" => instance} = _params) do
	if can_access_stream(user, oauth_token, :federated) do
	{:ok, "public:remote:media:" <> instance}
	else
	{:error, :unauthorized}
	end
	end

	# Expand user streams.
	def get_topic(
	stream,
	%User{id: user_id} = user,
	%Token{user_id: user_id} = oauth_token,
	_params
	)
	when stream in @user_streams do
	# Note: "read" works for all user streams (not mentioning it since it's an ancestor scope)
	required_scopes =
	if stream == "user:notification" do
	["read:notifications"]
	else
	["read:statuses"]
	end

	if OAuthScopesPlug.filter_descendants(required_scopes, oauth_token.scopes) == [] do
	{:error, :unauthorized}
	else
	{:ok, stream <> ":" <> to_string(user.id)}
	end
	end

	def get_topic(stream, _user, _oauth_token, _params) when stream in @user_streams do
	{:error, :unauthorized}
	end

	# List streams.
	def get_topic(
	"list",
	%User{id: user_id} = user,
	%Token{user_id: user_id} = oauth_token,
	%{"list" => id}
	) do
	cond do
	OAuthScopesPlug.filter_descendants(["read", "read:lists"], oauth_token.scopes) == [] ->
	{:error, :unauthorized}

	Pleroma.List.get(id, user) ->
	{:ok, "list:" <> to_string(id)}

	true ->
	{:error, :bad_topic}
	end
	end

	def get_topic("list", _user, _oauth_token, _params) do
	{:error, :unauthorized}
	end

	def get_topic(_stream, _user, _oauth_token, _params) do
	{:error, :bad_topic}
	end

	@doc "Registers the process for streaming. Use `get_topic/3` to get the full authorized topic."
	def add_socket(topic, oauth_token) do
	if should_env_send?() do
	oauth_token_id = if oauth_token, do: oauth_token.id, else: false
	Registry.register(@registry, topic, oauth_token_id)
	end

	{:ok, topic}
	end

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

	def stream(topics, items) do
	if should_env_send?() do
	for topic <- List.wrap(topics), item <- List.wrap(items) do
	fun = fn -> do_stream(topic, item) end

	if Config.get([__MODULE__, :sync_streaming], false) do
	fun.()
	else
	spawn(fun)
	end
	end
	end
	end

	def filtered_by_user?(user, item, streamed_type \\ :activity)

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

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

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

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

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

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

	defp do_stream("follow_relationship", item) do
	user_topic = "user:#{item.follower.id}"
	text = StreamerView.render("follow_relationships_update.json", item, user_topic)

	Logger.debug("Trying to push follow relationship update to #{user_topic}\n\n")

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

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

	push_to_socket(user_topic, participation)
	end

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

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

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

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

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

	defp do_stream(topic, %Notification{} = item)
	when topic in ["user", "user:notification"] do
	user_topic = "#{topic}:#{item.user_id}"

	Registry.dispatch(@registry, user_topic, fn list ->
	Enum.each(list, fn {pid, _auth} ->
	send(pid, {:render_with_user, StreamerView, "notification.json", item, user_topic})
	end)
	end)
	end

	defp do_stream(topic, {user, %MessageReference{} = cm_ref})
	when topic in ["user", "user:pleroma_chat"] do
	topic = "#{topic}:#{user.id}"

	text = StreamerView.render("chat_update.json", %{chat_message_reference: cm_ref}, topic)

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

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

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

	hashtag_recipients =
	if Pleroma.Constants.as_public() in item.recipients do
	Pleroma.Hashtag.get_recipients_for_activity(item)
	\|> Enum.map(fn id -> "user:#{id}" end)
	else
	[]
	end

	all_recipients = Enum.uniq(recipient_topics ++ hashtag_recipients)

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

	+ defp do_stream(topic, %Marker{} = marker) do
	+ Registry.dispatch(@registry, "#{topic}:#{marker.user_id}", fn list ->
	+ Enum.each(list, fn {pid, _auth} ->
	+ text = StreamerView.render("marker.json", marker)
	+
	+ send(pid, {:text, text})
	+ end)
	+ end)
	+ end
	+
	defp do_stream(topic, item) do
	Logger.debug("Trying to push to #{topic}")
	Logger.debug("Pushing item to #{topic}")
	push_to_socket(topic, item)
	end

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

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

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

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

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

	defp push_to_socket(topic, %Activity{data: %{"type" => "Update"}} = item) do
	create_activity =
	Pleroma.Activity.get_create_by_object_ap_id(item.object.data["id"])
	\|> Map.put(:object, item.object)

	anon_render = StreamerView.render("status_update.json", create_activity, topic)

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

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

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

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

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

	def close_streams_by_oauth_token(oauth_token) do
	if should_env_send?() do
	Registry.select(
	@registry,
	[
	{
	{:"$1", :"$2", :"$3"},
	[{:==, :"$3", oauth_token.id}],
	[:"$2"]
	}
	]
	)
	\|> Enum.each(fn pid -> send(pid, :close) end)
	end
	end

	# In dev/prod the streamer registry is expected to be started, so return true
	# In test it is possible to have the registry started for a test so it will check
	# In benchmark it will never find the process alive and return false
	def should_env_send? do
	if Application.get_env(:pleroma, Pleroma.Application)[:streamer_registry] do
	true
	else
	case Process.whereis(@registry) do
	nil ->
	false

	pid ->
	Process.alive?(pid)
	end
	end
	end
	end
	diff --git a/lib/pleroma/web/views/streamer_view.ex b/lib/pleroma/web/views/streamer_view.ex
	index 079a37351..5806ba9ab 100644
	--- a/lib/pleroma/web/views/streamer_view.ex
	+++ b/lib/pleroma/web/views/streamer_view.ex
	@@ -1,201 +1,215 @@
	# Pleroma: A lightweight social networking server
	# Copyright © 2017-2022 Pleroma Authors <https://pleroma.social/>
	# SPDX-License-Identifier: AGPL-3.0-only

	defmodule Pleroma.Web.StreamerView do
	use Pleroma.Web, :view

	alias Pleroma.Activity
	alias Pleroma.Conversation.Participation
	+ alias Pleroma.Marker
	alias Pleroma.Notification
	alias Pleroma.User
	alias Pleroma.Web.MastodonAPI.NotificationView

	require Pleroma.Constants

	def render("update.json", %Activity{} = activity, %User{} = user, topic) do
	%{
	stream: render("stream.json", %{topic: topic}),
	event: "update",
	payload:
	Pleroma.Web.MastodonAPI.StatusView.render(
	"show.json",
	activity: activity,
	for: user
	)
	\|> Jason.encode!()
	}
	\|> Jason.encode!()
	end

	def render("status_update.json", %Activity{} = activity, %User{} = user, topic) do
	%{
	stream: render("stream.json", %{topic: topic}),
	event: "status.update",
	payload:
	Pleroma.Web.MastodonAPI.StatusView.render(
	"show.json",
	activity: activity,
	for: user
	)
	\|> Jason.encode!()
	}
	\|> Jason.encode!()
	end

	def render("notification.json", %Notification{} = notify, %User{} = user, topic) do
	%{
	stream: render("stream.json", %{topic: topic}),
	event: "notification",
	payload:
	NotificationView.render(
	"show.json",
	%{notification: notify, for: user}
	)
	\|> Jason.encode!()
	}
	\|> Jason.encode!()
	end

	def render("update.json", %Activity{} = activity, topic) do
	%{
	stream: render("stream.json", %{topic: topic}),
	event: "update",
	payload:
	Pleroma.Web.MastodonAPI.StatusView.render(
	"show.json",
	activity: activity
	)
	\|> Jason.encode!()
	}
	\|> Jason.encode!()
	end

	def render("status_update.json", %Activity{} = activity, topic) do
	%{
	stream: render("stream.json", %{topic: topic}),
	event: "status.update",
	payload:
	Pleroma.Web.MastodonAPI.StatusView.render(
	"show.json",
	activity: activity
	)
	\|> Jason.encode!()
	}
	\|> Jason.encode!()
	end

	def render("chat_update.json", %{chat_message_reference: cm_ref}, topic) do
	# Explicitly giving the cmr for the object here, so we don't accidentally
	# send a later 'last_message' that was inserted between inserting this and
	# streaming it out
	#
	# It also contains the chat with a cache of the correct unread count
	Logger.debug("Trying to stream out #{inspect(cm_ref)}")

	representation =
	Pleroma.Web.PleromaAPI.ChatView.render(
	"show.json",
	%{last_message: cm_ref, chat: cm_ref.chat}
	)

	%{
	stream: render("stream.json", %{topic: topic}),
	event: "pleroma:chat_update",
	payload:
	representation
	\|> Jason.encode!()
	}
	\|> Jason.encode!()
	end

	def render(
	"follow_relationships_update.json",
	%{follower: follower, following: following} = item,
	topic
	) do
	following_follower_count =
	if Enum.any?([following.hide_followers_count, following.hide_followers]) do
	0
	else
	following.follower_count
	end

	following_following_count =
	if Enum.any?([following.hide_follows_count, following.hide_follows]) do
	0
	else
	following.following_count
	end

	%{
	stream: render("stream.json", %{topic: topic}),
	event: "pleroma:follow_relationships_update",
	payload:
	%{
	state: item.state,
	follower: %{
	id: follower.id,
	follower_count: follower.follower_count,
	following_count: follower.following_count
	},
	following: %{
	id: following.id,
	follower_count: following_follower_count,
	following_count: following_following_count
	}
	}
	\|> Jason.encode!()
	}
	\|> Jason.encode!()
	end

	def render("conversation.json", %Participation{} = participation, topic) do
	%{
	stream: render("stream.json", %{topic: topic}),
	event: "conversation",
	payload:
	Pleroma.Web.MastodonAPI.ConversationView.render("participation.json", %{
	participation: participation,
	for: participation.user
	})
	\|> Jason.encode!()
	}
	\|> Jason.encode!()
	end

	+ def render("marker.json", %Marker{} = marker) do
	+ %{
	+ event: "marker",
	+ payload:
	+ Pleroma.Web.MastodonAPI.MarkerView.render(
	+ "markers.json",
	+ markers: [marker]
	+ )
	+ \|> Jason.encode!()
	+ }
	+ \|> Jason.encode!()
	+ end
	+
	def render("pleroma_respond.json", %{type: type, result: result} = params) do
	%{
	event: "pleroma:respond",
	payload:
	%{
	result: result,
	type: type
	}
	\|> Map.merge(maybe_error(params))
	\|> Jason.encode!()
	}
	\|> Jason.encode!()
	end

	def render("stream.json", %{topic: "user:pleroma_chat:" <> _}), do: ["user:pleroma_chat"]
	def render("stream.json", %{topic: "user:notification:" <> _}), do: ["user:notification"]
	def render("stream.json", %{topic: "user:" <> _}), do: ["user"]
	def render("stream.json", %{topic: "direct:" <> _}), do: ["direct"]
	def render("stream.json", %{topic: "list:" <> id}), do: ["list", id]
	def render("stream.json", %{topic: "hashtag:" <> tag}), do: ["hashtag", tag]

	def render("stream.json", %{topic: "public:remote:media:" <> instance}),
	do: ["public:remote:media", instance]

	def render("stream.json", %{topic: "public:remote:" <> instance}),
	do: ["public:remote", instance]

	def render("stream.json", %{topic: stream}) when stream in Pleroma.Constants.public_streams(),
	do: [stream]

	defp maybe_error(%{error: :bad_topic}), do: %{error: "bad_topic"}
	defp maybe_error(%{error: :unauthorized}), do: %{error: "unauthorized"}
	defp maybe_error(%{error: :already_authenticated}), do: %{error: "already_authenticated"}
	defp maybe_error(_), do: %{}
	end
	diff --git a/test/pleroma/notification_test.exs b/test/pleroma/notification_test.exs
	index 4b20e07cf..2ad950e84 100644
	--- a/test/pleroma/notification_test.exs
	+++ b/test/pleroma/notification_test.exs
	@@ -1,1125 +1,1156 @@
	# Pleroma: A lightweight social networking server
	# Copyright © 2017-2022 Pleroma Authors <https://pleroma.social/>
	# SPDX-License-Identifier: AGPL-3.0-only

	defmodule Pleroma.NotificationTest do
	use Pleroma.DataCase, async: false

	import Pleroma.Factory

	alias Pleroma.FollowingRelationship
	alias Pleroma.Notification
	alias Pleroma.Repo
	alias Pleroma.Tests.ObanHelpers
	alias Pleroma.User
	alias Pleroma.Web.ActivityPub.ActivityPub
	alias Pleroma.Web.ActivityPub.Builder
	alias Pleroma.Web.ActivityPub.Transmogrifier
	alias Pleroma.Web.CommonAPI
	alias Pleroma.Web.MastodonAPI.NotificationView
	+ alias Pleroma.Web.Streamer

	setup do
	Mox.stub_with(Pleroma.UnstubbedConfigMock, Pleroma.Test.StaticConfig)
	:ok
	end

	describe "create_notifications" do
	test "never returns nil" do
	user = insert(:user)
	other_user = insert(:user, %{invisible: true})

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

	refute {:ok, [nil]} == Notification.create_notifications(activity)
	end

	test "creates a report notification only for privileged users" do
	reporting_user = insert(:user)
	reported_user = insert(:user)
	moderator_user = insert(:user, is_moderator: true)

	clear_config([:instance, :moderator_privileges], [])
	{:ok, activity1} = CommonAPI.report(reporting_user, %{account_id: reported_user.id})
	{:ok, []} = Notification.create_notifications(activity1)

	clear_config([:instance, :moderator_privileges], [:reports_manage_reports])
	{:ok, activity2} = CommonAPI.report(reporting_user, %{account_id: reported_user.id})
	{:ok, [notification]} = Notification.create_notifications(activity2)

	assert notification.user_id == moderator_user.id
	assert notification.type == "pleroma:report"
	end

	test "suppresses notifications for own reports" do
	clear_config([:instance, :admin_privileges], [:reports_manage_reports])

	reporting_admin = insert(:user, is_admin: true)
	reported_user = insert(:user)
	other_admin = insert(:user, is_admin: true)

	{:ok, activity} = CommonAPI.report(reporting_admin, %{account_id: reported_user.id})

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

	refute notification.user_id == reporting_admin.id
	assert notification.user_id == other_admin.id
	assert notification.type == "pleroma:report"
	end

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

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

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

	assert notification.user_id == user.id
	assert notification.type == "pleroma:emoji_reaction"
	end

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

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

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

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

	assert [%Pleroma.Marker{unread_count: 2}] =
	Pleroma.Marker.get_markers(other_user, ["notifications"])
	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
	assert notification.type == "status"
	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

	test "does not create subscriber notification if mentioned" do
	user = insert(:user)
	subscriber = insert(:user)

	User.subscribe(subscriber, user)

	{:ok, status} = CommonAPI.post(user, %{status: "mentioning @#{subscriber.nickname}"})
	{:ok, [notification] = notifications} = Notification.create_notifications(status)

	assert length(notifications) == 1

	assert notification.user_id == subscriber.id
	assert notification.type == "mention"
	end

	test "it sends edited notifications to those who repeated a status" do
	user = insert(:user)
	repeated_user = insert(:user)
	other_user = insert(:user)

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

	{:ok, _activity_two} = CommonAPI.repeat(activity_one.id, repeated_user)

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

	assert [%{type: "reblog"}] = Notification.for_user(user)
	assert [%{type: "update"}] = Notification.for_user(repeated_user)
	assert [%{type: "mention"}] = Notification.for_user(other_user)
	end
	end

	test "create_poll_notifications/1" do
	[user1, user2, user3, _, _] = insert_list(5, :user)
	question = insert(:question, user: user1)
	activity = insert(:question_activity, question: question)

	{:ok, _, _} = CommonAPI.vote(question, user2, [0])
	{:ok, _, _} = CommonAPI.vote(question, user3, [1])

	{:ok, notifications} = Notification.create_poll_notifications(activity)

	assert [user2.id, user3.id, user1.id] == Enum.map(notifications, & &1.user_id)
	end

	describe "create_notification" do
	test "it disables notifications from strangers" do
	follower = insert(:user)

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

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

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

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

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

	test "it allows notifications from followees" do
	poster = insert(:user)

	receiver =
	insert(:user,
	notification_settings: %Pleroma.User.NotificationSetting{block_from_strangers: true}
	)

	CommonAPI.follow(poster, receiver)
	{:ok, activity} = CommonAPI.post(poster, %{status: "hey @#{receiver.nickname}"})
	assert Notification.create_notification(activity, receiver)
	end

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

	refute Notification.create_notification(activity, author)
	end

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

	{:ok, _, _, _} = CommonAPI.follow(user, subscriber)
	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

	test "it disables notifications from people who are invisible" do
	author = insert(:user, invisible: true)
	user = insert(:user)

	{:ok, status} = CommonAPI.post(author, %{status: "hey @#{user.nickname}"})
	refute Notification.create_notification(status, user)
	end

	test "it doesn't create notifications if content matches with an irreversible filter" do
	user = insert(:user)
	subscriber = insert(:user)

	User.subscribe(subscriber, user)
	insert(:filter, user: subscriber, phrase: "cofe", hide: true)

	{:ok, status} = CommonAPI.post(user, %{status: "got cofe?"})

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

	test "it creates notifications if content matches with a not irreversible filter" do
	user = insert(:user)
	subscriber = insert(:user)

	User.subscribe(subscriber, user)
	insert(:filter, user: subscriber, phrase: "cofe", hide: false)

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

	assert notification
	refute notification.seen
	end

	test "it creates notifications when someone likes user's status with a filtered word" do
	user = insert(:user)
	other_user = insert(:user)
	insert(:filter, user: user, phrase: "tesla", hide: true)

	{:ok, activity_one} = CommonAPI.post(user, %{status: "wow tesla"})
	{:ok, activity_two} = CommonAPI.favorite(activity_one.id, other_user)

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

	assert notification
	refute notification.seen
	end
	end

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

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

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

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

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

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

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

	notification =
	Repo.get(Notification, notification.id)
	\|> Repo.preload(:activity)

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

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

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

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

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

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

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

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

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

	assert notification.user_id == other_user.id
	end

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

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

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

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

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

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

	assert notification.user_id == other_user.id
	end

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

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

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

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

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

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

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

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

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

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

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

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

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

	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

	assert %Pleroma.Marker{} =
	m =
	Pleroma.Repo.get_by(
	Pleroma.Marker,
	user_id: other_user.id,
	timeline: "notifications"
	)

	assert m.last_read_id == to_string(n2.id)
	end
	+
	+ @tag needs_streamer: true
	+ test "it send updated marker to the 'user' and the 'user:notification' stream" do
	+ %{user: user, token: oauth_token} = oauth_access(["read"])
	+ other_user = insert(:user)
	+
	+ {:ok, _activity} =
	+ CommonAPI.post(other_user, %{
	+ status: "hi @#{user.nickname}!"
	+ })
	+
	+ [%{id: notification_id}] = Notification.for_user(user)
	+
	+ notification_id = to_string(notification_id)
	+
	+ task =
	+ Task.async(fn ->
	+ {:ok, _topic} =
	+ Streamer.get_topic_and_add_socket("user:notification", user, oauth_token)
	+
	+ assert_receive {:text, event}, 4_000
	+
	+ assert %{"event" => "marker", "payload" => payload} = Jason.decode!(event)
	+
	+ assert %{"notifications" => %{"last_read_id" => ^notification_id}} =
	+ Jason.decode!(payload)
	+ end)
	+
	+ Notification.set_read_up_to(user, notification_id)
	+ Task.await(task)
	+ end
	end

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

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

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

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

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

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

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

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

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

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

	enabled_receivers = Notification.get_notified_from_activity(activity)

	assert other_user in enabled_receivers
	end

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

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

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

	enabled_receivers = Notification.get_notified_from_activity(activity)

	assert other_user in enabled_receivers
	end

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

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

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

	enabled_receivers = Notification.get_notified_from_activity(activity)

	assert other_user not in enabled_receivers
	end

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

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

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

	enabled_receivers = Notification.get_notified_from_activity(activity_two)

	assert other_user not in enabled_receivers
	end

	test "it only notifies the post's author 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, like_data, _} = Builder.like(third_user, activity_one.object)

	{:ok, like, _} =
	like_data
	\|> Map.put("to", [other_user.ap_id \| like_data["to"]])
	\|> ActivityPub.persist(local: true)

	enabled_receivers = Notification.get_notified_from_activity(like)

	assert other_user not in enabled_receivers
	end

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

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

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

	enabled_receivers = Notification.get_notified_from_activity(activity_two)

	assert other_user not in enabled_receivers
	end

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

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

	enabled_receivers = Notification.get_notified_from_activity(activity)

	assert [] == enabled_receivers
	end

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

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

	enabled_receivers = Notification.get_notified_from_activity(activity)

	assert [] == enabled_receivers
	end

	test "it does not return thread-muting recipient in recipients list" do
	user = insert(:user)
	other_user = insert(:user)

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

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

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

	enabled_receivers = Notification.get_notified_from_activity(same_context_activity)

	refute other_user in enabled_receivers
	end

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

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

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

	enabled_receivers = Notification.get_notified_from_activity(activity)

	assert [] == enabled_receivers
	end

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

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

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

	enabled_receivers = Notification.get_notified_from_activity(activity)

	assert [other_user] == enabled_receivers
	end

	test "it sends edited notifications to those who repeated a status" do
	user = insert(:user)
	repeated_user = insert(:user)
	other_user = insert(:user)

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

	{:ok, _activity_two} = CommonAPI.repeat(activity_one.id, repeated_user)

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

	enabled_receivers = Notification.get_notified_from_activity(edit_activity)

	assert repeated_user in enabled_receivers
	refute other_user in enabled_receivers
	end
	end

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

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

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

	{:ok, _} = CommonAPI.favorite(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, :not_found} = 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 "notifications are deleted if a local user is deleted" do
	user = insert(:user)
	other_user = insert(:user)

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

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

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

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

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

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

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

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

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

	remote_user_url = remote_user.ap_id

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

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

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

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

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

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

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

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

	describe "for_user" do
	setup do
	user = insert(:user)

	{:ok, %{user: user}}
	end

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

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

	[notification] = Notification.for_user(user)

	assert notification.activity.object
	assert notification.seen
	end

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

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

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

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

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

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

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

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

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

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

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

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

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

	test "it doesn't return notifications for muted thread", %{user: user} do
	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 doesn't return notifications from a muted user when with_muted is set", %{user: user} do
	muted = insert(:user)
	{:ok, _user_relationships} = User.mute(user, muted)

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

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

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

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

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

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

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

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

	test "it returns notifications from muted threads when with_muted is set", %{user: user} do
	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

	test "it doesn't return notifications about mentions with filtered word", %{user: user} do
	insert(:filter, user: user, phrase: "cofe", hide: true)
	another_user = insert(:user)

	{:ok, _activity} = CommonAPI.post(another_user, %{status: "@#{user.nickname} got cofe?"})

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

	test "it returns notifications about mentions with not hidden filtered word", %{user: user} do
	insert(:filter, user: user, phrase: "test", hide: false)
	another_user = insert(:user)

	{:ok, _} = CommonAPI.post(another_user, %{status: "@#{user.nickname} test"})

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

	test "it returns notifications about favorites with filtered word", %{user: user} do
	insert(:filter, user: user, phrase: "cofe", hide: true)
	another_user = insert(:user)

	{:ok, activity} = CommonAPI.post(user, %{status: "Give me my cofe!"})
	{:ok, _} = CommonAPI.favorite(activity.id, another_user)

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

	test "it returns notifications when related object is without content and filters are defined",
	%{user: user} do
	followed_user = insert(:user, is_locked: true)

	insert(:filter, user: followed_user, phrase: "test", hide: true)

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

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

	assert {:ok, _} = CommonAPI.accept_follow_request(user, followed_user)

	assert [notification] = Notification.for_user(followed_user)

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

File Metadata

Mime Type: text/x-diff
Expires: Fri, Oct 30, 11:51 AM (19 h, 46 m)
Storage Engine: blob
Storage Format: Raw Data
Storage Handle: 1800093
Default Alt Text: (82 KB)

No OneTemporaryActions

View Options

File Metadata

Event Timeline

No OneTemporary
Actions