No OneTemporary
Actions

Size

66 KB

Referenced Files

None

Subscribers

None

View Options

	diff --git a/lib/pleroma/web/mastodon_api/websocket_handler.ex b/lib/pleroma/web/mastodon_api/websocket_handler.ex
	index 2707673ba..07c2b62e3 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, 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, 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,
	+ %{"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",
	+ type: "pleroma:authenticate",
	result: "success"
	})}
	], %{state \| user: user, oauth_token: oauth_token}}
	else
	{:auth, _, _} ->
	{[
	{:text,
	StreamerView.render("pleroma_respond.json", %{
	- type: "pleroma.authenticate",
	+ type: "pleroma:authenticate",
	result: "error",
	error: :already_authenticated
	})}
	], state}

	_ ->
	{[
	{:text,
	StreamerView.render("pleroma_respond.json", %{
	- type: "pleroma.authenticate",
	+ 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/views/streamer_view.ex b/lib/pleroma/web/views/streamer_view.ex
	index f591da9a6..f97570b0a 100644
	--- a/lib/pleroma/web/views/streamer_view.ex
	+++ b/lib/pleroma/web/views/streamer_view.ex
	@@ -1,183 +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

	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", 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, 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",
	+ 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 d2e98df3b..21e7ca2b0 100644
	--- a/test/pleroma/integration/mastodon_websocket_test.exs
	+++ b/test/pleroma/integration/mastodon_websocket_test.exs
	@@ -1,357 +1,361 @@
	# 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",
	+ "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",
	+ "event" => "pleroma:respond",
	"payload" => %{"type" => "subscribe", "result" => "success"}
	}} = decode_json(raw_json)

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

	assert {:ok,
	%{
	- "event" => "pleroma.respond",
	+ "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",
	+ "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",
	+ "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",
	+ "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",
	+ "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!()
	+ %{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"}
	+ "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",
	+ "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!()
	+ %{type: "pleroma:authenticate", token: "Something else"} \|> Jason.encode!()
	)

	assert_receive {:text, raw_json}, 1_000

	assert {:ok,
	%{
	- "event" => "pleroma.respond",
	+ "event" => "pleroma:respond",
	"payload" => %{
	- "type" => "pleroma.authenticate",
	+ "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!()
	+ %{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"}
	+ "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!()
	+ %{type: "pleroma:authenticate", token: "Something else"} \|> Jason.encode!()
	)

	assert_receive {:text, raw_json}, 1_000

	assert {:ok,
	%{
	- "event" => "pleroma.respond",
	+ "event" => "pleroma:respond",
	"payload" => %{
	- "type" => "pleroma.authenticate",
	+ "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 cc57a2989..d85358fd4 100644
	--- a/test/pleroma/web/streamer_test.exs
	+++ b/test/pleroma/web/streamer_test.exs
	@@ -1,1072 +1,1082 @@
	# 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, _}
	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, _}
	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, _}
	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, _}
	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, _}
	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}, "user:pleroma_chat:#{user.id}")
	+ 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}, "user:#{user.id}")
	+ 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, _}
	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 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 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, _}
	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, _}
	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, _}
	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 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, _}
	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, _}
	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 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 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 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 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, _}
	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 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 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, _}
	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 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 {: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 {: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", _, _}

	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: Sun, Oct 11, 12:49 AM (1 d, 15 h)
Storage Engine: blob
Storage Format: Raw Data
Storage Handle: 1785515
Default Alt Text: (66 KB)

No OneTemporaryActions

View Options

File Metadata

Event Timeline

No OneTemporary
Actions