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