1 # Pleroma: A lightweight social networking server
2 # Copyright © 2017-2020 Pleroma Authors <https://pleroma.social/>
3 # SPDX-License-Identifier: AGPL-3.0-only
5 defmodule Pleroma.Web.Streamer do
10 alias Pleroma.Conversation.Participation
11 alias Pleroma.Notification
15 alias Pleroma.Web.ActivityPub.ActivityPub
16 alias Pleroma.Web.ActivityPub.Visibility
17 alias Pleroma.Web.CommonAPI
18 alias Pleroma.Web.StreamerView
21 @registry Pleroma.Web.StreamerRegistry
23 def registry, do: @registry
25 @public_streams ["public", "public:local", "public:media", "public:local:media"]
26 @user_streams ["user", "user:notification", "direct", "user:pleroma_chat"]
28 @doc "Expands and authorizes a stream, and registers the process for streaming."
29 @spec get_topic_and_add_socket(stream :: String.t(), User.t() | nil, Map.t() | nil) ::
30 {:ok, topic :: String.t()} | {:error, :bad_topic} | {:error, :unauthorized}
31 def get_topic_and_add_socket(stream, user, params \\ %{}) do
32 case get_topic(stream, user, params) do
33 {:ok, topic} -> add_socket(topic, user)
38 @doc "Expand and authorizes a stream"
39 @spec get_topic(stream :: String.t(), User.t() | nil, Map.t()) ::
40 {:ok, topic :: String.t()} | {:error, :bad_topic}
41 def get_topic(stream, user, params \\ %{})
43 # Allow all public steams.
44 def get_topic(stream, _, _) when stream in @public_streams do
48 # Allow all hashtags streams.
49 def get_topic("hashtag", _, %{"tag" => tag}) do
50 {:ok, "hashtag:" <> tag}
53 # Expand user streams.
54 def get_topic(stream, %User{} = user, _) when stream in @user_streams do
55 {:ok, stream <> ":" <> to_string(user.id)}
58 def get_topic(stream, _, _) when stream in @user_streams do
59 {:error, :unauthorized}
63 def get_topic("list", %User{} = user, %{"list" => id}) do
64 if Pleroma.List.get(id, user) do
65 {:ok, "list:" <> to_string(id)}
71 def get_topic("list", _, _) do
72 {:error, :unauthorized}
75 def get_topic(_, _, _) do
79 @doc "Registers the process for streaming. Use `get_topic/3` to get the full authorized topic."
80 def add_socket(topic, user) do
81 if should_env_send?() do
82 auth? = if user, do: true
83 Registry.register(@registry, topic, auth?)
89 def remove_socket(topic) do
90 if should_env_send?(), do: Registry.unregister(@registry, topic)
93 def stream(topics, item) when is_list(topics) do
94 if should_env_send?() do
95 Enum.each(topics, fn t ->
96 spawn(fn -> do_stream(t, item) end)
103 def stream(topic, items) when is_list(items) do
104 if should_env_send?() do
105 Enum.each(items, fn i ->
106 spawn(fn -> do_stream(topic, i) end)
113 def stream(topic, item) do
114 if should_env_send?() do
115 spawn(fn -> do_stream(topic, item) end)
121 def filtered_by_user?(%User{} = user, %Activity{} = item) do
122 %{block: blocked_ap_ids, mute: muted_ap_ids, reblog_mute: reblog_muted_ap_ids} =
123 User.outgoing_relationships_ap_ids(user, [:block, :mute, :reblog_mute])
125 recipient_blocks = MapSet.new(blocked_ap_ids ++ muted_ap_ids)
126 recipients = MapSet.new(item.recipients)
127 domain_blocks = Pleroma.Web.ActivityPub.MRF.subdomains_regex(user.domain_blocks)
129 with parent <- Object.normalize(item) || item,
131 Enum.all?([blocked_ap_ids, muted_ap_ids], &(item.actor not in &1)),
132 true <- item.data["type"] != "Announce" || item.actor not in reblog_muted_ap_ids,
133 true <- Enum.all?([blocked_ap_ids, muted_ap_ids], &(parent.data["actor"] not in &1)),
134 true <- MapSet.disjoint?(recipients, recipient_blocks),
135 %{host: item_host} <- URI.parse(item.actor),
136 %{host: parent_host} <- URI.parse(parent.data["actor"]),
137 false <- Pleroma.Web.ActivityPub.MRF.subdomain_match?(domain_blocks, item_host),
138 false <- Pleroma.Web.ActivityPub.MRF.subdomain_match?(domain_blocks, parent_host),
139 true <- thread_containment(item, user),
140 false <- CommonAPI.thread_muted?(user, parent) do
147 def filtered_by_user?(%User{} = user, %Notification{activity: activity}) do
148 filtered_by_user?(user, activity)
151 defp do_stream("direct", item) do
153 User.get_recipients_from_activity(item)
154 |> Enum.map(fn %{id: id} -> "direct:#{id}" end)
156 Enum.each(recipient_topics, fn user_topic ->
157 Logger.debug("Trying to push direct message to #{user_topic}\n\n")
158 push_to_socket(user_topic, item)
162 defp do_stream("participation", participation) do
163 user_topic = "direct:#{participation.user_id}"
164 Logger.debug("Trying to push a conversation participation to #{user_topic}\n\n")
166 push_to_socket(user_topic, participation)
169 defp do_stream("list", item) do
170 # filter the recipient list if the activity is not public, see #270.
172 case Visibility.is_public?(item) do
174 Pleroma.List.get_lists_from_activity(item)
177 Pleroma.List.get_lists_from_activity(item)
178 |> Enum.filter(fn list ->
179 owner = User.get_cached_by_id(list.user_id)
181 Visibility.visible_for_user?(item, owner)
187 |> Enum.map(fn %{id: id} -> "list:#{id}" end)
189 Enum.each(recipient_topics, fn list_topic ->
190 Logger.debug("Trying to push message to #{list_topic}\n\n")
191 push_to_socket(list_topic, item)
195 defp do_stream(topic, %Notification{} = item)
196 when topic in ["user", "user:notification"] do
197 Registry.dispatch(@registry, "#{topic}:#{item.user_id}", fn list ->
198 Enum.each(list, fn {pid, _auth} ->
199 send(pid, {:render_with_user, StreamerView, "notification.json", item})
204 defp do_stream(topic, %{data: %{"type" => "ChatMessage"}} = object)
205 when topic in ["user", "user:pleroma_chat"] do
206 recipients = [object.data["actor"] | object.data["to"]]
209 %{ap_id: recipients, local: true}
210 |> Pleroma.User.Query.build()
212 |> Enum.map(fn %{id: id} = user -> {user, "#{topic}:#{id}"} end)
214 Enum.each(topics, fn {user, topic} ->
215 Registry.dispatch(@registry, topic, fn list ->
216 Enum.each(list, fn {pid, _auth} ->
217 text = StreamerView.render("chat_update.json", object, user, recipients)
218 send(pid, {:text, text})
224 defp do_stream("user", item) do
225 Logger.debug("Trying to push to users")
228 User.get_recipients_from_activity(item)
229 |> Enum.map(fn %{id: id} -> "user:#{id}" end)
231 Enum.each(recipient_topics, fn topic ->
232 push_to_socket(topic, item)
236 defp do_stream(topic, item) do
237 Logger.debug("Trying to push to #{topic}")
238 Logger.debug("Pushing item to #{topic}")
239 push_to_socket(topic, item)
242 defp push_to_socket(topic, %Participation{} = participation) do
243 rendered = StreamerView.render("conversation.json", participation)
245 Registry.dispatch(@registry, topic, fn list ->
246 Enum.each(list, fn {pid, _} ->
247 send(pid, {:text, rendered})
252 defp push_to_socket(topic, %Activity{
253 data: %{"type" => "Delete", "deleted_activity_id" => deleted_activity_id}
255 rendered = Jason.encode!(%{event: "delete", payload: to_string(deleted_activity_id)})
257 Registry.dispatch(@registry, topic, fn list ->
258 Enum.each(list, fn {pid, _} ->
259 send(pid, {:text, rendered})
264 defp push_to_socket(_topic, %Activity{data: %{"type" => "Delete"}}), do: :noop
266 defp push_to_socket(topic, item) do
267 anon_render = StreamerView.render("update.json", item)
269 Registry.dispatch(@registry, topic, fn list ->
270 Enum.each(list, fn {pid, auth?} ->
272 send(pid, {:render_with_user, StreamerView, "update.json", item})
274 send(pid, {:text, anon_render})
280 defp thread_containment(_activity, %User{skip_thread_containment: true}), do: true
282 defp thread_containment(activity, user) do
283 if Config.get([:instance, :skip_thread_containment]) do
286 ActivityPub.contain_activity(activity, user)
290 # In test environement, only return true if the registry is started.
291 # In benchmark environment, returns false.
292 # In any other environment, always returns true.
295 def should_env_send? do
296 case Process.whereis(@registry) do
305 @mix_env == :benchmark ->
306 def should_env_send?, do: false
309 def should_env_send?, do: true