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
14 alias Pleroma.Web.ActivityPub.ActivityPub
15 alias Pleroma.Web.ActivityPub.Visibility
16 alias Pleroma.Web.CommonAPI
17 alias Pleroma.Web.StreamerView
20 @registry Pleroma.Web.StreamerRegistry
22 def registry, do: @registry
24 def add_socket(topic, %User{} = user) do
25 if should_env_send?(), do: Registry.register(@registry, user_topic(topic, user), true)
28 def add_socket(topic, _) do
29 if should_env_send?(), do: Registry.register(@registry, topic, false)
32 def remove_socket(topic) do
33 if should_env_send?(), do: Registry.unregister(@registry, topic)
36 def stream(topics, item) when is_list(topics) do
37 if should_env_send?() do
38 Enum.each(topics, fn t ->
39 spawn(fn -> do_stream(t, item) end)
46 def stream(topic, items) when is_list(items) do
47 if should_env_send?() do
48 Enum.each(items, fn i ->
49 spawn(fn -> do_stream(topic, i) end)
56 def stream(topic, item) do
57 if should_env_send?() do
58 spawn(fn -> do_stream(topic, item) end)
64 def filtered_by_user?(%User{} = user, %Activity{} = item) do
65 %{block: blocked_ap_ids, mute: muted_ap_ids, reblog_mute: reblog_muted_ap_ids} =
66 User.outgoing_relationships_ap_ids(user, [:block, :mute, :reblog_mute])
68 recipient_blocks = MapSet.new(blocked_ap_ids ++ muted_ap_ids)
69 recipients = MapSet.new(item.recipients)
70 domain_blocks = Pleroma.Web.ActivityPub.MRF.subdomains_regex(user.domain_blocks)
72 with parent <- Object.normalize(item) || item,
74 Enum.all?([blocked_ap_ids, muted_ap_ids], &(item.actor not in &1)),
75 true <- item.data["type"] != "Announce" || item.actor not in reblog_muted_ap_ids,
76 true <- Enum.all?([blocked_ap_ids, muted_ap_ids], &(parent.data["actor"] not in &1)),
77 true <- MapSet.disjoint?(recipients, recipient_blocks),
78 %{host: item_host} <- URI.parse(item.actor),
79 %{host: parent_host} <- URI.parse(parent.data["actor"]),
80 false <- Pleroma.Web.ActivityPub.MRF.subdomain_match?(domain_blocks, item_host),
81 false <- Pleroma.Web.ActivityPub.MRF.subdomain_match?(domain_blocks, parent_host),
82 true <- thread_containment(item, user),
83 false <- CommonAPI.thread_muted?(user, item) do
90 def filtered_by_user?(%User{} = user, %Notification{activity: activity}) do
91 filtered_by_user?(user, activity)
94 defp do_stream("direct", item) do
96 User.get_recipients_from_activity(item)
97 |> Enum.map(fn %{id: id} -> "direct:#{id}" end)
99 Enum.each(recipient_topics, fn user_topic ->
100 Logger.debug("Trying to push direct message to #{user_topic}\n\n")
101 push_to_socket(user_topic, item)
105 defp do_stream("participation", participation) do
106 user_topic = "direct:#{participation.user_id}"
107 Logger.debug("Trying to push a conversation participation to #{user_topic}\n\n")
109 push_to_socket(user_topic, participation)
112 defp do_stream("list", item) do
113 # filter the recipient list if the activity is not public, see #270.
115 case Visibility.is_public?(item) do
117 Pleroma.List.get_lists_from_activity(item)
120 Pleroma.List.get_lists_from_activity(item)
121 |> Enum.filter(fn list ->
122 owner = User.get_cached_by_id(list.user_id)
124 Visibility.visible_for_user?(item, owner)
130 |> Enum.map(fn %{id: id} -> "list:#{id}" end)
132 Enum.each(recipient_topics, fn list_topic ->
133 Logger.debug("Trying to push message to #{list_topic}\n\n")
134 push_to_socket(list_topic, item)
138 defp do_stream(topic, %Notification{} = item)
139 when topic in ["user", "user:notification"] do
140 Registry.dispatch(@registry, "#{topic}:#{item.user_id}", fn list ->
141 Enum.each(list, fn {pid, _auth} ->
142 send(pid, {:render_with_user, StreamerView, "notification.json", item})
147 defp do_stream("user", item) do
148 Logger.debug("Trying to push to users")
151 User.get_recipients_from_activity(item)
152 |> Enum.map(fn %{id: id} -> "user:#{id}" end)
154 Enum.each(recipient_topics, fn topic ->
155 push_to_socket(topic, item)
159 defp do_stream(topic, item) do
160 Logger.debug("Trying to push to #{topic}")
161 Logger.debug("Pushing item to #{topic}")
162 push_to_socket(topic, item)
165 defp push_to_socket(topic, %Participation{} = participation) do
166 rendered = StreamerView.render("conversation.json", participation)
168 Registry.dispatch(@registry, topic, fn list ->
169 Enum.each(list, fn {pid, _} ->
170 send(pid, {:text, rendered})
175 defp push_to_socket(topic, %Activity{
176 data: %{"type" => "Delete", "deleted_activity_id" => deleted_activity_id}
178 rendered = Jason.encode!(%{event: "delete", payload: to_string(deleted_activity_id)})
180 Registry.dispatch(@registry, topic, fn list ->
181 Enum.each(list, fn {pid, _} ->
182 send(pid, {:text, rendered})
187 defp push_to_socket(_topic, %Activity{data: %{"type" => "Delete"}}), do: :noop
189 defp push_to_socket(topic, item) do
190 anon_render = StreamerView.render("update.json", item)
192 Registry.dispatch(@registry, topic, fn list ->
193 Enum.each(list, fn {pid, auth?} ->
195 send(pid, {:render_with_user, StreamerView, "update.json", item})
197 send(pid, {:text, anon_render})
203 defp thread_containment(_activity, %User{skip_thread_containment: true}), do: true
205 defp thread_containment(activity, user) do
206 if Config.get([:instance, :skip_thread_containment]) do
209 ActivityPub.contain_activity(activity, user)
213 # In test environement, only return true if the registry is started.
214 # In benchmark environment, returns false.
215 # In any other environment, always returns true.
218 def should_env_send? do
219 case Process.whereis(@registry) do
228 @mix_env == :benchmark ->
229 def should_env_send?, do: false
232 def should_env_send?, do: true
235 defp user_topic(topic, user)
236 when topic in ~w[user user:notification direct] do
237 "#{topic}:#{user.id}"
240 defp user_topic(topic, _) do