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) 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