git.squeep.com Git - akkoma/blob - 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),
 117            false <- CommonAPI.thread_muted?(user, item.activity) do
 118         send(
 119           socket.transport_pid,
 120           {:text, represent_notification(socket.assigns[:user], item)}
 121         )
 122       end
 123     end)
 124
 125     {:noreply, topics}
 126   end
 127
 128   def handle_cast(%{action: :stream, topic: "user", item: item}, topics) do
 129     Logger.debug("Trying to push to users")
 130
 131     recipient_topics =
 132       User.get_recipients_from_activity(item)
 133       |> Enum.map(fn %{id: id} -> "user:#{id}" end)
 134
 135     Enum.each(recipient_topics, fn topic ->
 136       push_to_socket(topics, topic, item)
 137     end)
 138
 139     {:noreply, topics}
 140   end
 141
 142   def handle_cast(%{action: :stream, topic: topic, item: item}, topics) do
 143     Logger.debug("Trying to push to #{topic}")
 144     Logger.debug("Pushing item to #{topic}")
 145     push_to_socket(topics, topic, item)
 146     {:noreply, topics}
 147   end
 148
 149   def handle_cast(%{action: :add, topic: topic, socket: socket}, sockets) do
 150     topic = internal_topic(topic, socket)
 151     sockets_for_topic = sockets[topic] || []
 152     sockets_for_topic = Enum.uniq([socket | sockets_for_topic])
 153     sockets = Map.put(sockets, topic, sockets_for_topic)
 154     Logger.debug("Got new conn for #{topic}")
 155     {:noreply, sockets}
 156   end
 157
 158   def handle_cast(%{action: :remove, topic: topic, socket: socket}, sockets) do
 159     topic = internal_topic(topic, socket)
 160     sockets_for_topic = sockets[topic] || []
 161     sockets_for_topic = List.delete(sockets_for_topic, socket)
 162     sockets = Map.put(sockets, topic, sockets_for_topic)
 163     Logger.debug("Removed conn for #{topic}")
 164     {:noreply, sockets}
 165   end
 166
 167   def handle_cast(m, state) do
 168     Logger.info("Unknown: #{inspect(m)}, #{inspect(state)}")
 169     {:noreply, state}
 170   end
 171
 172   defp represent_update(%Activity{} = activity, %User{} = user) do
 173     %{
 174       event: "update",
 175       payload:
 176         Pleroma.Web.MastodonAPI.StatusView.render(
 177           "status.json",
 178           activity: activity,
 179           for: user
 180         )
 181         |> Jason.encode!()
 182     }
 183     |> Jason.encode!()
 184   end
 185
 186   defp represent_update(%Activity{} = activity) do
 187     %{
 188       event: "update",
 189       payload:
 190         Pleroma.Web.MastodonAPI.StatusView.render(
 191           "status.json",
 192           activity: activity
 193         )
 194         |> Jason.encode!()
 195     }
 196     |> Jason.encode!()
 197   end
 198
 199   def represent_conversation(%Participation{} = participation) do
 200     %{
 201       event: "conversation",
 202       payload:
 203         Pleroma.Web.MastodonAPI.ConversationView.render("participation.json", %{
 204           participation: participation,
 205           user: participation.user
 206         })
 207         |> Jason.encode!()
 208     }
 209     |> Jason.encode!()
 210   end
 211
 212   @spec represent_notification(User.t(), Notification.t()) :: binary()
 213   defp represent_notification(%User{} = user, %Notification{} = notify) do
 214     %{
 215       event: "notification",
 216       payload:
 217         NotificationView.render(
 218           "show.json",
 219           %{notification: notify, for: user}
 220         )
 221         |> Jason.encode!()
 222     }
 223     |> Jason.encode!()
 224   end
 225
 226   defp should_send?(%User{} = user, %Activity{} = item) do
 227     blocks = user.info.blocks || []
 228     mutes = user.info.mutes || []
 229     reblog_mutes = user.info.muted_reblogs || []
 230     domain_blocks = Pleroma.Web.ActivityPub.MRF.subdomains_regex(user.info.domain_blocks)
 231
 232     with parent when not is_nil(parent) <- Object.normalize(item),
 233          true <- Enum.all?([blocks, mutes, reblog_mutes], &(item.actor not in &1)),
 234          true <- Enum.all?([blocks, mutes], &(parent.data["actor"] not in &1)),
 235          %{host: item_host} <- URI.parse(item.actor),
 236          %{host: parent_host} <- URI.parse(parent.data["actor"]),
 237          false <- Pleroma.Web.ActivityPub.MRF.subdomain_match?(domain_blocks, item_host),
 238          false <- Pleroma.Web.ActivityPub.MRF.subdomain_match?(domain_blocks, parent_host),
 239          true <- thread_containment(item, user) 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