No OneTemporary
Actions

Authored By

Unknown

Size

84 KB

Referenced Files

None

Subscribers

None

View Options

	diff --git a/lib/pleroma/constants.ex b/lib/pleroma/constants.ex
	index ed60bcc37..77bc4bfac 100644
	--- a/lib/pleroma/constants.ex
	+++ b/lib/pleroma/constants.ex
	@@ -1,97 +1,101 @@
	# Pleroma: A lightweight social networking server
	# Copyright © 2017-2022 Pleroma Authors <https://pleroma.social/>
	# SPDX-License-Identifier: AGPL-3.0-only

	defmodule Pleroma.Constants do
	use Const

	const(as_public, do: "https://www.w3.org/ns/activitystreams#Public")

	const(object_internal_fields,
	do: [
	"reactions",
	"reaction_count",
	"likes",
	"like_count",
	"announcements",
	"announcement_count",
	"emoji",
	"context_id",
	"deleted_activity_id",
	"pleroma_internal",
	"generator"
	]
	)

	const(static_only_files,
	do:
	~w(index.html robots.txt static static-fe finmoji emoji packs sounds images instance sw.js sw-pleroma.js favicon.png schemas doc embed.js embed.css)
	)

	const(status_updatable_fields,
	do: [
	"source",
	"tag",
	"updated",
	"emoji",
	"content",
	"summary",
	"sensitive",
	"attachment",
	"generator"
	]
	)

	const(status_object_types,
	do: [
	"Note",
	"Question",
	"Audio",
	"Video",
	"Event",
	"Article",
	"Page"
	]
	)

	const(updatable_object_types,
	do: [
	"Note",
	"Question",
	"Audio",
	"Video",
	"Event",
	"Article",
	"Page"
	]
	)

	const(actor_types,
	do: [
	"Application",
	"Group",
	"Organization",
	"Person",
	"Service"
	]
	)

	# basic regex, just there to weed out potential mistakes
	# https://datatracker.ietf.org/doc/html/rfc2045#section-5.1
	const(mime_regex,
	do: ~r/^[^[:cntrl:] ()<>@,;:\\"\/\[\]?=]+\/[^[:cntrl:] ()<>@,;:\\"\/\[\]?=]+(; .*)?$/
	)

	const(upload_object_types, do: ["Document", "Image"])

	const(activity_json_canonical_mime_type,
	do: "application/ld+json; profile=\"https://www.w3.org/ns/activitystreams\""
	)

	const(activity_json_mime_types,
	do: [
	"application/ld+json; profile=\"https://www.w3.org/ns/activitystreams\"",
	"application/activity+json"
	]
	)
	+
	+ const(public_streams,
	+ do: ["public", "public:local", "public:media", "public:local:media"]
	+ )
	end
	diff --git a/lib/pleroma/web/mastodon_api/websocket_handler.ex b/lib/pleroma/web/mastodon_api/websocket_handler.ex
	index 6233c3340..2707673ba 100644
	--- a/lib/pleroma/web/mastodon_api/websocket_handler.ex
	+++ b/lib/pleroma/web/mastodon_api/websocket_handler.ex
	@@ -1,257 +1,257 @@
	# Pleroma: A lightweight social networking server
	# Copyright © 2017-2022 Pleroma Authors <https://pleroma.social/>
	# SPDX-License-Identifier: AGPL-3.0-only

	defmodule Pleroma.Web.MastodonAPI.WebsocketHandler do
	require Logger

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

	@behaviour :cowboy_websocket

	# Client ping period.
	@tick :timer.seconds(30)
	# Cowboy timeout period.
	@timeout :timer.seconds(60)
	# Hibernate every X messages
	@hibernate_every 100

	def init(%{qs: qs} = req, state) do
	with params <- Enum.into(:cow_qs.parse_qs(qs), %{}),
	sec_websocket <- :cowboy_req.header("sec-websocket-protocol", req, nil),
	access_token <- Map.get(params, "access_token"),
	{:ok, user, oauth_token} <- authenticate_request(access_token, sec_websocket),
	{:ok, topic} <- Streamer.get_topic(params["stream"], user, oauth_token, params) do
	req =
	if sec_websocket do
	:cowboy_req.set_resp_header("sec-websocket-protocol", sec_websocket, req)
	else
	req
	end

	topics =
	if topic do
	[topic]
	else
	[]
	end

	{:cowboy_websocket, req,
	%{user: user, topics: topics, oauth_token: oauth_token, count: 0, timer: nil},
	%{idle_timeout: @timeout}}
	else
	{:error, :bad_topic} ->
	Logger.debug("#{__MODULE__} bad topic #{inspect(req)}")
	req = :cowboy_req.reply(404, req)
	{:ok, req, state}

	{:error, :unauthorized} ->
	Logger.debug("#{__MODULE__} authentication error: #{inspect(req)}")
	req = :cowboy_req.reply(401, req)
	{:ok, req, state}
	end
	end

	def websocket_init(state) do
	Logger.debug(
	"#{__MODULE__} accepted websocket connection for user #{(state.user \|\| %{id: "anonymous"}).id}, topics #{state.topics}"
	)

	Enum.each(state.topics, fn topic -> Streamer.add_socket(topic, state.oauth_token) end)
	{:ok, %{state \| timer: timer()}}
	end

	# Client's Pong frame.
	def websocket_handle(:pong, state) do
	if state.timer, do: Process.cancel_timer(state.timer)
	{:ok, %{state \| timer: timer()}}
	end

	# We only receive pings for now
	def websocket_handle(:ping, state), do: {:ok, state}

	def websocket_handle({:text, text}, state) do
	with {:ok, %{} = event} <- Jason.decode(text) do
	handle_client_event(event, state)
	else
	_ ->
	Logger.error("#{__MODULE__} received non-JSON event: #{inspect(text)}")
	{:ok, state}
	end
	end

	def websocket_handle(frame, state) do
	Logger.error("#{__MODULE__} received frame: #{inspect(frame)}")
	{:ok, state}
	end

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

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

	def websocket_info({:text, message}, state) do
	# If the websocket processed X messages, force an hibernate/GC.
	# We don't hibernate at every message to balance CPU usage/latency with RAM usage.
	if state.count > @hibernate_every do
	{:reply, {:text, message}, %{state \| count: 0}, :hibernate}
	else
	{:reply, {:text, message}, %{state \| count: state.count + 1}}
	end
	end

	# Ping tick. We don't re-queue a timer there, it is instead queued when :pong is received.
	# As we hibernate there, reset the count to 0.
	# If the client misses :pong, Cowboy will automatically timeout the connection after
	# `@idle_timeout`.
	def websocket_info(:tick, state) do
	{:reply, :ping, %{state \| timer: nil, count: 0}, :hibernate}
	end

	def websocket_info(:close, state) do
	{:stop, state}
	end

	# State can be `[]` only in case we terminate before switching to websocket,
	# we already log errors for these cases in `init/1`, so just do nothing here
	def terminate(_reason, _req, []), do: :ok

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

	Enum.each(state.topics, fn topic -> Streamer.remove_socket(topic) end)
	:ok
	end

	# Public streams without authentication.
	defp authenticate_request(nil, nil) do
	{:ok, nil, nil}
	end

	# Authenticated streams.
	defp authenticate_request(access_token, sec_websocket) do
	token = access_token \|\| sec_websocket

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

	defp timer do
	Process.send_after(self(), :tick, @tick)
	end

	defp handle_client_event(%{"type" => "subscribe", "stream" => _topic} = params, state) do
	with {_, {:ok, topic}} <-
	{:topic, Streamer.get_topic(params["stream"], state.user, state.oauth_token, params)},
	{_, false} <- {:subscribed, topic in state.topics} do
	Streamer.add_socket(topic, state.oauth_token)

	{[
	{:text,
	StreamerView.render("pleroma_respond.json", %{type: "subscribe", result: "success"})}
	], %{state \| topics: [topic \| state.topics]}}
	else
	{:subscribed, true} ->
	{[
	{:text,
	StreamerView.render("pleroma_respond.json", %{type: "subscribe", result: "ignored"})}
	], state}

	{:topic, {:error, error}} ->
	{[
	{:text,
	StreamerView.render("pleroma_respond.json", %{
	type: "subscribe",
	result: "error",
	error: error
	})}
	], state}
	end
	end

	defp handle_client_event(%{"type" => "unsubscribe", "stream" => _topic} = params, state) do
	with {_, {:ok, topic}} <-
	{:topic, Streamer.get_topic(params["stream"], state.user, state.oauth_token, params)},
	{_, true} <- {:subscribed, topic in state.topics} do
	Streamer.remove_socket(topic)

	{[
	{:text,
	StreamerView.render("pleroma_respond.json", %{type: "unsubscribe", result: "success"})}
	], %{state \| topics: List.delete(state.topics, topic)}}
	else
	{:subscribed, false} ->
	{[
	{:text,
	StreamerView.render("pleroma_respond.json", %{type: "unsubscribe", result: "ignored"})}
	], state}

	{:topic, {:error, error}} ->
	{[
	{:text,
	StreamerView.render("pleroma_respond.json", %{
	type: "unsubscribe",
	result: "error",
	error: error
	})}
	], state}
	end
	end

	defp handle_client_event(
	%{"type" => "pleroma.authenticate", "token" => access_token} = _params,
	state
	) do
	with {:auth, nil, nil} <- {:auth, state.user, state.oauth_token},
	{:ok, user, oauth_token} <- authenticate_request(access_token, nil) do
	{[
	{:text,
	StreamerView.render("pleroma_respond.json", %{
	type: "pleroma.authenticate",
	result: "success"
	})}
	], %{state \| user: user, oauth_token: oauth_token}}
	else
	{:auth, _, _} ->
	{[
	{:text,
	StreamerView.render("pleroma_respond.json", %{
	type: "pleroma.authenticate",
	result: "error",
	error: :already_authenticated
	})}
	], state}

	_ ->
	{[
	{:text,
	StreamerView.render("pleroma_respond.json", %{
	type: "pleroma.authenticate",
	result: "error",
	error: :unauthorized
	})}
	], state}
	end
	end

	defp handle_client_event(params, state) do
	Logger.error("#{__MODULE__} received unknown event: #{inspect(params)}")
	{[], state}
	end
	end
	diff --git a/lib/pleroma/web/streamer.ex b/lib/pleroma/web/streamer.ex
	index 6b5fdbf0c..70e46617a 100644
	--- a/lib/pleroma/web/streamer.ex
	+++ b/lib/pleroma/web/streamer.ex
	@@ -1,414 +1,421 @@
	# 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.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

	@mix_env Mix.env()
	@registry Pleroma.Web.StreamerRegistry

	def registry, do: @registry

	- @public_streams ["public", "public:local", "public:media", "public:local:media"]
	+ @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.t() \| 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.t()) ::
	{: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
	spawn(fn -> do_stream(topic, item) 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?(user, parent) 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
	- text = StreamerView.render("follow_relationships_update.json", item)
	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.is_public?(item) do
	true ->
	Pleroma.List.get_lists_from_activity(item)

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

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

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

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

	defp do_stream(topic, %Notification{} = item)
	when topic in ["user", "user:notification"] do
	- Registry.dispatch(@registry, "#{topic}:#{item.user_id}", fn list ->
	+ 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})
	+ 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})
	+ 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)

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

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

	defp push_to_socket(topic, %Participation{} = participation) do
	- rendered = StreamerView.render("conversation.json", participation)
	+ 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)
	+ 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})
	+ 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)
	+ Logger.debug("topic=#{topic}")
	+ 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})
	+ 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 test environement, only return true if the registry is started.
	# In benchmark environment, returns false.
	# In any other environment, always returns true.
	cond do
	@mix_env == :test ->
	def should_env_send? do
	case Process.whereis(@registry) do
	nil ->
	false

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

	@mix_env == :benchmark ->
	def should_env_send?, do: false

	true ->
	def should_env_send?, do: true
	end
	end
	diff --git a/lib/pleroma/web/views/streamer_view.ex b/lib/pleroma/web/views/streamer_view.ex
	index 0cdcb1918..f591da9a6 100644
	--- a/lib/pleroma/web/views/streamer_view.ex
	+++ b/lib/pleroma/web/views/streamer_view.ex
	@@ -1,157 +1,183 @@
	# 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.Notification
	alias Pleroma.User
	alias Pleroma.Web.MastodonAPI.NotificationView

	- def render("update.json", %Activity{} = activity, %User{} = user) do
	+ 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) do
	+ 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) do
	+ 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) do
	+ 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) do
	+ 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}) do
	+ 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", item) do
	+ def render("follow_relationships_update.json", item, topic) do
	%{
	+ stream: render("stream.json", %{topic: topic}),
	event: "pleroma:follow_relationships_update",
	payload:
	%{
	state: item.state,
	follower: %{
	id: item.follower.id,
	follower_count: item.follower.follower_count,
	following_count: item.follower.following_count
	},
	following: %{
	id: item.following.id,
	follower_count: item.following.follower_count,
	following_count: item.following.following_count
	}
	}
	\|> Jason.encode!()
	}
	\|> Jason.encode!()
	end

	- def render("conversation.json", %Participation{} = participation) do
	+ 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("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/integration/mastodon_websocket_test.exs b/test/pleroma/integration/mastodon_websocket_test.exs
	index 827c7b5b0..d2e98df3b 100644
	--- a/test/pleroma/integration/mastodon_websocket_test.exs
	+++ b/test/pleroma/integration/mastodon_websocket_test.exs
	@@ -1,324 +1,357 @@
	# Pleroma: A lightweight social networking server
	# Copyright © 2017-2022 Pleroma Authors <https://pleroma.social/>
	# SPDX-License-Identifier: AGPL-3.0-only

	defmodule Pleroma.Integration.MastodonWebsocketTest do
	# Needs a streamer, needs to stay synchronous
	use Pleroma.DataCase

	import ExUnit.CaptureLog
	import Pleroma.Factory

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

	@moduletag needs_streamer: true, capture_log: true

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

	def start_socket(qs \\ nil, headers \\ []) do
	path =
	case qs do
	nil -> @path
	qs -> @path <> qs
	end

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

	defp decode_json(json) do
	with {:ok, %{"event" => event, "payload" => payload_text}} <- Jason.decode(json),
	{:ok, payload} <- Jason.decode(payload_text) do
	{:ok, %{"event" => event, "payload" => payload}}
	end
	end

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

	test "requires authentication and a valid token for protected streams" do
	capture_log(fn ->
	assert {:error, %WebSockex.RequestError{code: 401}} =
	start_socket("?stream=user&access_token=aaaaaaaaaaaa")

	assert {:error, %WebSockex.RequestError{code: 401}} = start_socket("?stream=user")
	Process.sleep(30)
	end)
	end

	test "allows unified stream" do
	assert {:ok, _} = start_socket()
	end

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

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

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

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

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

	assert json == view_json
	end

	describe "subscribing via WebSocket" do
	test "can subscribe" do
	user = insert(:user)
	{:ok, pid} = start_socket()
	WebsocketClient.send_text(pid, %{type: "subscribe", stream: "public"} \|> Jason.encode!())
	assert_receive {:text, raw_json}, 1_000

	assert {:ok,
	%{
	"event" => "pleroma.respond",
	"payload" => %{"type" => "subscribe", "result" => "success"}
	}} = decode_json(raw_json)

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

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

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

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

	assert json == view_json
	end

	+ test "can subscribe to multiple streams" do
	+ user = insert(:user)
	+ {:ok, pid} = start_socket()
	+ WebsocketClient.send_text(pid, %{type: "subscribe", stream: "public"} \|> Jason.encode!())
	+ assert_receive {:text, raw_json}, 1_000
	+
	+ assert {:ok,
	+ %{
	+ "event" => "pleroma.respond",
	+ "payload" => %{"type" => "subscribe", "result" => "success"}
	+ }} = decode_json(raw_json)
	+
	+ WebsocketClient.send_text(pid, %{type: "subscribe", stream: "hashtag", tag: "mew"} \|> Jason.encode!())
	+ assert_receive {:text, raw_json}, 1_000
	+
	+ assert {:ok,
	+ %{
	+ "event" => "pleroma.respond",
	+ "payload" => %{"type" => "subscribe", "result" => "success"}
	+ }} = decode_json(raw_json)
	+
	+ {:ok, _activity} = CommonAPI.post(user, %{status: "nice echo chamber #mew"})
	+
	+ assert_receive {:text, raw_json}, 1_000
	+ assert {:ok, %{"stream" => stream1}} = Jason.decode(raw_json)
	+ assert_receive {:text, raw_json}, 1_000
	+ assert {:ok, %{"stream" => stream2}} = Jason.decode(raw_json)
	+
	+ streams = [stream1, stream2]
	+ assert ["hashtag", "mew"] in streams
	+ assert ["public"] in streams
	+ end
	+
	test "won't double subscribe" do
	user = insert(:user)
	{:ok, pid} = start_socket()
	WebsocketClient.send_text(pid, %{type: "subscribe", stream: "public"} \|> Jason.encode!())
	assert_receive {:text, raw_json}, 1_000

	assert {:ok,
	%{
	"event" => "pleroma.respond",
	"payload" => %{"type" => "subscribe", "result" => "success"}
	}} = decode_json(raw_json)

	WebsocketClient.send_text(pid, %{type: "subscribe", stream: "public"} \|> Jason.encode!())
	assert_receive {:text, raw_json}, 1_000

	assert {:ok,
	%{
	"event" => "pleroma.respond",
	"payload" => %{"type" => "subscribe", "result" => "ignored"}
	}} = decode_json(raw_json)

	{:ok, _activity} = CommonAPI.post(user, %{status: "nice echo chamber"})

	assert_receive {:text, _}, 1_000
	refute_receive {:text, _}, 1_000
	end

	test "can unsubscribe" do
	user = insert(:user)
	{:ok, pid} = start_socket()
	WebsocketClient.send_text(pid, %{type: "subscribe", stream: "public"} \|> Jason.encode!())
	assert_receive {:text, raw_json}, 1_000

	assert {:ok,
	%{
	"event" => "pleroma.respond",
	"payload" => %{"type" => "subscribe", "result" => "success"}
	}} = decode_json(raw_json)

	WebsocketClient.send_text(pid, %{type: "unsubscribe", stream: "public"} \|> Jason.encode!())
	assert_receive {:text, raw_json}, 1_000

	assert {:ok,
	%{
	"event" => "pleroma.respond",
	"payload" => %{"type" => "unsubscribe", "result" => "success"}
	}} = decode_json(raw_json)

	{:ok, _activity} = CommonAPI.post(user, %{status: "nice echo chamber"})
	refute_receive {:text, _}, 1_000
	end
	end

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

	user = insert(:user)

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

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

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

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

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

	capture_log(fn ->
	assert {:error, %WebSockex.RequestError{code: 401}} = start_socket("?stream=user")
	Process.sleep(30)
	end)
	end

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

	capture_log(fn ->
	assert {:error, %WebSockex.RequestError{code: 401}} =
	start_socket("?stream=user:notification")

	Process.sleep(30)
	end)
	end

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

	capture_log(fn ->
	assert {:error, %WebSockex.RequestError{code: 401}} =
	start_socket("?stream=user", [{"Sec-WebSocket-Protocol", "I am a friend"}])

	Process.sleep(30)
	end)
	end

	test "accepts valid token on client-sent event", %{token: token} do
	assert {:ok, pid} = start_socket()

	WebsocketClient.send_text(
	pid,
	%{type: "pleroma.authenticate", token: token.token} \|> Jason.encode!()
	)

	assert_receive {:text, raw_json}, 1_000

	assert {:ok,
	%{
	"event" => "pleroma.respond",
	"payload" => %{"type" => "pleroma.authenticate", "result" => "success"}
	}} = decode_json(raw_json)

	WebsocketClient.send_text(pid, %{type: "subscribe", stream: "user"} \|> Jason.encode!())
	assert_receive {:text, raw_json}, 1_000

	assert {:ok,
	%{
	"event" => "pleroma.respond",
	"payload" => %{"type" => "subscribe", "result" => "success"}
	}} = decode_json(raw_json)
	end

	test "rejects invalid token on client-sent event" do
	assert {:ok, pid} = start_socket()

	WebsocketClient.send_text(
	pid,
	%{type: "pleroma.authenticate", token: "Something else"} \|> Jason.encode!()
	)

	assert_receive {:text, raw_json}, 1_000

	assert {:ok,
	%{
	"event" => "pleroma.respond",
	"payload" => %{
	"type" => "pleroma.authenticate",
	"result" => "error",
	"error" => "unauthorized"
	}
	}} = decode_json(raw_json)
	end

	test "rejects new authenticate request if already logged-in", %{token: token} do
	assert {:ok, pid} = start_socket()

	WebsocketClient.send_text(
	pid,
	%{type: "pleroma.authenticate", token: token.token} \|> Jason.encode!()
	)

	assert_receive {:text, raw_json}, 1_000

	assert {:ok,
	%{
	"event" => "pleroma.respond",
	"payload" => %{"type" => "pleroma.authenticate", "result" => "success"}
	}} = decode_json(raw_json)

	WebsocketClient.send_text(
	pid,
	%{type: "pleroma.authenticate", token: "Something else"} \|> Jason.encode!()
	)

	assert_receive {:text, raw_json}, 1_000

	assert {:ok,
	%{
	"event" => "pleroma.respond",
	"payload" => %{
	"type" => "pleroma.authenticate",
	"result" => "error",
	"error" => "already_authenticated"
	}
	}} = decode_json(raw_json)
	end

	test "disconnect when token is revoked", %{app: app, user: user, token: token} do
	assert {:ok, _} = start_socket("?stream=user:notification&access_token=#{token.token}")
	assert {:ok, _} = start_socket("?stream=user&access_token=#{token.token}")

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

	{:ok, token2} = OAuth.Token.exchange_token(app, auth)
	assert {:ok, _} = start_socket("?stream=user&access_token=#{token2.token}")

	OAuth.Token.Strategy.Revoke.revoke(token)

	assert_receive {:close, _}
	assert_receive {:close, _}
	refute_receive {:close, _}
	end
	end
	end
	diff --git a/test/pleroma/web/streamer_test.exs b/test/pleroma/web/streamer_test.exs
	index da97b87d8..cc57a2989 100644
	--- a/test/pleroma/web/streamer_test.exs
	+++ b/test/pleroma/web/streamer_test.exs
	@@ -1,1072 +1,1072 @@
	# Pleroma: A lightweight social networking server
	# Copyright © 2017-2022 Pleroma Authors <https://pleroma.social/>
	# SPDX-License-Identifier: AGPL-3.0-only

	defmodule Pleroma.Web.StreamerTest do
	use Pleroma.DataCase

	import Pleroma.Factory

	alias Pleroma.Chat
	alias Pleroma.Chat.MessageReference
	alias Pleroma.Conversation.Participation
	alias Pleroma.List
	alias Pleroma.Object
	alias Pleroma.User
	alias Pleroma.Web.CommonAPI
	alias Pleroma.Web.Streamer
	alias Pleroma.Web.StreamerView

	@moduletag needs_streamer: true, capture_log: true

	setup do: clear_config([:instance, :skip_thread_containment])

	describe "get_topic/_ (unauthenticated)" do
	test "allows no stream" do
	assert {:ok, nil} = Streamer.get_topic(nil, nil, nil)
	end

	test "allows public" do
	assert {:ok, "public"} = Streamer.get_topic("public", nil, nil)
	assert {:ok, "public:local"} = Streamer.get_topic("public:local", nil, nil)
	assert {:ok, "public:media"} = Streamer.get_topic("public:media", nil, nil)
	assert {:ok, "public:local:media"} = Streamer.get_topic("public:local:media", nil, nil)
	end

	test "rejects local public streams if restricted_unauthenticated is on" do
	clear_config([:restrict_unauthenticated, :timelines, :local], true)

	assert {:error, :unauthorized} = Streamer.get_topic("public:local", nil, nil)
	assert {:error, :unauthorized} = Streamer.get_topic("public:local:media", nil, nil)
	end

	test "rejects remote public streams if restricted_unauthenticated is on" do
	clear_config([:restrict_unauthenticated, :timelines, :federated], true)

	assert {:error, :unauthorized} = Streamer.get_topic("public", nil, nil)
	assert {:error, :unauthorized} = Streamer.get_topic("public:media", nil, nil)

	assert {:error, :unauthorized} =
	Streamer.get_topic("public:remote", nil, nil, %{"instance" => "lain.com"})

	assert {:error, :unauthorized} =
	Streamer.get_topic("public:remote:media", nil, nil, %{"instance" => "lain.com"})
	end

	test "allows instance streams" do
	assert {:ok, "public:remote:lain.com"} =
	Streamer.get_topic("public:remote", nil, nil, %{"instance" => "lain.com"})

	assert {:ok, "public:remote:media:lain.com"} =
	Streamer.get_topic("public:remote:media", nil, nil, %{"instance" => "lain.com"})
	end

	test "allows hashtag streams" do
	assert {:ok, "hashtag:cofe"} = Streamer.get_topic("hashtag", nil, nil, %{"tag" => "cofe"})
	end

	test "disallows user streams" do
	assert {:error, _} = Streamer.get_topic("user", nil, nil)
	assert {:error, _} = Streamer.get_topic("user:notification", nil, nil)
	assert {:error, _} = Streamer.get_topic("direct", nil, nil)
	end

	test "disallows list streams" do
	assert {:error, _} = Streamer.get_topic("list", nil, nil, %{"list" => 42})
	end
	end

	describe "get_topic/_ (authenticated)" do
	setup do: oauth_access(["read"])

	test "allows public streams (regardless of OAuth token scopes)", %{
	user: user,
	token: read_oauth_token
	} do
	with oauth_token <- [nil, read_oauth_token] do
	assert {:ok, "public"} = Streamer.get_topic("public", user, oauth_token)
	assert {:ok, "public:local"} = Streamer.get_topic("public:local", user, oauth_token)
	assert {:ok, "public:media"} = Streamer.get_topic("public:media", user, oauth_token)

	assert {:ok, "public:local:media"} =
	Streamer.get_topic("public:local:media", user, oauth_token)
	end
	end

	test "allows local public streams if restricted_unauthenticated is on", %{
	user: user,
	token: oauth_token
	} do
	clear_config([:restrict_unauthenticated, :timelines, :local], true)

	%{token: read_notifications_token} = oauth_access(["read:notifications"], user: user)
	%{token: badly_scoped_token} = oauth_access(["irrelevant:scope"], user: user)

	assert {:ok, "public:local"} = Streamer.get_topic("public:local", user, oauth_token)

	assert {:ok, "public:local:media"} =
	Streamer.get_topic("public:local:media", user, oauth_token)

	for token <- [read_notifications_token, badly_scoped_token] do
	assert {:error, :unauthorized} = Streamer.get_topic("public:local", user, token)

	assert {:error, :unauthorized} = Streamer.get_topic("public:local:media", user, token)
	end
	end

	test "allows remote public streams if restricted_unauthenticated is on", %{
	user: user,
	token: oauth_token
	} do
	clear_config([:restrict_unauthenticated, :timelines, :federated], true)

	%{token: read_notifications_token} = oauth_access(["read:notifications"], user: user)
	%{token: badly_scoped_token} = oauth_access(["irrelevant:scope"], user: user)

	assert {:ok, "public"} = Streamer.get_topic("public", user, oauth_token)
	assert {:ok, "public:media"} = Streamer.get_topic("public:media", user, oauth_token)

	assert {:ok, "public:remote:lain.com"} =
	Streamer.get_topic("public:remote", user, oauth_token, %{"instance" => "lain.com"})

	assert {:ok, "public:remote:media:lain.com"} =
	Streamer.get_topic("public:remote:media", user, oauth_token, %{
	"instance" => "lain.com"
	})

	for token <- [read_notifications_token, badly_scoped_token] do
	assert {:error, :unauthorized} = Streamer.get_topic("public", user, token)
	assert {:error, :unauthorized} = Streamer.get_topic("public:media", user, token)

	assert {:error, :unauthorized} =
	Streamer.get_topic("public:remote", user, token, %{
	"instance" => "lain.com"
	})

	assert {:error, :unauthorized} =
	Streamer.get_topic("public:remote:media", user, token, %{
	"instance" => "lain.com"
	})
	end
	end

	test "allows user streams (with proper OAuth token scopes)", %{
	user: user,
	token: read_oauth_token
	} do
	%{token: read_notifications_token} = oauth_access(["read:notifications"], user: user)
	%{token: read_statuses_token} = oauth_access(["read:statuses"], user: user)
	%{token: badly_scoped_token} = oauth_access(["irrelevant:scope"], user: user)

	expected_user_topic = "user:#{user.id}"
	expected_notification_topic = "user:notification:#{user.id}"
	expected_direct_topic = "direct:#{user.id}"
	expected_pleroma_chat_topic = "user:pleroma_chat:#{user.id}"

	for valid_user_token <- [read_oauth_token, read_statuses_token] do
	assert {:ok, ^expected_user_topic} = Streamer.get_topic("user", user, valid_user_token)

	assert {:ok, ^expected_direct_topic} =
	Streamer.get_topic("direct", user, valid_user_token)

	assert {:ok, ^expected_pleroma_chat_topic} =
	Streamer.get_topic("user:pleroma_chat", user, valid_user_token)
	end

	for invalid_user_token <- [read_notifications_token, badly_scoped_token],
	user_topic <- ["user", "direct", "user:pleroma_chat"] do
	assert {:error, :unauthorized} = Streamer.get_topic(user_topic, user, invalid_user_token)
	end

	for valid_notification_token <- [read_oauth_token, read_notifications_token] do
	assert {:ok, ^expected_notification_topic} =
	Streamer.get_topic("user:notification", user, valid_notification_token)
	end

	for invalid_notification_token <- [read_statuses_token, badly_scoped_token] do
	assert {:error, :unauthorized} =
	Streamer.get_topic("user:notification", user, invalid_notification_token)
	end
	end

	test "allows hashtag streams (regardless of OAuth token scopes)", %{
	user: user,
	token: read_oauth_token
	} do
	for oauth_token <- [nil, read_oauth_token] do
	assert {:ok, "hashtag:cofe"} =
	Streamer.get_topic("hashtag", user, oauth_token, %{"tag" => "cofe"})
	end
	end

	test "disallows registering to another user's stream", %{user: user, token: read_oauth_token} do
	another_user = insert(:user)
	assert {:error, _} = Streamer.get_topic("user:#{another_user.id}", user, read_oauth_token)

	assert {:error, _} =
	Streamer.get_topic("user:notification:#{another_user.id}", user, read_oauth_token)

	assert {:error, _} = Streamer.get_topic("direct:#{another_user.id}", user, read_oauth_token)
	end

	test "allows list stream that are owned by the user (with `read` or `read:lists` scopes)", %{
	user: user,
	token: read_oauth_token
	} do
	%{token: read_lists_token} = oauth_access(["read:lists"], user: user)
	%{token: invalid_token} = oauth_access(["irrelevant:scope"], user: user)
	{:ok, list} = List.create("Test", user)

	assert {:error, _} = Streamer.get_topic("list:#{list.id}", user, read_oauth_token)

	for valid_token <- [read_oauth_token, read_lists_token] do
	assert {:ok, _} = Streamer.get_topic("list", user, valid_token, %{"list" => list.id})
	end

	assert {:error, _} = Streamer.get_topic("list", user, invalid_token, %{"list" => list.id})
	end

	test "disallows list stream that are not owned by the user", %{user: user, token: oauth_token} do
	another_user = insert(:user)
	{:ok, list} = List.create("Test", another_user)

	assert {:error, _} = Streamer.get_topic("list:#{list.id}", user, oauth_token)
	assert {:error, _} = Streamer.get_topic("list", user, oauth_token, %{"list" => list.id})
	end
	end

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

	test "it streams the user's post in the 'user' stream", %{user: user, token: oauth_token} do
	Streamer.get_topic_and_add_socket("user", user, oauth_token)
	{:ok, activity} = CommonAPI.post(user, %{status: "hey"})

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

	test "it streams boosts of the user in the 'user' stream", %{user: user, token: oauth_token} do
	Streamer.get_topic_and_add_socket("user", user, oauth_token)

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

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

	test "it does not stream announces of the user's own posts in the 'user' stream", %{
	user: user,
	token: oauth_token
	} do
	Streamer.get_topic_and_add_socket("user", user, oauth_token)

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

	assert Streamer.filtered_by_user?(user, announce)
	end

	test "it does stream notifications announces of the user's own posts in the 'user' stream", %{
	user: user,
	token: oauth_token
	} do
	Streamer.get_topic_and_add_socket("user", user, oauth_token)

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

	notification =
	Pleroma.Notification
	\|> Repo.get_by(%{user_id: user.id, activity_id: announce.id})
	\|> Repo.preload(:activity)

	refute Streamer.filtered_by_user?(user, notification)
	end

	test "it streams boosts of mastodon user in the 'user' stream", %{
	user: user,
	token: oauth_token
	} do
	Streamer.get_topic_and_add_socket("user", user, oauth_token)

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

	data =
	File.read!("test/fixtures/mastodon-announce.json")
	\|> Jason.decode!()
	\|> Map.put("object", activity.data["object"])
	\|> Map.put("actor", user.ap_id)

	{:ok, %Pleroma.Activity{data: _data, local: false} = announce} =
	Pleroma.Web.ActivityPub.Transmogrifier.handle_incoming(data)

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

	test "it sends notify to in the 'user' stream", %{
	user: user,
	token: oauth_token,
	notify: notify
	} do
	Streamer.get_topic_and_add_socket("user", user, oauth_token)
	Streamer.stream("user", notify)

	- assert_receive {:render_with_user, _, _, ^notify}
	+ assert_receive {:render_with_user, _, _, ^notify, _}
	refute Streamer.filtered_by_user?(user, notify)
	end

	test "it sends notify to in the 'user:notification' stream", %{
	user: user,
	token: oauth_token,
	notify: notify
	} do
	Streamer.get_topic_and_add_socket("user:notification", user, oauth_token)
	Streamer.stream("user:notification", notify)

	- assert_receive {:render_with_user, _, _, ^notify}
	+ assert_receive {:render_with_user, _, _, ^notify, _}
	refute Streamer.filtered_by_user?(user, notify)
	end

	test "it sends chat messages to the 'user:pleroma_chat' stream", %{
	user: user,
	token: oauth_token
	} do
	other_user = insert(:user)

	{:ok, create_activity} =
	CommonAPI.post_chat_message(other_user, user, "hey cirno", idempotency_key: "123")

	object = Object.normalize(create_activity, fetch: false)
	chat = Chat.get(user.id, other_user.ap_id)
	cm_ref = MessageReference.for_chat_and_object(chat, object)
	cm_ref = %{cm_ref \| chat: chat, object: object}

	Streamer.get_topic_and_add_socket("user:pleroma_chat", user, oauth_token)
	Streamer.stream("user:pleroma_chat", {user, cm_ref})

	- text = StreamerView.render("chat_update.json", %{chat_message_reference: cm_ref})
	+ text = StreamerView.render("chat_update.json", %{chat_message_reference: cm_ref}, "user:pleroma_chat:#{user.id}")

	assert text =~ "hey cirno"
	assert_receive {:text, ^text}
	end

	test "it sends chat messages to the 'user' stream", %{user: user, token: oauth_token} do
	other_user = insert(:user)

	{:ok, create_activity} = CommonAPI.post_chat_message(other_user, user, "hey cirno")
	object = Object.normalize(create_activity, fetch: false)
	chat = Chat.get(user.id, other_user.ap_id)
	cm_ref = MessageReference.for_chat_and_object(chat, object)
	cm_ref = %{cm_ref \| chat: chat, object: object}

	Streamer.get_topic_and_add_socket("user", user, oauth_token)
	Streamer.stream("user", {user, cm_ref})

	- text = StreamerView.render("chat_update.json", %{chat_message_reference: cm_ref})
	+ text = StreamerView.render("chat_update.json", %{chat_message_reference: cm_ref}, "user:#{user.id}")

	assert text =~ "hey cirno"
	assert_receive {:text, ^text}
	end

	test "it sends chat message notifications to the 'user:notification' stream", %{
	user: user,
	token: oauth_token
	} do
	other_user = insert(:user)

	{:ok, create_activity} = CommonAPI.post_chat_message(other_user, user, "hey")

	notify =
	Repo.get_by(Pleroma.Notification, user_id: user.id, activity_id: create_activity.id)
	\|> Repo.preload(:activity)

	Streamer.get_topic_and_add_socket("user:notification", user, oauth_token)
	Streamer.stream("user:notification", notify)

	- assert_receive {:render_with_user, _, _, ^notify}
	+ assert_receive {:render_with_user, _, _, ^notify, _}
	refute Streamer.filtered_by_user?(user, notify)
	end

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

	Streamer.get_topic_and_add_socket("user:notification", user, oauth_token)

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

	refute_receive _
	end

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

	{:ok, activity} = CommonAPI.post(user, %{status: "super hot take"})
	{:ok, _} = CommonAPI.add_mute(user, activity)

	Streamer.get_topic_and_add_socket("user:notification", user, oauth_token)

	{:ok, favorite_activity} = CommonAPI.favorite(user2, activity.id)

	refute_receive _
	assert Streamer.filtered_by_user?(user, favorite_activity)
	end

	test "it sends favorite to 'user:notification' stream'", %{
	user: user,
	token: oauth_token
	} do
	user2 = insert(:user, %{ap_id: "https://hecking-lewd-place.com/user/meanie"})

	{:ok, activity} = CommonAPI.post(user, %{status: "super hot take"})
	Streamer.get_topic_and_add_socket("user:notification", user, oauth_token)
	{:ok, favorite_activity} = CommonAPI.favorite(user2, activity.id)

	- assert_receive {:render_with_user, _, "notification.json", notif}
	+ assert_receive {:render_with_user, _, "notification.json", notif, _}
	assert notif.activity.id == favorite_activity.id
	refute Streamer.filtered_by_user?(user, notif)
	end

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

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

	refute_receive _
	assert Streamer.filtered_by_user?(user, favorite_activity)
	end

	test "it sends follow activities to the 'user:notification' stream", %{
	user: user,
	token: oauth_token
	} do
	user2 = insert(:user)

	Streamer.get_topic_and_add_socket("user:notification", user, oauth_token)
	{:ok, _follower, _followed, follow_activity} = CommonAPI.follow(user2, user)

	- assert_receive {:render_with_user, _, "notification.json", notif}
	+ assert_receive {:render_with_user, _, "notification.json", notif, _}
	assert notif.activity.id == follow_activity.id
	refute Streamer.filtered_by_user?(user, notif)
	end

	test "it sends follow relationships updates to the 'user' stream", %{
	user: user,
	token: oauth_token
	} do
	user_id = user.id
	other_user = insert(:user)
	other_user_id = other_user.id

	Streamer.get_topic_and_add_socket("user", user, oauth_token)
	{:ok, _follower, _followed, _follow_activity} = CommonAPI.follow(user, other_user)

	assert_receive {:text, event}

	assert %{"event" => "pleroma:follow_relationships_update", "payload" => payload} =
	Jason.decode!(event)

	assert %{
	"follower" => %{
	"follower_count" => 0,
	"following_count" => 0,
	"id" => ^user_id
	},
	"following" => %{
	"follower_count" => 0,
	"following_count" => 0,
	"id" => ^other_user_id
	},
	"state" => "follow_pending"
	} = Jason.decode!(payload)

	assert_receive {:text, event}

	assert %{"event" => "pleroma:follow_relationships_update", "payload" => payload} =
	Jason.decode!(event)

	assert %{
	"follower" => %{
	"follower_count" => 0,
	"following_count" => 1,
	"id" => ^user_id
	},
	"following" => %{
	"follower_count" => 1,
	"following_count" => 0,
	"id" => ^other_user_id
	},
	"state" => "follow_accept"
	} = Jason.decode!(payload)
	end

	test "it streams edits in the 'user' stream", %{user: user, token: oauth_token} do
	sender = insert(:user)
	{:ok, _, _, _} = CommonAPI.follow(user, sender)

	{:ok, activity} = CommonAPI.post(sender, %{status: "hey"})

	Streamer.get_topic_and_add_socket("user", user, oauth_token)
	{:ok, edited} = CommonAPI.update(sender, activity, %{status: "mew mew"})
	create = Pleroma.Activity.get_create_by_object_ap_id_with_object(activity.object.data["id"])

	- assert_receive {:render_with_user, _, "status_update.json", ^create}
	+ assert_receive {:render_with_user, _, "status_update.json", ^create, _}
	refute Streamer.filtered_by_user?(user, edited)
	end

	test "it streams own edits in the 'user' stream", %{user: user, token: oauth_token} do
	{:ok, activity} = CommonAPI.post(user, %{status: "hey"})

	Streamer.get_topic_and_add_socket("user", user, oauth_token)
	{:ok, edited} = CommonAPI.update(user, activity, %{status: "mew mew"})
	create = Pleroma.Activity.get_create_by_object_ap_id_with_object(activity.object.data["id"])

	- assert_receive {:render_with_user, _, "status_update.json", ^create}
	+ assert_receive {:render_with_user, _, "status_update.json", ^create, _}
	refute Streamer.filtered_by_user?(user, edited)
	end
	end

	describe "public streams" do
	test "it sends to public (authenticated)" do
	%{user: user, token: oauth_token} = oauth_access(["read"])
	other_user = insert(:user)

	Streamer.get_topic_and_add_socket("public", user, oauth_token)

	{:ok, activity} = CommonAPI.post(other_user, %{status: "Test"})
	- assert_receive {:render_with_user, _, _, ^activity}
	+ assert_receive {:render_with_user, _, _, ^activity, _}
	refute Streamer.filtered_by_user?(other_user, activity)
	end

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

	Streamer.get_topic_and_add_socket("public", nil, nil)

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

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

	test "handles deletions" do
	%{user: user, token: oauth_token} = oauth_access(["read"])
	other_user = insert(:user)
	{:ok, activity} = CommonAPI.post(other_user, %{status: "Test"})

	Streamer.get_topic_and_add_socket("public", user, oauth_token)

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

	test "it streams edits in the 'public' stream" do
	sender = insert(:user)

	Streamer.get_topic_and_add_socket("public", nil, nil)
	{:ok, activity} = CommonAPI.post(sender, %{status: "hey"})
	assert_receive {:text, _}

	{:ok, edited} = CommonAPI.update(sender, activity, %{status: "mew mew"})

	edited = Pleroma.Activity.normalize(edited)

	%{id: activity_id} = Pleroma.Activity.get_create_by_object_ap_id(edited.object.data["id"])

	assert_receive {:text, event}
	assert %{"event" => "status.update", "payload" => payload} = Jason.decode!(event)
	assert %{"id" => ^activity_id} = Jason.decode!(payload)
	refute Streamer.filtered_by_user?(sender, edited)
	end

	test "it streams multiple edits in the 'public' stream correctly" do
	sender = insert(:user)

	Streamer.get_topic_and_add_socket("public", nil, nil)
	{:ok, activity} = CommonAPI.post(sender, %{status: "hey"})
	assert_receive {:text, _}

	{:ok, edited} = CommonAPI.update(sender, activity, %{status: "mew mew"})

	edited = Pleroma.Activity.normalize(edited)

	%{id: activity_id} = Pleroma.Activity.get_create_by_object_ap_id(edited.object.data["id"])

	assert_receive {:text, event}
	assert %{"event" => "status.update", "payload" => payload} = Jason.decode!(event)
	assert %{"id" => ^activity_id} = Jason.decode!(payload)
	refute Streamer.filtered_by_user?(sender, edited)

	{:ok, edited} = CommonAPI.update(sender, activity, %{status: "mew mew 2"})

	edited = Pleroma.Activity.normalize(edited)

	%{id: activity_id} = Pleroma.Activity.get_create_by_object_ap_id(edited.object.data["id"])
	assert_receive {:text, event}
	assert %{"event" => "status.update", "payload" => payload} = Jason.decode!(event)
	assert %{"id" => ^activity_id, "content" => "mew mew 2"} = Jason.decode!(payload)
	refute Streamer.filtered_by_user?(sender, edited)
	end
	end

	describe "thread_containment/2" do
	test "it filters to user if recipients invalid and thread containment is enabled" do
	clear_config([:instance, :skip_thread_containment], false)
	author = insert(:user)
	%{user: user, token: oauth_token} = oauth_access(["read"])
	User.follow(user, author, :follow_accept)

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

	Streamer.get_topic_and_add_socket("public", user, oauth_token)
	Streamer.stream("public", activity)
	- assert_receive {:render_with_user, _, _, ^activity}
	+ assert_receive {:render_with_user, _, _, ^activity, _}
	assert Streamer.filtered_by_user?(user, activity)
	end

	test "it sends message if recipients invalid and thread containment is disabled" do
	clear_config([:instance, :skip_thread_containment], true)
	author = insert(:user)
	%{user: user, token: oauth_token} = oauth_access(["read"])
	User.follow(user, author, :follow_accept)

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

	Streamer.get_topic_and_add_socket("public", user, oauth_token)
	Streamer.stream("public", activity)

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

	test "it sends message if recipients invalid and thread containment is enabled but user's thread containment is disabled" do
	clear_config([:instance, :skip_thread_containment], false)
	author = insert(:user)
	user = insert(:user, skip_thread_containment: true)
	%{token: oauth_token} = oauth_access(["read"], user: user)
	User.follow(user, author, :follow_accept)

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

	Streamer.get_topic_and_add_socket("public", user, oauth_token)
	Streamer.stream("public", activity)

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

	describe "blocks" do
	setup do: oauth_access(["read"])

	test "it filters messages involving blocked users", %{user: user, token: oauth_token} do
	blocked_user = insert(:user)
	{:ok, _user_relationship} = User.block(user, blocked_user)

	Streamer.get_topic_and_add_socket("public", user, oauth_token)
	{:ok, activity} = CommonAPI.post(blocked_user, %{status: "Test"})
	- assert_receive {:render_with_user, _, _, ^activity}
	+ assert_receive {:render_with_user, _, _, ^activity, _}
	assert Streamer.filtered_by_user?(user, activity)
	end

	test "it filters messages transitively involving blocked users", %{
	user: blocker,
	token: blocker_token
	} do
	blockee = insert(:user)
	friend = insert(:user)

	Streamer.get_topic_and_add_socket("public", blocker, blocker_token)

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

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

	- assert_receive {:render_with_user, _, _, ^activity_one}
	+ assert_receive {:render_with_user, _, _, ^activity_one, _}
	assert Streamer.filtered_by_user?(blocker, activity_one)

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

	- assert_receive {:render_with_user, _, _, ^activity_two}
	+ assert_receive {:render_with_user, _, _, ^activity_two, _}
	assert Streamer.filtered_by_user?(blocker, activity_two)

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

	- assert_receive {:render_with_user, _, _, ^activity_three}
	+ assert_receive {:render_with_user, _, _, ^activity_three, _}
	assert Streamer.filtered_by_user?(blocker, activity_three)
	end
	end

	describe "lists" do
	setup do: oauth_access(["read"])

	test "it doesn't send unwanted DMs to list", %{user: user_a, token: user_a_token} do
	user_b = insert(:user)
	user_c = insert(:user)

	{:ok, user_a, user_b} = User.follow(user_a, user_b)

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

	Streamer.get_topic_and_add_socket("list", user_a, user_a_token, %{"list" => list.id})

	{:ok, _activity} =
	CommonAPI.post(user_b, %{
	status: "@#{user_c.nickname} Test",
	visibility: "direct"
	})

	refute_receive _
	end

	test "it doesn't send unwanted private posts to list", %{user: user_a, token: user_a_token} do
	user_b = insert(:user)

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

	Streamer.get_topic_and_add_socket("list", user_a, user_a_token, %{"list" => list.id})

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

	refute_receive _
	end

	test "it sends wanted private posts to list", %{user: user_a, token: user_a_token} do
	user_b = insert(:user)

	{:ok, user_a, user_b} = User.follow(user_a, user_b)

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

	Streamer.get_topic_and_add_socket("list", user_a, user_a_token, %{"list" => list.id})

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

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

	describe "muted reblogs" do
	setup do: oauth_access(["read"])

	test "it filters muted reblogs", %{user: user1, token: user1_token} do
	user2 = insert(:user)
	user3 = insert(:user)
	CommonAPI.follow(user1, user2)
	CommonAPI.hide_reblogs(user1, user2)

	{:ok, create_activity} = CommonAPI.post(user3, %{status: "I'm kawen"})

	Streamer.get_topic_and_add_socket("user", user1, user1_token)
	{:ok, announce_activity} = CommonAPI.repeat(create_activity.id, user2)
	- assert_receive {:render_with_user, _, _, ^announce_activity}
	+ assert_receive {:render_with_user, _, _, ^announce_activity, _}
	assert Streamer.filtered_by_user?(user1, announce_activity)
	end

	test "it filters reblog notification for reblog-muted actors", %{
	user: user1,
	token: user1_token
	} do
	user2 = insert(:user)
	CommonAPI.follow(user1, user2)
	CommonAPI.hide_reblogs(user1, user2)

	{:ok, create_activity} = CommonAPI.post(user1, %{status: "I'm kawen"})
	Streamer.get_topic_and_add_socket("user", user1, user1_token)
	{:ok, _announce_activity} = CommonAPI.repeat(create_activity.id, user2)

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

	test "it send non-reblog notification for reblog-muted actors", %{
	user: user1,
	token: user1_token
	} do
	user2 = insert(:user)
	CommonAPI.follow(user1, user2)
	CommonAPI.hide_reblogs(user1, user2)

	{:ok, create_activity} = CommonAPI.post(user1, %{status: "I'm kawen"})
	Streamer.get_topic_and_add_socket("user", user1, user1_token)
	{:ok, _favorite_activity} = CommonAPI.favorite(user2, create_activity.id)

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

	describe "muted threads" do
	test "it filters posts from muted threads" do
	user = insert(:user)
	%{user: user2, token: user2_token} = oauth_access(["read"])
	Streamer.get_topic_and_add_socket("user", user2, user2_token)

	{:ok, user2, user, _activity} = CommonAPI.follow(user2, user)
	{:ok, activity} = CommonAPI.post(user, %{status: "super hot take"})
	{:ok, _} = CommonAPI.add_mute(user2, activity)

	- assert_receive {:render_with_user, _, _, ^activity}
	+ assert_receive {:render_with_user, _, _, ^activity, _}
	assert Streamer.filtered_by_user?(user2, activity)
	end
	end

	describe "direct streams" do
	setup do: oauth_access(["read"])

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

	Streamer.get_topic_and_add_socket("direct", user, oauth_token)

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

	assert_receive {:text, received_event}

	assert %{"event" => "conversation", "payload" => received_payload} =
	Jason.decode!(received_event)

	assert %{"last_status" => last_status} = Jason.decode!(received_payload)
	[participation] = Participation.for_user(user)
	assert last_status["pleroma"]["direct_conversation_id"] == participation.id
	end

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

	Streamer.get_topic_and_add_socket("direct", user, oauth_token)

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

	create_activity_id = create_activity.id
	- assert_receive {:render_with_user, _, _, ^create_activity}
	+ assert_receive {:render_with_user, _, _, ^create_activity, _}
	assert_receive {:text, received_conversation1}
	assert %{"event" => "conversation", "payload" => _} = Jason.decode!(received_conversation1)

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

	assert_receive {:text, received_event}

	assert %{"event" => "delete", "payload" => ^create_activity_id} =
	Jason.decode!(received_event)

	refute_receive _
	end

	@tag :erratic
	test "it sends conversation update to the 'direct' stream when a message is deleted", %{
	user: user,
	token: oauth_token
	} do
	another_user = insert(:user)
	Streamer.get_topic_and_add_socket("direct", user, oauth_token)

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

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

	- assert_receive {:render_with_user, _, _, ^create_activity}
	- assert_receive {:render_with_user, _, _, ^create_activity2}
	+ assert_receive {:render_with_user, _, _, ^create_activity, _}
	+ assert_receive {:render_with_user, _, _, ^create_activity2, _}
	assert_receive {:text, received_conversation1}
	assert %{"event" => "conversation", "payload" => _} = Jason.decode!(received_conversation1)
	assert_receive {:text, received_conversation1}
	assert %{"event" => "conversation", "payload" => _} = Jason.decode!(received_conversation1)

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

	assert_receive {:text, received_event}
	assert %{"event" => "delete", "payload" => _} = Jason.decode!(received_event)

	assert_receive {:text, received_event}

	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
	end

	describe "stop streaming if token got revoked" do
	setup do
	child_proc = fn start, finalize ->
	fn ->
	start.()

	receive do
	{StreamerTest, :ready} ->
	- assert_receive {:render_with_user, _, "update.json", _}
	+ assert_receive {:render_with_user, _, "update.json", _, _}

	receive do
	{StreamerTest, :revoked} -> finalize.()
	end
	end
	end
	end

	starter = fn user, token ->
	fn -> Streamer.get_topic_and_add_socket("user", user, token) end
	end

	hit = fn -> assert_receive :close end
	miss = fn -> refute_receive :close end

	send_all = fn tasks, thing -> Enum.each(tasks, &send(&1.pid, thing)) end

	%{
	child_proc: child_proc,
	starter: starter,
	hit: hit,
	miss: miss,
	send_all: send_all
	}
	end

	test "do not revoke other tokens", %{
	child_proc: child_proc,
	starter: starter,
	hit: hit,
	miss: miss,
	send_all: send_all
	} do
	%{user: user, token: token} = oauth_access(["read"])
	%{token: token2} = oauth_access(["read"], user: user)
	%{user: user2, token: user2_token} = oauth_access(["read"])

	post_user = insert(:user)
	CommonAPI.follow(user, post_user)
	CommonAPI.follow(user2, post_user)

	tasks = [
	Task.async(child_proc.(starter.(user, token), hit)),
	Task.async(child_proc.(starter.(user, token2), miss)),
	Task.async(child_proc.(starter.(user2, user2_token), miss))
	]

	{:ok, _} =
	CommonAPI.post(post_user, %{
	status: "hi"
	})

	send_all.(tasks, {StreamerTest, :ready})

	Pleroma.Web.OAuth.Token.Strategy.Revoke.revoke(token)

	send_all.(tasks, {StreamerTest, :revoked})

	Enum.each(tasks, &Task.await/1)
	end

	test "revoke all streams for this token", %{
	child_proc: child_proc,
	starter: starter,
	hit: hit,
	send_all: send_all
	} do
	%{user: user, token: token} = oauth_access(["read"])

	post_user = insert(:user)
	CommonAPI.follow(user, post_user)

	tasks = [
	Task.async(child_proc.(starter.(user, token), hit)),
	Task.async(child_proc.(starter.(user, token), hit))
	]

	{:ok, _} =
	CommonAPI.post(post_user, %{
	status: "hi"
	})

	send_all.(tasks, {StreamerTest, :ready})

	Pleroma.Web.OAuth.Token.Strategy.Revoke.revoke(token)

	send_all.(tasks, {StreamerTest, :revoked})

	Enum.each(tasks, &Task.await/1)
	end
	end
	end

File Metadata

Mime Type: text/x-diff
Expires: Thu, Oct 29, 11:32 PM (1 d, 16 h)
Storage Engine: blob
Storage Format: Raw Data
Storage Handle: 1798867
Default Alt Text: (84 KB)

No OneTemporaryActions

View Options

File Metadata

Event Timeline

No OneTemporary
Actions