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