git.squeep.com Git - akkoma/blob - lib/pleroma/web/streamer.ex

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