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