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