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(%{action: :stream, topic: "user", item: %Notification{} = item}, topics) do
 114     topic = "user:#{item.user_id}"
 115
 116     Enum.each(topics[topic] || [], fn socket ->
 117       json =
 118         %{
 119           event: "notification",
 120           payload:
 121             NotificationView.render("show.json", %{
 122               notification: item,
 123               for: socket.assigns["user"]
 124             })
 125             |> Jason.encode!()
 126         }
 127         |> Jason.encode!()
 128
 129       send(socket.transport_pid, {:text, json})
 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   def push_to_socket(topics, topic, %Activity{data: %{"type" => "Announce"}} = item) do
 220     Enum.each(topics[topic] || [], fn socket ->
 221       # Get the current user so we have up-to-date blocks etc.
 222       if socket.assigns[:user] do
 223         user = User.get_cached_by_ap_id(socket.assigns[:user].ap_id)
 224         blocks = user.info.blocks || []
 225         mutes = user.info.mutes || []
 226         reblog_mutes = user.info.muted_reblogs || []
 227
 228         with parent when not is_nil(parent) <- Object.normalize(item),
 229              true <- Enum.all?([blocks, mutes, reblog_mutes], &(item.actor not in &1)),
 230              true <- Enum.all?([blocks, mutes], &(parent.data["actor"] not in &1)),
 231              true <- thread_containment(item, user) do
 232           send(socket.transport_pid, {:text, represent_update(item, user)})
 233         end
 234       else
 235         send(socket.transport_pid, {:text, represent_update(item)})
 236       end
 237     end)
 238   end
 239
 240   def push_to_socket(topics, topic, %Participation{} = participation) do
 241     Enum.each(topics[topic] || [], fn socket ->
 242       send(socket.transport_pid, {:text, represent_conversation(participation)})
 243     end)
 244   end
 245
 246   def push_to_socket(topics, topic, %Activity{
 247         data: %{"type" => "Delete", "deleted_activity_id" => deleted_activity_id}
 248       }) do
 249     Enum.each(topics[topic] || [], fn socket ->
 250       send(
 251         socket.transport_pid,
 252         {:text, %{event: "delete", payload: to_string(deleted_activity_id)} |> Jason.encode!()}
 253       )
 254     end)
 255   end
 256
 257   def push_to_socket(_topics, _topic, %Activity{data: %{"type" => "Delete"}}), do: :noop
 258
 259   def push_to_socket(topics, topic, item) do
 260     Enum.each(topics[topic] || [], fn socket ->
 261       # Get the current user so we have up-to-date blocks etc.
 262       if socket.assigns[:user] do
 263         user = User.get_cached_by_ap_id(socket.assigns[:user].ap_id)
 264         blocks = user.info.blocks || []
 265         mutes = user.info.mutes || []
 266
 267         with true <- Enum.all?([blocks, mutes], &(item.actor not in &1)),
 268              true <- thread_containment(item, user) do
 269           send(socket.transport_pid, {:text, represent_update(item, user)})
 270         end
 271       else
 272         send(socket.transport_pid, {:text, represent_update(item)})
 273       end
 274     end)
 275   end
 276
 277   defp internal_topic(topic, socket) when topic in ~w[user direct] do
 278     "#{topic}:#{socket.assigns[:user].id}"
 279   end
 280
 281   defp internal_topic(topic, _), do: topic
 282
 283   @spec thread_containment(Activity.t(), User.t()) :: boolean()
 284   defp thread_containment(_activity, %User{info: %{skip_thread_containment: true}}), do: true
 285
 286   defp thread_containment(activity, user) do
 287     if Config.get([:instance, :skip_thread_containment]) do
 288       true
 289     else
 290       ActivityPub.contain_activity(activity, user)
 291     end
 292   end
 293 end