No OneTemporary
Actions

Authored By

Unknown

Size

175 KB

Referenced Files

None

Subscribers

None

View Options

	diff --git a/changelog.d/unified-streaming.add b/changelog.d/unified-streaming.add
	new file mode 100644
	index 000000000..84821fcc8
	--- /dev/null
	+++ b/changelog.d/unified-streaming.add
	@@ -0,0 +1 @@
	+Add unified streaming endpoint
	diff --git a/docs/development/API/differences_in_mastoapi_responses.md b/docs/development/API/differences_in_mastoapi_responses.md
	index 4007c63c8..48a9c104c 100644
	--- a/docs/development/API/differences_in_mastoapi_responses.md
	+++ b/docs/development/API/differences_in_mastoapi_responses.md
	@@ -1,390 +1,506 @@
	# Differences in Mastodon API responses from vanilla Mastodon

	A Pleroma instance can be identified by "<Mastodon version> (compatible; Pleroma <version>)" present in `version` field in response from `/api/v1/instance`

	## Flake IDs

	Pleroma uses 128-bit ids as opposed to Mastodon's 64 bits. However, just like Mastodon's ids, they are lexically sortable strings

	## Timelines

	Adding the parameter `with_muted=true` to the timeline queries will also return activities by muted (not by blocked!) users.

	Adding the parameter `exclude_visibilities` to the timeline queries will exclude the statuses with the given visibilities. The parameter accepts an array of visibility types (`public`, `unlisted`, `private`, `direct`), e.g., `exclude_visibilities[]=direct&exclude_visibilities[]=private`.

	Adding the parameter `reply_visibility` to the public and home timelines queries will filter replies. Possible values: without parameter (default) shows all replies, `following` - replies directed to you or users you follow, `self` - replies directed to you.

	Adding the parameter `instance=lain.com` to the public timeline will show only statuses originating from `lain.com` (or any remote instance).

	Home, public, hashtag & list timelines accept these parameters:

	- `only_media`: show only statuses with media attached
	- `local`: show only local statuses
	- `remote`: show only remote statuses

	## Statuses

	- `visibility`: has additional possible values `list` and `local` (for local-only statuses)

	Has these additional fields under the `pleroma` object:

	- `local`: true if the post was made on the local instance
	- `conversation_id`: the ID of the AP context the status is associated with (if any)
	- `direct_conversation_id`: the ID of the Mastodon direct message conversation the status is associated with (if any)
	- `in_reply_to_account_acct`: the `acct` property of User entity for replied user (if any)
	- `content`: a map consisting of alternate representations of the `content` property with the key being its mimetype. Currently, the only alternate representation supported is `text/plain`
	- `spoiler_text`: a map consisting of alternate representations of the `spoiler_text` property with the key being its mimetype. Currently, the only alternate representation supported is `text/plain`
	- `expires_at`: a datetime (iso8601) that states when the post will expire (be deleted automatically), or empty if the post won't expire
	- `thread_muted`: true if the thread the post belongs to is muted
	- `emoji_reactions`: A list with emoji / reaction maps. The format is `{name: "☕", count: 1, me: true}`. Contains no information about the reacting users, for that use the `/statuses/:id/reactions` endpoint.
	- `parent_visible`: If the parent of this post is visible to the user or not.
	- `pinned_at`: a datetime (iso8601) when status was pinned, `null` otherwise.

	The `GET /api/v1/statuses/:id/source` endpoint additionally has the following attributes:

	- `content_type`: The content type of the status source.

	## Scheduled statuses

	Has these additional fields in `params`:

	- `expires_in`: the number of seconds the posted activity should expire in.

	## Media Attachments

	Has these additional fields under the `pleroma` object:

	- `mime_type`: mime type of the attachment.

	### Attachment cap

	Some apps operate under the assumption that no more than 4 attachments can be returned or uploaded. Pleroma however does not enforce any limits on attachment count neither when returning the status object nor when posting.

	### Limitations

	Pleroma does not process remote images and therefore cannot include fields such as `meta` and `blurhash`. It does not support focal points or aspect ratios. The frontend is expected to handle it.

	## Accounts

	The `id` parameter can also be the `nickname` of the user. This only works in these endpoints, not the deeper nested ones for following etc.

	- `/api/v1/accounts/:id`
	- `/api/v1/accounts/:id/statuses`

	`/api/v1/accounts/:id/statuses` endpoint accepts these parameters:

	- `pinned`: include only pinned statuses
	- `tagged`: with tag
	- `only_media`: include only statuses with media attached
	- `with_muted`: include statuses/reactions from muted accounts
	- `exclude_reblogs`: exclude reblogs
	- `exclude_replies`: exclude replies
	- `exclude_visibilities`: exclude visibilities

	Endpoints which accept `with_relationships` parameter:

	- `/api/v1/accounts/:id`
	- `/api/v1/accounts/:id/followers`
	- `/api/v1/accounts/:id/following`
	- `/api/v1/mutes`

	Has these additional fields under the `pleroma` object:

	- `ap_id`: nullable URL string, ActivityPub id of the user
	- `background_image`: nullable URL string, background image of the user
	- `tags`: Lists an array of tags for the user
	- `relationship` (object): Includes fields as documented for Mastodon API https://docs.joinmastodon.org/entities/relationship/
	- `is_moderator`: boolean, nullable, true if user is a moderator
	- `is_admin`: boolean, nullable, true if user is an admin
	- `confirmation_pending`: boolean, true if a new user account is waiting on email confirmation to be activated
	- `hide_favorites`: boolean, true when the user has hiding favorites enabled
	- `hide_followers`: boolean, true when the user has follower hiding enabled
	- `hide_follows`: boolean, true when the user has follow hiding enabled
	- `hide_followers_count`: boolean, true when the user has follower stat hiding enabled
	- `hide_follows_count`: boolean, true when the user has follow stat hiding enabled
	- `settings_store`: A generic map of settings for frontends. Opaque to the backend. Only returned in `/api/v1/accounts/verify_credentials` and `/api/v1/accounts/update_credentials`
	- `chat_token`: The token needed for Pleroma shoutbox. Only returned in `/api/v1/accounts/verify_credentials`
	- `deactivated`: boolean, true when the user is deactivated
	- `allow_following_move`: boolean, true when the user allows automatically follow moved following accounts
	- `unread_conversation_count`: The count of unread conversations. Only returned to the account owner.
	- `unread_notifications_count`: The count of unread notifications. Only returned to the account owner.
	- `notification_settings`: object, can be absent. See `/api/v1/pleroma/notification_settings` for the parameters/keys returned.
	- `accepts_chat_messages`: boolean, but can be null if we don't have that information about a user
	- `favicon`: nullable URL string, Favicon image of the user's instance

	### Source

	Has these additional fields under the `pleroma` object:

	- `show_role`: boolean, nullable, true when the user wants his role (e.g admin, moderator) to be shown
	- `no_rich_text` - boolean, nullable, true when html tags are stripped from all statuses requested from the API
	- `discoverable`: boolean, true when the user allows external services (search bots) etc. to index / list the account (regardless of this setting, user will still appear in regular search results)
	- `actor_type`: string, the type of this account.

	## Conversations

	Has an additional field under the `pleroma` object:

	- `recipients`: The list of the recipients of this Conversation. These will be addressed when replying to this conversation.

	## GET `/api/v1/conversations`

	Accepts additional parameters:

	- `recipients`: Only return conversations with the given recipients (a list of user ids). Usage example: `GET /api/v1/conversations?recipients[]=1&recipients[]=2`

	## Account Search

	Behavior has changed:

	- `/api/v1/accounts/search`: Does not require authentication

	## Search (global)

	Unlisted posts are available in search results, they are considered to be public posts that shouldn't be shown in local/federated timeline.

	## Notifications

	Has these additional fields under the `pleroma` object:

	- `is_seen`: true if the notification was read by the user

	### Move Notification

	The `type` value is `move`. Has an additional field:

	- `target`: new account

	### EmojiReact Notification

	The `type` value is `pleroma:emoji_reaction`. Has these fields:

	- `emoji`: The used emoji
	- `account`: The account of the user who reacted
	- `status`: The status that was reacted on

	### ChatMention Notification (not default)

	This notification has to be requested explicitly.

	The `type` value is `pleroma:chat_mention`

	- `account`: The account who sent the message
	- `chat_message`: The chat message

	### Report Notification (not default)

	This notification has to be requested explicitly.

	The `type` value is `pleroma:report`

	- `account`: The account who reported
	- `report`: The report

	## GET `/api/v1/notifications`

	Accepts additional parameters:

	- `exclude_visibilities`: will exclude the notifications for activities with the given visibilities. The parameter accepts an array of visibility types (`public`, `unlisted`, `private`, `direct`). Usage example: `GET /api/v1/notifications?exclude_visibilities[]=direct&exclude_visibilities[]=private`.
	- `include_types`: will include the notifications for activities with the given types. The parameter accepts an array of types (`mention`, `follow`, `reblog`, `favourite`, `move`, `pleroma:emoji_reaction`, `pleroma:chat_mention`, `pleroma:report`). Usage example: `GET /api/v1/notifications?include_types[]=mention&include_types[]=reblog`.

	## DELETE `/api/v1/notifications/destroy_multiple`

	An endpoint to delete multiple statuses by IDs.

	Required parameters:

	- `ids`: array of activity ids

	Usage example: `DELETE /api/v1/notifications/destroy_multiple/?ids[]=1&ids[]=2`.

	Returns on success: 200 OK `{}`

	## POST `/api/v1/statuses`

	Additional parameters can be added to the JSON body/Form data:

	- `preview`: boolean, if set to `true` the post won't be actually posted, but the status entity would still be rendered back. This could be useful for previewing rich text/custom emoji, for example.
	- `content_type`: string, contain the MIME type of the status, it is transformed into HTML by the backend. You can get the list of the supported MIME types with the nodeinfo endpoint.
	- `to`: A list of nicknames (like `lain@soykaf.club` or `lain` on the local server) that will be used to determine who is going to be addressed by this post. Using this will disable the implicit addressing by mentioned names in the `status` body, only the people in the `to` list will be addressed. The normal rules for post visibility are not affected by this and will still apply.
	- `visibility`: string, besides standard MastoAPI values (`direct`, `private`, `unlisted`, `local` or `public`) it can be used to address a List by setting it to `list:LIST_ID`.
	- `expires_in`: The number of seconds the posted activity should expire in. When a posted activity expires it will be deleted from the server, and a delete request for it will be federated. This needs to be longer than an hour.
	- `in_reply_to_conversation_id`: Will reply to a given conversation, addressing only the people who are part of the recipient set of that conversation. Sets the visibility to `direct`.

	## GET `/api/v1/statuses`

	An endpoint to get multiple statuses by IDs.

	Required parameters:

	- `ids`: array of activity ids

	Usage example: `GET /api/v1/statuses/?ids[]=1&ids[]=2`.

	Returns: array of Status.

	The maximum number of statuses is limited to 100 per request.

	## PATCH `/api/v1/accounts/update_credentials`

	Additional parameters can be added to the JSON body/Form data:

	- `no_rich_text` - if true, html tags are stripped from all statuses requested from the API
	- `hide_followers` - if true, user's followers will be hidden
	- `hide_follows` - if true, user's follows will be hidden
	- `hide_followers_count` - if true, user's follower count will be hidden
	- `hide_follows_count` - if true, user's follow count will be hidden
	- `hide_favorites` - if true, user's favorites timeline will be hidden
	- `show_role` - if true, user's role (e.g admin, moderator) will be exposed to anyone in the API
	- `default_scope` - the scope returned under `privacy` key in Source subentity
	- `pleroma_settings_store` - Opaque user settings to be saved on the backend.
	- `skip_thread_containment` - if true, skip filtering out broken threads
	- `allow_following_move` - if true, allows automatically follow moved following accounts
	- `also_known_as` - array of ActivityPub IDs, needed for following move
	- `pleroma_background_image` - sets the background image of the user. Can be set to "" (an empty string) to reset.
	- `discoverable` - if true, external services (search bots) etc. are allowed to index / list the account (regardless of this setting, user will still appear in regular search results).
	- `actor_type` - the type of this account.
	- `accepts_chat_messages` - if false, this account will reject all chat messages.
	- `language` - user's preferred language for receiving emails (digest, confirmation, etc.)

	All images (avatar, banner and background) can be reset to the default by sending an empty string ("") instead of a file.

	### Pleroma Settings Store

	Pleroma has mechanism that allows frontends to save blobs of json for each user on the backend. This can be used to save frontend-specific settings for a user that the backend does not need to know about.

	The parameter should have a form of `{frontend_name: {...}}`, with `frontend_name` identifying your type of client, e.g. `pleroma_fe`. It will overwrite everything under this property, but will not overwrite other frontend's settings.

	This information is returned in the `/api/v1/accounts/verify_credentials` endpoint.

	## Authentication

	Pleroma supports refreshing tokens.

	### POST `/oauth/token`

	You can obtain access tokens for a user in a few additional ways.

	#### Refreshing a token

	To obtain a new access token from a refresh token, pass `grant_type=refresh_token` with the following extra parameters:

	- `refresh_token`: The refresh token.

	#### Getting a token with a password

	To obtain a token from a user's password, pass `grant_type=password` with the following extra parameters:

	- `username`: Username to authenticate.
	- `password`: The user's password.

	#### Response body

	Additional fields are returned in the response:

	- `id`: The primary key of this token in Pleroma's database.
	- `me` (user tokens only): The ActivityPub ID of the user who owns the token.

	## Account Registration

	`POST /api/v1/accounts`

	Has these additional parameters (which are the same as in Pleroma-API):

	- `fullname`: optional
	- `bio`: optional
	- `captcha_solution`: optional, contains provider-specific captcha solution,
	- `captcha_token`: optional, contains provider-specific captcha token
	- `captcha_answer_data`: optional, contains provider-specific captcha data
	- `token`: invite token required when the registrations aren't public.
	- `language`: optional, user's preferred language for receiving emails (digest, confirmation, etc.), default to the language set in the `userLanguage` cookies or `Accept-Language` header.

	## Instance

	`GET /api/v1/instance` has additional fields

	- `max_toot_chars`: The maximum characters per post
	- `chat_limit`: The maximum characters per chat message
	- `description_limit`: The maximum characters per image description
	- `poll_limits`: The limits of polls
	- `upload_limit`: The maximum upload file size
	- `avatar_upload_limit`: The same for avatars
	- `background_upload_limit`: The same for backgrounds
	- `banner_upload_limit`: The same for banners
	- `background_image`: A background image that frontends can use
	- `pleroma.metadata.features`: A list of supported features
	- `pleroma.metadata.federation`: The federation restrictions of this instance
	- `pleroma.metadata.fields_limits`: A list of values detailing the length and count limitation for various instance-configurable fields.
	- `pleroma.metadata.post_formats`: A list of the allowed post format types
	- `vapid_public_key`: The public key needed for push messages

	## Push Subscription

	`POST /api/v1/push/subscription`
	`PUT /api/v1/push/subscription`

	Permits these additional alert types:

	- pleroma:chat_mention
	- pleroma:emoji_reaction

	## Markers

	Has these additional fields under the `pleroma` object:

	- `unread_count`: contains number unread notifications

	## Streaming

	### Chats

	There is an additional `user:pleroma_chat` stream. Incoming chat messages will make the current chat be sent to this `user` stream. The `event` of an incoming chat message is `pleroma:chat_update`. The payload is the updated chat with the incoming chat message in the `last_message` field.

	### Remote timelines

	For viewing remote server timelines, there are `public:remote` and `public:remote:media` streams. Each of these accept a parameter like `?instance=lain.com`.

	### Follow relationships updates

	Pleroma streams follow relationships updates as `pleroma:follow_relationships_update` events to the `user` stream.

	The message payload consist of:

	- `state`: a relationship state, one of `follow_pending`, `follow_accept` or `follow_reject`.

	- `follower` and `following` maps with following fields:
	- `id`: user ID
	- `follower_count`: follower count
	- `following_count`: following count

	+### Authenticating via `sec-websocket-protocol` header
	+
	+Pleroma allows to authenticate via the `sec-websocket-protocol` header, for example, if your access token is `your-access-token`, you can authenticate using the following:
	+
	+```
	+sec-websocket-protocol: your-access-token
	+```
	+
	+### Authenticating after connection via `pleroma:authenticate` event
	+
	+Pleroma allows to authenticate after connection is established, via the `pleroma:authenticate` event. For example, if your access token is `your-access-token`, you can send the following after the connection is established:
	+
	+```
	+{"type": "pleroma:authenticate", "token": "your-access-token"}
	+```
	+
	+### Response to client-sent events
	+
	+Pleroma will respond to client-sent events that it recognizes. Supported event types are:
	+
	+- `subscribe`
	+- `unsubscribe`
	+- `pleroma:authenticate`
	+
	+The reply will be in the following format:
	+
	+```
	+{
	+ "event": "pleroma:respond",
	+ "payload": "{\"type\": \"<type of the client-sent event>\", \"result\": \"<result of the action>\", \"error\": \"<error code>\"}"
	+}
	+```
	+
	+Result of the action can be either `success`, `ignored` or `error`. If it is `error`, the `error` property will contain the error code. Otherwise, the `error` property will not be present. Below are some examples:
	+
	+```
	+{
	+ "event": "pleroma:respond",
	+ "payload": "{\"type\": \"pleroma:authenticate\", \"result\": \"success\"}"
	+}
	+
	+{
	+ "event": "pleroma:respond",
	+ "payload": "{\"type\": \"subscribe\", \"result\": \"ignored\"}"
	+}
	+
	+{
	+ "event": "pleroma:respond",
	+ "payload": "{\"type\": \"unsubscribe\", \"result\": \"error\", \"error\": \"bad_topic\"}"
	+}
	+```
	+
	+If the sent event is not of a type that Pleroma supports, it will not reply.
	+
	+### The `stream` attribute of a server-sent event
	+
	+Technically, this is in Mastodon, but its documentation does nothing to specify its format.
	+
	+This attribute appears on every event type except `pleroma:respond` and `delete`. It helps clients determine where they should display the new statuses.
	+
	+The value of the attribute is an array containing one or two elements. The first element is the type of the stream. The second is the identifier related to that specific stream, if applicable.
	+
	+For the following stream types, there is a second element in the array:
	+
	+- `list`: The second element is the id of the list, as a string.
	+- `hashtag`: The second element is the name of the hashtag.
	+- `public:remote:media` and `public:remote`: The second element is the domain of the corresponding instance.
	+
	+For all other stream types, there is no second element.
	+
	+Some examples of valid `stream` values:
	+
	+- `["list", "1"]`: List of id 1.
	+- `["hashtag", "mew"]`: The hashtag #mew.
	+- `["user:notifications"]`: Notifications for the current user.
	+- `["user"]`: Home timeline.
	+- `["public:remote", "mew.moe"]`: Public posts from the instance mew.moe .
	+
	+### The unified streaming endpoint
	+
	+If you do not specify a stream to connect to when requesting `/api/v1/streaming`, you will enter a connection that subscribes to no streams. After the connection is established, you can authenticate and then subscribe to different streams.
	+
	+### List of supported streams
	+
	+Below is a list of supported streams by Pleroma. To make a single-stream WebSocket connection, append the string specified in "Query style" to the streaming endpoint url.
	+To subscribe to a stream after the connection is established, merge the JSON object specified in "Subscribe style" with `{"type": "subscribe"}`. To unsubscribe, merge it with `{"type": "unsubscribe"}`.
	+
	+For example, to receive updates on the list 1, you can connect to `/api/v1/streaming/?stream=list&list=1`, or send
	+
	+```
	+{"type": "subscribe", "stream": "list", "list": "1"}
	+```
	+
	+upon establishing the websocket connection.
	+
	+To unsubscribe to list 1, send
	+
	+```
	+{"type": "unsubscribe", "stream": "list", "list": "1"}
	+```
	+
	+Note that if you specify a stream that requires a logged-in user in the query string (for example, `user` or `list`), you have to specify the access token when you are trying to establish the connection, i.e. in the query string or via the `sec-websocket-protocol` header.
	+
	+- `list`
	+ - Query style: `?stream=list&list=<id>`
	+ - Subscribe style: `{"stream": "list", "list": "<id>"}`
	+- `public`, `public:local`, `public:media`, `public:local:media`, `user`, `user:pleroma_chat`, `user:notifications`, `direct`
	+ - Query style: `?stream=<stream name>`
	+ - Subscribe style: `{"stream": "<stream name>"}`
	+- `hashtag`
	+ - Query style: `?stream=hashtag&tag=<name>`
	+ - Subscribe style: `{"stream": "hashtag", "tag": "<name>"}`
	+- `public:remote`, `public:remote:media`
	+ - Query style: `?stream=<stream name>&instance=<instance domain>`
	+ - Subscribe style: `{"stream": "<stream name>", "instance": "<instance domain>"}`
	+
	## User muting and thread muting

	Both user muting and thread muting can be done for only a certain time by adding an `expires_in` parameter to the API calls and giving the expiration time in seconds.

	## Not implemented

	Pleroma is generally compatible with the Mastodon 2.7.2 API, but some newer features and non-essential features are omitted. These features usually return an HTTP 200 status code, but with an empty response. While they may be added in the future, they are considered low priority.

	### Suggestions

	Added in Mastodon 2.4.3

	- `GET /api/v1/suggestions`: Returns an empty array, `[]`

	### Trends

	Added in Mastodon 3.0.0

	- `GET /api/v1/trends`: Returns an empty array, `[]`

	### Identity proofs

	Added in Mastodon 2.8.0

	- `GET /api/v1/identity_proofs`: Returns an empty array, `[]`

	### Featured tags

	Added in Mastodon 3.0.0

	- `GET /api/v1/featured_tags`: Returns HTTP 404
	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/api_spec.ex b/lib/pleroma/web/api_spec.ex
	index 2d56dc643..163226ce5 100644
	--- a/lib/pleroma/web/api_spec.ex
	+++ b/lib/pleroma/web/api_spec.ex
	@@ -1,152 +1,160 @@
	# Pleroma: A lightweight social networking server
	# Copyright © 2017-2022 Pleroma Authors <https://pleroma.social/>
	# SPDX-License-Identifier: AGPL-3.0-only

	defmodule Pleroma.Web.ApiSpec do
	alias OpenApiSpex.OpenApi
	alias OpenApiSpex.Operation
	alias Pleroma.Web.Endpoint
	alias Pleroma.Web.Router

	@behaviour OpenApi

	+ defp streaming_paths do
	+ %{
	+ "/api/v1/streaming" => %OpenApiSpex.PathItem{
	+ get: Pleroma.Web.ApiSpec.StreamingOperation.streaming_operation()
	+ }
	+ }
	+ end
	+
	@impl OpenApi
	def spec(opts \\ []) do
	%OpenApi{
	servers:
	if opts[:server_specific] do
	[
	# Populate the Server info from a phoenix endpoint
	OpenApiSpex.Server.from_endpoint(Endpoint)
	]
	else
	[]
	end,
	info: %OpenApiSpex.Info{
	title: "Pleroma API",
	description: """
	This is documentation for client Pleroma API. Most of the endpoints and entities come
	from Mastodon API and have custom extensions on top.

	While this document aims to be a complete guide to the client API Pleroma exposes,
	the details are still being worked out. Some endpoints may have incomplete or poorly worded documentation.
	You might want to check the following resources if something is not clear:
	- [Legacy Pleroma-specific endpoint documentation](https://docs-develop.pleroma.social/backend/development/API/pleroma_api/)
	- [Mastodon API documentation](https://docs.joinmastodon.org/client/intro/)
	- [Differences in Mastodon API responses from vanilla Mastodon](https://docs-develop.pleroma.social/backend/development/API/differences_in_mastoapi_responses/)

	Please report such occurences on our [issue tracker](https://git.pleroma.social/pleroma/pleroma/-/issues). Feel free to submit API questions or proposals there too!
	""",
	# Strip environment from the version
	version: Application.spec(:pleroma, :vsn) \|> to_string() \|> String.replace(~r/\+.*$/, ""),
	extensions: %{
	# Logo path should be picked so that the path exists both on Pleroma instances and on api.pleroma.social
	"x-logo": %{"url" => "/static/logo.svg", "altText" => "Pleroma logo"}
	}
	},
	# populate the paths from a phoenix router
	- paths: OpenApiSpex.Paths.from_router(Router),
	+ paths: Map.merge(streaming_paths(), OpenApiSpex.Paths.from_router(Router)),
	components: %OpenApiSpex.Components{
	parameters: %{
	"accountIdOrNickname" =>
	Operation.parameter(:id, :path, :string, "Account ID or nickname",
	example: "123",
	required: true
	)
	},
	securitySchemes: %{
	"oAuth" => %OpenApiSpex.SecurityScheme{
	type: "oauth2",
	flows: %OpenApiSpex.OAuthFlows{
	password: %OpenApiSpex.OAuthFlow{
	authorizationUrl: "/oauth/authorize",
	tokenUrl: "/oauth/token",
	scopes: %{
	# TODO: Document granular scopes
	"read" => "Read everything",
	"write" => "Write everything",
	"follow" => "Manage relationships",
	"push" => "Web Push API subscriptions"
	}
	}
	}
	}
	}
	},
	extensions: %{
	# Redoc-specific extension, every time a new tag is added it should be reflected here,
	# otherwise it won't be shown.
	"x-tagGroups": [
	%{
	"name" => "Accounts",
	"tags" => ["Account actions", "Retrieve account information", "Scrobbles"]
	},
	%{
	"name" => "Administration",
	"tags" => [
	"Chat administration",
	"Emoji pack administration",
	"Frontend managment",
	"Instance configuration",
	"Instance documents",
	"Invites",
	"MediaProxy cache",
	"OAuth application managment",
	"Relays",
	"Report managment",
	"Status administration",
	"User administration",
	"Announcement management"
	]
	},
	%{"name" => "Applications", "tags" => ["Applications", "Push subscriptions"]},
	%{
	"name" => "Current account",
	"tags" => [
	"Account credentials",
	"Backups",
	"Blocks and mutes",
	"Data import",
	"Domain blocks",
	"Follow requests",
	"Mascot",
	"Markers",
	"Notifications",
	"Filters",
	"Settings"
	]
	},
	%{"name" => "Instance", "tags" => ["Custom emojis", "Instance misc"]},
	%{"name" => "Messaging", "tags" => ["Chats", "Conversations"]},
	%{
	"name" => "Statuses",
	"tags" => [
	"Emoji reactions",
	"Lists",
	"Polls",
	"Timelines",
	"Retrieve status information",
	"Scheduled statuses",
	"Search",
	"Status actions",
	"Media attachments"
	]
	},
	%{
	"name" => "Miscellaneous",
	"tags" => [
	"Emoji packs",
	"Reports",
	"Suggestions",
	"Announcements",
	"Remote interaction",
	"Others"
	]
	}
	]
	}
	}
	# discover request/response schemas from path specs
	\|> OpenApiSpex.resolve_schema_modules()
	end
	end
	diff --git a/lib/pleroma/web/api_spec/operations/streaming_operation.ex b/lib/pleroma/web/api_spec/operations/streaming_operation.ex
	new file mode 100644
	index 000000000..b580bc2f0
	--- /dev/null
	+++ b/lib/pleroma/web/api_spec/operations/streaming_operation.ex
	@@ -0,0 +1,464 @@
	+# Pleroma: A lightweight social networking server
	+# Copyright © 2017-2022 Pleroma Authors <https://pleroma.social/>
	+# SPDX-License-Identifier: AGPL-3.0-only
	+
	+defmodule Pleroma.Web.ApiSpec.StreamingOperation do
	+ alias OpenApiSpex.Operation
	+ alias OpenApiSpex.Response
	+ alias OpenApiSpex.Schema
	+ alias Pleroma.Web.ApiSpec.NotificationOperation
	+ alias Pleroma.Web.ApiSpec.Schemas.Chat
	+ alias Pleroma.Web.ApiSpec.Schemas.Conversation
	+ alias Pleroma.Web.ApiSpec.Schemas.FlakeID
	+ alias Pleroma.Web.ApiSpec.Schemas.Status
	+
	+ require Pleroma.Constants
	+
	+ @spec open_api_operation(atom) :: Operation.t()
	+ def open_api_operation(action) do
	+ operation = String.to_existing_atom("#{action}_operation")
	+ apply(__MODULE__, operation, [])
	+ end
	+
	+ @spec streaming_operation() :: Operation.t()
	+ def streaming_operation do
	+ %Operation{
	+ tags: ["Timelines"],
	+ summary: "Establish streaming connection",
	+ description: """
	+ Receive statuses in real-time via WebSocket.
	+
	+ You can specify the access token on the query string or through the `sec-websocket-protocol` header. Using
	+ the query string to authenticate is considered unsafe and should not be used unless you have to (e.g. to maintain
	+ your client's compatibility with Mastodon).
	+
	+ You may specify a stream on the query string. If you do so and you are connecting to a stream that requires logged-in users,
	+ you must specify the access token at the time of the connection (i.e. via query string or header).
	+
	+ Otherwise, you have the option to authenticate after you have established the connection through client-sent events.
	+
	+ The "Request body" section below describes what events clients can send through WebSocket, and the "Responses" section
	+ describes what events server will send through WebSocket.
	+ """,
	+ security: [%{"oAuth" => ["read:statuses", "read:notifications"]}],
	+ operationId: "WebsocketHandler.streaming",
	+ parameters:
	+ [
	+ Operation.parameter(:connection, :header, %Schema{type: :string}, "connection header",
	+ required: true
	+ ),
	+ Operation.parameter(:upgrade, :header, %Schema{type: :string}, "upgrade header",
	+ required: true
	+ ),
	+ Operation.parameter(
	+ :"sec-websocket-key",
	+ :header,
	+ %Schema{type: :string},
	+ "sec-websocket-key header",
	+ required: true
	+ ),
	+ Operation.parameter(
	+ :"sec-websocket-version",
	+ :header,
	+ %Schema{type: :string},
	+ "sec-websocket-version header",
	+ required: true
	+ )
	+ ] ++ stream_params() ++ access_token_params(),
	+ requestBody: request_body("Client-sent events", client_sent_events()),
	+ responses: %{
	+ 101 => switching_protocols_response(),
	+ 200 =>
	+ Operation.response(
	+ "Server-sent events",
	+ "application/json",
	+ server_sent_events()
	+ )
	+ }
	+ }
	+ end
	+
	+ defp stream_params do
	+ stream_specifier()
	+ \|> Enum.map(fn {name, schema} ->
	+ Operation.parameter(name, :query, schema, get_schema(schema).description)
	+ end)
	+ end
	+
	+ defp access_token_params do
	+ [
	+ Operation.parameter(:access_token, :query, token(), token().description),
	+ Operation.parameter(:"sec-websocket-protocol", :header, token(), token().description)
	+ ]
	+ end
	+
	+ defp switching_protocols_response do
	+ %Response{
	+ description: "Switching protocols",
	+ headers: %{
	+ "connection" => %OpenApiSpex.Header{required: true},
	+ "upgrade" => %OpenApiSpex.Header{required: true},
	+ "sec-websocket-accept" => %OpenApiSpex.Header{required: true}
	+ }
	+ }
	+ end
	+
	+ defp server_sent_events do
	+ %Schema{
	+ oneOf: [
	+ update_event(),
	+ status_update_event(),
	+ notification_event(),
	+ chat_update_event(),
	+ follow_relationships_update_event(),
	+ conversation_event(),
	+ delete_event(),
	+ pleroma_respond_event()
	+ ]
	+ }
	+ end
	+
	+ defp stream do
	+ %Schema{
	+ type: :array,
	+ title: "Stream",
	+ description: """
	+ The stream identifier.
	+ The first item is the name of the stream. If the stream needs a differentiator, the second item will be the corresponding identifier.
	+ Currently, for the following stream types, there is a second element in the array:
	+
	+ - `list`: The second element is the id of the list, as a string.
	+ - `hashtag`: The second element is the name of the hashtag.
	+ - `public:remote:media` and `public:remote`: The second element is the domain of the corresponding instance.
	+ """,
	+ maxItems: 2,
	+ minItems: 1,
	+ items: %Schema{type: :string},
	+ example: ["hashtag", "mew"]
	+ }
	+ end
	+
	+ defp get_schema(%Schema{} = schema), do: schema
	+ defp get_schema(schema), do: schema.schema
	+
	+ defp server_sent_event_helper(name, description, type, payload, opts \\ []) do
	+ payload_type = Keyword.get(opts, :payload_type, :json)
	+ has_stream = Keyword.get(opts, :has_stream, true)
	+
	+ stream_properties =
	+ if has_stream do
	+ %{stream: stream()}
	+ else
	+ %{}
	+ end
	+
	+ stream_example = if has_stream, do: %{"stream" => get_schema(stream()).example}, else: %{}
	+
	+ stream_required = if has_stream, do: [:stream], else: []
	+
	+ payload_schema =
	+ if payload_type == :json do
	+ %Schema{
	+ title: "Event payload",
	+ description: "JSON-encoded string of #{get_schema(payload).title}",
	+ allOf: [payload]
	+ }
	+ else
	+ payload
	+ end
	+
	+ payload_example =
	+ if payload_type == :json do
	+ get_schema(payload).example \|> Jason.encode!()
	+ else
	+ get_schema(payload).example
	+ end
	+
	+ %Schema{
	+ type: :object,
	+ title: name,
	+ description: description,
	+ required: [:event, :payload] ++ stream_required,
	+ properties:
	+ %{
	+ event: %Schema{
	+ title: "Event type",
	+ description: "Type of the event.",
	+ type: :string,
	+ required: true,
	+ enum: [type]
	+ },
	+ payload: payload_schema
	+ }
	+ \|> Map.merge(stream_properties),
	+ example:
	+ %{
	+ "event" => type,
	+ "payload" => payload_example
	+ }
	+ \|> Map.merge(stream_example)
	+ }
	+ end
	+
	+ defp update_event do
	+ server_sent_event_helper("New status", "A newly-posted status.", "update", Status)
	+ end
	+
	+ defp status_update_event do
	+ server_sent_event_helper("Edit", "A status that was just edited", "status.update", Status)
	+ end
	+
	+ defp notification_event do
	+ server_sent_event_helper(
	+ "Notification",
	+ "A new notification.",
	+ "notification",
	+ NotificationOperation.notification()
	+ )
	+ end
	+
	+ defp follow_relationships_update_event do
	+ server_sent_event_helper(
	+ "Follow relationships update",
	+ "An update to follow relationships.",
	+ "pleroma:follow_relationships_update",
	+ %Schema{
	+ type: :object,
	+ title: "Follow relationships update",
	+ required: [:state, :follower, :following],
	+ properties: %{
	+ state: %Schema{
	+ type: :string,
	+ description: "Follow state of the relationship.",
	+ enum: ["follow_pending", "follow_accept", "follow_reject", "unfollow"]
	+ },
	+ follower: %Schema{
	+ type: :object,
	+ description: "Information about the follower.",
	+ required: [:id, :follower_count, :following_count],
	+ properties: %{
	+ id: FlakeID,
	+ follower_count: %Schema{type: :integer},
	+ following_count: %Schema{type: :integer}
	+ }
	+ },
	+ following: %Schema{
	+ type: :object,
	+ description: "Information about the following person.",
	+ required: [:id, :follower_count, :following_count],
	+ properties: %{
	+ id: FlakeID,
	+ follower_count: %Schema{type: :integer},
	+ following_count: %Schema{type: :integer}
	+ }
	+ }
	+ },
	+ example: %{
	+ "state" => "follow_pending",
	+ "follower" => %{
	+ "id" => "someUser1",
	+ "follower_count" => 1,
	+ "following_count" => 1
	+ },
	+ "following" => %{
	+ "id" => "someUser2",
	+ "follower_count" => 1,
	+ "following_count" => 1
	+ }
	+ }
	+ }
	+ )
	+ end
	+
	+ defp chat_update_event do
	+ server_sent_event_helper(
	+ "Chat update",
	+ "A new chat message.",
	+ "pleroma:chat_update",
	+ Chat
	+ )
	+ end
	+
	+ defp conversation_event do
	+ server_sent_event_helper(
	+ "Conversation update",
	+ "An update about a conversation",
	+ "conversation",
	+ Conversation
	+ )
	+ end
	+
	+ defp delete_event do
	+ server_sent_event_helper(
	+ "Delete",
	+ "A status that was just deleted.",
	+ "delete",
	+ %Schema{
	+ type: :string,
	+ title: "Status id",
	+ description: "Id of the deleted status",
	+ allOf: [FlakeID],
	+ example: "some-opaque-id"
	+ },
	+ payload_type: :string,
	+ has_stream: false
	+ )
	+ end
	+
	+ defp pleroma_respond_event do
	+ server_sent_event_helper(
	+ "Server response",
	+ "A response to a client-sent event.",
	+ "pleroma:respond",
	+ %Schema{
	+ type: :object,
	+ title: "Results",
	+ required: [:result, :type],
	+ properties: %{
	+ result: %Schema{
	+ type: :string,
	+ title: "Result of the request",
	+ enum: ["success", "error", "ignored"]
	+ },
	+ error: %Schema{
	+ type: :string,
	+ title: "Error code",
	+ description: "An error identifier. Only appears if `result` is `error`."
	+ },
	+ type: %Schema{
	+ type: :string,
	+ description: "Type of the request."
	+ }
	+ },
	+ example: %{"result" => "success", "type" => "pleroma:authenticate"}
	+ },
	+ has_stream: false
	+ )
	+ end
	+
	+ defp client_sent_events do
	+ %Schema{
	+ oneOf: [
	+ subscribe_event(),
	+ unsubscribe_event(),
	+ authenticate_event()
	+ ]
	+ }
	+ end
	+
	+ defp request_body(description, schema, opts \\ []) do
	+ %OpenApiSpex.RequestBody{
	+ description: description,
	+ content: %{
	+ "application/json" => %OpenApiSpex.MediaType{
	+ schema: schema,
	+ example: opts[:example],
	+ examples: opts[:examples]
	+ }
	+ }
	+ }
	+ end
	+
	+ defp client_sent_event_helper(name, description, type, properties, opts) do
	+ required = opts[:required] \|\| []
	+
	+ %Schema{
	+ type: :object,
	+ title: name,
	+ required: [:type] ++ required,
	+ description: description,
	+ properties:
	+ %{
	+ type: %Schema{type: :string, enum: [type], description: "Type of the event."}
	+ }
	+ \|> Map.merge(properties),
	+ example: opts[:example]
	+ }
	+ end
	+
	+ defp subscribe_event do
	+ client_sent_event_helper(
	+ "Subscribe",
	+ "Subscribe to a stream.",
	+ "subscribe",
	+ stream_specifier(),
	+ required: [:stream],
	+ example: %{"type" => "subscribe", "stream" => "list", "list" => "1"}
	+ )
	+ end
	+
	+ defp unsubscribe_event do
	+ client_sent_event_helper(
	+ "Unsubscribe",
	+ "Unsubscribe from a stream.",
	+ "unsubscribe",
	+ stream_specifier(),
	+ required: [:stream],
	+ example: %{
	+ "type" => "unsubscribe",
	+ "stream" => "public:remote:media",
	+ "instance" => "example.org"
	+ }
	+ )
	+ end
	+
	+ defp authenticate_event do
	+ client_sent_event_helper(
	+ "Authenticate",
	+ "Authenticate via an access token.",
	+ "pleroma:authenticate",
	+ %{
	+ token: token()
	+ },
	+ required: [:token]
	+ )
	+ end
	+
	+ defp token do
	+ %Schema{
	+ type: :string,
	+ description: "An OAuth access token with corresponding permissions.",
	+ example: "some token"
	+ }
	+ end
	+
	+ defp stream_specifier do
	+ %{
	+ stream: %Schema{
	+ type: :string,
	+ description: "The name of the stream.",
	+ enum:
	+ Pleroma.Constants.public_streams() ++
	+ [
	+ "public:remote",
	+ "public:remote:media",
	+ "user",
	+ "user:pleroma_chat",
	+ "user:notification",
	+ "direct",
	+ "list",
	+ "hashtag"
	+ ]
	+ },
	+ list: %Schema{
	+ type: :string,
	+ title: "List id",
	+ description: "The id of the list. Required when `stream` is `list`.",
	+ example: "some-id"
	+ },
	+ tag: %Schema{
	+ type: :string,
	+ title: "Hashtag name",
	+ description: "The name of the hashtag. Required when `stream` is `hashtag`.",
	+ example: "mew"
	+ },
	+ instance: %Schema{
	+ type: :string,
	+ title: "Domain name",
	+ description:
	+ "Domain name of the instance. Required when `stream` is `public:remote` or `public:remote:media`.",
	+ example: "example.org"
	+ }
	+ }
	+ end
	+end
	diff --git a/lib/pleroma/web/mastodon_api/websocket_handler.ex b/lib/pleroma/web/mastodon_api/websocket_handler.ex
	index 88444106d..07c2b62e3 100644
	--- a/lib/pleroma/web/mastodon_api/websocket_handler.ex
	+++ b/lib/pleroma/web/mastodon_api/websocket_handler.ex
	@@ -1,140 +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, topic: topic, oauth_token: oauth_token, count: 0, timer: nil},
	+ %{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}, topic #{state.topic}"
	+ "#{__MODULE__} accepted websocket connection for user #{(state.user \|\| %{id: "anonymous"}).id}, topics #{state.topics}"
	)

	- Streamer.add_socket(state.topic, state.oauth_token)
	+ 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}, topic #{state.topic \|\| "?"}: #{inspect(reason)}"
	+ "#{__MODULE__} terminating websocket connection for user #{(state.user \|\| %{id: "anonymous"}).id}, topics #{state.topics \|\| "?"}: #{inspect(reason)}"
	)

	- Streamer.remove_socket(state.topic)
	+ 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 b9a04cc76..48ca82421 100644
	--- a/lib/pleroma/web/streamer.ex
	+++ b/lib/pleroma/web/streamer.ex
	@@ -1,410 +1,420 @@
	# 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(), User.t() \| nil, Token.t() \| nil, Map.t()) ::
	- {:ok, topic :: String.t()} \| {:error, :bad_topic}
	+ @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)
	+ 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 6a55242b0..f97570b0a 100644
	--- a/lib/pleroma/web/views/streamer_view.ex
	+++ b/lib/pleroma/web/views/streamer_view.ex
	@@ -1,138 +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 9be0445c0..a2c20f0a6 100644
	--- a/test/pleroma/integration/mastodon_websocket_test.exs
	+++ b/test/pleroma/integration/mastodon_websocket_test.exs
	@@ -1,150 +1,496 @@
	# 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
	+
	+ # Turns atom keys to strings
	+ defp atom_key_to_string(json) do
	+ json
	+ \|> Jason.encode!()
	+ \|> Jason.decode!()
	+ end
	+
	test "refuses invalid requests" do
	capture_log(fn ->
	- assert {:error, %WebSockex.RequestError{code: 404}} = start_socket()
	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!()
	+ \|> atom_key_to_string()

	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 "rejects invalid streams" do
	+ {:ok, pid} = start_socket()
	+ WebsocketClient.send_text(pid, %{type: "subscribe", stream: "nonsense"} \|> Jason.encode!())
	+ assert_receive {:text, raw_json}, 1_000
	+
	+ assert {:ok,
	+ %{
	+ "event" => "pleroma:respond",
	+ "payload" => %{"type" => "subscribe", "result" => "error", "error" => "bad_topic"}
	+ }} = decode_json(raw_json)
	+ 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 "accepts the 'list' stream", %{token: token, user: user} do
	+ posting_user = insert(:user)
	+
	+ {:ok, list} = Pleroma.List.create("test", user)
	+ Pleroma.List.follow(list, posting_user)
	+
	+ assert {:ok, _} = start_socket("?stream=list&access_token=#{token.token}&list=#{list.id}")
	+
	+ assert {:ok, pid} = start_socket("?access_token=#{token.token}")
	+
	+ WebsocketClient.send_text(
	+ pid,
	+ %{type: "subscribe", stream: "list", list: list.id} \|> 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: "list", list: to_string(list.id)} \|> Jason.encode!()
	+ )
	+
	+ assert_receive {:text, raw_json}, 1_000
	+
	+ assert {:ok,
	+ %{
	+ "event" => "pleroma:respond",
	+ "payload" => %{"type" => "subscribe", "result" => "ignored"}
	+ }} = 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
	+
	+ test "receives private statuses", %{user: reading_user, token: token} do
	+ user = insert(:user)
	+ CommonAPI.follow(reading_user, user)
	+
	+ {:ok, _} = start_socket("?stream=user&access_token=#{token.token}")
	+
	+ {:ok, activity} =
	+ CommonAPI.post(user, %{status: "nice echo chamber", visibility: "private"})
	+
	+ 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: reading_user
	+ )
	+ \|> Jason.encode!()
	+ \|> Jason.decode!()
	+
	+ assert json == view_json
	+ end
	+
	+ test "receives edits", %{user: reading_user, token: token} do
	+ user = insert(:user)
	+ CommonAPI.follow(reading_user, user)
	+
	+ {:ok, _} = start_socket("?stream=user&access_token=#{token.token}")
	+
	+ {:ok, activity} =
	+ CommonAPI.post(user, %{status: "nice echo chamber", visibility: "private"})
	+
	+ assert_receive {:text, _raw_json}, 1_000
	+
	+ {:ok, _} = CommonAPI.update(user, activity, %{status: "mew mew", visibility: "private"})
	+
	+ assert_receive {:text, raw_json}, 1_000
	+
	+ activity = Pleroma.Activity.normalize(activity)
	+
	+ view_json =
	+ Pleroma.Web.MastodonAPI.StatusView.render("show.json",
	+ activity: activity,
	+ for: reading_user
	+ )
	+ \|> Jason.encode!()
	+ \|> Jason.decode!()
	+
	+ assert {:ok, %{"event" => "status.update", "payload" => ^view_json}} = decode_json(raw_json)
	+ end
	+
	+ test "receives notifications", %{user: reading_user, token: token} do
	+ user = insert(:user)
	+ CommonAPI.follow(reading_user, user)
	+
	+ {:ok, _} = start_socket("?stream=user:notification&access_token=#{token.token}")
	+
	+ {:ok, %Pleroma.Activity{id: activity_id} = _activity} =
	+ CommonAPI.post(user, %{
	+ status: "nice echo chamber @#{reading_user.nickname}",
	+ visibility: "private"
	+ })
	+
	+ assert_receive {:text, raw_json}, 1_000
	+
	+ assert {:ok,
	+ %{
	+ "event" => "notification",
	+ "payload" => %{
	+ "status" => %{
	+ "id" => ^activity_id
	+ }
	+ }
	+ }} = decode_json(raw_json)
	+ end
	end
	end
	diff --git a/test/pleroma/notification_test.exs b/test/pleroma/notification_test.exs
	index e55aa3a08..71af9acb8 100644
	--- a/test/pleroma/notification_test.exs
	+++ b/test/pleroma/notification_test.exs
	@@ -1,1282 +1,1282 @@
	# Pleroma: A lightweight social networking server
	# Copyright © 2017-2022 Pleroma Authors <https://pleroma.social/>
	# SPDX-License-Identifier: AGPL-3.0-only

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

	import Pleroma.Factory
	import Mock

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

	assert [%Pleroma.Marker{unread_count: 2}] =
	Pleroma.Marker.get_markers(other_user, ["notifications"])
	end

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

	User.subscribe(subscriber, user)

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

	assert notification.user_id == subscriber.id
	end

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

	User.subscribe(subscriber, other_user)

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

	pre_mute_notification_id = pre_mute_notification.id
	post_mute_notification_id = post_mute_notification.id

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

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

	describe "create_notification" do
	@tag needs_streamer: true
	test "it creates a notification for user and send to the 'user' and the 'user:notification' stream" do
	%{user: user, token: oauth_token} = oauth_access(["read"])

	task =
	Task.async(fn ->
	{:ok, _topic} = Streamer.get_topic_and_add_socket("user", user, oauth_token)
	- assert_receive {:render_with_user, _, _, _}, 4_000
	+ assert_receive {:render_with_user, _, _, _, _}, 4_000
	end)

	task_user_notification =
	Task.async(fn ->
	{:ok, _topic} =
	Streamer.get_topic_and_add_socket("user:notification", user, oauth_token)

	- assert_receive {:render_with_user, _, _, _}, 4_000
	+ assert_receive {:render_with_user, _, _, _, _}, 4_000
	end)

	activity = insert(:note_activity)

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

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

	assert Notification.create_notification(activity, user)
	end

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

	notification = Notification.create_notification(activity, muter)

	assert notification.id
	assert notification.seen
	end

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

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

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

	assert Notification.create_notification(activity, muter)
	end

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

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

	notification = Notification.create_notification(activity, muter)

	assert notification.id
	assert notification.seen
	end

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

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

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

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

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

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

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

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

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

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

	refute Notification.create_notification(activity, author)
	end

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

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

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

	User.subscribe(subscriber, user)

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

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

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

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

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

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

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

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

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

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

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

	assert notification
	refute notification.seen
	end

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

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

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

	assert notification
	refute notification.seen
	end
	end

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

	assert notification.user_id == other_user.id
	end

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

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

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

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

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

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

	assert notification.user_id == other_user.id
	end

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

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

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

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

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

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

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

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

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

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

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

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

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

	assert n2.id > n1.id

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

	[_, read_notification] = Notification.set_read_up_to(other_user, n2.id)

	assert read_notification.activity.object

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

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

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

	assert m.last_read_id == to_string(n2.id)
	end
	end

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

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

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

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

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

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

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

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

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

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

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

	assert other_user in enabled_receivers
	end

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

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

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

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

	assert other_user in enabled_receivers
	end

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

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

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

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

	assert other_user not in enabled_receivers
	end

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

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

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

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

	assert other_user not in enabled_receivers
	end

	test "it only notifies the post's author in likes" do
	user = insert(:user)
	other_user = insert(:user)
	third_user = insert(:user)

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

	{:ok, like_data, _} = Builder.like(third_user, activity_one.object)

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

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

	assert other_user not in enabled_receivers
	end

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

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

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

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

	assert other_user not in enabled_receivers
	end

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

	assert repeated_user in enabled_receivers
	assert other_user not in enabled_receivers
	end
	end

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

	remote_user_url = remote_user.ap_id

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

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

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

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

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

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

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

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

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

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

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

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

	[notification] = Notification.for_user(user)

	assert notification.activity.object
	assert notification.seen
	end

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

	assert [notification] = Notification.for_user(followed_user)

	assert %{type: "follow"} =
	NotificationView.render("show.json", %{
	notification: notification,
	for: followed_user
	})
	end
	end
	end
	diff --git a/test/pleroma/web/streamer_test.exs b/test/pleroma/web/streamer_test.exs
	index 7ab0e379b..d85358fd4 100644
	--- a/test/pleroma/web/streamer_test.exs
	+++ b/test/pleroma/web/streamer_test.exs
	@@ -1,1068 +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}
	+ 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, 10:22 PM (1 d, 17 h)
Storage Engine: blob
Storage Format: Raw Data
Storage Handle: 1798764
Default Alt Text: (175 KB)

No OneTemporaryActions

View Options

File Metadata

Event Timeline

No OneTemporary
Actions