fix formatting
[akkoma] / lib / pleroma / web / streamer.ex
1 # Pleroma: A lightweight social networking server
2 # Copyright © 2017-2019 Pleroma Authors <https://pleroma.social/>
3 # SPDX-License-Identifier: AGPL-3.0-only
4
5 defmodule Pleroma.Web.Streamer do
6 use GenServer
7 require Logger
8 alias Pleroma.Activity
9 alias Pleroma.Config
10 alias Pleroma.Conversation.Participation
11 alias Pleroma.Notification
12 alias Pleroma.Object
13 alias Pleroma.User
14 alias Pleroma.Web.ActivityPub.ActivityPub
15 alias Pleroma.Web.ActivityPub.Visibility
16 alias Pleroma.Web.CommonAPI
17 alias Pleroma.Web.MastodonAPI.NotificationView
18
19 @keepalive_interval :timer.seconds(30)
20
21 def start_link(_) do
22 GenServer.start_link(__MODULE__, %{}, name: __MODULE__)
23 end
24
25 def add_socket(topic, socket) do
26 GenServer.cast(__MODULE__, %{action: :add, socket: socket, topic: topic})
27 end
28
29 def remove_socket(topic, socket) do
30 GenServer.cast(__MODULE__, %{action: :remove, socket: socket, topic: topic})
31 end
32
33 def stream(topic, item) do
34 GenServer.cast(__MODULE__, %{action: :stream, topic: topic, item: item})
35 end
36
37 def init(args) do
38 Process.send_after(self(), %{action: :ping}, @keepalive_interval)
39
40 {:ok, args}
41 end
42
43 def handle_info(%{action: :ping}, topics) do
44 topics
45 |> Map.values()
46 |> List.flatten()
47 |> Enum.each(fn socket ->
48 Logger.debug("Sending keepalive ping")
49 send(socket.transport_pid, {:text, ""})
50 end)
51
52 Process.send_after(self(), %{action: :ping}, @keepalive_interval)
53
54 {:noreply, topics}
55 end
56
57 def handle_cast(%{action: :stream, topic: "direct", item: item}, topics) do
58 recipient_topics =
59 User.get_recipients_from_activity(item)
60 |> Enum.map(fn %{id: id} -> "direct:#{id}" end)
61
62 Enum.each(recipient_topics || [], fn user_topic ->
63 Logger.debug("Trying to push direct message to #{user_topic}\n\n")
64 push_to_socket(topics, user_topic, item)
65 end)
66
67 {:noreply, topics}
68 end
69
70 def handle_cast(%{action: :stream, topic: "participation", item: participation}, topics) do
71 user_topic = "direct:#{participation.user_id}"
72 Logger.debug("Trying to push a conversation participation to #{user_topic}\n\n")
73
74 push_to_socket(topics, user_topic, participation)
75
76 {:noreply, topics}
77 end
78
79 def handle_cast(%{action: :stream, topic: "list", item: item}, topics) do
80 # filter the recipient list if the activity is not public, see #270.
81 recipient_lists =
82 case Visibility.is_public?(item) do
83 true ->
84 Pleroma.List.get_lists_from_activity(item)
85
86 _ ->
87 Pleroma.List.get_lists_from_activity(item)
88 |> Enum.filter(fn list ->
89 owner = User.get_cached_by_id(list.user_id)
90
91 Visibility.visible_for_user?(item, owner)
92 end)
93 end
94
95 recipient_topics =
96 recipient_lists
97 |> Enum.map(fn %{id: id} -> "list:#{id}" end)
98
99 Enum.each(recipient_topics || [], fn list_topic ->
100 Logger.debug("Trying to push message to #{list_topic}\n\n")
101 push_to_socket(topics, list_topic, item)
102 end)
103
104 {:noreply, topics}
105 end
106
107 def handle_cast(
108 %{action: :stream, topic: topic, item: %Notification{} = item},
109 topics
110 )
111 when topic in ["user", "user:notification"] do
112 topics
113 |> Map.get("#{topic}:#{item.user_id}", [])
114 |> Enum.each(fn socket ->
115 with %User{} = user <- User.get_cached_by_ap_id(socket.assigns[:user].ap_id),
116 true <- should_send?(user, item) do
117 send(
118 socket.transport_pid,
119 {:text, represent_notification(socket.assigns[:user], item)}
120 )
121 end
122 end)
123
124 {:noreply, topics}
125 end
126
127 def handle_cast(%{action: :stream, topic: "user", item: item}, topics) do
128 Logger.debug("Trying to push to users")
129
130 recipient_topics =
131 User.get_recipients_from_activity(item)
132 |> Enum.map(fn %{id: id} -> "user:#{id}" end)
133
134 Enum.each(recipient_topics, fn topic ->
135 push_to_socket(topics, topic, item)
136 end)
137
138 {:noreply, topics}
139 end
140
141 def handle_cast(%{action: :stream, topic: topic, item: item}, topics) do
142 Logger.debug("Trying to push to #{topic}")
143 Logger.debug("Pushing item to #{topic}")
144 push_to_socket(topics, topic, item)
145 {:noreply, topics}
146 end
147
148 def handle_cast(%{action: :add, topic: topic, socket: socket}, sockets) do
149 topic = internal_topic(topic, socket)
150 sockets_for_topic = sockets[topic] || []
151 sockets_for_topic = Enum.uniq([socket | sockets_for_topic])
152 sockets = Map.put(sockets, topic, sockets_for_topic)
153 Logger.debug("Got new conn for #{topic}")
154 {:noreply, sockets}
155 end
156
157 def handle_cast(%{action: :remove, topic: topic, socket: socket}, sockets) do
158 topic = internal_topic(topic, socket)
159 sockets_for_topic = sockets[topic] || []
160 sockets_for_topic = List.delete(sockets_for_topic, socket)
161 sockets = Map.put(sockets, topic, sockets_for_topic)
162 Logger.debug("Removed conn for #{topic}")
163 {:noreply, sockets}
164 end
165
166 def handle_cast(m, state) do
167 Logger.info("Unknown: #{inspect(m)}, #{inspect(state)}")
168 {:noreply, state}
169 end
170
171 defp represent_update(%Activity{} = activity, %User{} = user) do
172 %{
173 event: "update",
174 payload:
175 Pleroma.Web.MastodonAPI.StatusView.render(
176 "status.json",
177 activity: activity,
178 for: user
179 )
180 |> Jason.encode!()
181 }
182 |> Jason.encode!()
183 end
184
185 defp represent_update(%Activity{} = activity) do
186 %{
187 event: "update",
188 payload:
189 Pleroma.Web.MastodonAPI.StatusView.render(
190 "status.json",
191 activity: activity
192 )
193 |> Jason.encode!()
194 }
195 |> Jason.encode!()
196 end
197
198 def represent_conversation(%Participation{} = participation) do
199 %{
200 event: "conversation",
201 payload:
202 Pleroma.Web.MastodonAPI.ConversationView.render("participation.json", %{
203 participation: participation,
204 for: participation.user
205 })
206 |> Jason.encode!()
207 }
208 |> Jason.encode!()
209 end
210
211 @spec represent_notification(User.t(), Notification.t()) :: binary()
212 defp represent_notification(%User{} = user, %Notification{} = notify) do
213 %{
214 event: "notification",
215 payload:
216 NotificationView.render(
217 "show.json",
218 %{notification: notify, for: user}
219 )
220 |> Jason.encode!()
221 }
222 |> Jason.encode!()
223 end
224
225 defp should_send?(%User{} = user, %Activity{} = item) do
226 blocks = user.info.blocks || []
227 mutes = user.info.mutes || []
228 reblog_mutes = user.info.muted_reblogs || []
229 domain_blocks = Pleroma.Web.ActivityPub.MRF.subdomains_regex(user.info.domain_blocks)
230
231 with parent when not is_nil(parent) <- Object.normalize(item),
232 true <- Enum.all?([blocks, mutes, reblog_mutes], &(item.actor not in &1)),
233 true <- Enum.all?([blocks, mutes], &(parent.data["actor"] not in &1)),
234 %{host: item_host} <- URI.parse(item.actor),
235 %{host: parent_host} <- URI.parse(parent.data["actor"]),
236 false <- Pleroma.Web.ActivityPub.MRF.subdomain_match?(domain_blocks, item_host),
237 false <- Pleroma.Web.ActivityPub.MRF.subdomain_match?(domain_blocks, parent_host),
238 true <- thread_containment(item, user),
239 false <- CommonAPI.thread_muted?(user, item) do
240 true
241 else
242 _ -> false
243 end
244 end
245
246 defp should_send?(%User{} = user, %Notification{activity: activity}) do
247 should_send?(user, activity)
248 end
249
250 def push_to_socket(topics, topic, %Activity{data: %{"type" => "Announce"}} = item) do
251 Enum.each(topics[topic] || [], fn socket ->
252 # Get the current user so we have up-to-date blocks etc.
253 if socket.assigns[:user] do
254 user = User.get_cached_by_ap_id(socket.assigns[:user].ap_id)
255
256 if should_send?(user, item) do
257 send(socket.transport_pid, {:text, represent_update(item, user)})
258 end
259 else
260 send(socket.transport_pid, {:text, represent_update(item)})
261 end
262 end)
263 end
264
265 def push_to_socket(topics, topic, %Participation{} = participation) do
266 Enum.each(topics[topic] || [], fn socket ->
267 send(socket.transport_pid, {:text, represent_conversation(participation)})
268 end)
269 end
270
271 def push_to_socket(topics, topic, %Activity{
272 data: %{"type" => "Delete", "deleted_activity_id" => deleted_activity_id}
273 }) do
274 Enum.each(topics[topic] || [], fn socket ->
275 send(
276 socket.transport_pid,
277 {:text, %{event: "delete", payload: to_string(deleted_activity_id)} |> Jason.encode!()}
278 )
279 end)
280 end
281
282 def push_to_socket(_topics, _topic, %Activity{data: %{"type" => "Delete"}}), do: :noop
283
284 def push_to_socket(topics, topic, item) do
285 Enum.each(topics[topic] || [], fn socket ->
286 # Get the current user so we have up-to-date blocks etc.
287 if socket.assigns[:user] do
288 user = User.get_cached_by_ap_id(socket.assigns[:user].ap_id)
289 blocks = user.info.blocks || []
290 mutes = user.info.mutes || []
291
292 with true <- Enum.all?([blocks, mutes], &(item.actor not in &1)),
293 true <- thread_containment(item, user) do
294 send(socket.transport_pid, {:text, represent_update(item, user)})
295 end
296 else
297 send(socket.transport_pid, {:text, represent_update(item)})
298 end
299 end)
300 end
301
302 defp internal_topic(topic, socket) when topic in ~w[user user:notification direct] do
303 "#{topic}:#{socket.assigns[:user].id}"
304 end
305
306 defp internal_topic(topic, _), do: topic
307
308 @spec thread_containment(Activity.t(), User.t()) :: boolean()
309 defp thread_containment(_activity, %User{info: %{skip_thread_containment: true}}), do: true
310
311 defp thread_containment(activity, user) do
312 if Config.get([:instance, :skip_thread_containment]) do
313 true
314 else
315 ActivityPub.contain_activity(activity, user)
316 end
317 end
318 end